mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(hanzo-tools-vcs): repo lists Hanzo git repositories and syncs one now
This commit is contained in:
@@ -1,10 +1,15 @@
|
||||
# hanzo-tools-vcs
|
||||
|
||||
Unified version control tool for Hanzo AI MCP (HIP-0300).
|
||||
Version control tools for Hanzo AI MCP (HIP-0300).
|
||||
|
||||
## Tools
|
||||
|
||||
- `git` - Unified version control (status, diff, commit, branch, log, …)
|
||||
- `repo` - Your repositories on Hanzo git and the GitHub links that feed them,
|
||||
through `api.hanzo.ai` with your Hanzo credential: `list` (`GET /v1/sync`; each
|
||||
link's id, scope, source, target, direction, trigger, status, branch, url and
|
||||
last sync; `search` and `status` narrow it) and `sync` (`POST /v1/sync/{id}/run`,
|
||||
sync one now).
|
||||
|
||||
## Installation
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ Tools:
|
||||
- branch: Branch operations (list, create, delete)
|
||||
- checkout: Switch branches
|
||||
- log: Commit history
|
||||
- repo: Repositories on Hanzo git linked to GitHub (list, sync now)
|
||||
|
||||
Outputs diffs in unified patch format for use with fs.apply_patch.
|
||||
|
||||
@@ -28,17 +29,19 @@ Usage:
|
||||
from hanzo_tools.core import BaseTool, ToolRegistry
|
||||
|
||||
from .git_tool import GitTool, git_tool
|
||||
from .repo_tool import RepoTool
|
||||
|
||||
# Backward compat
|
||||
VcsTool = GitTool
|
||||
vcs_tool = git_tool
|
||||
|
||||
# Export list for tool discovery - HIP-0300 unified tool
|
||||
TOOLS = [GitTool]
|
||||
# Export list for tool discovery - HIP-0300 unified tools
|
||||
TOOLS = [GitTool, RepoTool]
|
||||
|
||||
__all__ = [
|
||||
"GitTool",
|
||||
"git_tool",
|
||||
"RepoTool",
|
||||
"VcsTool",
|
||||
"vcs_tool",
|
||||
"register_tools",
|
||||
@@ -56,7 +59,7 @@ def register_tools(mcp_server, **kwargs) -> list[BaseTool]:
|
||||
Returns:
|
||||
List of registered tool instances
|
||||
"""
|
||||
cwd = kwargs.get("cwd")
|
||||
tool = VcsTool(cwd=cwd)
|
||||
ToolRegistry.register_tool(mcp_server, tool)
|
||||
return [tool]
|
||||
tools = [VcsTool(cwd=kwargs.get("cwd")), RepoTool()]
|
||||
for tool in tools:
|
||||
ToolRegistry.register_tool(mcp_server, tool)
|
||||
return tools
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
"""repo — your repositories on Hanzo git and the GitHub links that feed them (HIP-0300).
|
||||
|
||||
list GET /v1/sync every link: source, target, direction, where it stands
|
||||
sync POST /v1/sync/{id}/run sync one now; the reconcile is queued and runs in the background
|
||||
|
||||
A link is a GitHub repository (or a whole account) declared once and kept in step
|
||||
with its copy on git.hanzo.ai. The calls go to api.hanzo.ai through the one
|
||||
HanzoCloud client and the caller's Hanzo credential.
|
||||
"""
|
||||
|
||||
from typing import Any, ClassVar
|
||||
from urllib.parse import quote
|
||||
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
|
||||
from hanzo_tools.core import BaseTool, HanzoCloud, InvalidParamsError, ToolError
|
||||
from hanzo_tools.core.cloud import NO_KEY, CloudError
|
||||
|
||||
|
||||
def row(v: Any) -> dict[str, Any]:
|
||||
"""One link as an agent reads it: where it comes from, where it goes, how it stands."""
|
||||
v = v if isinstance(v, dict) else {}
|
||||
native = v.get("native") if isinstance(v.get("native"), dict) else {}
|
||||
return {
|
||||
"id": v.get("id"),
|
||||
"scope": v.get("scope"),
|
||||
"source": (v.get("source") or {}).get("locator"),
|
||||
"target": (v.get("target") or {}).get("locator"),
|
||||
"direction": v.get("direction"),
|
||||
"trigger": v.get("trigger"),
|
||||
"status": native.get("status"),
|
||||
"branch": native.get("branch"),
|
||||
"url": native.get("url"),
|
||||
# bumped by every reconcile, so it reads as the last sync
|
||||
"synced_at": v.get("updatedAt"),
|
||||
}
|
||||
|
||||
|
||||
LIST_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"search": {
|
||||
"type": "string",
|
||||
"description": "list: words that must all appear in the source or target",
|
||||
},
|
||||
"status": {
|
||||
"type": "string",
|
||||
"enum": ["synced", "conflict", "pending", "paused"],
|
||||
"description": "list: only links that stand so",
|
||||
},
|
||||
},
|
||||
"required": [],
|
||||
}
|
||||
SYNC_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "string", "description": "sync: the link's id from list (sync_...)"}
|
||||
},
|
||||
"required": [],
|
||||
}
|
||||
|
||||
|
||||
class RepoTool(BaseTool):
|
||||
"""Repositories on Hanzo git linked to GitHub: list them, sync one now."""
|
||||
|
||||
name: ClassVar[str] = "repo"
|
||||
VERSION: ClassVar[str] = "0.1.0"
|
||||
DEFAULT_ACTION: ClassVar[str] = "list"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._cloud: HanzoCloud | None = None
|
||||
|
||||
@self.action("list", "Every repository link and where it stands", schema=LIST_SCHEMA)
|
||||
async def list_(
|
||||
ctx: MCPContext, search: str | None = None, status: str | None = None
|
||||
) -> dict:
|
||||
body = await self._send("GET", "/v1/sync")
|
||||
rows = body.get("data") if isinstance(body, dict) else None
|
||||
words = (search or "").lower().split()
|
||||
links = []
|
||||
for v in rows if isinstance(rows, list) else []:
|
||||
r = row(v)
|
||||
if status and r["status"] != status:
|
||||
continue
|
||||
hay = f"{r['source'] or ''} {r['target'] or ''}".lower()
|
||||
if all(w in hay for w in words):
|
||||
links.append(r)
|
||||
return {"count": len(links), "repos": links}
|
||||
|
||||
@self.action("sync", "Sync one repository link now", schema=SYNC_SCHEMA)
|
||||
async def sync(ctx: MCPContext, id: str | None = None) -> dict:
|
||||
if not isinstance(id, str) or not id.strip():
|
||||
raise InvalidParamsError(
|
||||
"id required: a link's id from repo list (sync_...)", param="id"
|
||||
)
|
||||
return await self._send("POST", f"/v1/sync/{quote(id.strip(), safe='')}/run")
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return (
|
||||
"Your repositories on Hanzo git and the GitHub links that keep them in step (/v1/sync). "
|
||||
"list (default): every link with its id, scope (repo or account), source, target, direction "
|
||||
"(both, pull, push, off), trigger, status (synced, conflict, pending, paused), branch, url and "
|
||||
"synced_at; search (words in source or target) and status narrow it. "
|
||||
"sync: sync one link now by its id; answers {queued, id} and the reconcile runs in the background."
|
||||
)
|
||||
|
||||
async def _send(self, method: str, path: str) -> Any:
|
||||
if self._cloud is None:
|
||||
self._cloud = HanzoCloud()
|
||||
if not self._cloud.configured():
|
||||
raise ToolError(code="INVALID_PARAMS", message=NO_KEY)
|
||||
try:
|
||||
return await self._cloud.call(method, path)
|
||||
except CloudError as e:
|
||||
if e.status == 404:
|
||||
raise ToolError(code="NOT_FOUND", message=str(e))
|
||||
raise ToolError(code="UPSTREAM", message=str(e))
|
||||
@@ -4,8 +4,8 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "hanzo-tools-vcs"
|
||||
version = "0.1.2"
|
||||
description = "Unified version control tool for Hanzo AI (HIP-0300)"
|
||||
version = "0.1.3"
|
||||
description = "Version control tools for Hanzo AI (HIP-0300): git, and repo for repositories on Hanzo git linked to GitHub"
|
||||
readme = "README.md"
|
||||
license = "MIT"
|
||||
requires-python = ">=3.12"
|
||||
@@ -30,7 +30,7 @@ classifiers = [
|
||||
"Programming Language :: Python :: 3.12",
|
||||
]
|
||||
dependencies = [
|
||||
"hanzo-tools>=0.3.0",
|
||||
"hanzo-tools>=0.3.8", # HanzoCloud: the api.hanzo.ai client repo calls /v1/sync through
|
||||
"mcp>=1.0.0",
|
||||
]
|
||||
|
||||
|
||||
@@ -0,0 +1,161 @@
|
||||
"""Tests for repo: repositories on Hanzo git and their GitHub links, over /v1/sync.
|
||||
|
||||
The tool runs against a real HanzoCloud whose httpx transport answers as
|
||||
api.hanzo.ai does, so each test reads the request that went out and what the
|
||||
tool made of the answer. The live test is skipped when no Hanzo credential
|
||||
resolves.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from hanzo_tools.core import HanzoCloud
|
||||
|
||||
from hanzo_tools.vcs import TOOLS, RepoTool
|
||||
|
||||
UI = {
|
||||
"id": "sync_713c4bb98167c774be9d6d01a35ac69f",
|
||||
"kind": "git",
|
||||
"source": {"provider": "github", "locator": "https://github.com/hanzoai/ui.git"},
|
||||
"target": {"provider": "hanzo-git", "locator": "ui"},
|
||||
"direction": "pull",
|
||||
"trigger": "poll",
|
||||
"createdAt": "2026-07-22T10:42:41Z",
|
||||
"updatedAt": "2026-10-04T23:06:13Z",
|
||||
"scope": "repo",
|
||||
"native": {
|
||||
"url": "https://git.hanzo.ai/hanzoai/hanzoai_ui",
|
||||
"clone": "https://git.hanzo.ai/hanzoai/hanzoai_ui.git",
|
||||
"branch": "main",
|
||||
"status": "synced",
|
||||
"sizeBytes": 131655680,
|
||||
},
|
||||
}
|
||||
ARC = {
|
||||
"id": "sync_0b1d",
|
||||
"source": {"provider": "github", "locator": "https://github.com/hanzoai/arc.git"},
|
||||
"target": {"provider": "hanzo-git", "locator": "arc"},
|
||||
"direction": "both",
|
||||
"trigger": "webhook",
|
||||
"scope": "repo",
|
||||
"native": {
|
||||
"status": "conflict",
|
||||
"branch": "main",
|
||||
"url": "https://git.hanzo.ai/hanzoai/hanzoai_arc",
|
||||
},
|
||||
}
|
||||
ACCOUNT = {
|
||||
"id": "sync_acct",
|
||||
"source": {"provider": "github", "locator": "https://github.com/hanzoai"},
|
||||
"direction": "pull",
|
||||
"trigger": "webhook",
|
||||
"scope": "account",
|
||||
}
|
||||
|
||||
|
||||
class Gateway:
|
||||
def __init__(self):
|
||||
self.seen: list[httpx.Request] = []
|
||||
|
||||
def __call__(self, request: httpx.Request) -> httpx.Response:
|
||||
self.seen.append(request)
|
||||
path = request.url.path
|
||||
if request.method == "GET" and path == "/v1/sync":
|
||||
return httpx.Response(200, json={"data": [UI, ARC, ACCOUNT]})
|
||||
if request.method == "POST" and path == f"/v1/sync/{UI['id']}/run":
|
||||
return httpx.Response(202, json={"queued": True, "id": UI["id"]})
|
||||
return httpx.Response(404, json={"error": "no such sync"})
|
||||
|
||||
|
||||
def wired():
|
||||
gw = Gateway()
|
||||
cloud = HanzoCloud(base_url="https://api.hanzo.ai", key="sk-test")
|
||||
cloud._client = httpx.AsyncClient(
|
||||
base_url="https://api.hanzo.ai", transport=httpx.MockTransport(gw)
|
||||
)
|
||||
tool = RepoTool()
|
||||
tool._cloud = cloud
|
||||
return tool, gw
|
||||
|
||||
|
||||
def run(tool, **kw):
|
||||
async def call():
|
||||
try:
|
||||
return await tool.call(None, **kw)
|
||||
finally:
|
||||
if tool._cloud is not None:
|
||||
await tool._cloud.aclose()
|
||||
|
||||
return asyncio.run(call())
|
||||
|
||||
|
||||
def test_the_package_carries_git_and_repo():
|
||||
assert [t.name for t in TOOLS] == ["git", "repo"]
|
||||
assert RepoTool.DEFAULT_ACTION == "list"
|
||||
|
||||
|
||||
def test_list_reads_every_link_and_where_it_stands():
|
||||
tool, gw = wired()
|
||||
env = run(tool)
|
||||
assert env["ok"], env
|
||||
assert (gw.seen[-1].method, gw.seen[-1].url.path) == ("GET", "/v1/sync")
|
||||
assert gw.seen[-1].headers["authorization"] == "Bearer sk-test"
|
||||
assert env["data"]["count"] == 3
|
||||
assert env["data"]["repos"][0] == {
|
||||
"id": "sync_713c4bb98167c774be9d6d01a35ac69f",
|
||||
"scope": "repo",
|
||||
"source": "https://github.com/hanzoai/ui.git",
|
||||
"target": "ui",
|
||||
"direction": "pull",
|
||||
"trigger": "poll",
|
||||
"status": "synced",
|
||||
"branch": "main",
|
||||
"url": "https://git.hanzo.ai/hanzoai/hanzoai_ui",
|
||||
"synced_at": "2026-10-04T23:06:13Z",
|
||||
}
|
||||
# An account link has no native copy of its own: its standing reads None.
|
||||
assert (env["data"]["repos"][2]["scope"], env["data"]["repos"][2]["status"]) == (
|
||||
"account",
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
def test_list_narrows_by_words_and_by_status():
|
||||
tool, _ = wired()
|
||||
assert [r["id"] for r in run(tool, action="list", search="hanzoai arc")["data"]["repos"]] == [
|
||||
"sync_0b1d"
|
||||
]
|
||||
tool, _ = wired()
|
||||
assert [r["id"] for r in run(tool, action="list", status="conflict")["data"]["repos"]] == [
|
||||
"sync_0b1d"
|
||||
]
|
||||
|
||||
|
||||
def test_sync_runs_one_link_now():
|
||||
tool, gw = wired()
|
||||
env = run(tool, action="sync", id=UI["id"])
|
||||
assert env["ok"], env
|
||||
assert (gw.seen[-1].method, gw.seen[-1].url.path) == ("POST", f"/v1/sync/{UI['id']}/run")
|
||||
assert env["data"] == {"queued": True, "id": UI["id"]}
|
||||
|
||||
|
||||
def test_sync_needs_an_id_and_names_one_it_does_not_know():
|
||||
tool, gw = wired()
|
||||
env = run(tool, action="sync")
|
||||
assert not env["ok"] and "id required" in env["error"]["message"]
|
||||
assert gw.seen == []
|
||||
tool, gw = wired()
|
||||
env = run(tool, action="sync", id="sync_nope/../x")
|
||||
assert not env["ok"] and env["error"]["code"] == "NOT_FOUND"
|
||||
assert gw.seen[-1].url.raw_path.decode() == "/v1/sync/sync_nope%2F..%2Fx/run"
|
||||
|
||||
|
||||
@pytest.mark.skipif(not HanzoCloud().configured(), reason="no Hanzo credential")
|
||||
def test_live_list_reads_the_org_s_links():
|
||||
env = run(RepoTool())
|
||||
assert env["ok"], env
|
||||
repos = env["data"]["repos"]
|
||||
assert env["data"]["count"] == len(repos)
|
||||
assert all(r["id"].startswith("sync_") for r in repos)
|
||||
assert all(r["status"] in (None, "synced", "conflict", "pending", "paused") for r in repos)
|
||||
Reference in New Issue
Block a user