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(tools): hybrid local+cloud tools -- HanzoCloud client, cloud code/vector actions, vision

Browse Source 
Adds a shared HanzoCloud client (core/cloud.py; lazy httpx, api.hanzo.ai /v1)
and wires the cloud half of the hybrid tools: code gains cross-repo
search/context/ask/index over the cloud index, vector gains a cloud-backed
store plus an infinity-embedded local option, net gains a vision tool. Bumps
hanzo-tools 0.3.2->0.3.3, -code 0.1.2->0.1.3, -net 0.1.2->0.1.3, -vector
0.2.0->0.2.1 and declares httpx where the cloud path is used. New tests cover
the code/net cloud actions. import hanzo_tools.core is green.
This commit is contained in:
zeekay
2026-07-19 20:02:48 -07:00
parent d8dd174ee0
commit b98e6de447
17 changed files with 3853 additions and 2594 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-tools-code/hanzo_tools/code/code_tool.py
+154 -4
View File
@@ -27,6 +27,7 @@ from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import (
BaseTool,
HanzoCloud,
InvalidParamsError,
NotFoundError,
ToolError,
@@ -111,22 +112,29 @@ class CodeTool(BaseTool):
self.cwd = cwd or os.getcwd()
self._tree_sitter = None
self._lsp_clients: dict[str, Any] = {}
self._cloud: HanzoCloud | None = None
self._register_code_actions()
self._register_cloud_actions()
@property
def description(self) -> str:
return """Unified code semantics tool (HIP-0300).
return """Unified code semantics tool (HIP-0300). Hybrid: local AST + cloud index.
Actions:
- parse: Parse source code to AST (tree-sitter)
Local (tree-sitter/LSP, instant, single-file — PURE):
- parse: Parse source code to AST
- serialize: Convert AST back to text
- symbols: List symbols in file/scope
- definition: Go to symbol definition (LSP)
- references: Find all references to symbol (LSP)
- transform: Pure codemod producing Patch (no side effects)
- summarize: Compress Diff/Log/Report to summary
- search_symbol/outline/metrics/exports/types/hierarchy/rename/grep_replace
All operations are PURE - safe to cache and parallelize.
Cloud (api.hanzo.ai /v1/code, cross-repo semantic index over real Qdrant):
- search: Cross-repo code search (type: symbol|text|semantic|hybrid)
- context: Budget-packed context bundle for a query
- ask: Cited RAG answer over indexed code
- index: (Re)index files/dirs into the cloud code index
"""
def _detect_lang(self, path: str | None, text: str | None = None) -> str:
@@ -311,6 +319,148 @@ All operations are PURE - safe to cache and parallelize.
f"Invalid regex pattern: {e}", param="match_pattern"
)
def _get_cloud(self) -> HanzoCloud:
"""Lazily build the shared Hanzo Cloud client for /v1/code/*."""
if self._cloud is None:
self._cloud = HanzoCloud()
if not self._cloud.configured():
raise ToolError(
code="INVALID_PARAMS",
message="Hanzo Cloud not configured: set HANZO_API_KEY "
"or add .apiKey to ~/.hanzo/config.json",
)
return self._cloud
def _collect_index_files(
self, path: str, max_files: int, max_bytes: int
) -> list[dict]:
"""Read a file or walk a directory into index payload records."""
root = Path(path) if Path(path).is_absolute() else Path(self.cwd) / path
if not root.exists():
raise NotFoundError(f"Path not found: {path}", uri=str(root))
skip_dirs = {".git", "node_modules", "target", "dist", "__pycache__", ".venv"}
files: list[dict] = []
candidates = [root] if root.is_file() else sorted(root.rglob("*"))
for p in candidates:
if len(files) >= max_files:
break
if p.is_dir() or any(sd in p.parts for sd in skip_dirs):
continue
if p.suffix not in LANG_MAP:
continue
try:
if p.stat().st_size > max_bytes:
continue
content = p.read_text(errors="ignore")
except (OSError, UnicodeDecodeError):
continue
rel = str(p.relative_to(root)) if root.is_dir() else p.name
files.append({"path": rel, "content": content})
return files
def _register_cloud_actions(self):
"""Register cloud-backed actions calling api.hanzo.ai /v1/code/*.
The cloud index is the cross-repo, semantic half of the hybrid: local
AST answers single-file questions instantly; these answer questions that
span repos (real Qdrant vectors + zen embeddings, server-side).
"""
@self.action("search", "Cross-repo code search via the cloud index")
async def search(
ctx: MCPContext,
query: str,
type: str = "hybrid",
repo: str | None = None,
limit: int = 20,
) -> dict:
"""Search indexed code across repos.
Args:
query: Search query
type: symbol | text | semantic | hybrid
repo: Restrict to a single repo (optional)
limit: Max results
Returns:
{query, type, results: [{repo, file, line, endLine, kind, symbol, snippet, score}]}
Effect: NONDETERMINISTIC_EFFECT (cloud I/O)
"""
data = await self._get_cloud().get(
"/v1/code/search",
{"q": query, "type": type, "repo": repo, "limit": limit},
)
return data
@self.action("context", "Budget-packed context bundle for a query")
async def context(
ctx: MCPContext,
query: str,
budgetTokens: int = 4000,
repo: str | None = None,
) -> dict:
"""Get a token-budgeted bundle of the most relevant code spans.
Returns:
{query, budgetTokens, usedTokens, spans: [...]}
"""
body: dict[str, Any] = {"query": query, "budgetTokens": budgetTokens}
if repo:
body["repo"] = repo
return await self._get_cloud().post("/v1/code/context", body)
@self.action("ask", "Cited RAG answer over indexed code")
async def ask(
ctx: MCPContext,
query: str,
repo: str | None = None,
) -> dict:
"""Ask a question over the code index; get a cited answer.
Returns:
{question, answer, citations: [{repo, file, line, endLine, symbol}]}
"""
body: dict[str, Any] = {"query": query}
if repo:
body["repo"] = repo
return await self._get_cloud().post("/v1/code/ask", body)
@self.action("index", "Index files/dir into the cloud code index")
async def index(
ctx: MCPContext,
repo: str,
path: str | None = None,
files: list[dict] | None = None,
prune: bool = False,
max_files: int = 2000,
max_bytes: int = 1_000_000,
) -> dict:
"""(Re)index code into the cloud index (incremental, content-hashed).
Provide either an explicit `files` list ([{path, content}]) or a
`path` (file or directory) to read and index.
Returns:
{repo, indexed, skipped, pruned, files, symbols, chunks, vectors, semantic}
"""
payload_files = files
if payload_files is None:
if not path:
raise InvalidParamsError(
"index requires either 'files' or 'path'", param="path"
)
payload_files = self._collect_index_files(path, max_files, max_bytes)
if not payload_files:
raise InvalidParamsError("no indexable files found", param="path")
return await self._get_cloud().post(
"/v1/code/index",
{"repo": repo, "files": payload_files, "prune": prune},
)
def _register_code_actions(self):
"""Register all code actions."""
pkg/hanzo-tools-code/pyproject.toml
+2 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "hanzo-tools-code"
version = "0.1.2"
version = "0.1.3"
description = "Code semantic tools for Hanzo AI (HIP-0300)"
readme = "README.md"
license = "MIT"
@@ -33,6 +33,7 @@ classifiers = [
dependencies = [
"hanzo-tools>=0.3.0",
"mcp>=1.0.0",
"httpx>=0.25.0",
]
[project.optional-dependencies]
pkg/hanzo-tools-code/tests/test_code_cloud.py
+118
View File
@@ -0,0 +1,118 @@
"""Offline tests for the cloud-backed code actions (no network).
A FakeCloud is injected so the /v1/code/* wiring is exercised deterministically:
correct paths, params, and body shapes, plus the fail-closed not-configured path.
"""
import asyncio
from hanzo_tools.core import CloudError
from hanzo_tools.code import CodeTool
class FakeCloud:
def __init__(self, get_result=None, post_result=None, configured=True):
self.get_result = get_result if get_result is not None else {}
self.post_result = post_result if post_result is not None else {}
self.calls = []
self._configured = configured
def configured(self):
return self._configured
async def get(self, path, params=None):
self.calls.append(("GET", path, params))
if isinstance(self.get_result, Exception):
raise self.get_result
return self.get_result
async def post(self, path, json_body=None):
self.calls.append(("POST", path, json_body))
if isinstance(self.post_result, Exception):
raise self.post_result
return self.post_result
def _run(coro):
return asyncio.run(coro)
def test_has_cloud_actions():
t = CodeTool()
for a in ("search", "context", "ask", "index"):
assert a in t._handlers
def test_search_maps_to_code_search():
t = CodeTool()
t._cloud = FakeCloud(get_result={"query": "q", "results": [{"file": "a.go"}]})
env = _run(t.call(None, action="search", query="parse", type="hybrid", limit=5))
assert env["ok"] is True
method, path, params = t._cloud.calls[0]
assert method == "GET" and path == "/v1/code/search"
assert params["q"] == "parse" and params["type"] == "hybrid" and params["limit"] == 5
assert env["data"]["results"] == [{"file": "a.go"}]
def test_context_posts_body():
t = CodeTool()
t._cloud = FakeCloud(post_result={"usedTokens": 10, "spans": []})
env = _run(t.call(None, action="context", query="how", budgetTokens=123, repo="r"))
assert env["ok"] is True
method, path, body = t._cloud.calls[0]
assert method == "POST" and path == "/v1/code/context"
assert body == {"query": "how", "budgetTokens": 123, "repo": "r"}
def test_ask_posts_body():
t = CodeTool()
t._cloud = FakeCloud(post_result={"question": "q", "answer": "a", "citations": []})
env = _run(t.call(None, action="ask", query="what does X do"))
assert env["ok"] is True
method, path, body = t._cloud.calls[0]
assert method == "POST" and path == "/v1/code/ask"
assert body == {"query": "what does X do"}
def test_index_collects_dir(tmp_path):
(tmp_path / "a.py").write_text("def a():\n return 1\n")
(tmp_path / "b.txt").write_text("ignored ext") # .txt not in LANG_MAP
(tmp_path / "node_modules").mkdir()
(tmp_path / "node_modules" / "c.py").write_text("skip me")
t = CodeTool()
t._cloud = FakeCloud(post_result={"repo": "demo", "indexed": 1})
env = _run(t.call(None, action="index", repo="demo", path=str(tmp_path)))
assert env["ok"] is True
method, path, body = t._cloud.calls[0]
assert method == "POST" and path == "/v1/code/index"
assert body["repo"] == "demo"
paths = {f["path"] for f in body["files"]}
assert paths == {"a.py"} # only .py, node_modules skipped
def test_index_explicit_files():
t = CodeTool()
t._cloud = FakeCloud(post_result={"indexed": 1})
files = [{"path": "x.go", "content": "package x"}]
env = _run(t.call(None, action="index", repo="r", files=files))
assert env["ok"] is True
_, _, body = t._cloud.calls[0]
assert body["files"] == files
def test_not_configured_fails_closed():
t = CodeTool()
t._cloud = FakeCloud(configured=False)
env = _run(t.call(None, action="search", query="x"))
assert env["ok"] is False
assert env["error"]["code"] == "INVALID_PARAMS"
def test_cloud_error_surfaces():
t = CodeTool()
t._cloud = FakeCloud(get_result=CloudError("boom", status=500))
env = _run(t.call(None, action="search", query="x"))
assert env["ok"] is False
assert env["error"]["code"] == "INTERNAL_ERROR"
pkg/hanzo-tools-net/hanzo_tools/net/__init__.py
+9 -5
View File
@@ -28,17 +28,20 @@ Usage:
from hanzo_tools.core import BaseTool, ToolRegistry
from .fetch_tool import FetchTool, fetch_tool
from .vision_tool import VisionTool, vision_tool
# Backward compat
NetTool = FetchTool
net_tool = fetch_tool
# Export list for tool discovery - HIP-0300 unified tool
TOOLS = [FetchTool]
# Export list for tool discovery - HIP-0300 unified tools
TOOLS = [FetchTool, VisionTool]
__all__ = [
"FetchTool",
"fetch_tool",
"VisionTool",
"vision_tool",
"NetTool",
"net_tool",
"register_tools",
@@ -57,6 +60,7 @@ def register_tools(mcp_server, **kwargs) -> list[BaseTool]:
List of registered tool instances
"""
cwd = kwargs.get("cwd")
tool = FetchTool(cwd=cwd)
ToolRegistry.register_tool(mcp_server, tool)
return [tool]
tools: list[BaseTool] = [FetchTool(cwd=cwd), VisionTool(cwd=cwd)]
for tool in tools:
ToolRegistry.register_tool(mcp_server, tool)
return tools
pkg/hanzo-tools-net/hanzo_tools/net/fetch_tool.py
+165 -50
View File
@@ -14,19 +14,63 @@ import asyncio
import os
import re
from pathlib import Path
from typing import ClassVar
from typing import Any, ClassVar
from urllib.parse import urljoin, urlparse
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import (
BaseTool,
InvalidParamsError,
CloudError,
HanzoCloud,
ToolError,
content_hash,
)
def _extract_markdown(resp: Any) -> str:
"""Best-effort markdown out of a Hanzo Crawl / Firecrawl response.
The crawl `data` is shape-polymorphic across versions — a bare string, a
{markdown|fit_markdown|raw_markdown|content} object, or a list of those — so
this accepts any of them and returns the first non-empty markdown found.
"""
def from_obj(obj: Any) -> str:
if isinstance(obj, str):
return obj.strip()
if isinstance(obj, dict):
for k in ("markdown", "fit_markdown", "raw_markdown", "content", "text"):
v = obj.get(k)
if isinstance(v, str) and v.strip():
return v.strip()
if isinstance(v, dict):
inner = from_obj(v)
if inner:
return inner
return ""
if not isinstance(resp, dict):
return from_obj(resp)
payload = resp.get("data", resp) # {status,msg,data} or {success,data:{...}}
if isinstance(payload, list):
for item in payload:
md = from_obj(item)
if md:
return md
return ""
return from_obj(payload)
def _crawl_metadata(resp: Any) -> dict:
"""Pull a metadata dict out of a crawl response, if present."""
if isinstance(resp, dict):
payload = resp.get("data", resp)
if isinstance(payload, dict) and isinstance(payload.get("metadata"), dict):
return payload["metadata"]
return {}
class FetchTool(BaseTool):
"""Network operations tool (HIP-0300).
@@ -46,14 +90,23 @@ class FetchTool(BaseTool):
super().__init__()
self.cwd = cwd or os.getcwd()
self._client = None
self._cloud: HanzoCloud | None = None
self._register_net_actions()
def _get_cloud(self) -> HanzoCloud | None:
"""Shared Hanzo Cloud client, or None when no API key is configured."""
if self._cloud is None:
self._cloud = HanzoCloud()
return self._cloud if self._cloud.configured() else None
@property
def description(self) -> str:
return """Network operations tool (HIP-0300).
Actions:
- search: Web search (Query → [URL, title, snippet])
- search: Web search — cloud Bing meta-search (api.hanzo.ai) with a local
DuckDuckGo fallback (Query → [URL, title, snippet])
- web_read: Fetch a URL → clean markdown (cloud Crawl4AI, local fallback)
- fetch: Retrieve URL content (URL → {text, mime, status})
- download: Save page with assets (URL → Path)
- crawl: Recursive site mirror (URL, depth → [Path])
@@ -119,6 +172,39 @@ Effect: NONDETERMINISTIC_EFFECT (network I/O)
matches = re.findall(pattern, html)
return [urljoin(base_url, m) for m in matches]
async def _duckduckgo_search(self, query: str, limit: int) -> list[dict]:
"""Local DuckDuckGo HTML search (no API key). Used as a fallback."""
client = await self._get_client()
response = await client.get(f"https://html.duckduckgo.com/html/?q={query}")
results: list[dict] = []
try:
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.text, "lxml")
for result in soup.select(".result")[:limit]:
title_el = result.select_one(".result__title")
snippet_el = result.select_one(".result__snippet")
link_el = result.select_one(".result__url")
if title_el and link_el:
results.append(
{
"title": title_el.get_text(strip=True),
"url": link_el.get("href", ""),
"snippet": (
snippet_el.get_text(strip=True) if snippet_el else ""
),
}
)
except ImportError:
results = [
{
"note": "Install beautifulsoup4 for better parsing",
"raw_length": len(response.text),
}
]
return results
def _register_net_actions(self):
"""Register all network actions."""
@@ -126,73 +212,102 @@ Effect: NONDETERMINISTIC_EFFECT (network I/O)
async def search(
ctx: MCPContext,
query: str,
engine: str = "duckduckgo",
engine: str = "hanzo",
limit: int = 10,
) -> dict:
"""Perform web search.
Prefers the cloud Bing meta-search (api.hanzo.ai /v1/websearch/search)
and falls back to local DuckDuckGo when the cloud is unreachable or
no API key is configured.
Args:
query: Search query
engine: Search engine (duckduckgo, google)
engine: hanzo (cloud, default) | duckduckgo (local)
limit: Max results
Returns:
{results: [{url, title, snippet}]}
{results: [{url, title, snippet}], query, engine, count}
Effect: NONDETERMINISTIC_EFFECT
"""
client = await self._get_client()
if engine == "duckduckgo":
# DuckDuckGo HTML search (no API key needed)
url = f"https://html.duckduckgo.com/html/?q={query}"
response = await client.get(url)
results = []
try:
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.text, "lxml")
for result in soup.select(".result")[:limit]:
title_el = result.select_one(".result__title")
snippet_el = result.select_one(".result__snippet")
link_el = result.select_one(".result__url")
if title_el and link_el:
results.append(
{
"title": title_el.get_text(strip=True),
"url": link_el.get("href", ""),
"snippet": (
snippet_el.get_text(strip=True)
if snippet_el
else ""
),
}
)
except ImportError:
# Fallback without beautifulsoup
results = [
{
"note": "Install beautifulsoup4 for better parsing",
"raw_length": len(response.text),
if engine in ("hanzo", "auto", "bing"):
cloud = self._get_cloud()
if cloud is not None:
try:
data = await cloud.get("/v1/websearch/search", {"q": query})
results = [
{
"url": r.get("url", ""),
"title": r.get("title", ""),
"snippet": r.get("content") or r.get("snippet", ""),
}
for r in (data.get("results") or [])[:limit]
]
return {
"results": results,
"query": query,
"engine": "hanzo",
"count": len(results),
}
]
else:
raise InvalidParamsError(
f"Unknown engine: {engine}",
param="engine",
expected="duckduckgo",
)
except CloudError:
pass # fall through to local DuckDuckGo
results = await self._duckduckgo_search(query, limit)
return {
"results": results,
"query": query,
"engine": engine,
"engine": "duckduckgo",
"count": len(results),
}
@self.action("web_read", "Fetch a URL → clean markdown")
async def web_read(
ctx: MCPContext,
url: str,
) -> dict:
"""Read a web page as clean markdown.
Prefers the cloud reader (api.hanzo.ai /v1/crawl, Crawl4AI) and falls
back to a local fetch + text extraction when the cloud is unreachable.
Returns:
{url, markdown, source: "cloud" | "local", metadata?}
Effect: NONDETERMINISTIC_EFFECT
"""
cloud = self._get_cloud()
if cloud is not None:
try:
data = await cloud.post("/v1/crawl", {"url": url})
markdown = _extract_markdown(data)
if markdown:
return {
"url": url,
"markdown": markdown,
"source": "cloud",
"metadata": _crawl_metadata(data),
}
except CloudError:
pass # fall through to local fetch
client = await self._get_client()
try:
response = await client.get(url)
except Exception as e:
raise ToolError(
code="INTERNAL_ERROR",
message=f"web_read failed (cloud + local): {e}",
details={"url": url},
)
text = self._extract_text(response.text)
return {
"url": str(response.url),
"markdown": text,
"source": "local",
"status": response.status_code,
}
@self.action("fetch", "Retrieve URL content")
async def fetch(
ctx: MCPContext,
pkg/hanzo-tools-net/hanzo_tools/net/vision_tool.py
+165
View File
@@ -0,0 +1,165 @@
"""Vision tool for HIP-0300 architecture.
Sends an image to a Hanzo vision model (api.hanzo.ai /v1/chat/completions,
OpenAI-compatible) and returns the model's answer. The default SKU is
``zen3-vl`` (``GET /v1/models`` also lists ``zen3-omni``); override via ``model``.
Effect lattice position: NONDETERMINISTIC_EFFECT (cloud inference).
"""
import base64
import mimetypes
import os
from pathlib import Path
from typing import Any, ClassVar
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import (
BaseTool,
HanzoCloud,
InvalidParamsError,
NotFoundError,
ToolError,
)
DEFAULT_VISION_MODEL = "zen3-vl"
class VisionTool(BaseTool):
"""Vision-language tool: image (+ prompt) → text answer (HIP-0300)."""
name: ClassVar[str] = "vision"
VERSION: ClassVar[str] = "0.1.0"
def __init__(self, cwd: str | None = None):
super().__init__()
self.cwd = cwd or os.getcwd()
self._cloud: HanzoCloud | None = None
self._register_vision_actions()
@property
def description(self) -> str:
return """Vision-language tool (HIP-0300).
Actions:
- ask: Ask a question about an image (image + prompt → text)
- describe: Describe an image (image → text)
Image accepts a local file path, an http(s) URL, or a data: URI.
Backed by api.hanzo.ai /v1/chat/completions (default model: zen3-vl).
Effect: NONDETERMINISTIC_EFFECT (cloud inference)
"""
def _get_cloud(self) -> HanzoCloud:
if self._cloud is None:
self._cloud = HanzoCloud()
if not self._cloud.configured():
raise ToolError(
code="INVALID_PARAMS",
message="Hanzo Cloud not configured: set HANZO_API_KEY "
"or add .apiKey to ~/.hanzo/config.json",
)
return self._cloud
async def _image_url(self, image: str) -> str:
"""Normalize an image reference to an inline data: URI.
A data: URI passes through; a local path is read and encoded; an http(s)
URL is fetched and encoded (the vision backend takes inline images, not
remote URLs). Encoding here makes every input path work uniformly.
"""
if image.startswith("data:"):
return image
if image.startswith(("http://", "https://")):
try:
import httpx
async with httpx.AsyncClient(
follow_redirects=True,
timeout=30.0,
headers={"User-Agent": "Mozilla/5.0 (compatible; HanzoBot/1.0)"},
) as client:
resp = await client.get(image)
resp.raise_for_status()
except Exception as e:
raise ToolError(
code="INTERNAL_ERROR",
message=f"could not fetch image: {e}",
details={"image": image},
)
mime = resp.headers.get("content-type", "image/png").split(";")[0].strip()
b64 = base64.b64encode(resp.content).decode("ascii")
return f"data:{mime};base64,{b64}"
p = Path(image) if Path(image).is_absolute() else Path(self.cwd) / image
if not p.exists():
raise NotFoundError(f"Image not found: {image}", uri=str(p))
mime = mimetypes.guess_type(p.name)[0] or "image/png"
b64 = base64.b64encode(p.read_bytes()).decode("ascii")
return f"data:{mime};base64,{b64}"
async def _complete(
self, image: str, prompt: str, model: str, max_tokens: int
) -> dict:
content = [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": await self._image_url(image)}},
]
body: dict[str, Any] = {
"model": model,
"messages": [{"role": "user", "content": content}],
"max_tokens": max_tokens,
}
data = await self._get_cloud().post("/v1/chat/completions", body)
choices = data.get("choices") or []
answer = ""
if choices:
answer = (choices[0].get("message") or {}).get("content") or ""
return {
"answer": answer,
"model": data.get("model", model),
"usage": data.get("usage"),
}
def _register_vision_actions(self):
@self.action("ask", "Ask a question about an image")
async def ask(
ctx: MCPContext,
image: str,
prompt: str = "Describe this image in detail.",
model: str = DEFAULT_VISION_MODEL,
max_tokens: int = 1024,
) -> dict:
"""Ask a vision model about an image.
Args:
image: Local path, http(s) URL, or data: URI
prompt: Question / instruction
model: Vision SKU (default zen3-vl)
max_tokens: Max response tokens
Returns:
{answer, model, usage}
"""
if not image:
raise InvalidParamsError("image is required", param="image")
return await self._complete(image, prompt, model, max_tokens)
@self.action("describe", "Describe an image")
async def describe(
ctx: MCPContext,
image: str,
model: str = DEFAULT_VISION_MODEL,
max_tokens: int = 1024,
) -> dict:
"""Describe an image (convenience wrapper over ask)."""
if not image:
raise InvalidParamsError("image is required", param="image")
return await self._complete(
image, "Describe this image in detail.", model, max_tokens
)
# Singleton
vision_tool = VisionTool
pkg/hanzo-tools-net/pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "hanzo-tools-net"
version = "0.1.2"
version = "0.1.3"
description = "Network tools for Hanzo AI (HIP-0300)"
readme = "README.md"
license = "MIT"
pkg/hanzo-tools-net/tests/test_net_cloud.py
+202
View File
@@ -0,0 +1,202 @@
"""Offline tests for the cloud-backed net + vision wiring (no network).
Cloud calls are faked; the local DuckDuckGo/fetch fallbacks are patched so the
routing logic (cloud-first, local-fallback) is exercised deterministically.
"""
import asyncio
import base64
from hanzo_tools.core import CloudError
from hanzo_tools.net import FetchTool, VisionTool
from hanzo_tools.net.fetch_tool import _crawl_metadata, _extract_markdown
class FakeCloud:
def __init__(self, get_result=None, post_result=None, configured=True):
self.get_result = get_result if get_result is not None else {}
self.post_result = post_result if post_result is not None else {}
self.calls = []
self._configured = configured
def configured(self):
return self._configured
async def get(self, path, params=None):
self.calls.append(("GET", path, params))
if isinstance(self.get_result, Exception):
raise self.get_result
return self.get_result
async def post(self, path, json_body=None):
self.calls.append(("POST", path, json_body))
if isinstance(self.post_result, Exception):
raise self.post_result
return self.post_result
class FakeResponse:
def __init__(self, text="<html><body>hi there</body></html>", url="https://x/"):
self.text = text
self.url = url
self.status_code = 200
class FakeClient:
async def get(self, url, **kw):
return FakeResponse(url=url)
def _run(coro):
return asyncio.run(coro)
# ── search ───────────────────────────────────────────────────────────────
def test_search_cloud_normalizes_searxng():
t = FetchTool()
t._cloud = FakeCloud(
get_result={
"results": [
{"url": "u1", "title": "t1", "content": "c1", "engine": "bing"},
{"url": "u2", "title": "t2", "content": "c2"},
]
}
)
env = _run(t.call(None, action="search", query="hanzo", limit=5))
assert env["ok"] is True
_, path, params = t._cloud.calls[0]
assert path == "/v1/websearch/search" and params["q"] == "hanzo"
data = env["data"]
assert data["engine"] == "hanzo" and data["count"] == 2
assert data["results"][0] == {"url": "u1", "title": "t1", "snippet": "c1"}
def test_search_falls_back_to_duckduckgo_on_cloud_error():
t = FetchTool()
t._cloud = FakeCloud(get_result=CloudError("down"))
async def fake_ddg(query, limit):
return [{"url": "d", "title": "ddg", "snippet": "s"}]
t._duckduckgo_search = fake_ddg
env = _run(t.call(None, action="search", query="q"))
assert env["ok"] is True
assert env["data"]["engine"] == "duckduckgo"
assert env["data"]["results"][0]["title"] == "ddg"
def test_search_engine_duckduckgo_skips_cloud():
t = FetchTool()
t._cloud = FakeCloud() # configured, but should not be used
async def fake_ddg(query, limit):
return [{"url": "d", "title": "ddg", "snippet": "s"}]
t._duckduckgo_search = fake_ddg
env = _run(t.call(None, action="search", query="q", engine="duckduckgo"))
assert env["ok"] is True
assert env["data"]["engine"] == "duckduckgo"
assert t._cloud.calls == [] # cloud never touched
# ── web_read ─────────────────────────────────────────────────────────────
def test_web_read_cloud_markdown():
t = FetchTool()
t._cloud = FakeCloud(
post_result={"status": "ok", "data": {"markdown": "# Title\n\nBody"}}
)
env = _run(t.call(None, action="web_read", url="https://x"))
assert env["ok"] is True
assert env["data"]["source"] == "cloud"
assert env["data"]["markdown"] == "# Title\n\nBody"
_, path, body = t._cloud.calls[0]
assert path == "/v1/crawl" and body == {"url": "https://x"}
def test_web_read_falls_back_local_when_cloud_errors():
t = FetchTool()
t._cloud = FakeCloud(post_result=CloudError("dns fail"))
async def fake_get_client():
return FakeClient()
t._get_client = fake_get_client
env = _run(t.call(None, action="web_read", url="https://x"))
assert env["ok"] is True
assert env["data"]["source"] == "local"
assert "hi there" in env["data"]["markdown"]
def test_web_read_falls_back_when_cloud_returns_no_markdown():
# Hanzo Crawl error envelope (data=null) → treated as no markdown → local.
t = FetchTool()
t._cloud = FakeCloud(post_result={"status": "error", "msg": "boom", "data": None})
async def fake_get_client():
return FakeClient()
t._get_client = fake_get_client
env = _run(t.call(None, action="web_read", url="https://x"))
assert env["ok"] is True and env["data"]["source"] == "local"
def test_extract_markdown_shapes():
assert _extract_markdown({"data": {"markdown": "m"}}) == "m"
assert _extract_markdown({"success": True, "data": {"fit_markdown": "fm"}}) == "fm"
assert _extract_markdown({"data": [{"raw_markdown": "rm"}]}) == "rm"
assert _extract_markdown("bare string") == "bare string"
assert _extract_markdown({"status": "error", "data": None}) == ""
assert _crawl_metadata({"data": {"metadata": {"title": "T"}}}) == {"title": "T"}
# ── vision ───────────────────────────────────────────────────────────────
def test_vision_encodes_local_file(tmp_path):
png = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
)
p = tmp_path / "pix.png"
p.write_bytes(png)
t = VisionTool()
uri = _run(t._image_url(str(p)))
assert uri.startswith("data:image/png;base64,")
def test_vision_data_uri_passthrough():
t = VisionTool()
duri = "data:image/png;base64,AAAA"
assert _run(t._image_url(duri)) == duri
def test_vision_complete_extracts_answer():
t = VisionTool()
t._cloud = FakeCloud(
post_result={
"model": "zen3-vl",
"choices": [{"message": {"content": "a cat"}}],
"usage": {"total_tokens": 5},
}
)
env = _run(
t.call(None, action="ask", image="data:image/png;base64,AAAA", prompt="what?")
)
assert env["ok"] is True
assert env["data"]["answer"] == "a cat"
_, path, body = t._cloud.calls[0]
assert path == "/v1/chat/completions"
assert body["messages"][0]["content"][1]["image_url"]["url"] == (
"data:image/png;base64,AAAA"
)
def test_vision_not_configured_fails_closed():
t = VisionTool()
t._cloud = FakeCloud(configured=False)
env = _run(t.call(None, action="ask", image="data:image/png;base64,AAAA"))
assert env["ok"] is False and env["error"]["code"] == "INVALID_PARAMS"
pkg/hanzo-tools-vector/hanzo_tools/vector/__init__.py
+17 -40
View File
@@ -1,72 +1,49 @@
"""Vector/embedding tools for Hanzo AI.
Tools:
- vector_index: Index documents for semantic search
- vector_search: Search indexed documents
- index: Project indexing
Default surface is the cloud-backed `VectorTool` (real Qdrant + zen embeddings
via api.hanzo.ai). Actions: search, index, embed.
The local Infinity store (infinity_store / vector_search / vector_index) remains
importable for offline use but is NOT the default and never substitutes a mock
silently — the random-vector mock is opt-in via HANZO_VECTOR_ALLOW_MOCK=1.
Install:
pip install hanzo-tools-vector[full]
pip install hanzo-tools-vector # cloud (default)
pip install hanzo-tools-vector[full] # + local Infinity store
"""
import logging
logger = logging.getLogger(__name__)
_tools = []
VECTOR_AVAILABLE = False
# Primary surface: cloud-backed, real vectors. Depends only on hanzo_tools.core.
from .cloud_vector import VectorTool, vector_tool
try:
from .index_tool import IndexTool
_tools.append(IndexTool)
from .vector_index import VectorIndexTool
_tools.append(VectorIndexTool)
from .vector_search import VectorSearchTool
_tools.append(VectorSearchTool)
VECTOR_AVAILABLE = True
except ImportError as e:
logger.debug(f"Vector tools not available: {e}")
TOOLS = _tools
TOOLS = [VectorTool]
VECTOR_AVAILABLE = True
__all__ = [
"TOOLS",
"VECTOR_AVAILABLE",
"VectorTool",
"vector_tool",
"register_tools",
]
if VECTOR_AVAILABLE:
__all__.extend(["IndexTool", "VectorIndexTool", "VectorSearchTool"])
def register_tools(
mcp_server, permission_manager=None, enabled_tools: dict[str, bool] | None = None
):
"""Register vector tools with MCP server."""
if not VECTOR_AVAILABLE:
logger.warning("Vector tools not available - missing dependencies")
return []
def register_tools(mcp_server, permission_manager=None, enabled_tools=None):
"""Register the cloud vector tool with the MCP server."""
from hanzo_tools.core import ToolRegistry
enabled = enabled_tools or {}
registered = []
for tool_class in TOOLS:
tool_name = getattr(tool_class, "name", tool_class.__name__.lower())
if enabled.get(tool_name, True):
try:
tool = (
tool_class(permission_manager)
if permission_manager
else tool_class()
)
tool = tool_class()
ToolRegistry.register_tool(mcp_server, tool)
registered.append(tool)
except Exception as e:
logger.warning(f"Failed to register {tool_name}: {e}")
return registered
pkg/hanzo-tools-vector/hanzo_tools/vector/cloud_vector.py
+211
View File
@@ -0,0 +1,211 @@
"""Cloud-backed vector tool (HIP-0300) — the default semantic surface.
Semantic search, indexing, and raw embeddings run against the live Hanzo Cloud
backend (api.hanzo.ai) so vectors are REAL: zen embeddings into the in-cluster
Qdrant, never a local random-vector mock.
Backend map (all verified on api.hanzo.ai /v1):
- search → GET /v1/code/search (type=semantic|hybrid|symbol|text over Qdrant)
- index → POST /v1/code/index ({repo, files:[{path,content}]})
- embed → POST /v1/embeddings ({model, input} → 1024-dim vectors)
A `collection` names the index namespace (the backend's `repo`). The local
Infinity store remains available behind an explicit opt-in (see infinity_store);
this tool never falls back to a mock silently.
"""
import os
from typing import Any, ClassVar
from pathlib import Path
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import (
BaseTool,
ToolError,
HanzoCloud,
NotFoundError,
InvalidParamsError,
)
DEFAULT_EMBED_MODEL = "zen-embedding"
# Text-ish extensions worth indexing when walking a directory.
_INDEX_EXTS = {
".py", ".js", ".ts", ".tsx", ".jsx", ".go", ".rs", ".c", ".cpp", ".h",
".hpp", ".java", ".rb", ".php", ".swift", ".kt", ".scala", ".cs", ".lua",
".sh", ".bash", ".zsh", ".json", ".yaml", ".yml", ".toml", ".md", ".rst",
".txt", ".html", ".css", ".sql",
}
_SKIP_DIRS = {".git", "node_modules", "target", "dist", "__pycache__", ".venv"}
def _collect_files(root: Path, max_files: int, max_bytes: int) -> list[dict]:
"""Read a file or walk a directory into [{path, content}] index records."""
candidates = [root] if root.is_file() else sorted(root.rglob("*"))
files: list[dict] = []
for p in candidates:
if len(files) >= max_files:
break
if p.is_dir() or any(sd in p.parts for sd in _SKIP_DIRS):
continue
if p.suffix not in _INDEX_EXTS:
continue
try:
if p.stat().st_size > max_bytes:
continue
content = p.read_text(errors="ignore")
except (OSError, UnicodeDecodeError):
continue
rel = str(p.relative_to(root)) if root.is_dir() else p.name
files.append({"path": rel, "content": content})
return files
class VectorTool(BaseTool):
"""Unified cloud vector tool: semantic search, index, embed (HIP-0300)."""
name: ClassVar[str] = "vector"
VERSION: ClassVar[str] = "0.2.0"
def __init__(self, cwd: str | None = None):
super().__init__()
self.cwd = cwd or os.getcwd()
self._cloud: HanzoCloud | None = None
self._register_vector_actions()
@property
def description(self) -> str:
return """Semantic vector tool (HIP-0300), cloud-backed (real Qdrant).
Actions:
- search: Semantic/hybrid search over an indexed collection
- index: Index a file or directory into a collection
- embed: Compute embeddings for text (1024-dim zen vectors)
Vectors are computed server-side (zen embeddings) and stored in the in-cluster
Qdrant. There is no silent local mock. Set a collection to namespace an index.
Effect: NONDETERMINISTIC_EFFECT (cloud I/O)
"""
def _get_cloud(self) -> HanzoCloud:
if self._cloud is None:
self._cloud = HanzoCloud()
if not self._cloud.configured():
raise ToolError(
code="INVALID_PARAMS",
message="Hanzo Cloud not configured: set HANZO_API_KEY "
"or add .apiKey to ~/.hanzo/config.json",
)
return self._cloud
def _register_vector_actions(self):
@self.action("search", "Semantic search over an indexed collection")
async def search(
ctx: MCPContext,
query: str,
collection: str | None = None,
limit: int = 10,
type: str = "hybrid",
) -> dict:
"""Search a collection by meaning (real Qdrant vectors).
Defaults to `hybrid` (RRF over semantic + lexical + symbolic), which
uses the Qdrant vector signal while staying robust when the pure
semantic path is warming up. Pass type="semantic" to force vectors.
Args:
query: Natural-language query
collection: Index namespace (backend repo); optional
limit: Max results
type: hybrid (default) | semantic | symbol | text
Returns:
{query, collection, results: [{file, repo, line, endLine, kind,
symbol, snippet, score}]}
"""
if not query:
raise InvalidParamsError("query is required", param="query")
data = await self._get_cloud().get(
"/v1/code/search",
{"q": query, "type": type, "repo": collection, "limit": limit},
)
return {
"query": query,
"collection": collection,
"results": data.get("results", []),
"count": len(data.get("results", [])),
}
@self.action("index", "Index a file or directory into a collection")
async def index(
ctx: MCPContext,
collection: str,
path: str | None = None,
files: list[dict] | None = None,
prune: bool = False,
max_files: int = 2000,
max_bytes: int = 1_000_000,
) -> dict:
"""Index content into a collection (server-side embeddings → Qdrant).
Provide either `files` ([{path, content}]) or a `path` (file or dir).
Returns:
{collection, indexed, skipped, files, symbols, chunks, vectors,
semantic}
"""
payload_files = files
if payload_files is None:
if not path:
raise InvalidParamsError(
"index requires either 'files' or 'path'", param="path"
)
root = Path(path) if Path(path).is_absolute() else Path(self.cwd) / path
if not root.exists():
raise NotFoundError(f"Path not found: {path}", uri=str(root))
payload_files = _collect_files(root, max_files, max_bytes)
if not payload_files:
raise InvalidParamsError("no indexable files found", param="path")
data = await self._get_cloud().post(
"/v1/code/index",
{"repo": collection, "files": payload_files, "prune": prune},
)
data.setdefault("collection", collection)
return data
@self.action("embed", "Compute embeddings for text")
async def embed(
ctx: MCPContext,
input: Any,
model: str = DEFAULT_EMBED_MODEL,
) -> dict:
"""Compute embeddings (1024-dim zen vectors).
Args:
input: A string or list of strings
model: Embedding model (default zen-embedding)
Returns:
{model, count, dimensions, vectors: [[float, ...], ...]}
"""
if input is None or input == "":
raise InvalidParamsError("input is required", param="input")
data = await self._get_cloud().post(
"/v1/embeddings", {"model": model, "input": input}
)
rows = data.get("data") or []
vectors = [r.get("embedding", []) for r in rows]
return {
"model": data.get("model", model),
"count": len(vectors),
"dimensions": len(vectors[0]) if vectors else 0,
"vectors": vectors,
"usage": data.get("usage"),
}
# Singleton
vector_tool = VectorTool
pkg/hanzo-tools-vector/hanzo_tools/vector/infinity_store.py
+22 -10
View File
@@ -1,6 +1,7 @@
"""Infinity vector database integration for Hanzo AI."""
import json
import os
import hashlib
from typing import Any, Dict, List, Tuple, Optional
from pathlib import Path
@@ -11,10 +12,15 @@ try:
INFINITY_AVAILABLE = True
except ImportError:
# Use mock implementation when infinity_embedded is not available
from . import mock_infinity as infinity_embedded
# No silent mock: the random-vector mock makes search meaningless, so it is
# OPT-IN only (HANZO_VECTOR_ALLOW_MOCK=1) for tests. Otherwise the local store
# is simply unavailable — callers should use the cloud VectorTool (default).
infinity_embedded = None
INFINITY_AVAILABLE = False
if os.environ.get("HANZO_VECTOR_ALLOW_MOCK") == "1":
from . import mock_infinity as infinity_embedded
INFINITY_AVAILABLE = True # Mock is always available
INFINITY_AVAILABLE = True
from .ast_analyzer import Symbol, FileAST, ASTAnalyzer, create_symbol_embedding_text
@@ -772,16 +778,22 @@ class InfinityVectorStore:
return chunks
def _generate_embedding(self, text: str) -> List[float]:
"""Generate embedding for text.
"""Generate an embedding for text.
For now, this returns a dummy embedding. In a real implementation,
you would call an embedding API (OpenAI, Cohere, etc.) or use a local model.
The local Infinity store ships no embedder. Rather than return random
vectors (which make similarity search meaningless), this raises unless
the mock is explicitly enabled (HANZO_VECTOR_ALLOW_MOCK=1). For real
vectors, use the cloud VectorTool (api.hanzo.ai /v1/embeddings).
"""
# This is a placeholder - you would implement actual embedding generation here
# For now, return a random embedding of the correct dimension
import random
if os.environ.get("HANZO_VECTOR_ALLOW_MOCK") == "1":
import random
return [random.random() for _ in range(self.dimension)]
return [random.random() for _ in range(self.dimension)]
raise NotImplementedError(
"Local Infinity store has no embedder; use the cloud VectorTool "
"(default) for real embeddings, or set HANZO_VECTOR_ALLOW_MOCK=1 "
"for random test vectors."
)
async def get_stats(self) -> Dict[str, Any]:
"""Get statistics about the vector store.
pkg/hanzo-tools-vector/pyproject.toml
+3 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools-vector"
version = "0.2.0"
version = "0.2.1"
description = "Vector/embedding tools for Hanzo AI - indexing, search, RAG"
readme = "README.md"
requires-python = ">=3.12"
@@ -16,10 +16,12 @@ dependencies = [
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
"httpx>=0.25.0",
]
[project.optional-dependencies]
full = ["sentence-transformers>=2.0.0", "faiss-cpu>=1.7.0"]
infinity = ["infinity-embedded>=0.5.0"]
[project.entry-points."hanzo.tools"]
vector = "hanzo_tools.vector:TOOLS"
pkg/hanzo-tools-vector/tests/test_vector_tools.py
+123 -1
View File
@@ -1,4 +1,10 @@
"""Tests for hanzo-tools-vector."""
"""Tests for hanzo-tools-vector.
Import/registration checks plus offline tests of the cloud-backed VectorTool
wiring (/v1/code/search, /v1/code/index, /v1/embeddings) and the mock-kill.
"""
import asyncio
import pytest
@@ -15,3 +21,119 @@ class TestImports:
from hanzo_tools.vector import TOOLS
assert len(TOOLS) > 0
class FakeCloud:
def __init__(self, get_result=None, post_result=None, configured=True):
self.get_result = get_result if get_result is not None else {}
self.post_result = post_result if post_result is not None else {}
self.calls = []
self._configured = configured
def configured(self):
return self._configured
async def get(self, path, params=None):
self.calls.append(("GET", path, params))
if isinstance(self.get_result, Exception):
raise self.get_result
return self.get_result
async def post(self, path, json_body=None):
self.calls.append(("POST", path, json_body))
if isinstance(self.post_result, Exception):
raise self.post_result
return self.post_result
def _run(coro):
return asyncio.run(coro)
class TestCloudVectorTool:
def test_actions_present(self):
from hanzo_tools.vector import VectorTool
t = VectorTool()
for a in ("search", "index", "embed"):
assert a in t._handlers
def test_search_maps_to_code_search_hybrid_default(self):
from hanzo_tools.vector import VectorTool
t = VectorTool()
t._cloud = FakeCloud(get_result={"results": [{"file": "a.go"}]})
env = _run(t.call(None, action="search", query="qdrant", collection="c"))
assert env["ok"] is True
method, path, params = t._cloud.calls[0]
assert method == "GET" and path == "/v1/code/search"
assert params["q"] == "qdrant"
assert params["type"] == "hybrid" # robust default
assert params["repo"] == "c"
assert env["data"]["count"] == 1
def test_embed_maps_to_embeddings(self):
from hanzo_tools.vector import VectorTool
t = VectorTool()
t._cloud = FakeCloud(
post_result={
"model": "zen-embedding",
"data": [{"embedding": [0.1, 0.2, 0.3]}],
"usage": {"total_tokens": 2},
}
)
env = _run(t.call(None, action="embed", input="hello"))
assert env["ok"] is True
method, path, body = t._cloud.calls[0]
assert method == "POST" and path == "/v1/embeddings"
assert body == {"model": "zen-embedding", "input": "hello"}
assert env["data"]["dimensions"] == 3 and env["data"]["count"] == 1
def test_index_collects_dir(self, tmp_path):
from hanzo_tools.vector import VectorTool
(tmp_path / "a.py").write_text("print(1)")
(tmp_path / "README.md").write_text("# hi")
(tmp_path / ".git").mkdir()
(tmp_path / ".git" / "config").write_text("x")
t = VectorTool()
t._cloud = FakeCloud(post_result={"indexed": 2})
env = _run(t.call(None, action="index", collection="c", path=str(tmp_path)))
assert env["ok"] is True
method, path, body = t._cloud.calls[0]
assert method == "POST" and path == "/v1/code/index"
assert body["repo"] == "c"
names = {f["path"] for f in body["files"]}
assert names == {"a.py", "README.md"} # .git skipped
def test_not_configured_fails_closed(self):
from hanzo_tools.vector import VectorTool
t = VectorTool()
t._cloud = FakeCloud(configured=False)
env = _run(t.call(None, action="search", query="x"))
assert env["ok"] is False
assert env["error"]["code"] == "INVALID_PARAMS"
class TestNoSilentMock:
def test_infinity_mock_is_opt_in(self):
"""When infinity_embedded is absent, the store is UNAVAILABLE (no mock)."""
from hanzo_tools.vector import infinity_store as ins
if ins.infinity_embedded is None:
assert ins.INFINITY_AVAILABLE is False
else:
pytest.skip("infinity_embedded is installed in this environment")
def test_local_embedding_raises_without_mock(self, monkeypatch):
"""The local store must not silently return random vectors."""
from hanzo_tools.vector.infinity_store import InfinityVectorStore
monkeypatch.delenv("HANZO_VECTOR_ALLOW_MOCK", raising=False)
store = InfinityVectorStore.__new__(InfinityVectorStore) # no __init__/DB
store.dimension = 8
with pytest.raises(NotImplementedError):
store._generate_embedding("hello")
pkg/hanzo-tools/hanzo_tools/core/__init__.py
+11
View File
@@ -29,6 +29,12 @@ from hanzo_tools.core.unified import (
content_hash,
)
from hanzo_tools.core.id_tool import IdTool, id_tool
from hanzo_tools.core.cloud import (
HanzoCloud,
CloudError,
cloud_api_key,
cloud_api_base,
)
from hanzo_tools.core.decorators import auto_timeout
from hanzo_tools.core.validation import ValidationResult, validate_path_parameter
from hanzo_tools.core.permissions import PermissionManager
@@ -54,6 +60,11 @@ __all__ = [
# Identity tool
"IdTool",
"id_tool",
# Cloud client (api.hanzo.ai)
"HanzoCloud",
"CloudError",
"cloud_api_key",
"cloud_api_base",
# Context
"ToolContext",
"create_tool_context",
pkg/hanzo-tools/hanzo_tools/core/cloud.py
+152
View File
@@ -0,0 +1,152 @@
"""One client for the live Hanzo Cloud backend (api.hanzo.ai).
Every hanzo-tools package that reaches the cloud — code index/search, vector
search, web search, vision — talks to it through this single seam. There is
exactly one place that knows the base URL, the auth header, and how to turn a
non-2xx into a typed error; tools compose it, never re-implement it.
Auth resolves in order: ``HANZO_API_KEY`` env, then the ``apiKey`` (hk- key) in
``~/.hanzo/config.json``. Base URL is ``https://api.hanzo.ai``, overridable via
``HANZO_API_BASE``.
Reference: HIP-0300 unified tools; api.hanzo.ai /v1 surface.
"""
import os
import json
from typing import Any, ClassVar
from pathlib import Path
DEFAULT_BASE = "https://api.hanzo.ai"
def cloud_api_key() -> str | None:
"""Resolve the hk- API key: env first, then ~/.hanzo/config.json."""
env = os.environ.get("HANZO_API_KEY") or os.environ.get("HANZO_KEY")
if env and env.strip():
return env.strip()
cfg = Path.home() / ".hanzo" / "config.json"
if cfg.exists():
try:
key = json.loads(cfg.read_text()).get("apiKey")
if key and str(key).strip():
return str(key).strip()
except (OSError, ValueError):
return None
return None
def cloud_api_base() -> str:
"""Resolve the API base URL (no trailing slash)."""
base = (
os.environ.get("HANZO_API_BASE")
or os.environ.get("HANZO_BASE_URL")
or DEFAULT_BASE
)
return base.rstrip("/")
class CloudError(Exception):
"""A Hanzo Cloud call failed (network error or non-2xx response).
``status`` is the HTTP status when the server answered, else None (a
transport/DNS failure). Tools inspect it to decide fall-back vs surface.
"""
def __init__(self, message: str, status: int | None = None):
super().__init__(message)
self.status = status
class HanzoCloud:
"""Async client for the api.hanzo.ai /v1 surface.
Lazily builds one ``httpx.AsyncClient`` (reused for the process lifetime,
mirroring the other network tools). ``configured()`` reports whether a key
is present so tools can fall back to a local path instead of failing.
"""
USER_AGENT: ClassVar[str] = "hanzo-tools/cloud"
def __init__(
self,
*,
base_url: str | None = None,
key: str | None = None,
timeout: float = 60.0,
):
self.base_url = (base_url or cloud_api_base()).rstrip("/")
self.key = key if key is not None else cloud_api_key()
self.timeout = timeout
self._client: Any = None
def configured(self) -> bool:
"""True when an API key is available."""
return bool(self.key)
def _headers(self) -> dict[str, str]:
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": self.USER_AGENT,
}
if self.key:
headers["Authorization"] = f"Bearer {self.key}"
return headers
async def _get_client(self):
if self._client is None:
try:
import httpx
except ImportError as e:
raise CloudError("httpx not installed. Run: pip install httpx") from e
self._client = httpx.AsyncClient(
base_url=self.base_url,
timeout=self.timeout,
follow_redirects=True,
)
return self._client
async def _request(self, method: str, path: str, **kw: Any) -> Any:
client = await self._get_client()
try:
resp = await client.request(
method, path, headers=self._headers(), **kw
)
except Exception as e: # transport/DNS/timeout — no HTTP status
raise CloudError(f"{method} {path} failed: {e}") from e
if resp.status_code >= 400:
raise CloudError(
f"{method} {path} → {resp.status_code}: {_short_body(resp)}",
status=resp.status_code,
)
try:
return resp.json()
except ValueError:
return {"text": resp.text}
async def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
"""GET path, returning parsed JSON. Raises CloudError on failure."""
clean = {k: v for k, v in (params or {}).items() if v is not None}
return await self._request("GET", path, params=clean)
async def post(self, path: str, json_body: dict[str, Any] | None = None) -> Any:
"""POST json_body to path, returning parsed JSON. Raises on failure."""
return await self._request("POST", path, json=json_body or {})
async def aclose(self) -> None:
if self._client is not None:
await self._client.aclose()
self._client = None
def _short_body(resp: Any, limit: int = 300) -> str:
"""A trimmed, single-line error body for CloudError messages."""
try:
body = resp.text
except Exception:
return ""
body = " ".join(body.split())
return body[:limit] + ("…" if len(body) > limit else "")
pkg/hanzo-tools/pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools"
version = "0.3.2"
version = "0.3.3"
description = "Hanzo AI tools - core infrastructure and tool bundles"
readme = "README.md"
requires-python = ">=3.12"
uv.lock
Generated
+2497 -2480
View File
File diff suppressed because it is too large Load Diff