hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(lsp): repo answers from /v1/code/lsp, file stays local

Browse Source 
Mirrors hanzoai/mcp's Rust `lsp` exactly: one tool, two planes, picked by data
the caller already has. `file` names the file; `repo` (a git.hanzo.ai slug,
optional `rev`) says which world it lives in. Without `repo` nothing changes —
the same language server on the same tree. With it, the question goes to the
indexed corpus behind /v1/code/lsp, which reaches across a repo's dependencies
without checking anything out.

One table maps actions onto ops: locate carries a relation (definition|
reference|type|implementation) because "where is X" is one question with four
answers, not four routes; hover, symbols, diagnostics and complete stand on
their own. The body is {repo, rev?, path, line, character, relation?} in LSP's
own frame — 0-based line, UTF-16 character — shifted from the tool's 1-based
`line` by the same expression the local plane uses.

The call goes through the shared HanzoCloud client hanzo-tools-code and
hanzo-tools-net already compose, so there is no second client and no second
lsp tool. Actions a plane cannot serve say so up front rather than spawning a
server or calling out.

test_lsp_tools.py asked for a class named `LspTool`; the class is `LSPTool`,
so those two never ran. One name, and they run.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zooqueen
2026-08-06 03:19:28 -07:00
co-authored by Hanzo Dev
parent edeb65d5af
commit e1635ebf98
4 changed files with 350 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-tools-lsp/hanzo_tools/lsp/lsp_tool.py
+124 -24
View File
@@ -1,9 +1,15 @@
"""Language Server Protocol (LSP) tool for code intelligence.
This tool provides on-demand LSP configuration and installation for various
programming languages. It automatically installs language servers as needed
and provides code intelligence features like go-to-definition, find references,
rename symbol, and diagnostics.
One tool, two planes, chosen by the data the caller already supplies:
- no ``repo`` — a language server on the local tree, installed on demand and
kept alive per ``language:root``. Answers what a working tree knows,
including the edits it can apply back to disk.
- with ``repo`` — the indexed corpus behind ``/v1/code/lsp`` on api.hanzo.ai,
which answers across a repo's dependencies without checking anything out.
``file`` names the file either way; ``repo`` says which world that file lives
in.
"""
import os
@@ -19,7 +25,7 @@ from pathlib import Path
from dataclasses import field, dataclass
from urllib.parse import unquote, urlparse
from hanzo_tools.core import BaseTool, MCPResourceDocument
from hanzo_tools.core import BaseTool, CloudError, HanzoCloud, MCPResourceDocument
# LSP server configurations
LSP_SERVERS = {
@@ -167,6 +173,33 @@ LSP_SERVERS = {
}
# Actions a local language server answers.
LOCAL_ACTIONS = [
"definition",
"references",
"rename",
"diagnostics",
"hover",
"completion",
"code_action",
"organize_imports",
"status",
]
# The /v1/code/lsp op each action asks for, and the `relation` a locate op
# carries. "Where is X" is one question with four answers, so it is one op.
CLOUD_OPS: Dict[str, tuple] = {
"hover": ("hover", None),
"definition": ("locate", "definition"),
"references": ("locate", "reference"),
"type": ("locate", "type"),
"implementation": ("locate", "implementation"),
"symbols": ("symbols", None),
"diagnostics": ("diagnostics", None),
"completion": ("complete", None),
}
# Global LSP server registry - singleton per language:root_uri
_GLOBAL_SERVERS: Dict[str, "LSPServer"] = {}
_GLOBAL_LOCK = asyncio.Lock()
@@ -213,8 +246,8 @@ class LSPTool(BaseTool):
@property
def description(self) -> str:
return """Language Server Protocol tool for code intelligence.
Actions:
Local — a language server on your tree, installed as needed:
- definition: Go to definition of symbol at position
- references: Find all references to symbol
- rename: Rename symbol across codebase
@@ -224,13 +257,17 @@ class LSPTool(BaseTool):
- code_action: Run LSP code actions
- organize_imports: Organize imports for a file
- status: Check LSP server status
The tool automatically installs language servers as needed.
Supported languages: Go, Python, TypeScript/JavaScript, Rust, Java, C/C++, Ruby, Lua
Cloud — pass `repo` (a git.hanzo.ai slug, optionally `rev`) to answer from
the indexed corpus instead, which reaches across the repo's dependencies:
definition, references, type, implementation, symbols, hover, completion,
diagnostics. `file` is then the path within the repo.
"""
def __init__(self):
super().__init__()
self._cloud: Optional[HanzoCloud] = None
global _CLEANUP_REGISTERED
if not _CLEANUP_REGISTERED:
atexit.register(_cleanup_all_servers)
@@ -477,29 +514,41 @@ class LSPTool(BaseTool):
async def run(
self,
action: str,
file: str,
file: Optional[str] = None,
line: Optional[int] = None,
character: Optional[int] = None,
new_name: Optional[str] = None,
apply_edits: bool = False,
repo: Optional[str] = None,
rev: Optional[str] = None,
**kwargs,
) -> MCPResourceDocument:
"""Execute LSP action."""
valid_actions = [
"definition",
"references",
"rename",
"diagnostics",
"hover",
"completion",
"code_action",
"organize_imports",
"status",
]
if action not in valid_actions:
"""Execute LSP action, locally or against the cloud index."""
if not file or not file.strip():
return MCPResourceDocument(
data={
"error": f"Invalid action. Must be one of: {', '.join(valid_actions)}"
"error": "file is required: a local path, or the path within `repo`"
}
)
if action not in LOCAL_ACTIONS and action not in CLOUD_OPS:
known = LOCAL_ACTIONS + [a for a in CLOUD_OPS if a not in LOCAL_ACTIONS]
return MCPResourceDocument(
data={"error": f"Invalid action. Must be one of: {', '.join(known)}"}
)
if repo and repo.strip():
return MCPResourceDocument(
data=await self._cloud_action(
action, repo.strip(), file, line, character, rev
)
)
if action not in LOCAL_ACTIONS:
return MCPResourceDocument(
data={
"error": f"Action '{action}' is answered by the cloud index; pass `repo`",
"local_actions": LOCAL_ACTIONS,
}
)
@@ -559,6 +608,53 @@ class LSPTool(BaseTool):
)
return MCPResourceDocument(data=result)
async def _cloud_action(
self,
action: str,
repo: str,
file: str,
line: Optional[int],
character: Optional[int],
rev: Optional[str],
) -> Dict[str, Any]:
"""Ask the cloud index. The answer is whatever /v1/code/lsp returns."""
op, relation = CLOUD_OPS.get(action, (None, None))
if op is None:
return {
"action": action,
"repo": repo,
"error": f"Action '{action}' needs a working tree; call it without `repo`",
"cloud_actions": list(CLOUD_OPS),
}
if self._cloud is None:
self._cloud = HanzoCloud()
if not self._cloud.configured():
return {
"action": action,
"repo": repo,
"error": "no hk- key: set HANZO_API_KEY or ~/.hanzo/config.json .apiKey",
}
# The wire speaks LSP's own frame: 0-based line, 0-based UTF-16
# character. The tool's `line` is 1-based, so it is shifted here exactly
# as the local plane shifts it before a JSON-RPC request.
body: Dict[str, Any] = {
"repo": repo,
"path": file,
"line": (line - 1) if line else 0,
"character": character if character else 0,
}
if rev and rev.strip():
body["rev"] = rev.strip()
if relation:
body["relation"] = relation
try:
return await self._cloud.post(f"/v1/code/lsp/{op}", body)
except CloudError as e:
return {"action": action, "repo": repo, "error": str(e)}
async def call(self, **kwargs) -> str:
"""Tool interface for MCP - converts result to JSON string."""
result = await self.run(**kwargs)
@@ -577,6 +673,8 @@ class LSPTool(BaseTool):
apply_edits: bool = False,
only: Optional[List[str]] = None,
range: Optional[Dict[str, Dict[str, int]]] = None,
repo: Optional[str] = None,
rev: Optional[str] = None,
) -> str:
return await self.call(
action=action,
@@ -587,6 +685,8 @@ class LSPTool(BaseTool):
apply_edits=apply_edits,
only=only,
range=range,
repo=repo,
rev=rev,
)
def _path_to_uri(self, path: str) -> str:
pkg/hanzo-tools-lsp/pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools-lsp"
version = "0.2.0"
version = "0.2.1"
description = "Language Server Protocol tools for code intelligence"
readme = "README.md"
requires-python = ">=3.12"
pkg/hanzo-tools-lsp/tests/test_lsp_cloud.py
+219
View File
@@ -0,0 +1,219 @@
"""`lsp` against a stand-in /v1/code/lsp (no network).
One tool, two planes. These fake the wire, not the client: a real HanzoCloud
runs over a canned transport, so what is under test is what actually goes out
when a caller names a `repo` — the op the action maps to, the body it carries
(0-based line, UTF-16 character), and the answer coming back verbatim. With no
`repo`, nothing leaves the process.
"""
import json
import asyncio
from hanzo_tools.lsp import TOOLS, LSPTool
from hanzo_tools.core import HanzoCloud
from hanzo_tools.lsp.lsp_tool import CLOUD_OPS, LOCAL_ACTIONS
def _run(coro):
return asyncio.run(coro)
class FakeResponse:
def __init__(self, payload, status=200):
self.status_code = status
self.payload = payload
self.text = json.dumps(payload)
def json(self):
return self.payload
class FakeClient:
"""An httpx-shaped client that records requests and replays one reply."""
def __init__(self, reply, status=200):
self.reply = reply
self.status = status
self.calls = []
async def request(self, method, path, **kw):
self.calls.append((method, path, kw))
return FakeResponse(self.reply, self.status)
def tool_over(reply=None, key="hk-test", status=200):
"""An LSPTool whose cloud client answers from a canned transport."""
tool = LSPTool()
cloud = HanzoCloud(key=key)
client = FakeClient(reply if reply is not None else {"locations": []}, status)
async def fake_get_client():
return client
cloud._get_client = fake_get_client
tool._cloud = cloud
return tool, client
def sent(client):
"""The one request the tool made: (path, body)."""
assert len(client.calls) == 1, f"expected one request, saw {client.calls}"
method, path, kw = client.calls[0]
assert method == "POST"
return path, kw["json"]
# ── repo → the cloud index ───────────────────────────────────────────────
def test_definition_locates_with_a_relation_in_lsp_frame():
hit = {"locations": [{"repo": "hanzoai/mcp", "path": "rust/src/hanzo_api.rs", "line": 18}]}
tool, client = tool_over(hit)
out = _run(
tool.run(
action="definition",
repo="hanzoai/mcp",
rev="main",
file="rust/src/tools/lsp_tool.rs",
line=42,
character=7,
)
)
assert out.data == hit, "the cloud's answer comes back verbatim"
path, body = sent(client)
assert path == "/v1/code/lsp/locate"
assert body == {
"repo": "hanzoai/mcp",
"rev": "main",
"path": "rust/src/tools/lsp_tool.rs",
# the tool's 1-based line 42 leaves as LSP's 0-based 41
"line": 41,
"character": 7,
"relation": "definition",
}
def test_locate_relations_share_one_op():
for action, relation in [
("definition", "definition"),
("references", "reference"),
("type", "type"),
("implementation", "implementation"),
]:
tool, client = tool_over()
_run(tool.run(action=action, repo="hanzoai/mcp", file="a.py", line=1, character=0))
path, body = sent(client)
assert path == "/v1/code/lsp/locate"
assert body["relation"] == relation
assert body["line"] == 0
assert "rev" not in body, "an unnamed rev is absent, not empty"
def test_standalone_ops_carry_no_relation():
for action, op in [
("hover", "hover"),
("symbols", "symbols"),
("diagnostics", "diagnostics"),
("completion", "complete"),
]:
tool, client = tool_over({"ok": True})
_run(tool.run(action=action, repo="hanzoai/mcp", file="a.py", line=9, character=4))
path, body = sent(client)
assert path == f"/v1/code/lsp/{op}"
assert "relation" not in body
assert body["path"] == "a.py"
assert body["line"] == 8
def test_every_cloud_action_is_mapped():
assert set(CLOUD_OPS) == {
"definition",
"references",
"type",
"implementation",
"hover",
"symbols",
"diagnostics",
"completion",
}
for op, relation in CLOUD_OPS.values():
assert op in {"locate", "hover", "symbols", "diagnostics", "complete"}
assert (relation is None) == (op != "locate")
def test_upstream_failure_is_reported_not_raised():
tool, client = tool_over({"error": "nope"}, status=502)
out = _run(tool.run(action="hover", repo="hanzoai/mcp", file="a.py", line=1))
assert out.data["repo"] == "hanzoai/mcp"
assert "502" in out.data["error"]
def test_without_a_key_nothing_is_sent():
tool, client = tool_over(key="")
out = _run(tool.run(action="hover", repo="hanzoai/mcp", file="a.py", line=1))
assert "hk-" in out.data["error"]
assert client.calls == []
# ── file → the local plane ───────────────────────────────────────────────
def test_file_alone_stays_local_and_sends_nothing():
tool, client = tool_over()
out = _run(tool.run(action="definition", file="notes.txt"))
assert "Unsupported file type" in out.data["error"]
assert "go" in out.data["supported_languages"]
assert client.calls == []
def test_neither_file_nor_repo_fails_cleanly():
tool, client = tool_over()
out = _run(tool.run(action="definition"))
assert out.data["error"].startswith("file is required")
assert client.calls == []
def test_an_invalid_action_lists_both_planes():
tool, _ = tool_over()
out = _run(tool.run(action="frobnicate", file="a.py"))
assert "Invalid action" in out.data["error"]
assert "implementation" in out.data["error"]
def test_a_local_only_action_with_repo_says_so():
tool, client = tool_over()
out = _run(tool.run(action="rename", repo="hanzoai/mcp", file="a.py", new_name="x"))
assert "needs a working tree" in out.data["error"]
assert client.calls == []
def test_a_cloud_only_action_without_repo_says_so():
tool, client = tool_over()
out = _run(tool.run(action="symbols", file="a.py"))
assert "pass `repo`" in out.data["error"]
assert out.data["local_actions"] == LOCAL_ACTIONS
assert client.calls == []
# ── one tool ─────────────────────────────────────────────────────────────
def test_lsp_is_one_tool_with_both_planes_on_its_handler():
assert [t.name for t in TOOLS] == ["lsp"]
handlers = {}
class Server:
def tool(self, name, description):
def register(fn):
handlers[name] = fn
return fn
return register
LSPTool().register(Server())
assert set(handlers) == {"lsp"}
params = handlers["lsp"].__annotations__
assert "repo" in params and "rev" in params and "file" in params
pkg/hanzo-tools-lsp/tests/test_lsp_tools.py
+6 -6
View File
@@ -17,19 +17,19 @@ class TestImports:
assert len(TOOLS) > 0
def test_import_lsp_tool(self):
from hanzo_tools.lsp import LspTool
from hanzo_tools.lsp import LSPTool
assert LspTool.name == "lsp"
assert LSPTool.name == "lsp"
class TestLspTool:
"""Tests for LspTool."""
class TestLSPTool:
"""Tests for LSPTool."""
@pytest.fixture
def tool(self):
from hanzo_tools.lsp import LspTool
from hanzo_tools.lsp import LSPTool
return LspTool()
return LSPTool()
def test_has_description(self, tool):
assert tool.description