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(mcp): unify tool surfaces, rename iching→zen, fix test infrastructure

Browse Source 
- Consolidate memory tools into single unified surface
- Rename IChingTool → ZenTool across agent tools
- Rename tool files: proc_tool→exec_tool, net_tool→fetch_tool,
  vcs_tool→git_tool, todo_tool→tasks_tool
- Add hanzo_tool.py for unified Hanzo API surface
- Add workspace_tool.py for config management
- Add hanzo-tools-ui package with GitHub API integration
- Add unified.py core tool dispatcher
- Update IAM client to RFC-compliant OAuth endpoints
- Fix pytest-asyncio plugin conflict: pin <1.0.0, fix addopts
  double-registration, and align test APIs to current signatures
- Fix integration tests: PermissionManager API, ReviewProtocol
  focus values, AgentTool parameter names, mock llm module for CI
This commit is contained in:
zeekay
2026-03-06 03:13:54 -08:00
parent aff3ec6e8a
commit d5818d2dc9
58 changed files with 2081 additions and 1190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-cli/hanzo_cli/auth.py
+1 -1
View File
@@ -318,7 +318,7 @@ def password_login(
# Use ROPC grant to get tokens directly
resp = httpx.post(
f"{server_url}/api/login/oauth/access_token",
f"{server_url}/oauth/token",
data={
"grant_type": "password",
"client_id": client_id,
pkg/hanzo-cli/hanzo_cli/bot/commands.py
+2 -2
View File
@@ -683,7 +683,7 @@ def _bot_password_login(
password = click.prompt("Password", hide_input=True)
resp = httpx.post(
f"{BOT_IAM_SERVER_URL}/api/login/oauth/access_token",
f"{BOT_IAM_SERVER_URL}/oauth/token",
data={
"grant_type": "password",
"client_id": BOT_IAM_CLIENT_ID,
@@ -743,7 +743,7 @@ def _get_iam_token() -> str:
import httpx
resp = httpx.post(
f"{BOT_IAM_SERVER_URL}/api/login/oauth/access_token",
f"{BOT_IAM_SERVER_URL}/oauth/token",
data={
"grant_type": "client_credentials",
"client_id": client_id,
pkg/hanzo-iam/hanzo_iam/async_client.py
+5 -5
View File
@@ -201,7 +201,7 @@ class AsyncIAMClient:
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}/login/oauth/authorize?{urlencode(params)}"
return f"{base_url}/oauth/authorize?{urlencode(params)}"
async def exchange_code(
self,
@@ -232,7 +232,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/access_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -257,7 +257,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/refresh_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -286,7 +286,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/access_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -393,7 +393,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/introspect",
"/oauth/introspect",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
pkg/hanzo-iam/hanzo_iam/client.py
+10 -10
View File
@@ -218,7 +218,7 @@ class IAMClient:
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}/login/oauth/authorize?{urlencode(params)}"
return f"{base_url}/oauth/authorize?{urlencode(params)}"
def exchange_code(
self,
@@ -248,7 +248,7 @@ class IAMClient:
data["code_verifier"] = code_verifier
response = self.http.post(
"/api/login/oauth/access_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -272,7 +272,7 @@ class IAMClient:
}
response = self.http.post(
"/api/login/oauth/refresh_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -300,7 +300,7 @@ class IAMClient:
}
response = self.http.post(
"/api/login/oauth/access_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -407,7 +407,7 @@ class IAMClient:
}
response = self.http.post(
"/api/login/oauth/introspect",
"/oauth/introspect",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -945,7 +945,7 @@ class AsyncIAMClient:
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}/login/oauth/authorize?{urlencode(params)}"
return f"{base_url}/oauth/authorize?{urlencode(params)}"
async def exchange_code(
self,
@@ -967,7 +967,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/access_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -985,7 +985,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/refresh_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -1003,7 +1003,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/access_token",
"/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
@@ -1050,7 +1050,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/api/login/oauth/introspect",
"/oauth/introspect",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
pkg/hanzo-iam/hanzo_iam/config.py
+2 -2
View File
@@ -75,12 +75,12 @@ class IAMConfig(BaseModel):
@property
def token_endpoint(self) -> str:
"""OAuth2 token endpoint URL."""
return f"{self.server_url}/api/login/oauth/access_token"
return f"{self.server_url}/oauth/token"
@property
def authorize_endpoint(self) -> str:
"""OAuth2 authorization endpoint URL."""
return f"{self.server_url}/login/oauth/authorize"
return f"{self.server_url}/oauth/authorize"
@property
def userinfo_endpoint(self) -> str:
pkg/hanzo-mcp/hanzo_mcp/analytics/insights_analytics.py
+5 -5
View File
@@ -18,9 +18,9 @@ from datetime import datetime
from importlib.metadata import PackageNotFoundError, version
from typing import Any, Callable, Dict, Optional, TypeVar
# Try to import PostHog client (used as backend), but make it optional
# Try to import Insights client (used as backend), but make it optional
try:
from posthog import Posthog
from insights import Posthog
INSIGHTS_AVAILABLE = True
except ImportError:
@@ -36,7 +36,7 @@ class AnalyticsConfig:
"""Configuration for analytics."""
api_key: Optional[str] = None
host: str = "https://us.i.posthog.com"
host: str = "https://us.i.insights.com"
enabled: bool = True
debug: bool = False
capture_errors: bool = True
@@ -52,10 +52,10 @@ class InsightsAnalytics:
self.config = config or AnalyticsConfig()
self._client = None
# Load from environment if not provided (INSIGHTS_API_KEY preferred, POSTHOG_API_KEY fallback)
# Load from environment if not provided (INSIGHTS_API_KEY preferred, INSIGHTS_API_KEY fallback)
if not self.config.api_key:
self.config.api_key = os.environ.get("INSIGHTS_API_KEY") or os.environ.get(
"POSTHOG_API_KEY"
"INSIGHTS_API_KEY"
)
if not self.config.distinct_id:
pkg/hanzo-mcp/hanzo_mcp/prompts/tool_explorer.py
+4 -4
View File
@@ -13,7 +13,7 @@ Tools for delegating tasks to specialized AI agents:
- **claude_cli**: Interact with Claude via CLI
- **critic**: Critical analysis and review
- **review**: Code review and feedback
- **iching**: Decision-making guidance using I Ching
- **zen**: Decision-making guidance using Hanzo Zen
### 📁 Filesystem Tools
Tools for file and directory operations:
@@ -350,11 +350,11 @@ review(
)
```
#### I Ching Guidance
#### Zen Guidance
```python
# Get decision-making guidance
iching(
question="Should we refactor the authentication system now or after the release?"
zen(
challenge="Should we refactor the authentication system now or after the release?"
)
```
pkg/hanzo-mcp/hanzo_mcp/tools/__init__.py
+26 -26
View File
@@ -118,8 +118,8 @@ def register_all_tools(
for tool in PACKAGE_TOOL_PREFIXES.get("memory", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# Todo tools
resolved_enabled_tools["todo"] = is_tool_enabled("todo", True)
# Tasks tools
resolved_enabled_tools["tasks"] = is_tool_enabled("tasks", True)
# Reasoning tools
resolved_enabled_tools["think"] = is_tool_enabled("think", True)
@@ -137,7 +137,7 @@ def register_all_tools(
# Agent tools
resolved_enabled_tools["agent"] = enable_agent_tool or is_tool_enabled(
"agent", False
"agent", True
)
resolved_enabled_tools["swarm"] = is_tool_enabled("swarm", False)
for tool in ["claude", "codex", "gemini", "grok", "code_auth"]:
@@ -159,33 +159,27 @@ def register_all_tools(
# Config tools
resolved_enabled_tools["config"] = is_tool_enabled("config", True)
resolved_enabled_tools["mode"] = is_tool_enabled("mode", True)
resolved_enabled_tools["workspace"] = is_tool_enabled("workspace", True)
# MCP tools
resolved_enabled_tools["mcp"] = is_tool_enabled("mcp", True)
# Billing tools
for tool in PACKAGE_TOOL_PREFIXES.get("billing", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# Commerce tools
for tool in PACKAGE_TOOL_PREFIXES.get("commerce", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# IAM tools
for tool in PACKAGE_TOOL_PREFIXES.get("iam_tools", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# Ingress tools
for tool in PACKAGE_TOOL_PREFIXES.get("ingress", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# MPC tools
for tool in PACKAGE_TOOL_PREFIXES.get("mpc", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# Team tools
for tool in PACKAGE_TOOL_PREFIXES.get("team", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# Unified Hanzo platform surface.
# No backwards compatibility: expose only `hanzo` and hide legacy per-service tools.
resolved_enabled_tools["hanzo"] = is_tool_enabled("hanzo", True)
for legacy_tool in [
"api",
"auth",
"billing",
"commerce",
"iam",
"ingress",
"kms",
"mpc",
"paas",
"team",
]:
resolved_enabled_tools[legacy_tool] = False
# Jupyter tools
for tool in PACKAGE_TOOL_PREFIXES.get("jupyter", []):
@@ -195,6 +189,9 @@ def register_all_tools(
for tool in PACKAGE_TOOL_PREFIXES.get("computer", []):
resolved_enabled_tools[tool] = is_tool_enabled(tool, True)
# UI component registry tool
resolved_enabled_tools["ui"] = is_tool_enabled("ui", True)
# Create loader and discover packages
loader = EntryPointToolLoader(permission_manager=permission_manager)
discovered = loader.discover_packages()
@@ -208,6 +205,9 @@ def register_all_tools(
loaded = loader.load_all(
mcp_server,
enabled_tools=resolved_enabled_tools,
# Database package tools currently require explicit db_manager wiring.
# Keep disabled by default unless explicitly enabled.
enabled_packages={"database": is_tool_enabled("database", False)},
)
all_tools.update(loaded)
logger.info(f"Loaded {len(loaded)} tools from entry points")
pkg/hanzo-mcp/hanzo_mcp/tools/common/entrypoint_loader.py
+11 -14
View File
@@ -33,29 +33,30 @@ PACKAGE_TOOL_PREFIXES: dict[str, list[str]] = {
"filesystem": ["fs"], # Bytes + Paths axis
"core": ["id"], # Identity axis (hash, uri, ref, verify)
"code": ["code"], # Symbols + Structure axis (parse, transform, summarize)
"shell": ["proc"], # Execution axis
"vcs": ["vcs"], # History + Diffs axis
"shell": ["exec"], # Execution axis
"vcs": ["git"], # History + Diffs axis
"test": ["test"], # Validation axis (check, build, test)
"net": ["net"], # Network axis (search, fetch, download, crawl)
"net": ["fetch"], # Network axis (search, fetch, download, crawl)
"plan": ["plan"], # Orchestration axis (intent, route, compose)
# Control surfaces
"browser": ["browser"], # Web content control (Playwright/extension)
"computer": ["computer"], # OS/desktop control (computer use)
"ui": ["ui"], # UI component registry (browse, search, install)
# Extended operators
"lsp": ["lsp"], # Semantic stream (diagnostics, code_actions)
"memory": ["memory"], # Knowledge persistence
"todo": ["todo"], # Task tracking
"todo": ["tasks"], # Task tracking
"reasoning": ["think", "critic"],
"refactor": ["refactor"],
"database": ["sql", "graph"],
"agent": ["agent", "iching", "review"],
"agent": ["agent", "zen", "review"],
"jupyter": ["jupyter"],
"editor": ["neovim_edit", "neovim_command", "neovim_session"],
"llm": ["llm", "consensus"],
"vector": ["index", "vector_index", "vector_search"],
"config": ["config", "mode"],
"config": ["config", "mode", "workspace"],
"mcp_tools": ["mcp"],
"api": ["api"],
"api": ["api", "hanzo"],
"auth": ["auth"],
"kms": ["kms"],
"paas": ["paas"],
@@ -250,16 +251,12 @@ class EntryPointToolLoader:
logger.debug(f"Skipping disabled package: {package_name}")
continue
# Build package-specific enabled_tools
package_tool_names = PACKAGE_TOOL_PREFIXES.get(package_name, [])
package_enabled = {
name: enabled_tools.get(name, True) for name in package_tool_names
}
self.load_package(
package_name,
mcp_server,
enabled_tools=package_enabled,
# Apply global resolved tool state directly. Package-prefix mapping is
# advisory for UX/grouping but not authoritative for runtime gating.
enabled_tools=enabled_tools,
**kwargs,
)
pkg/hanzo-mcp/hanzo_mcp/tools/common/persona_adapter.py
+9 -9
View File
@@ -33,20 +33,20 @@ CATEGORY_TOOL_MAPPINGS = {
"scientist": ESSENTIAL_TOOLS + ["jupyter", "critic"] + AI_TOOLS + VECTOR_TOOLS,
"philosopher": ESSENTIAL_TOOLS + ["critic", "think", "search"],
"artist": ESSENTIAL_TOOLS + ["watch", "jupyter"] + AI_TOOLS,
"leader": ESSENTIAL_TOOLS + ["todo", "rules", "agent"] + AI_TOOLS,
"writer": ESSENTIAL_TOOLS + ["search", "critic", "todo"],
"leader": ESSENTIAL_TOOLS + ["tasks", "rules", "agent"] + AI_TOOLS,
"writer": ESSENTIAL_TOOLS + ["search", "critic", "tasks"],
"mathematician": ESSENTIAL_TOOLS + ["jupyter", "symbols", "critic"],
"musician": ESSENTIAL_TOOLS + ["watch", "todo"],
"athlete": ESSENTIAL_TOOLS + ["todo", "watch", "process"],
"musician": ESSENTIAL_TOOLS + ["watch", "tasks"],
"athlete": ESSENTIAL_TOOLS + ["tasks", "watch", "process"],
"entrepreneur": ESSENTIAL_TOOLS
+ ["todo", "agent", "consensus"]
+ ["tasks", "agent", "consensus"]
+ BUILD_TOOLS
+ DATABASE_TOOLS,
"activist": ESSENTIAL_TOOLS + ["search", "todo", "rules"],
"activist": ESSENTIAL_TOOLS + ["search", "tasks", "rules"],
"religious_leader": ESSENTIAL_TOOLS + ["search", "critic", "think"],
"military_leader": ESSENTIAL_TOOLS + ["todo", "process", "critic"] + UNIX_TOOLS,
"explorer": ESSENTIAL_TOOLS + ["search", "watch", "todo"],
"comedian": ESSENTIAL_TOOLS + ["todo", "watch", "critic"],
"military_leader": ESSENTIAL_TOOLS + ["tasks", "process", "critic"] + UNIX_TOOLS,
"explorer": ESSENTIAL_TOOLS + ["search", "watch", "tasks"],
"comedian": ESSENTIAL_TOOLS + ["tasks", "watch", "critic"],
"default": ESSENTIAL_TOOLS + AI_TOOLS,
}
pkg/hanzo-mcp/hanzo_mcp/tools/common/personality.py
+4 -4
View File
@@ -184,7 +184,7 @@ def _register_builtin_personalities() -> None:
ESSENTIAL_TOOLS
+ [
"agent",
"todo",
"tasks",
"browser",
"computer",
"search",
@@ -193,7 +193,7 @@ def _register_builtin_personalities() -> None:
"jupyter",
"refactor",
"lsp",
"iching",
"zen",
"review", # Agent sub-tools
]
+ BUILD_TOOLS
@@ -224,7 +224,7 @@ def _register_builtin_personalities() -> None:
+ UNIX_TOOLS
+ VECTOR_TOOLS
+ [
"todo",
"tasks",
"rules",
"browser",
"jupyter",
@@ -243,7 +243,7 @@ def _register_builtin_personalities() -> None:
description="Automate everything",
philosophy="You build it, you run it.",
tools=list(
set(ESSENTIAL_TOOLS + BUILD_TOOLS + UNIX_TOOLS + ["todo", "browser"])
set(ESSENTIAL_TOOLS + BUILD_TOOLS + UNIX_TOOLS + ["tasks", "browser"])
),
environment={"CI_CD": "enabled"},
cli_tools=[
pkg/hanzo-mcp/hanzo_mcp/tools/common/tool_list.py
+2 -1
View File
@@ -93,9 +93,10 @@ class ToolListTool(BaseTool):
("tool_list", "List all tools (this tool)"),
],
"productivity": [
("todo", "Todo management (list/add/update/remove/clear)"),
("tasks", "Task management (list/add/update/remove/clear)"),
("jupyter", "Jupyter notebooks (read/edit/create/delete/execute)"),
("think", "Structured thinking space"),
("ui", "UI component registry (browse/search/install components)"),
],
"system": [
("stats", "System and resource statistics"),
pkg/hanzo-mcp/pyproject.toml
+4 -4
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-mcp"
version = "0.12.4"
version = "0.12.5"
description = "The Zen of Hanzo MCP: One server to rule them all. The ultimate MCP that orchestrates all others."
readme = "README.md"
requires-python = ">=3.12"
@@ -30,7 +30,7 @@ dependencies = [
"hanzo-tools-fs>=0.1.0",
"hanzo-tools-shell>=0.6.1",
"hanzo-tools-browser[playwright]>=0.4.2", # Extension-first routing, Playwright fallback
"hanzo-tools-memory>=0.2.2",
"hanzo-tools-memory>=0.2.0",
"hanzo-tools-todo>=0.1.0",
"hanzo-tools-reasoning>=0.1.0",
"hanzo-tools-lsp>=0.1.0",
@@ -112,7 +112,7 @@ test = [
"pytest>=7.0.0",
"pytest-cov>=4.1.0",
"pytest-mock>=3.10.0",
"pytest-asyncio>=0.26.0",
"pytest-asyncio>=0.26.0,<1.0.0",
]
performance = ["ujson>=5.7.0", "orjson>=3.9.0"] # uvloop now standard
publish = ["twine>=4.0.2", "build>=1.0.3"]
@@ -136,7 +136,7 @@ include = ["hanzo_mcp"]
[tool.pytest.ini_options]
# Configuration for pytest
addopts = "--no-header --no-summary -p pytest_asyncio.plugin"
addopts = "--no-header --no-summary -p asyncio"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
# Note: --cov-fail-under=96.3 will be added after achieving target coverage
pkg/hanzo-mcp/tests/test_agent_tools_ci.py
+10 -14
View File
@@ -5,7 +5,7 @@ from unittest.mock import Mock
import pytest
from hanzo_tools.agent import TOOLS, register_tools
from hanzo_tools.agent.agent_tool import AgentTool
from hanzo_tools.agent.iching_tool import IChingTool
from hanzo_tools.agent.zen_tool import ZenTool
from hanzo_tools.agent.review_tool import ReviewTool
@@ -24,7 +24,7 @@ class TestAgentTools:
"""Test TOOLS exports the correct tools."""
tool_classes = [t.__name__ for t in TOOLS]
assert "AgentTool" in tool_classes
assert "IChingTool" in tool_classes
assert "ZenTool" in tool_classes
assert "ReviewTool" in tool_classes
assert len(TOOLS) == 3
@@ -35,16 +35,12 @@ class TestAgentTools:
assert tool.description is not None
assert len(tool.description) > 0
def test_iching_tool_creation(self):
"""Test IChingTool can be created."""
tool = IChingTool()
assert tool.name == "iching"
def test_zen_tool_creation(self):
"""Test ZenTool can be created."""
tool = ZenTool()
assert tool.name == "zen"
desc_lower = tool.description.lower()
assert (
"i ching" in desc_lower
or "wisdom" in desc_lower
or "hexagram" in desc_lower
)
assert "zen" in desc_lower or "64" in desc_lower or "guidance" in desc_lower
def test_review_tool_creation(self):
"""Test ReviewTool can be created."""
@@ -62,17 +58,17 @@ class TestAgentTools:
# Check tool names
tool_names = [t.name for t in tools]
assert "agent" in tool_names
assert "iching" in tool_names
assert "zen" in tool_names
assert "review" in tool_names
def test_tool_naming_consistency(self):
"""Ensure tool naming is consistent."""
agent = AgentTool()
iching = IChingTool()
zen = ZenTool()
review = ReviewTool()
assert agent.name == "agent"
assert iching.name == "iching"
assert zen.name == "zen"
assert review.name == "review"
pkg/hanzo-mcp/tests/test_integration/test_all_agent_tools.py
+27 -27
View File
@@ -1,12 +1,18 @@
"""Integration test for all agent tools working together."""
import json
import sys
from types import ModuleType
from unittest.mock import Mock, patch
import pytest
from hanzo_tools.agent.agent_tool import AgentTool
from hanzo_mcp.tools.common.permissions import PermissionManager
# Ensure 'llm' module is available for mocking even if not installed
if "llm" not in sys.modules:
_mock_llm = ModuleType("llm")
_mock_llm.completion = Mock()
sys.modules["llm"] = _mock_llm
@pytest.fixture
@@ -35,16 +41,11 @@ def test_project(tmp_path):
@pytest.mark.asyncio
@pytest.mark.skip(reason="Requires CLI agent backends (claude/codex/grok) installed")
async def test_agent_uses_all_tools(test_project, mock_context):
"""Test that an agent can use clarification, critic, review, and iching tools."""
permission_manager = PermissionManager(allowed_paths=[str(test_project)])
"""Test that an agent can use clarification, critic, review, and zen tools."""
# Create an agent tool
agent = AgentTool(
permission_manager=permission_manager,
model="test-model",
max_iterations=10, # Allow more iterations for all tools
)
agent = AgentTool()
# Mock the llm completion to simulate agent using all tools
with patch("llm.completion") as mock_completion:
@@ -75,7 +76,7 @@ async def test_agent_uses_all_tools(test_project, mock_context):
)
]
),
# 2. Use I Ching for creative guidance
# 2. Use zen guidance for creative direction
Mock(
choices=[
Mock(
@@ -84,7 +85,7 @@ async def test_agent_uses_all_tools(test_project, mock_context):
tool_calls=[
Mock(
function=Mock(
name="iching",
name="zen",
arguments=json.dumps(
{
"challenge": "How should I approach implementing this complex pathfinding algorithm?"
@@ -247,7 +248,7 @@ async def test_agent_uses_all_tools(test_project, mock_context):
message=Mock(
content="""I've implemented the pathfinding algorithm with guidance from multiple perspectives:
1. **I Ching Guidance**: Provided creative insights about balancing simplicity and power
1. **Zen Guidance**: Provided creative insights about balancing simplicity and power
2. **Clarification**: Confirmed we should optimize for lowest cost
3. **Implementation**: Created a hybrid Dijkstra/Bellman-Ford approach
4. **Balanced Review**: Highlighted good structure and suggested edge case handling
@@ -264,11 +265,11 @@ The final implementation addresses all concerns raised.""",
# Execute the agent
result = await agent.call(
mock_context,
prompts=f"Implement the complex pathfinding algorithm in {test_project / 'complex_algorithm.py'} using all available tools for guidance",
prompt=f"Implement the complex pathfinding algorithm in {test_project / 'complex_algorithm.py'} using all available tools for guidance",
)
# Verify the result mentions all tools
assert "i ching" in result.lower()
assert "zen" in result.lower()
assert "clarification" in result.lower()
assert "review" in result.lower()
assert "critic" in result.lower()
@@ -278,11 +279,11 @@ The final implementation addresses all concerns raised.""",
@pytest.mark.asyncio
async def test_iching_tool_directly():
"""Test the I Ching tool directly."""
from hanzo_tools.agent.iching_tool import IChingTool
async def test_zen_tool_directly():
"""Test the zen tool directly."""
from hanzo_tools.agent.zen_tool import ZenTool
tool = IChingTool()
tool = ZenTool()
ctx = Mock()
# Test with different challenges
@@ -299,11 +300,11 @@ async def test_iching_tool_directly():
result = result["output"]
# Verify result structure
tool_helper.assert_in_result("I CHING GUIDANCE", result)
tool_helper.assert_in_result("Hexagram Cast", result)
tool_helper.assert_in_result("Hanzo Principles", result)
tool_helper.assert_in_result("Synthesized Approach", result)
tool_helper.assert_in_result("The Way Forward", result)
assert "ZEN GUIDANCE" in result
assert "Hexagram Cast" in result
assert "Hanzo Principles" in result
assert "Synthesized Approach" in result
assert "The Way Forward" in result
# Verify it selected relevant principles
if "scale" in challenge.lower():
@@ -341,9 +342,9 @@ async def test_review_vs_critic_difference():
code_snippets=code_snippet,
)
# Get review feedback (balanced)
# Get review feedback (balanced) - use GENERAL since ReviewFocus has no SECURITY
review_feedback = reviewer.request_review(
focus="SECURITY", work_description=work_description, code_snippets=code_snippet
focus="GENERAL", work_description=work_description, code_snippets=code_snippet
)
# Critic should be harsher
@@ -354,9 +355,8 @@ async def test_review_vs_critic_difference():
assert "Positive Aspects" in review_feedback or "✓" in review_feedback
assert "Suggestions" in review_feedback
# Both should mention security
# Critic should mention security (it has SECURITY review type)
assert "security" in critic_feedback.lower()
assert "security" in review_feedback.lower()
@pytest.mark.asyncio
pkg/hanzo-mcp/uv.lock
Generated
+114 -19
View File
@@ -1290,7 +1290,7 @@ wheels = [
[[package]]
name = "hanzo-mcp"
version = "0.11.8"
version = "0.12.5"
source = { editable = "." }
dependencies = [
{ name = "fastmcp" },
@@ -1300,18 +1300,24 @@ dependencies = [
{ name = "hanzo-tools-agent" },
{ name = "hanzo-tools-api" },
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-billing" },
{ name = "hanzo-tools-browser", extra = ["playwright"] },
{ name = "hanzo-tools-commerce" },
{ name = "hanzo-tools-computer" },
{ name = "hanzo-tools-config" },
{ name = "hanzo-tools-fs" },
{ name = "hanzo-tools-iam" },
{ name = "hanzo-tools-ingress" },
{ name = "hanzo-tools-kms" },
{ name = "hanzo-tools-llm" },
{ name = "hanzo-tools-lsp" },
{ name = "hanzo-tools-memory" },
{ name = "hanzo-tools-mpc" },
{ name = "hanzo-tools-paas" },
{ name = "hanzo-tools-reasoning" },
{ name = "hanzo-tools-refactor" },
{ name = "hanzo-tools-shell" },
{ name = "hanzo-tools-team" },
{ name = "hanzo-tools-todo" },
{ name = "mcp" },
{ name = "pydantic" },
@@ -1387,22 +1393,28 @@ requires-dist = [
{ name = "hanzo-tools-agent", specifier = ">=0.3.1" },
{ name = "hanzo-tools-api", specifier = ">=0.3.1" },
{ name = "hanzo-tools-auth", specifier = ">=0.1.0" },
{ name = "hanzo-tools-billing", specifier = ">=0.1.0" },
{ name = "hanzo-tools-browser", marker = "extra == 'interactive'", specifier = ">=0.2.1" },
{ name = "hanzo-tools-browser", extras = ["playwright"], specifier = ">=0.3.0" },
{ name = "hanzo-tools-browser", extras = ["playwright"], specifier = ">=0.4.2" },
{ name = "hanzo-tools-commerce", specifier = ">=0.1.0" },
{ name = "hanzo-tools-computer", specifier = ">=0.1.0" },
{ name = "hanzo-tools-config", specifier = ">=0.1.0" },
{ name = "hanzo-tools-fs", specifier = ">=0.1.0" },
{ name = "hanzo-tools-iam", specifier = ">=0.1.0" },
{ name = "hanzo-tools-ide", marker = "extra == 'ide'", specifier = ">=0.1.0" },
{ name = "hanzo-tools-ingress", specifier = ">=0.1.0" },
{ name = "hanzo-tools-kms", specifier = ">=0.1.0" },
{ name = "hanzo-tools-llm", specifier = ">=0.1.0" },
{ name = "hanzo-tools-lsp", specifier = ">=0.1.0" },
{ name = "hanzo-tools-memory", specifier = ">=0.1.0" },
{ name = "hanzo-tools-memory", extras = ["full"], marker = "extra == 'memory'", specifier = ">=0.2.0" },
{ name = "hanzo-tools-memory", specifier = ">=0.2.0" },
{ name = "hanzo-tools-memory", extras = ["full"], marker = "extra == 'memory'", specifier = ">=0.2.2" },
{ name = "hanzo-tools-mpc", specifier = ">=0.1.0" },
{ name = "hanzo-tools-paas", specifier = ">=0.1.0" },
{ name = "hanzo-tools-reasoning", specifier = ">=0.1.0" },
{ name = "hanzo-tools-refactor", specifier = ">=0.1.0" },
{ name = "hanzo-tools-repl", marker = "extra == 'repl'", specifier = ">=0.1.0" },
{ name = "hanzo-tools-shell", specifier = ">=0.6.1" },
{ name = "hanzo-tools-team", specifier = ">=0.1.0" },
{ name = "hanzo-tools-todo", specifier = ">=0.1.0" },
{ name = "ipykernel", marker = "extra == 'repl'", specifier = ">=6.29.0" },
{ name = "jupyter-client", marker = "extra == 'repl'", specifier = ">=8.6.0" },
@@ -1414,7 +1426,7 @@ requires-dist = [
{ name = "pydantic-settings", specifier = ">=2.12.0" },
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" },
{ name = "pytest", marker = "extra == 'test'", specifier = ">=7.0.0" },
{ name = "pytest-asyncio", marker = "extra == 'test'", specifier = ">=0.26.0" },
{ name = "pytest-asyncio", marker = "extra == 'test'", specifier = ">=0.26.0,<1.0.0" },
{ name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.1.0" },
{ name = "pytest-cov", marker = "extra == 'test'", specifier = ">=4.1.0" },
{ name = "pytest-mock", marker = "extra == 'test'", specifier = ">=3.10.0" },
@@ -1541,9 +1553,23 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/2e/63/6b82a88840210518ce9a3eefa9f96dd6febb9656d49c68c5734a863aed6a/hanzo_tools_auth-0.1.0-py3-none-any.whl", hash = "sha256:38e75848892179dd5e3da184605d0dc7488070833cad21872a012e651a1b9f70", size = 7386, upload-time = "2026-02-25T05:57:51.519Z" },
]
[[package]]
name = "hanzo-tools-billing"
version = "0.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-core" },
{ name = "httpx" },
]
sdist = { url = "https://files.pythonhosted.org/packages/44/09/cff246b0d60fd80957bbe5f0d4b44efc8307677f1c5958e70aab96e9be72/hanzo_tools_billing-0.1.0.tar.gz", hash = "sha256:308bef7e148561ad147bfc0186b7afcc7de089cf6fcb1e44a5a7533f3c93661b", size = 4052, upload-time = "2026-03-02T04:50:34.787Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5f/2e/2db6833ddc12085b9435490165deeb21e3ea844c8759dc38b300d7351eb0/hanzo_tools_billing-0.1.0-py3-none-any.whl", hash = "sha256:43513c2c1408a1a7148914ce4237dadce410eb73f74b2025120e1edfd2ab6c54", size = 4748, upload-time = "2026-03-02T04:50:33.673Z" },
]
[[package]]
name = "hanzo-tools-browser"
version = "0.3.0"
version = "0.4.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -1552,9 +1578,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "websockets" },
]
sdist = { url = "https://files.pythonhosted.org/packages/61/37/1f539537f4c19536bcd36b14ba03ab51cc59e945543611fb9a96933dcbba/hanzo_tools_browser-0.3.0.tar.gz", hash = "sha256:6e907a8a775c80cb450c06975f47348bdcd4b5a776b350e202e271a8a3f8d7fe", size = 26152, upload-time = "2026-02-21T19:58:43.099Z" }
sdist = { url = "https://files.pythonhosted.org/packages/38/04/e61263085bbae5eb880c532b06021bca79b6178640590925167888789fc1/hanzo_tools_browser-0.4.2.tar.gz", hash = "sha256:5806550cb2e43304a4579f5c2e40a55be6eed5c8ed233011965c3762b851f102", size = 29037, upload-time = "2026-03-06T08:55:16.847Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ae/ca/1990211b1161add4000e3caa9481c50d55a9811071467730857c4f560296/hanzo_tools_browser-0.3.0-py3-none-any.whl", hash = "sha256:55d320af5107c9719270c07e411a3f0740f61d266dd5d8773dc6dc5e9aafbc18", size = 26470, upload-time = "2026-02-21T19:58:41.949Z" },
{ url = "https://files.pythonhosted.org/packages/70/5a/ece213056c5eea78e92a7300cabc1237f79c2190a2be7674ca8e543650c8/hanzo_tools_browser-0.4.2-py3-none-any.whl", hash = "sha256:6815312b28d80fd654b31636ede2ece70441a1abdc1154d30de478efc1c480b0", size = 29384, upload-time = "2026-03-06T08:55:15.663Z" },
]
[package.optional-dependencies]
@@ -1562,6 +1588,20 @@ playwright = [
{ name = "playwright" },
]
[[package]]
name = "hanzo-tools-commerce"
version = "0.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-core" },
{ name = "httpx" },
]
sdist = { url = "https://files.pythonhosted.org/packages/3e/da/bd05c7ef454c5cb7a87a2c3e4fef5d3f65b340bf71b1e1afb74dada2408e/hanzo_tools_commerce-0.1.0.tar.gz", hash = "sha256:cf3fbe7577bd3684af28df6b07379283212f21092b86b13b46a5ef0f71b3d652", size = 3618, upload-time = "2026-03-02T04:50:40.128Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/00/80/5af45745f0cb088597618c796cebe81638b432469f2b19e288004c73ada5/hanzo_tools_commerce-0.1.0-py3-none-any.whl", hash = "sha256:8311d4c0fa08c25bf75b1ede42e18c6b93c5b024ac1aa2e39ab053ffbdceca02", size = 4325, upload-time = "2026-03-02T04:50:38.92Z" },
]
[[package]]
name = "hanzo-tools-computer"
version = "0.5.2"
@@ -1627,6 +1667,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/bf/b2/4418ab22f8854992c786b145f87277a0a4ea2798b865af592a2725495348/hanzo_tools_fs-0.3.1-py3-none-any.whl", hash = "sha256:6892bf6ba2559cad90876276b7022bfc45923d86866ac08e83bbe86e0771081f", size = 14541, upload-time = "2026-01-04T01:40:44.447Z" },
]
[[package]]
name = "hanzo-tools-iam"
version = "0.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-core" },
{ name = "httpx" },
]
sdist = { url = "https://files.pythonhosted.org/packages/5d/b3/a0ae35cf9d859341982b2e9b87e63c51f12b6db8d0e69099afd54ebd40d5/hanzo_tools_iam-0.1.0.tar.gz", hash = "sha256:e901af6afcf5cf728fdf7cb099082e901a2870aacfc8b7613f6257a738c470d4", size = 5639, upload-time = "2026-03-02T04:50:44.574Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ac/ca/7fe1e7dd334bd7204007a712aeb380a016464f8f0513e1fbb4e8c535f490/hanzo_tools_iam-0.1.0-py3-none-any.whl", hash = "sha256:94b9c267fe0e80d3b00b93cf9921e071fd6cdf5ed4bafd5cf34e8b7f8eaf7ac7", size = 6287, upload-time = "2026-03-02T04:50:43.63Z" },
]
[[package]]
name = "hanzo-tools-ide"
version = "0.1.0"
@@ -1641,6 +1695,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/88/01/dc6ffba6b34403d8295a7225af55c2d891ca5db16cfb9420e76264145f75/hanzo_tools_ide-0.1.0-py3-none-any.whl", hash = "sha256:8b1a7fd381c6be5a265e1711b67bd007e1bc2ae5dc61e60bcbbc8d8e6a7d4bb7", size = 7984, upload-time = "2026-01-22T23:48:38.153Z" },
]
[[package]]
name = "hanzo-tools-ingress"
version = "0.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-core" },
{ name = "httpx" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d2/94/d942ab541b7a8d029ff1c3171f34827b539a5a8360277ab34bbeee8d6888/hanzo_tools_ingress-0.1.0.tar.gz", hash = "sha256:26ed087f9f5ec10b8159c86b623f052d06b2f6320aada917221a0545bb02bfc5", size = 3737, upload-time = "2026-03-02T04:50:54.997Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/dd/7b/0672340573d51ad915163f74043bfed7357cf9093d4e1927f3cf18f7f7c7/hanzo_tools_ingress-0.1.0-py3-none-any.whl", hash = "sha256:d57647444a6733d0dc2a9d97cb3041a75de6157010e5a37890e5ec4ca99ccdc2", size = 4423, upload-time = "2026-03-02T04:50:53.957Z" },
]
[[package]]
name = "hanzo-tools-kms"
version = "0.1.0"
@@ -1685,16 +1753,16 @@ wheels = [
[[package]]
name = "hanzo-tools-memory"
version = "0.2.0"
version = "0.2.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-core" },
{ name = "hanzo-tools" },
{ name = "mcp" },
{ name = "pydantic" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e9/95/0b78652310c7ae312375c2cd5bb683a101f3a617b301e104592adf6e462f/hanzo_tools_memory-0.2.0.tar.gz", hash = "sha256:5faf7afd9804b6f67c06740ccfd6a41997f40582565b37a6c4b60e10811c914f", size = 8504, upload-time = "2025-12-26T14:46:40.826Z" }
sdist = { url = "https://files.pythonhosted.org/packages/ed/c6/ddd1a702d436b5abc1c6b0aaf46a6f97036c06c2cb951fd31b1d6b65abd0/hanzo_tools_memory-0.2.2.tar.gz", hash = "sha256:4d73ae44c47ba02783588dde96d68c4d463cb299de253b93f948b04ccf5594dd", size = 12005, upload-time = "2026-03-06T10:42:16.408Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/de/c8/cea1e08fab38bb8082889b8e582bb49a8e5e7ee679541d4f60dc43ead354/hanzo_tools_memory-0.2.0-py3-none-any.whl", hash = "sha256:ce1d4c748c4c1326f8c4b019293c140bcc99c3b8d9baa3f6a37c9abe8df7544e", size = 10555, upload-time = "2025-12-26T14:46:36.382Z" },
{ url = "https://files.pythonhosted.org/packages/2a/f6/c4b380ec0b372b0c81820e2d71f303f248b55f37840fe20743de784c2964/hanzo_tools_memory-0.2.2-py3-none-any.whl", hash = "sha256:3b5d3a1cc19461d42d1cfcd12caaa6fa9235dfd79465bb31810dea22280433f2", size = 11415, upload-time = "2026-03-06T10:42:15.368Z" },
]
[package.optional-dependencies]
@@ -1702,6 +1770,20 @@ full = [
{ name = "hanzo-memory" },
]
[[package]]
name = "hanzo-tools-mpc"
version = "0.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-core" },
{ name = "httpx" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b7/89/da067ccc7549fa1f9fd21bd81962b5eb1d67266ab6003a1f7838d9847032/hanzo_tools_mpc-0.1.0.tar.gz", hash = "sha256:96a46d149594bf7ae33cfb89f24248fe2591daa1752b7c147a746e06d9f7ce38", size = 5082, upload-time = "2026-03-02T04:50:59.895Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/01/38/0265de9384a500e224841183d494a706f32ddfc5d9bfd20d1a90a8ca0adc/hanzo_tools_mpc-0.1.0-py3-none-any.whl", hash = "sha256:d16a41502e424fafd66503bee098239116ba4c1a830bec49776033baa6015877", size = 5692, upload-time = "2026-03-02T04:50:58.502Z" },
]
[[package]]
name = "hanzo-tools-paas"
version = "0.1.0"
@@ -1776,6 +1858,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/65/db/fe26bc4cace2900e000644d61cb2a37502312a4e20082d85cde11e419ba0/hanzo_tools_shell-0.6.1-py3-none-any.whl", hash = "sha256:d0a7eef3385744dba2e5dd0a268a44884a3276ef2ea4d8c96cc06e43c226e866", size = 46161, upload-time = "2026-01-12T03:21:55.028Z" },
]
[[package]]
name = "hanzo-tools-team"
version = "0.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hanzo-tools-auth" },
{ name = "hanzo-tools-core" },
{ name = "httpx" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b1/a4/ec214fd59735abc9d84303300b2548c8e841a098013efde0a3b27c7fe3e2/hanzo_tools_team-0.1.0.tar.gz", hash = "sha256:ebec2d8fd13d95b8e820d025e832460bad233f0212356e34611e8fe0a6aa27c6", size = 3952, upload-time = "2026-03-02T04:50:50.169Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c3/66/eb60ebb86ab73e03feec61548a0c4be95c21f95547d7b9c20817dface691/hanzo_tools_team-0.1.0-py3-none-any.whl", hash = "sha256:1ac1c76b3c0a06bcc937171cf3f7cf5c149bc936ce47f7ee4f1e09d755c6fd58", size = 4583, upload-time = "2026-03-02T04:50:48.513Z" },
]
[[package]]
name = "hanzo-tools-todo"
version = "0.2.0"
@@ -4295,7 +4391,7 @@ sdist = { url = "https://files.pythonhosted.org/packages/ee/f0/cb456ac4f1a73723d
[[package]]
name = "pytest"
version = "9.0.2"
version = "8.4.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
@@ -4304,22 +4400,21 @@ dependencies = [
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" }
sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618, upload-time = "2025-09-04T14:34:22.711Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" },
{ url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750, upload-time = "2025-09-04T14:34:20.226Z" },
]
[[package]]
name = "pytest-asyncio"
version = "1.3.0"
version = "0.26.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" }
sdist = { url = "https://files.pythonhosted.org/packages/8e/c4/453c52c659521066969523e87d85d54139bbd17b78f09532fb8eb8cdb58e/pytest_asyncio-0.26.0.tar.gz", hash = "sha256:c4df2a697648241ff39e7f0e4a73050b03f123f760673956cf0d72a4990e312f", size = 54156, upload-time = "2025-03-25T06:22:28.883Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
{ url = "https://files.pythonhosted.org/packages/20/7f/338843f449ace853647ace35870874f69a764d251872ed1b4de9f234822c/pytest_asyncio-0.26.0-py3-none-any.whl", hash = "sha256:7b51ed894f4fbea1340262bdae5135797ebbe21d8638978e35d31c6d19f72fb0", size = 19694, upload-time = "2025-03-25T06:22:27.807Z" },
]
[[package]]
pkg/hanzo-tools-agent/README.md
+2 -2
View File
@@ -56,9 +56,9 @@ Configure agents for direct API calls without CLI:
}
```
### iching - I Ching Wisdom
### zen - Hanzo Zen Guidance
```python
iching(challenge="How should I approach this refactoring?")
zen(challenge="How should I approach this refactoring?")
```
### review - Code Review
pkg/hanzo-tools-agent/hanzo_tools/agent/__init__.py
+4 -4
View File
@@ -2,7 +2,7 @@
Provides consolidated agent tools:
- agent: Multi-agent orchestration (run, dag, swarm, consensus, dispatch)
- iching: I Ching wisdom for decisions
- zen: Hanzo Zen guidance for decisions
- review: Code review tool
Install:
@@ -26,7 +26,7 @@ from .agent_tool import AgentTool
# Legacy imports for backwards compatibility
from .critic_tool import CriticTool # Legacy - use reasoning.critic instead
from .iching_tool import IChingTool
from .zen_tool import ZenTool
from .review_tool import ReviewTool
from .grok_cli_tool import GrokCLITool
from .cli_agent_base import CLIAgentBase
@@ -37,7 +37,7 @@ from .gemini_cli_tool import GeminiCLITool
# Export list for tool discovery
TOOLS = [
AgentTool, # Multi-agent orchestration
IChingTool, # I Ching wisdom
ZenTool, # Hanzo Zen guidance
ReviewTool, # Code review
]
@@ -46,7 +46,7 @@ __all__ = [
"TOOLS",
# Primary tools
"AgentTool",
"IChingTool",
"ZenTool",
"ReviewTool",
# Legacy (for backwards compatibility)
"CriticTool",
pkg/hanzo-tools-agent/hanzo_tools/agent/prompt.py
+2 -2
View File
@@ -81,8 +81,8 @@ BALANCED REVIEW:
- Use for regular code review and feedback
CREATIVE GUIDANCE:
- Use the iching tool when you need creative problem-solving approaches
- Combines ancient I Ching wisdom with Hanzo engineering principles
- Use the zen tool when you need creative problem-solving approaches
- Rolls one of 64 Hanzo Zen philosophies with engineering principles
- Provides unique perspectives and actionable guidance
- Use when stuck, need fresh ideas, or want philosophical alignment
pkg/hanzo-tools-agent/hanzo_tools/agent/iching_tool.py → pkg/hanzo-tools-agent/hanzo_tools/agent/zen_tool.py
+18 -18
View File
@@ -1,4 +1,4 @@
"""I Ching tool for creative problem solving using Hanzo principles."""
"""Zen guidance tool for creative problem solving using Hanzo principles."""
import random
from enum import Enum
@@ -102,7 +102,7 @@ class HanzoPrinciple(Enum):
class Hexagram:
"""I Ching hexagram with interpretation."""
"""64-path oracle hexagram with interpretation."""
HEXAGRAMS = {
"111111": (
@@ -405,7 +405,7 @@ class Hexagram:
class IChing:
"""I Ching oracle for engineering guidance."""
"""Zen oracle for engineering guidance."""
def __init__(self):
self.principles = list(HanzoPrinciple)
@@ -476,8 +476,8 @@ class IChing:
def generate_guidance(
self, hexagram: Hexagram, principles: List[HanzoPrinciple], challenge: str
) -> str:
"""Generate creative guidance combining I Ching wisdom and Hanzo principles."""
guidance = f"☯️ I CHING GUIDANCE FOR ENGINEERING CHALLENGE ☯️\n\n"
"""Generate creative guidance combining Hanzo Zen and engineering principles."""
guidance = f"☯️ ZEN GUIDANCE FOR ENGINEERING CHALLENGE ☯️\n\n"
guidance += f"**Your Challenge:** {challenge}\n\n"
guidance += f"**Hexagram Cast:** {hexagram.name} - {hexagram.title}\n"
@@ -542,7 +542,7 @@ class IChing:
guidance += "\n**The Way Forward:**\n"
guidance += self._synthesize_action_plan(hexagram, principles, challenge)
guidance += "\n\n*Remember: The I Ching reveals patterns, not prescriptions. "
guidance += "\n\n*Remember: Zen guidance reveals patterns, not prescriptions. "
guidance += "Let this wisdom guide your intuition as you craft your solution.*"
return guidance
@@ -584,31 +584,31 @@ class IChing:
return plan
class IChingTool(BaseTool):
"""Tool for applying I Ching wisdom to engineering challenges."""
class ZenTool(BaseTool):
"""Tool for applying Hanzo Zen guidance to engineering challenges."""
name = "iching"
name = "zen"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Apply I Ching wisdom and Hanzo principles to engineering challenges.
return """Apply Hanzo Zen guidance to engineering challenges.
This tool casts an I Ching hexagram and selects relevant Hanzo principles
to provide creative guidance for your engineering challenge.
This tool rolls a 64-path Zen oracle and selects relevant Hanzo principles
to provide creative, actionable guidance for your engineering challenge.
Parameters:
- challenge: Description of the engineering challenge or question
The oracle will:
1. Cast a hexagram representing the current situation
1. Roll one of 64 Zen philosophies for the current situation
2. Select relevant Hanzo principles
3. Synthesize actionable guidance
4. Provide specific recommendations
Example:
iching(
zen(
challenge="How should I approach refactoring this legacy codebase?"
)
@@ -619,13 +619,13 @@ Use this when you need:
- Alignment with Hanzo principles"""
def __init__(self):
"""Initialize the I Ching tool."""
"""Initialize the Zen tool."""
super().__init__()
self.oracle = IChing()
@auto_timeout("iching")
@auto_timeout("zen")
async def call(self, ctx: MCPContext, challenge: str) -> str:
"""Cast I Ching and provide guidance."""
"""Roll zen oracle and provide guidance."""
# Cast hexagram
hexagram = self.oracle.cast_hexagram()
@@ -642,5 +642,5 @@ Use this when you need:
tool_self = self
@server.tool(name=self.name, description=self.description)
async def iching(ctx: MCPContext, challenge: str) -> str:
async def zen(ctx: MCPContext, challenge: str) -> str:
return await tool_self.call(ctx, challenge)
pkg/hanzo-tools-api/README.md
+24 -1
View File
@@ -1,6 +1,8 @@
# hanzo-tools-api
Generic API tool for calling any REST API via OpenAPI specs.
Generic API tools for Hanzo MCP:
- `api`: OpenAPI-first generic REST tool
- `hanzo`: Unified Hanzo platform router (`auth`, `billing`, `commerce`, `iam`, `ingress`, `kms`, `mpc`, `paas`, `team`, `api`)
## Features
@@ -69,6 +71,27 @@ await tool.call(ctx,
)
```
### Unified Hanzo Platform Tool
```python
from hanzo_tools.api import HanzoTool
tool = HanzoTool()
# Discover available services
await tool.call(ctx, service="services")
# Route to service tools through one MCP surface
await tool.call(ctx, service="auth", action="status")
await tool.call(ctx, service="commerce", action="orders")
await tool.call(
ctx,
service="iam",
action="enforce",
args='{"owner":"hanzo","model":"rbac","resource":"/api/users","permission_action":"read"}',
)
```
### MCP Tool Usage
When registered with hanzo-mcp, use like:
pkg/hanzo-tools-api/hanzo_tools/api/__init__.py
+3 -1
View File
@@ -36,6 +36,7 @@ MCP Tool Usage:
# Models (structured data types)
# MCP Tool
from .api_tool import APITool
from .hanzo_tool import HanzoTool
# Main client (typed interface)
from .client import (
@@ -107,7 +108,7 @@ from .storage import (
)
# Tools list for entry point discovery - must be classes, not instances
TOOLS = [APITool]
TOOLS = [HanzoTool, APITool]
__all__ = [
# Models
@@ -159,6 +160,7 @@ __all__ = [
"get_api_client",
"reset_api_client",
# Tool
"HanzoTool",
"APITool",
"TOOLS",
]
pkg/hanzo-tools-api/hanzo_tools/api/hanzo_tool.py
+209
View File
@@ -0,0 +1,209 @@
"""Unified Hanzo platform tool.
Provides a compact `hanzo` surface that routes to Hanzo service tools
(`auth`, `billing`, `commerce`, `iam`, `ingress`, `kms`, `mpc`, `paas`,
`team`, and generic `api`).
"""
from __future__ import annotations
import importlib
import inspect
import json
from typing import Any, Annotated, final, override
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from pydantic import Field
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
SERVICE_TOOL_PATHS: dict[str, str] = {
"api": "hanzo_tools.api.api_tool:APITool",
"auth": "hanzo_tools.auth.login_tool:LoginTool",
"billing": "hanzo_tools.billing.billing_tool:BillingTool",
"commerce": "hanzo_tools.commerce.commerce_tool:CommerceTool",
"iam": "hanzo_tools.iam.iam_tool:IAMTool",
"ingress": "hanzo_tools.ingress.ingress_tool:IngressTool",
"kms": "hanzo_tools.kms.kms_tool:KMSTool",
"mpc": "hanzo_tools.mpc.mpc_tool:MPCTool",
"paas": "hanzo_tools.paas.paas_tool:PaaSTool",
"team": "hanzo_tools.team.team_tool:TeamTool",
}
SERVICE_ALIASES: dict[str, str] = {
"platform": "paas",
"identity": "iam",
"payments": "billing",
"store": "commerce",
}
@final
class HanzoTool(BaseTool):
"""Unified tool for Hanzo platform services."""
name = "hanzo"
def __init__(self) -> None:
self._delegates: dict[str, BaseTool] = {}
@property
@override
def description(self) -> str:
return """Unified Hanzo platform tool.
Use one `hanzo` tool surface for service operations across:
- auth
- billing
- commerce
- iam
- ingress
- kms
- mpc
- paas
- team
- api (generic OpenAPI bridge)
Parameters:
- service: Target Hanzo service
- action: Service-specific action
- args: JSON object string for service-specific parameters
Examples:
hanzo(service="auth", action="status")
hanzo(service="commerce", action="orders")
hanzo(service="iam", action="users", args='{"owner":"hanzo"}')
hanzo(service="api", action="list")
"""
def _normalize_service(self, service: str) -> str:
key = (service or "").strip().lower().replace("-", "_")
key = SERVICE_ALIASES.get(key, key)
return key
def _load_delegate(self, service: str) -> BaseTool:
if service in self._delegates:
return self._delegates[service]
path = SERVICE_TOOL_PATHS.get(service)
if not path:
available = ", ".join(sorted(SERVICE_TOOL_PATHS.keys()))
raise ValueError(
f"Unknown service '{service}'. Available services: {available}"
)
module_name, class_name = path.split(":")
module = importlib.import_module(module_name)
cls = getattr(module, class_name)
tool = cls()
self._delegates[service] = tool
return tool
def _parse_args(self, args: str | None) -> dict[str, Any]:
if not args:
return {}
try:
parsed = json.loads(args)
except json.JSONDecodeError as exc:
raise ValueError(f"args must be valid JSON object string: {exc}") from exc
if not isinstance(parsed, dict):
raise ValueError("args must decode to a JSON object")
return parsed
async def _delegate_call(
self,
tool: BaseTool,
ctx: MCPContext,
payload: dict[str, Any],
) -> str:
sig = inspect.signature(tool.call)
params = sig.parameters
accepts_kwargs = any(
p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()
)
if accepts_kwargs:
return await tool.call(ctx, **payload)
allowed = {
name
for name, param in params.items()
if name not in {"self", "ctx"}
and param.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
}
filtered = {k: v for k, v in payload.items() if k in allowed}
return await tool.call(ctx, **filtered)
@override
@auto_timeout("hanzo")
async def call(
self,
ctx: MCPContext,
service: str = "api",
action: str = "list",
args: str | None = None,
**kwargs: Any,
) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
service_key = self._normalize_service(service)
if service_key in {"services", "list"}:
return json.dumps(
{
"services": sorted(SERVICE_TOOL_PATHS.keys()),
"aliases": SERVICE_ALIASES,
"usage": 'hanzo(service="commerce", action="orders", args="{\\"query\\":\\"...\\\"}")',
},
indent=2,
)
try:
delegate = self._load_delegate(service_key)
payload = {"action": action}
payload.update(self._parse_args(args))
payload.update({k: v for k, v in kwargs.items() if v is not None})
return await self._delegate_call(delegate, ctx, payload)
except Exception as exc:
return json.dumps(
{
"error": str(exc),
"service": service_key,
"available_services": sorted(SERVICE_TOOL_PATHS.keys()),
},
indent=2,
)
def register(self, mcp_server: FastMCP) -> None:
"""Register unified hanzo tool with explicit compact params."""
tool_instance = self
@mcp_server.tool(name=self.name, description=self.description)
async def hanzo(
service: Annotated[
str,
Field(
description=(
"Target service: api, auth, billing, commerce, iam, ingress, "
"kms, mpc, paas, team. Use 'services' to list."
)
),
] = "api",
action: Annotated[
str,
Field(description="Service action to execute (service-specific)."),
] = "list",
args: Annotated[
str | None,
Field(
description=(
"JSON object string with service-specific parameters. "
'Example: "{\\"query\\":\\"foo\\",\\"owner\\":\\"hanzo\\"}"'
)
),
] = None,
ctx: MCPContext = None, # type: ignore[assignment]
) -> str:
return await tool_instance.call(ctx, service=service, action=action, args=args)
pkg/hanzo-tools-api/hanzo_tools/api/providers.py
+1 -1
View File
@@ -321,7 +321,7 @@ PROVIDER_CONFIGS: dict[str, ProviderConfig] = {
"hanzo-iam": ProviderConfig(
name="hanzo-iam",
display_name="Hanzo IAM",
base_url="https://iam.hanzo.ai",
base_url="https://hanzo.id",
auth_type=AuthType.BEARER,
spec_url="file:///Users/z/work/hanzo/openapi/iam/openapi.yaml",
env_vars=["HANZO_API_KEY", "HANZO_TOKEN"],
pkg/hanzo-tools-api/tests/test_api_tools.py
+33
View File
@@ -1,5 +1,6 @@
"""Tests for hanzo-tools-api v0.2.0."""
import json
import os
import tempfile
from pathlib import Path
@@ -12,6 +13,7 @@ from hanzo_tools.api import (
PROVIDER_CONFIGS,
APIClient,
APITool,
HanzoTool,
Credential,
CredentialManager,
OpenAPIClient,
@@ -412,6 +414,37 @@ class TestAPITool:
assert "credential" in desc.lower()
class TestHanzoTool:
"""Tests for unified HanzoTool surface."""
@pytest.mark.asyncio
async def test_services_listing(self):
"""Service discovery should return consolidated service list."""
tool = HanzoTool()
ctx = AsyncMock()
result = await tool.call(ctx, service="services")
payload = json.loads(result)
assert "services" in payload
assert "hanzo" not in payload["services"] # service router, not a nested service
assert "commerce" in payload["services"]
assert "iam" in payload["services"]
@pytest.mark.asyncio
async def test_invalid_args_json(self):
"""Invalid JSON args should return structured error."""
tool = HanzoTool()
ctx = AsyncMock()
result = await tool.call(
ctx,
service="iam",
action="users",
args="{invalid-json",
)
payload = json.loads(result)
assert "error" in payload
assert payload["service"] == "iam"
class TestIntegration:
"""Integration tests (require network, skip by default)."""
pkg/hanzo-tools-computer/hanzo_tools/computer/__init__.py
+8 -8
View File
@@ -37,25 +37,25 @@ Screen Limits (configurable via env vars):
from hanzo_tools.core import BaseTool, ToolRegistry, PermissionManager
# HIP-0300: Single unified 'ui' tool
from .ui_tool import UiTool, ui_tool
# HIP-0300: Single unified 'computer' tool
from .computer_tool import ComputerTool
# Internal utilities (used by screen_tool)
from .media_tool import MediaLimits, MediaResult, ActivitySegment, media_tool
from .screen_tool import ScreenTool, ScreenConfig, screen_tool
# Backward compatibility (deprecated)
from .computer_tool import ComputerTool
# Backward compat
from .ui_tool import UiTool, ui_tool
# Export list for tool discovery - single ui tool (HIP-0300)
TOOLS = [UiTool]
# Export list for tool discovery - single computer tool (HIP-0300)
TOOLS = [ComputerTool]
__all__ = [
# HIP-0300 unified tool
"ComputerTool",
# Backward compat
"UiTool",
"ui_tool",
# Backward compatibility (deprecated)
"ComputerTool",
"ScreenTool",
"ScreenConfig",
"screen_tool",
pkg/hanzo-tools-computer/hanzo_tools/computer/ui_tool.py
+1 -1
View File
@@ -37,7 +37,7 @@ class UiTool(ComputerTool):
"""
# Override name to 'ui' for HIP-0300
name: ClassVar[str] = "ui"
name: ClassVar[str] = "computer"
@property
def description(self) -> str:
pkg/hanzo-tools-config/hanzo_tools/config/__init__.py
+10
View File
@@ -3,6 +3,7 @@
Tools:
- config: Configuration management
- mode: Development mode switching
- workspace: Workspace context detection
Install:
pip install hanzo-tools-config
@@ -30,12 +31,21 @@ except ImportError as e:
logger.debug(f"ModeTool not available: {e}")
ModeTool = None
try:
from .workspace_tool import WorkspaceTool
_tools.append(WorkspaceTool)
except ImportError as e:
logger.debug(f"WorkspaceTool not available: {e}")
WorkspaceTool = None
TOOLS = _tools
__all__ = [
"TOOLS",
"ConfigTool",
"ModeTool",
"WorkspaceTool",
"register_tools",
]
pkg/hanzo-tools-config/hanzo_tools/config/workspace_tool.py
+152
View File
@@ -0,0 +1,152 @@
"""Workspace context tool (HIP-0300).
One tool for the Project Context axis.
Actions: detect, capabilities, help, schema
"""
from __future__ import annotations
import asyncio
import json
import os
import shutil
from pathlib import Path
from typing import Any, ClassVar
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool
class WorkspaceTool(BaseTool):
"""Workspace context tool (HIP-0300).
Detects project languages, build systems, test frameworks,
and available tool capabilities.
Actions:
- detect: Detect project languages, build systems, and VCS
- capabilities: List available system tools and runtimes
- help: Show all tool summaries
"""
name: ClassVar[str] = "workspace"
VERSION: ClassVar[str] = "0.1.0"
def __init__(self, cwd: str | None = None):
super().__init__()
self.cwd = cwd or os.getcwd()
self._register_workspace_actions()
@property
def description(self) -> str:
return """Workspace context tool (HIP-0300).
Actions:
- detect: Detect project languages, build systems, VCS, test frameworks
- capabilities: List available system tools and runtimes
- help: Show all tool summaries
"""
def _register_workspace_actions(self):
"""Register all workspace actions."""
@self.action("detect", "Detect project languages, build systems, and VCS")
async def detect(ctx: MCPContext, path: str | None = None) -> dict:
root = Path(path or self.cwd)
languages: list[str] = []
build: list[str] = []
test: list[str] = []
vcs: str | None = None
# VCS
if (root / ".git").exists():
vcs = "git"
# Languages & build systems
if (root / "package.json").exists():
languages.extend(["typescript", "javascript"])
build.append("npm")
if (root / "tsconfig.json").exists():
if "typescript" not in languages:
languages.append("typescript")
if (root / "pyproject.toml").exists():
languages.append("python")
build.append("uv")
if (root / "Cargo.toml").exists():
languages.append("rust")
build.append("cargo")
if (root / "go.mod").exists():
languages.append("go")
build.append("go")
if (root / "Makefile").exists():
build.append("make")
if (root / "compose.yml").exists():
build.append("docker-compose")
if (root / "Dockerfile").exists():
build.append("docker")
# Test frameworks
if (root / "jest.config.ts").exists() or (root / "jest.config.js").exists():
test.append("jest")
if (root / "vitest.config.ts").exists():
test.append("vitest")
if (root / "pytest.ini").exists() or (root / "conftest.py").exists():
test.append("pytest")
return {
"root": str(root),
"languages": list(dict.fromkeys(languages)),
"build": list(dict.fromkeys(build)),
"test": test,
"vcs": vcs,
}
@self.action("capabilities", "List available system tools and runtimes")
async def capabilities(ctx: MCPContext) -> dict:
has_rg = shutil.which("rg") is not None
has_git = shutil.which("git") is not None
has_node = shutil.which("node") is not None
has_python = shutil.which("python3") is not None
has_cargo = shutil.which("cargo") is not None
return {
"search": "ripgrep" if has_rg else "grep",
"vcs": "git" if has_git else None,
"runtimes": {
"node": has_node,
"python": has_python,
"rust": has_cargo,
},
"tools": [
"fs", "exec", "code", "git", "fetch", "computer", "workspace",
"browser", "think", "llm", "memory", "hanzo", "plan", "tasks", "mode",
],
}
@self.action("help", "Show all tool summaries")
async def help_action(ctx: MCPContext) -> dict:
return {
"tools": {
"fs": "Filesystem: read, write, stat, list, mkdir, rm, apply_patch, search_text",
"exec": "Processes: exec, ps, kill, logs",
"code": "Semantics: parse, symbols, definition, references, transform, summarize",
"git": "Version control: status, diff, apply, commit, branch, checkout, log",
"fetch": "Network: search, fetch, download, crawl, head",
"workspace": "Workspace: detect, capabilities, help",
"computer": "Native OS control: click, type, screenshot, window management",
"browser": "Web automation: navigate, click, fill, screenshot (Playwright)",
"think": "Structured reasoning: think, critic, review, summarize",
"llm": "LLM operations: query, consensus, models, enable, disable, test",
"memory": "Memory: store, recall, list, delete, search, stats, clear, export, import",
"hanzo": "Hanzo platform: iam, kms, paas, commerce, auth, api, billing, ingress, mpc, team",
"plan": "Planning: create, show, update, list, next, archive, add_step, remove_step",
"tasks": "Tasks: list, add, update, remove, clear",
"mode": "Modes: list, activate, show, current",
}
}
async def call(self, ctx: MCPContext, **kwargs: Any) -> str:
"""Route to action handler."""
action = kwargs.pop("action", "help")
return await self._dispatch(ctx, action, **kwargs)
pkg/hanzo-tools-fs/hanzo_tools/fs/fs_tool.py
-6
View File
@@ -27,12 +27,6 @@ import aiofiles
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
# Compatibility: some older hanzo-tools.core builds don't re-export Paging.
try:
from hanzo_tools.core import Paging
except ImportError:
from hanzo_tools.core.unified import Paging
from hanzo_tools.core import (
BaseTool,
ConflictError,
pkg/hanzo-tools-memory/hanzo_tools/memory/knowledge_tools.py
-511
View File
@@ -1,511 +0,0 @@
"""Knowledge base and fact management tools for MCP.
Uses hanzo-memory (full vector search) when available, otherwise falls back to
the lightweight MarkdownMemoryBackend which stores session facts in-memory.
IMPORTANT: All hanzo-memory imports are lazy to avoid slow startup.
"""
from typing import TYPE_CHECKING, Any, Dict, List, Optional, final, override
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
from hanzo_tools.memory.markdown_memory import get_markdown_backend
# Type hints only - no runtime import
if TYPE_CHECKING:
from hanzo_memory.services.memory import MemoryService
# Use lazy loading from memory_tools
from hanzo_tools.memory.memory_tools import (
_get_backend,
_check_memory_available,
_get_lazy_memory_service,
)
class KnowledgeToolBase(BaseTool):
"""Base class for knowledge tools."""
def __init__(self, user_id: str = "default", project_id: str = "default", **kwargs):
self.user_id = user_id
self.project_id = project_id
@final
class RecallFactsTool(KnowledgeToolBase):
"""Tool for recalling facts from knowledge bases."""
name = "recall_facts"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Recall facts from knowledge bases relevant to queries.
Facts are structured pieces of information stored in knowledge bases.
Supports different scopes: session, project, or global.
Usage:
recall_facts(queries=["Python best practices"], kb_name="coding_standards")
recall_facts(queries=["API endpoints"], scope="project")
recall_facts(queries=["company policies"], scope="global", limit=5)
"""
@override
@auto_timeout("knowledge_tools")
async def call(
self,
ctx: MCPContext,
queries: List[str],
kb_name: Optional[str] = None,
scope: str = "project",
limit: int = 10,
) -> str:
"""Recall facts matching queries.
Args:
ctx: MCP context
queries: Search queries
kb_name: Optional knowledge base name to search in
scope: Scope level (session, project, global)
limit: Max results per query
Returns:
Formatted fact results
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Searching for facts in scope: {scope}")
backend = _get_backend()
if _check_memory_available():
# Full vector search via hanzo-memory
if scope == "global":
user_id, project_id = "global", "global"
elif scope == "session":
user_id, project_id = f"session_{self.user_id}", self.project_id
else:
user_id, project_id = self.user_id, self.project_id
all_facts = []
seen: set = set()
for query in queries:
search_query = f"fact: {query}"
if kb_name:
search_query = f"kb:{kb_name} {search_query}"
memories = backend.search_memories( # type: ignore[union-attr]
user_id=user_id, query=search_query, project_id=project_id, limit=limit
)
for m in memories:
if m.memory_id not in seen and m.metadata and m.metadata.get("type") == "fact":
seen.add(m.memory_id)
all_facts.append(m)
if not all_facts:
return "No relevant facts found."
formatted = [f"Found {len(all_facts)} relevant facts:\n"]
for i, fact in enumerate(all_facts, 1):
kb_info = f" (KB: {fact.metadata['kb_name']})" if fact.metadata and fact.metadata.get("kb_name") else ""
formatted.append(f"{i}. {fact.content}{kb_info}")
return "\n".join(formatted)
else:
# Markdown fallback — session-only fact storage
facts = backend.recall_facts(queries=queries, kb_name=kb_name, limit=limit) # type: ignore[union-attr]
if not facts:
return "No relevant facts found."
formatted = [f"Found {len(facts)} relevant facts:\n"]
for i, fact in enumerate(facts, 1):
kb_info = f" (KB: {fact.kb_name})" if fact.kb_name and fact.kb_name != "general" else ""
formatted.append(f"{i}. {fact.statement}{kb_info}")
return "\n".join(formatted)
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def recall_facts(
ctx: MCPContext,
queries: List[str],
kb_name: Optional[str] = None,
scope: str = "project",
limit: int = 10,
) -> str:
return await tool_self.call(
ctx, queries=queries, kb_name=kb_name, scope=scope, limit=limit
)
@final
class StoreFactsTool(KnowledgeToolBase):
"""Tool for storing facts in knowledge bases."""
name = "store_facts"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Store new facts in knowledge bases.
Facts can be stored at different scopes (session, project, global) and
organized into knowledge bases for better categorization.
Usage:
store_facts(facts=["Python uses indentation for blocks"], kb_name="python_basics")
store_facts(facts=["API rate limit: 100/hour"], scope="project", kb_name="api_docs")
store_facts(facts=["Company founded in 2020"], scope="global", kb_name="company_info")
"""
@override
@auto_timeout("knowledge_tools")
async def call(
self,
ctx: MCPContext,
facts: List[str],
kb_name: str = "general",
scope: str = "project",
metadata: Optional[Dict[str, Any]] = None,
) -> str:
"""Store new facts.
Args:
ctx: MCP context
facts: Facts to store
kb_name: Knowledge base name
scope: Scope level (session, project, global)
metadata: Optional metadata for all facts
Returns:
Success message
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Storing {len(facts)} facts in {kb_name} (scope: {scope})")
backend = _get_backend()
if _check_memory_available():
if scope == "global":
user_id, project_id = "global", "global"
elif scope == "session":
user_id, project_id = f"session_{self.user_id}", self.project_id
else:
user_id, project_id = self.user_id, self.project_id
for fact_content in facts:
fact_metadata: Dict[str, Any] = {"type": "fact", "kb_name": kb_name}
if metadata:
fact_metadata.update(metadata)
backend.create_memory( # type: ignore[union-attr]
user_id=user_id,
project_id=project_id,
content=f"fact: {fact_content}",
metadata=fact_metadata,
importance=1.5,
)
else:
# Markdown fallback — store in session facts
for fact_content in facts:
backend.store_fact(statement=fact_content, kb_name=kb_name, scope=scope) # type: ignore[union-attr]
return f"Successfully stored {len(facts)} facts in {kb_name}."
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def store_facts(
ctx: MCPContext,
facts: List[str],
kb_name: str = "general",
scope: str = "project",
metadata: Optional[Dict[str, Any]] = None,
) -> str:
return await tool_self.call(
ctx, facts=facts, kb_name=kb_name, scope=scope, metadata=metadata
)
@final
class SummarizeToMemoryTool(KnowledgeToolBase):
"""Tool for summarizing information and storing in memory."""
name = "summarize_to_memory"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Summarize information and store it in memory for future reference.
This tool helps agents remember important information by summarizing it
and storing it at the appropriate scope level.
Usage:
summarize_to_memory(content="Long discussion about API design...", topic="API Design Decisions")
summarize_to_memory(content="User preferences...", topic="User Preferences", scope="session")
summarize_to_memory(content="Company guidelines...", topic="Guidelines", scope="global")
"""
@override
@auto_timeout("knowledge_tools")
async def call(
self,
ctx: MCPContext,
content: str,
topic: str,
scope: str = "project",
auto_facts: bool = True,
) -> str:
"""Summarize content and store in memory.
Args:
ctx: MCP context
content: Content to summarize
topic: Topic or title for the summary
scope: Scope level (session, project, global)
auto_facts: Whether to extract facts automatically
Returns:
Success message with summary
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Summarizing content about {topic}")
summary = (
f"Summary of {topic}:\n{content[:500]}..." if len(content) > 500 else content
)
backend = _get_backend()
if _check_memory_available():
if scope == "global":
user_id, project_id = "global", "global"
elif scope == "session":
user_id, project_id = f"session_{self.user_id}", self.project_id
else:
user_id, project_id = self.user_id, self.project_id
backend.create_memory( # type: ignore[union-attr]
user_id=user_id,
project_id=project_id,
content=summary,
metadata={"topic": topic, "type": "summary", "scope": scope},
)
else:
backend.add_memory(content=f"[{topic}] {summary}", scope=scope) # type: ignore[union-attr]
result = f"Stored summary of {topic} in {scope} memory."
if auto_facts:
result += "\n(Tip: install hanzo-memory for automatic fact extraction)"
return result
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def summarize_to_memory(
ctx: MCPContext,
content: str,
topic: str,
scope: str = "project",
auto_facts: bool = True,
) -> str:
return await tool_self.call(
ctx, content=content, topic=topic, scope=scope, auto_facts=auto_facts
)
@final
class ManageKnowledgeBasesTool(KnowledgeToolBase):
"""Tool for managing knowledge bases."""
name = "manage_knowledge_bases"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Create, list, and manage knowledge bases.
Knowledge bases help organize facts by topic or domain.
They can exist at different scopes for better organization.
Usage:
manage_knowledge_bases(action="create", kb_name="api_docs", description="API documentation")
manage_knowledge_bases(action="list", scope="project")
manage_knowledge_bases(action="delete", kb_name="old_docs")
"""
@override
@auto_timeout("knowledge_tools")
async def call(
self,
ctx: MCPContext,
action: str,
kb_name: Optional[str] = None,
description: Optional[str] = None,
scope: str = "project",
) -> str:
"""Manage knowledge bases.
Args:
ctx: MCP context
action: Action to perform (create, list, delete)
kb_name: Knowledge base name (for create/delete)
description: Description (for create)
scope: Scope level
Returns:
Result message
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
if not _check_memory_available():
# Markdown backend doesn't have persistent KB management
if action == "list":
return "Knowledge bases are session-only without hanzo-memory. Install hanzo-memory for persistent KBs."
elif action == "create":
return f"Knowledge base '{kb_name}' noted (session-only without hanzo-memory)."
elif action == "delete":
return f"Knowledge base '{kb_name}' removed from session."
return "Knowledge base management requires hanzo-memory for persistence."
backend = _get_lazy_memory_service()
# Determine scope
if scope == "global":
user_id = "global"
project_id = "global"
elif scope == "session":
user_id = f"session_{self.user_id}"
project_id = self.project_id
else:
user_id = self.user_id
project_id = self.project_id
if action == "create":
if not kb_name:
return "Error: kb_name required for create action"
# Create a knowledge base entry as a special memory
kb_metadata = {
"type": "knowledge_base",
"kb_name": kb_name,
"description": description or "",
"scope": scope,
}
memory = backend.create_memory(
user_id=user_id,
project_id=project_id,
content=f"Knowledge Base: {kb_name}\nDescription: {description or 'No description'}",
metadata=kb_metadata,
importance=2.0, # KBs have high importance
)
return f"Created knowledge base '{kb_name}' in {scope} scope."
elif action == "list":
# Search for knowledge base entries
kbs = backend.search_memories(
user_id=user_id,
query="type:knowledge_base",
project_id=project_id,
limit=100,
)
# Filter for KB-type memories
kb_list = []
for memory in kbs:
if memory.metadata and memory.metadata.get("type") == "knowledge_base":
kb_list.append(memory)
if not kb_list:
return f"No knowledge bases found in {scope} scope."
formatted = [f"Knowledge bases in {scope} scope:"]
for kb in kb_list:
name = kb.metadata.get("kb_name", "Unknown")
desc = kb.metadata.get("description", "")
desc_text = f" - {desc}" if desc else ""
formatted.append(f"- {name}{desc_text}")
return "\n".join(formatted)
elif action == "delete":
if not kb_name:
return "Error: kb_name required for delete action"
# Search for the KB entry
kbs = backend.search_memories(
user_id=user_id,
query=f"type:knowledge_base kb_name:{kb_name}",
project_id=project_id,
limit=10,
)
deleted_count = 0
for memory in kbs:
if (
memory.metadata
and memory.metadata.get("type") == "knowledge_base"
and memory.metadata.get("kb_name") == kb_name
):
# Note: delete_memory is not fully implemented
# but we'll call it anyway
backend.delete_memory(user_id, memory.memory_id)
deleted_count += 1
if deleted_count > 0:
return f"Deleted knowledge base '{kb_name}'."
else:
return f"Knowledge base '{kb_name}' not found."
else:
return f"Unknown action: {action}. Use create, list, or delete."
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def manage_knowledge_bases(
ctx: MCPContext,
action: str,
kb_name: Optional[str] = None,
description: Optional[str] = None,
scope: str = "project",
) -> str:
return await tool_self.call(
ctx,
action=action,
kb_name=kb_name,
description=description,
scope=scope,
)
# Export all knowledge tool classes
KNOWLEDGE_TOOL_CLASSES = [
RecallFactsTool,
StoreFactsTool,
SummarizeToMemoryTool,
ManageKnowledgeBasesTool,
]
pkg/hanzo-tools-memory/hanzo_tools/memory/memory_tool.py
+4 -4
View File
@@ -174,11 +174,11 @@ Examples:
elif action == "manage":
return await self._manage(create_list, update_list, delete_ids)
elif action == "facts":
return await self._recall_facts(
return await self._facts_lookup(
queries or ([query] if query else []), kb, limit
)
elif action == "store":
return await self._store_facts(
return await self._facts_store(
facts or ([fact] if fact else []), kb, metadata
)
elif action == "summarize":
@@ -371,7 +371,7 @@ Examples:
return "No operations specified. Provide create_list, update_list, or delete_ids."
return "\n".join(results)
async def _recall_facts(
async def _facts_lookup(
self, queries: List[str], kb: Optional[str], limit: int
) -> str:
"""Search knowledge bases."""
@@ -399,7 +399,7 @@ Examples:
except ImportError:
return "Knowledge service not available"
async def _store_facts(
async def _facts_store(
self,
facts: List[str],
kb: Optional[str],
pkg/hanzo-tools-memory/hanzo_tools/memory/memory_tools.py
-426
View File
@@ -1,426 +0,0 @@
"""Memory tools implementation for MCP.
Uses hanzo-memory (full vector search) when available, otherwise falls back to
the lightweight MarkdownMemoryBackend which reads from local .md files
(MEMORY.md, LLM.md, CLAUDE.md, etc.) with no external dependencies.
IMPORTANT: All hanzo-memory imports are lazy to avoid slow startup.
"""
from typing import TYPE_CHECKING, Dict, List, Union, Optional, final, override
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
from hanzo_tools.memory.markdown_memory import (
MarkdownMemoryBackend,
get_markdown_backend,
)
# Type hints only - no runtime import
if TYPE_CHECKING:
from hanzo_memory.services.memory import MemoryService
# Lazy loading - don't import at module load time
MEMORY_AVAILABLE: Optional[bool] = None
_memory_service: Optional["MemoryService"] = None
def _check_memory_available() -> bool:
"""Check if hanzo-memory is available (lazy check)."""
global MEMORY_AVAILABLE
if MEMORY_AVAILABLE is None:
try:
import hanzo_memory # noqa: F401
MEMORY_AVAILABLE = True
except ImportError:
MEMORY_AVAILABLE = False
return MEMORY_AVAILABLE
def _get_lazy_memory_service() -> "MemoryService":
"""Get hanzo-memory service lazily. Returns None if not available."""
global _memory_service
if _memory_service is None:
if not _check_memory_available():
return None # type: ignore[return-value]
from hanzo_memory.services.memory import get_memory_service
_memory_service = get_memory_service()
return _memory_service
def _get_backend() -> Union["MemoryService", MarkdownMemoryBackend]:
"""Get the best available memory backend."""
svc = _get_lazy_memory_service()
if svc is not None:
return svc
return get_markdown_backend()
class MemoryToolBase(BaseTool):
"""Base class for memory tools."""
def __init__(self, user_id: str = "default", project_id: str = "default", **kwargs):
self.user_id = user_id
self.project_id = project_id
@final
class RecallMemoriesTool(MemoryToolBase):
"""Tool for recalling memories."""
name = "recall_memories"
@property
@override
def description(self) -> str:
return """Recall memories relevant to one or more queries.
This tool searches through stored memories and returns relevant matches.
Supports different scopes: session, project, or global memories.
Multiple queries can be run in parallel for efficiency.
Usage:
recall_memories(queries=["user preferences", "previous conversations"])
recall_memories(queries=["project requirements"], scope="project")
recall_memories(queries=["coding standards"], scope="global")
"""
@override
@auto_timeout("memory_tools")
async def call(
self,
ctx: MCPContext,
queries: List[str],
limit: int = 10,
scope: str = "project",
) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Searching for {len(queries)} queries")
backend = _get_backend()
unique_results = []
if _check_memory_available():
# Full vector search via hanzo-memory
seen: set = set()
for query in queries:
results = backend.search_memories( # type: ignore[union-attr]
user_id=self.user_id,
query=query,
project_id=self.project_id,
limit=limit,
)
for r in results:
if r.memory_id not in seen:
seen.add(r.memory_id)
unique_results.append(r)
if not unique_results:
return "No relevant memories found."
formatted = [f"Found {len(unique_results)} relevant memories:\n"]
for i, m in enumerate(unique_results, 1):
score = getattr(m, "similarity_score", 0.0)
formatted.append(f"{i}. {m.content} (relevance: {score:.2f})")
return "\n".join(formatted)
else:
# Markdown fallback — keyword search across local .md files
results = backend.search_memories(queries=queries, limit=limit, scope=scope) # type: ignore[union-attr]
if not results:
return "No relevant memories found."
formatted = [f"Found {len(results)} relevant memories:\n"]
for i, m in enumerate(results, 1):
source = getattr(m, "source", "")
src_note = f" [{source}]" if source and source != "user" else ""
formatted.append(f"{i}. {m.content}{src_note}")
return "\n".join(formatted)
@override
def register(self, mcp_server: FastMCP) -> None:
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def recall_memories(
ctx: MCPContext, queries: List[str], limit: int = 10, scope: str = "project"
) -> str:
return await tool_self.call(ctx, queries=queries, limit=limit, scope=scope)
@final
class CreateMemoriesTool(MemoryToolBase):
"""Tool for creating memories."""
name = "create_memories"
@property
@override
def description(self) -> str:
return """Save one or more new pieces of information to memory.
This tool creates new memories from provided statements.
Each statement is stored as a separate memory.
Usage:
create_memories(statements=["User prefers dark mode", "User works in Python"])
"""
@override
@auto_timeout("memory_tools")
async def call(
self,
ctx: MCPContext,
statements: List[str],
scope: str = "project",
) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Creating {len(statements)} memories")
backend = _get_backend()
if _check_memory_available():
for statement in statements:
backend.create_memory( # type: ignore[union-attr]
user_id=self.user_id,
project_id=self.project_id,
content=statement,
metadata={"type": "statement"},
)
else:
for statement in statements:
backend.add_memory(content=statement, scope=scope) # type: ignore[union-attr]
return f"Successfully created {len(statements)} new memories."
@override
def register(self, mcp_server: FastMCP) -> None:
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def create_memories(
ctx: MCPContext, statements: List[str], scope: str = "project"
) -> str:
return await tool_self.call(ctx, statements=statements, scope=scope)
@final
class UpdateMemoriesTool(MemoryToolBase):
"""Tool for updating memories."""
name = "update_memories"
@property
@override
def description(self) -> str:
return """Update existing memories with corrected information.
This tool updates memories by ID with new content.
Usage:
update_memories(updates=[
{"id": "mem_1", "statement": "User prefers light mode"},
{"id": "mem_2", "statement": "User primarily works in TypeScript"}
])
"""
@override
@auto_timeout("memory_tools")
async def call(self, ctx: MCPContext, updates: List[Dict[str, str]]) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Updating {len(updates)} memories")
backend = _get_backend()
success_count = 0
for update in updates:
memory_id = update.get("id")
statement = update.get("statement")
if not memory_id or not statement:
continue
if _check_memory_available():
# hanzo-memory update not fully implemented yet
await tool_ctx.warning(f"Memory update pending implementation: {memory_id}")
success_count += 1
else:
result = backend.update_memory(memory_id, statement) # type: ignore[union-attr]
if result is not None:
success_count += 1
return f"Updated {success_count} of {len(updates)} memories."
@override
def register(self, mcp_server: FastMCP) -> None:
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def update_memories(
ctx: MCPContext, updates: List[Dict[str, str]]
) -> str:
return await tool_self.call(ctx, updates=updates)
@final
class DeleteMemoriesTool(MemoryToolBase):
"""Tool for deleting memories."""
name = "delete_memories"
@property
@override
def description(self) -> str:
return """Delete memories that are no longer relevant or incorrect.
This tool removes memories by their IDs.
Usage:
delete_memories(ids=["mem_1", "mem_2"])
"""
@override
@auto_timeout("memory_tools")
async def call(self, ctx: MCPContext, ids: List[str]) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Deleting {len(ids)} memories")
backend = _get_backend()
success_count = 0
for memory_id in ids:
if _check_memory_available():
success = backend.delete_memory(self.user_id, memory_id) # type: ignore[union-attr]
else:
success = backend.delete_memory(memory_id) # type: ignore[union-attr]
if success:
success_count += 1
return f"Successfully deleted {success_count} of {len(ids)} memories."
@override
def register(self, mcp_server: FastMCP) -> None:
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def delete_memories(ctx: MCPContext, ids: List[str]) -> str:
return await tool_self.call(ctx, ids=ids)
@final
class ManageMemoriesTool(MemoryToolBase):
"""Tool for managing memories atomically."""
name = "manage_memories"
@property
@override
def description(self) -> str:
return """Create, update, and/or delete memories in a single atomic operation.
This is the preferred way to modify memories as it allows multiple
operations to be performed together.
Usage:
manage_memories(
creations=["New fact 1", "New fact 2"],
updates=[{"id": "mem_1", "statement": "Updated fact"}],
deletions=["mem_old1", "mem_old2"]
)
"""
@override
@auto_timeout("memory_tools")
async def call(
self,
ctx: MCPContext,
creations: Optional[List[str]] = None,
updates: Optional[List[Dict[str, str]]] = None,
deletions: Optional[List[str]] = None,
scope: str = "project",
) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
backend = _get_backend()
full = _check_memory_available()
results = []
if creations:
await tool_ctx.info(f"Creating {len(creations)} memories")
for statement in creations:
if full:
backend.create_memory( # type: ignore[union-attr]
user_id=self.user_id,
project_id=self.project_id,
content=statement,
metadata={"type": "statement"},
)
else:
backend.add_memory(content=statement, scope=scope) # type: ignore[union-attr]
results.append(f"Created {len(creations)} memories")
if updates:
await tool_ctx.info(f"Updating {len(updates)} memories")
success_count = 0
for update in updates:
memory_id = update.get("id")
statement = update.get("statement")
if not memory_id or not statement:
continue
if full:
await tool_ctx.warning(f"Memory update pending implementation: {memory_id}")
success_count += 1
else:
result = backend.update_memory(memory_id, statement) # type: ignore[union-attr]
if result is not None:
success_count += 1
results.append(f"Updated {success_count} memories")
if deletions:
await tool_ctx.info(f"Deleting {len(deletions)} memories")
success_count = 0
for memory_id in deletions:
if full:
success = backend.delete_memory(self.user_id, memory_id) # type: ignore[union-attr]
else:
success = backend.delete_memory(memory_id) # type: ignore[union-attr]
if success:
success_count += 1
results.append(f"Deleted {success_count} memories")
return "Memory operations completed: " + ", ".join(results) if results else "No memory operations performed."
@override
def register(self, mcp_server: FastMCP) -> None:
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def manage_memories(
ctx: MCPContext,
creations: Optional[List[str]] = None,
updates: Optional[List[Dict[str, str]]] = None,
deletions: Optional[List[str]] = None,
scope: str = "project",
) -> str:
return await tool_self.call(
ctx, creations=creations, updates=updates, deletions=deletions, scope=scope
)
# Export all memory tool classes
MEMORY_TOOL_CLASSES = [
RecallMemoriesTool,
CreateMemoriesTool,
UpdateMemoriesTool,
DeleteMemoriesTool,
ManageMemoriesTool,
]
pkg/hanzo-tools-net/README.md
View File
pkg/hanzo-tools-net/hanzo_tools/net/__init__.py
+9 -3
View File
@@ -27,12 +27,18 @@ Usage:
from hanzo_tools.core import BaseTool, ToolRegistry
from .net_tool import NetTool, net_tool
from .fetch_tool import FetchTool, fetch_tool
# Backward compat
NetTool = FetchTool
net_tool = fetch_tool
# Export list for tool discovery - HIP-0300 unified tool
TOOLS = [NetTool]
TOOLS = [FetchTool]
__all__ = [
"FetchTool",
"fetch_tool",
"NetTool",
"net_tool",
"register_tools",
@@ -51,6 +57,6 @@ def register_tools(mcp_server, **kwargs) -> list[BaseTool]:
List of registered tool instances
"""
cwd = kwargs.get("cwd")
tool = NetTool(cwd=cwd)
tool = FetchTool(cwd=cwd)
ToolRegistry.register_tool(mcp_server, tool)
return [tool]
pkg/hanzo-tools-net/hanzo_tools/net/net_tool.py → pkg/hanzo-tools-net/hanzo_tools/net/fetch_tool.py
+4 -4
View File
@@ -1,6 +1,6 @@
"""Network tool for HIP-0300 architecture.
This module provides the 'net' tool for network operations:
This module provides the 'fetch' tool for network operations:
- search: Query → [URL, title, snippet]
- fetch: URL → {text, mime, status}
- download: URL → Path (with assets)
@@ -28,7 +28,7 @@ from hanzo_tools.core import (
)
class NetTool(BaseTool):
class FetchTool(BaseTool):
"""Network operations tool (HIP-0300).
Handles all network operations:
@@ -40,7 +40,7 @@ class NetTool(BaseTool):
Effect: NONDETERMINISTIC_EFFECT
"""
name: ClassVar[str] = "net"
name: ClassVar[str] = "fetch"
VERSION: ClassVar[str] = "0.1.0"
def __init__(self, cwd: str | None = None):
@@ -478,4 +478,4 @@ Effect: NONDETERMINISTIC_EFFECT (network I/O)
# Singleton
net_tool = NetTool
fetch_tool = FetchTool
pkg/hanzo-tools-plan/README.md
View File
pkg/hanzo-tools-shell/hanzo_tools/shell/__init__.py
+5 -5
View File
@@ -79,8 +79,8 @@ from hanzo_tools.shell.curl_tool import CurlTool
# Convenience tools
from hanzo_tools.shell.open_tool import OpenTool, open_tool
# HIP-0300: Unified proc tool
from hanzo_tools.shell.proc_tool import ProcTool, proc_tool
# HIP-0300: Unified exec tool
from hanzo_tools.shell.exec_tool import ExecTool, exec_tool
from hanzo_tools.shell.shellflow import (
parse as parse_shellflow,
compile as compile_shellflow,
@@ -195,9 +195,9 @@ TOOLS = _get_detected_shell_tools()
__all__ = [
# uvloop status
"_using_uvloop",
# HIP-0300: Unified proc tool
"ProcTool",
"proc_tool",
# HIP-0300: Unified exec tool
"ExecTool",
"exec_tool",
# Base classes
"ProcessManager",
"AutoBackgroundExecutor",
pkg/hanzo-tools-shell/hanzo_tools/shell/proc_tool.py → pkg/hanzo-tools-shell/hanzo_tools/shell/exec_tool.py
+4 -10
View File
@@ -1,6 +1,6 @@
"""Unified process tool for HIP-0300 architecture.
This module provides a single unified 'proc' tool that handles all process operations:
This module provides a single unified 'exec' tool that handles all process operations:
- exec: Execute commands (the ONE execution primitive)
- ps: List processes
- kill: Kill processes
@@ -21,12 +21,6 @@ from datetime import datetime
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
# Compatibility: some older hanzo-tools.core builds don't re-export Paging.
try:
from hanzo_tools.core import Paging
except ImportError:
from hanzo_tools.core.unified import Paging
from hanzo_tools.core import (
BaseTool,
ToolError,
@@ -40,7 +34,7 @@ from hanzo_tools.shell.base_process import (
)
class ProcTool(BaseTool):
class ExecTool(BaseTool):
"""Unified process execution tool (HIP-0300).
Handles all process operations on a single axis:
@@ -53,7 +47,7 @@ class ProcTool(BaseTool):
All command execution goes through proc.exec.
"""
name: ClassVar[str] = "proc"
name: ClassVar[str] = "exec"
VERSION: ClassVar[str] = "0.12.0"
def __init__(self, tools: Optional[Dict[str, Any]] = None):
@@ -486,4 +480,4 @@ Auto-backgrounds commands after {AUTO_BACKGROUND_TIMEOUT}s.
# Backward compatibility
proc_tool = ProcTool
exec_tool = ExecTool
pkg/hanzo-tools-todo/hanzo_tools/todo/__init__.py
+7 -3
View File
@@ -8,9 +8,13 @@ from mcp.server import FastMCP
from hanzo_tools.core import BaseTool, ToolRegistry
from hanzo_tools.todo.base import TodoStorage, TodoBaseTool
from hanzo_tools.todo.todo_tool import TodoTool
from hanzo_tools.todo.tasks_tool import TasksTool
# Backward compat
TodoTool = TasksTool
__all__ = [
"TasksTool",
"TodoTool",
"TodoStorage",
"TodoBaseTool",
@@ -26,7 +30,7 @@ def get_todo_tools() -> list[BaseTool]:
Returns:
List of todo tool instances
"""
return [TodoTool()]
return [TasksTool()]
def register_todo_tools(
@@ -48,4 +52,4 @@ def register_todo_tools(
# TOOLS list for entry point discovery
TOOLS = [TodoTool]
TOOLS = [TasksTool]
pkg/hanzo-tools-todo/hanzo_tools/todo/todo_tool.py → pkg/hanzo-tools-todo/hanzo_tools/todo/tasks_tool.py
+2 -2
View File
@@ -86,10 +86,10 @@ class TodoParams(TypedDict, total=False):
@final
class TodoTool(TodoBaseTool):
class TasksTool(TodoBaseTool):
"""Unified todo management tool."""
name = "todo"
name = "tasks"
def read_todos(self) -> list[Dict[str, Any]]:
"""Read todos from storage using default session."""
pkg/hanzo-tools-ui/README.md
+17
View File
@@ -0,0 +1,17 @@
# hanzo-tools-ui
UI component registry tools for Hanzo AI MCP.
Browse, search, install, and manage UI components from Hanzo and other registries (shadcn/ui, Vue, Svelte, React Native).
## Install
```bash
pip install hanzo-tools-ui
```
## Usage
```python
from hanzo_tools.ui import UiTool, TOOLS
```
pkg/hanzo-tools-ui/hanzo_tools/__init__.py
View File
pkg/hanzo-tools-ui/hanzo_tools/ui/__init__.py
+60
View File
@@ -0,0 +1,60 @@
"""UI component registry tools for Hanzo AI (HIP-0300).
Tools:
- ui: Unified UI component tool
- list_components: List available components
- get_component: Get component source code
- get_demo: Get component demo/example
- get_metadata: Get component metadata
- list_blocks: List UI blocks
- get_block: Get block implementation
- search: Search components
- get_structure: Browse repository structure
- install: Install component via CLI
- set_framework: Switch active framework
- get_framework: Show current and available frameworks
- create_composition: Scaffold from components
Frameworks: hanzo (default), shadcn, react, svelte, vue, react-native
Install:
pip install hanzo-tools-ui
Usage:
from hanzo_tools.ui import register_tools, TOOLS
"""
from hanzo_tools.core import BaseTool, ToolRegistry
from .ui_tool import UiTool, ui_tool
from .github_api import (
FRAMEWORK_CONFIGS,
FRAMEWORK_NAMES,
GitHubAPIClient,
)
TOOLS = [UiTool]
__all__ = [
"UiTool",
"ui_tool",
"FRAMEWORK_CONFIGS",
"FRAMEWORK_NAMES",
"GitHubAPIClient",
"register_tools",
"TOOLS",
]
def register_tools(mcp_server, **kwargs) -> list[BaseTool]:
"""Register UI tools with the MCP server.
Args:
mcp_server: The FastMCP server instance
Returns:
List of registered tool instances
"""
tool = UiTool()
ToolRegistry.register_tool(mcp_server, tool)
return [tool]
pkg/hanzo-tools-ui/hanzo_tools/ui/github_api.py
+327
View File
@@ -0,0 +1,327 @@
"""GitHub API client for fetching UI components from framework repositories."""
import os
import time
from typing import Any
import httpx
GITHUB_API_BASE = "https://api.github.com"
GITHUB_RAW_BASE = "https://raw.githubusercontent.com"
class FrameworkConfig:
"""Configuration for a UI framework repository."""
def __init__(
self,
owner: str,
repo: str,
branch: str,
components_path: str,
extension: str,
blocks_path: str | None = None,
examples_path: str | None = None,
):
self.owner = owner
self.repo = repo
self.branch = branch
self.components_path = components_path
self.blocks_path = blocks_path
self.examples_path = examples_path
self.extension = extension
FRAMEWORK_CONFIGS: dict[str, FrameworkConfig] = {
"hanzo": FrameworkConfig(
owner="hanzoai",
repo="ui",
branch="main",
components_path="packages/ui/src/components",
blocks_path="packages/ui/src/blocks",
extension=".tsx",
),
"hanzo-native": FrameworkConfig(
owner="hanzoai",
repo="ui-native",
branch="main",
components_path="packages/native/src/components",
extension=".tsx",
),
"hanzo-vue": FrameworkConfig(
owner="hanzoai",
repo="ui-vue",
branch="main",
components_path="packages/vue/src/components",
extension=".vue",
),
"hanzo-svelte": FrameworkConfig(
owner="hanzoai",
repo="ui-svelte",
branch="main",
components_path="packages/svelte/src/components",
extension=".svelte",
),
"shadcn": FrameworkConfig(
owner="shadcn-ui",
repo="ui",
branch="main",
components_path="apps/v4/registry/new-york-v4/ui",
blocks_path="apps/v4/registry/new-york-v4/blocks",
examples_path="apps/v4/registry/new-york-v4/examples",
extension=".tsx",
),
"react": FrameworkConfig(
owner="shadcn-ui",
repo="ui",
branch="main",
components_path="apps/v4/registry/new-york-v4/ui",
blocks_path="apps/v4/registry/new-york-v4/blocks",
examples_path="apps/v4/registry/new-york-v4/examples",
extension=".tsx",
),
"svelte": FrameworkConfig(
owner="huntabyte",
repo="shadcn-svelte",
branch="main",
components_path="apps/www/src/lib/registry/new-york/ui",
blocks_path="apps/www/src/lib/registry/new-york/blocks",
extension=".svelte",
),
"vue": FrameworkConfig(
owner="unovue",
repo="shadcn-vue",
branch="main",
components_path="apps/www/src/lib/registry/new-york/ui",
blocks_path="apps/www/src/lib/registry/new-york/blocks",
extension=".vue",
),
"react-native": FrameworkConfig(
owner="founded-labs",
repo="react-native-reusables",
branch="main",
components_path="packages/reusables/src",
extension=".tsx",
),
}
FRAMEWORK_NAMES: dict[str, str] = {
"hanzo": "Hanzo UI (React)",
"hanzo-native": "Hanzo UI Native (React Native)",
"hanzo-vue": "Hanzo UI Vue",
"hanzo-svelte": "Hanzo UI Svelte",
"shadcn": "shadcn/ui",
"react": "shadcn/ui (React)",
"svelte": "Svelte (shadcn)",
"vue": "Vue (shadcn)",
"react-native": "React Native Reusables",
}
class GitHubAPIClient:
"""GitHub API client with caching and rate limit tracking."""
CACHE_TTL = 15 * 60 # 15 minutes
def __init__(self):
self._token = os.environ.get("GITHUB_TOKEN") or os.environ.get(
"GITHUB_PERSONAL_ACCESS_TOKEN"
)
self._cache: dict[str, tuple[Any, float]] = {}
self._rate_limit_remaining = 60
self._rate_limit_reset = time.time()
self._client: httpx.AsyncClient | None = None
async def _get_client(self) -> httpx.AsyncClient:
if self._client is None:
headers = {
"User-Agent": "Hanzo-MCP-UI-Tool",
"Accept": "application/vnd.github.v3+json",
}
if self._token:
headers["Authorization"] = f"token {self._token}"
self._client = httpx.AsyncClient(headers=headers, timeout=30.0)
return self._client
def _cache_get(self, key: str) -> Any | None:
entry = self._cache.get(key)
if entry is None:
return None
data, ts = entry
if time.time() - ts > self.CACHE_TTL:
del self._cache[key]
return None
return data
def _cache_set(self, key: str, data: Any) -> None:
self._cache[key] = (data, time.time())
async def _api_request(self, url: str) -> Any:
if self._rate_limit_remaining <= 0 and time.time() < self._rate_limit_reset:
wait = int(self._rate_limit_reset - time.time())
raise RuntimeError(f"GitHub API rate limit exceeded. Reset in {wait}s.")
cached = self._cache_get(url)
if cached is not None:
return cached
client = await self._get_client()
resp = await client.get(url)
if "x-ratelimit-remaining" in resp.headers:
self._rate_limit_remaining = int(resp.headers["x-ratelimit-remaining"])
if "x-ratelimit-reset" in resp.headers:
self._rate_limit_reset = int(resp.headers["x-ratelimit-reset"])
if resp.status_code == 200:
data = resp.json()
self._cache_set(url, data)
return data
elif resp.status_code == 403:
raise RuntimeError("GitHub API rate limit exceeded or authentication required")
elif resp.status_code == 404:
raise FileNotFoundError("Resource not found")
else:
raise RuntimeError(f"GitHub API error: {resp.status_code}")
async def get_raw_content(
self, owner: str, repo: str, path: str, branch: str
) -> str:
url = f"{GITHUB_RAW_BASE}/{owner}/{repo}/{branch}/{path}"
cached = self._cache_get(url)
if cached is not None:
return cached
client = await self._get_client()
resp = await client.get(url)
if resp.status_code == 200:
self._cache_set(url, resp.text)
return resp.text
elif resp.status_code == 404:
raise FileNotFoundError(f"File not found: {path}")
else:
raise RuntimeError(f"Failed to fetch: {resp.status_code}")
async def get_directory_contents(
self, owner: str, repo: str, path: str, branch: str
) -> list[dict]:
url = f"{GITHUB_API_BASE}/repos/{owner}/{repo}/contents/{path}?ref={branch}"
return await self._api_request(url)
async def fetch_component(self, name: str, framework: str = "hanzo") -> str:
config = FRAMEWORK_CONFIGS[framework]
path = f"{config.components_path}/{name}{config.extension}"
try:
return await self.get_raw_content(config.owner, config.repo, path, config.branch)
except FileNotFoundError:
index_path = f"{config.components_path}/{name}/index{config.extension}"
try:
return await self.get_raw_content(
config.owner, config.repo, index_path, config.branch
)
except FileNotFoundError:
raise FileNotFoundError(
f"Component '{name}' not found in {framework} repository"
)
async def fetch_component_demo(self, name: str, framework: str = "hanzo") -> str:
config = FRAMEWORK_CONFIGS[framework]
if not config.examples_path and framework != "hanzo":
raise RuntimeError(f"Demo/examples not available for {framework}")
demo_path = (
f"{config.components_path}/{name}/demo{config.extension}"
if framework == "hanzo"
else f"{config.examples_path}/{name}-demo{config.extension}"
)
try:
return await self.get_raw_content(config.owner, config.repo, demo_path, config.branch)
except FileNotFoundError:
raise FileNotFoundError(
f"Demo for component '{name}' not found in {framework} repository"
)
async def fetch_component_metadata(self, name: str, framework: str = "hanzo") -> dict:
config = FRAMEWORK_CONFIGS[framework]
path = f"{config.components_path}/{name}/metadata.json"
try:
import json
content = await self.get_raw_content(config.owner, config.repo, path, config.branch)
return json.loads(content)
except (FileNotFoundError, Exception):
return {
"name": name,
"framework": framework,
"extension": config.extension,
"path": f"{config.components_path}/{name}",
}
async def fetch_block(self, name: str, framework: str = "hanzo") -> str:
config = FRAMEWORK_CONFIGS[framework]
if not config.blocks_path:
raise RuntimeError(f"Blocks not available for {framework}")
path = f"{config.blocks_path}/{name}{config.extension}"
try:
return await self.get_raw_content(config.owner, config.repo, path, config.branch)
except FileNotFoundError:
index_path = f"{config.blocks_path}/{name}/index{config.extension}"
try:
return await self.get_raw_content(
config.owner, config.repo, index_path, config.branch
)
except FileNotFoundError:
raise FileNotFoundError(
f"Block '{name}' not found in {framework} repository"
)
async def list_components(self, framework: str = "hanzo") -> list[dict]:
config = FRAMEWORK_CONFIGS[framework]
contents = await self.get_directory_contents(
config.owner, config.repo, config.components_path, config.branch
)
return [
{"name": item["name"].replace(config.extension, ""), "type": item["type"]}
for item in contents
if item["type"] == "dir"
or (item["type"] == "file" and item["name"].endswith(config.extension))
]
async def list_blocks(self, framework: str = "hanzo") -> list[dict]:
config = FRAMEWORK_CONFIGS[framework]
if not config.blocks_path:
return []
try:
contents = await self.get_directory_contents(
config.owner, config.repo, config.blocks_path, config.branch
)
return [
{"name": item["name"].replace(config.extension, ""), "type": item["type"]}
for item in contents
if item["type"] == "dir"
or (item["type"] == "file" and item["name"].endswith(config.extension))
]
except Exception:
return []
async def get_directory_structure(self, path: str, framework: str = "hanzo") -> dict:
config = FRAMEWORK_CONFIGS[framework]
contents = await self.get_directory_contents(
config.owner, config.repo, path, config.branch
)
children = []
for item in contents:
entry = {"name": item["name"], "type": item["type"], "path": item["path"]}
if item["type"] == "file" and "size" in item:
entry["size"] = item["size"]
children.append(entry)
return {"path": path, "children": children}
def get_rate_limit_info(self) -> dict:
return {
"remaining": self._rate_limit_remaining,
"reset": self._rate_limit_reset,
}
def clear_cache(self) -> None:
self._cache.clear()
pkg/hanzo-tools-ui/hanzo_tools/ui/ui_tool.py
+365
View File
@@ -0,0 +1,365 @@
"""Unified UI component registry tool (HIP-0300).
Single 'ui' tool for browsing, searching, installing, and managing
UI components from Hanzo and other registries.
Actions:
- list_components: List available components for a framework
- get_component: Get component source code
- get_demo: Get component demo/example
- get_metadata: Get component metadata
- list_blocks: List UI blocks
- get_block: Get block implementation
- search: Search components by name/description
- get_structure: Browse repository directory structure
- install: Install component via CLI
- set_framework: Switch active framework
- get_framework: Show current framework and available options
- create_composition: Scaffold a composition from components
"""
import asyncio
from typing import Any, ClassVar
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool
from .github_api import (
FRAMEWORK_CONFIGS,
FRAMEWORK_NAMES,
GitHubAPIClient,
)
# Module-level current framework state
_current_framework = "hanzo"
class UiTool(BaseTool):
"""UI component registry tool (HIP-0300).
Browse, search, install, and manage UI components from
Hanzo and other registries (shadcn/ui, Vue, Svelte, React Native).
"""
name: ClassVar[str] = "ui"
VERSION: ClassVar[str] = "0.1.0"
def __init__(self):
super().__init__()
self._client = GitHubAPIClient()
self._register_ui_actions()
@property
def description(self) -> str:
return """UI component registry tool (HIP-0300).
Actions:
- list_components: List available components (framework, category)
- get_component: Get component source code (name, framework)
- get_demo: Get component demo/example (name, framework)
- get_metadata: Get component metadata (name, framework)
- list_blocks: List UI blocks (framework, category)
- get_block: Get block implementation (name, framework)
- search: Search components (query, framework)
- get_structure: Browse repo directory structure (path, framework)
- install: Install component via CLI (name, framework)
- set_framework: Switch active framework (framework)
- get_framework: Show current and available frameworks
- create_composition: Scaffold from components (name, components)
Frameworks: hanzo (default), hanzo-native, hanzo-vue, hanzo-svelte,
shadcn, react, svelte, vue, react-native
"""
def _fw_name(self, framework: str) -> str:
return FRAMEWORK_NAMES.get(framework, framework)
def _register_ui_actions(self):
global _current_framework
@self.action("list_components", "List available components")
async def list_components(
ctx: MCPContext,
framework: str | None = None,
category: str | None = None,
) -> dict:
fw = framework or _current_framework
if fw not in FRAMEWORK_CONFIGS:
return {"error": f"Unknown framework: {fw}. Available: {', '.join(FRAMEWORK_CONFIGS)}"}
components = await self._client.list_components(fw)
if category:
components = [c for c in components if c.get("category") == category]
return {
"framework": self._fw_name(fw),
"total": len(components),
"components": components,
}
@self.action("get_component", "Get component source code")
async def get_component(
ctx: MCPContext,
name: str | None = None,
component: str | None = None,
framework: str | None = None,
) -> dict:
comp_name = name or component
if not comp_name:
return {"error": "Component name is required"}
fw = framework or _current_framework
source = await self._client.fetch_component(comp_name, fw)
return {
"framework": self._fw_name(fw),
"component": comp_name,
"source": source,
}
@self.action("get_demo", "Get component demo/example")
async def get_demo(
ctx: MCPContext,
name: str | None = None,
component: str | None = None,
framework: str | None = None,
) -> dict:
comp_name = name or component
if not comp_name:
return {"error": "Component name is required"}
fw = framework or _current_framework
demo = await self._client.fetch_component_demo(comp_name, fw)
return {
"framework": self._fw_name(fw),
"component": comp_name,
"demo": demo,
}
@self.action("get_metadata", "Get component metadata")
async def get_metadata(
ctx: MCPContext,
name: str | None = None,
component: str | None = None,
framework: str | None = None,
) -> dict:
comp_name = name or component
if not comp_name:
return {"error": "Component name is required"}
fw = framework or _current_framework
metadata = await self._client.fetch_component_metadata(comp_name, fw)
return {
"framework": self._fw_name(fw),
"component": comp_name,
"metadata": metadata,
}
@self.action("list_blocks", "List UI blocks")
async def list_blocks(
ctx: MCPContext,
framework: str | None = None,
category: str | None = None,
) -> dict:
fw = framework or _current_framework
blocks = await self._client.list_blocks(fw)
if category:
blocks = [b for b in blocks if b.get("category") == category]
return {
"framework": self._fw_name(fw),
"total": len(blocks),
"blocks": blocks,
}
@self.action("get_block", "Get block implementation")
async def get_block(
ctx: MCPContext,
name: str | None = None,
block: str | None = None,
framework: str | None = None,
) -> dict:
block_name = name or block
if not block_name:
return {"error": "Block name is required"}
fw = framework or _current_framework
content = await self._client.fetch_block(block_name, fw)
return {
"framework": self._fw_name(fw),
"block": block_name,
"implementation": content,
}
@self.action("search", "Search components")
async def search(
ctx: MCPContext,
query: str | None = None,
search: str | None = None,
framework: str | None = None,
) -> dict:
q = query or search
if not q:
return {"error": "Search query is required"}
fw = framework or _current_framework
components = await self._client.list_components(fw)
q_lower = q.lower()
matches = [
c
for c in components
if q_lower in c.get("name", "").lower()
or q_lower in c.get("description", "").lower()
or q_lower in c.get("category", "").lower()
]
return {
"framework": self._fw_name(fw),
"query": q,
"results": matches,
}
@self.action("get_structure", "Browse repository directory structure")
async def get_structure(
ctx: MCPContext,
path: str = "",
framework: str | None = None,
depth: int = 3,
) -> dict:
fw = framework or _current_framework
structure = await self._client.get_directory_structure(path, fw)
return {
"framework": self._fw_name(fw),
"path": path or "/",
"structure": structure,
}
@self.action("install", "Install component via CLI")
async def install(
ctx: MCPContext,
name: str | None = None,
component: str | None = None,
framework: str | None = None,
overwrite: bool = False,
) -> dict:
comp_name = name or component
if not comp_name:
return {"error": "Component name is required"}
fw = framework or _current_framework
if fw.startswith("hanzo"):
cmd = f"npx @hanzo/ui add {comp_name}"
elif fw in ("shadcn", "react"):
cmd = f"npx shadcn@latest add {comp_name}"
else:
return {"error": f"Installation not supported for framework: {fw}"}
if overwrite:
cmd += " --overwrite"
proc = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
return {
"framework": self._fw_name(fw),
"component": comp_name,
"command": cmd,
"output": stdout.decode() if stdout else "",
"warnings": stderr.decode() if stderr else "",
}
@self.action("set_framework", "Switch active framework")
async def set_framework(ctx: MCPContext, framework: str | None = None) -> dict:
global _current_framework
if not framework:
return {"error": "Framework is required"}
if framework not in FRAMEWORK_CONFIGS:
return {
"error": f"Unknown framework: {framework}. Available: {', '.join(FRAMEWORK_CONFIGS)}"
}
_current_framework = framework
return {
"success": True,
"framework": self._fw_name(framework),
"message": f"Switched to {self._fw_name(framework)}",
}
@self.action("get_framework", "Show current and available frameworks")
async def get_framework(ctx: MCPContext) -> dict:
return {
"current": self._fw_name(_current_framework),
"framework": _current_framework,
"available": [
{
"key": key,
"name": FRAMEWORK_NAMES.get(key, key),
"has_registry": key.startswith("hanzo"),
}
for key in FRAMEWORK_CONFIGS
],
}
@self.action("create_composition", "Scaffold a composition from components")
async def create_composition(
ctx: MCPContext,
name: str | None = None,
components: list[str] | None = None,
description: str | None = None,
framework: str | None = None,
) -> dict:
if not name:
return {"error": "Composition name is required"}
fw = framework or _current_framework
comps = components or []
lines = [f"/**", f" * {name}"]
if description:
lines.append(f" * {description}")
lines.append(f" * Framework: {self._fw_name(fw)}")
lines.append(f" * Components: {', '.join(comps)}")
lines.append(f" */")
lines.append("")
def pascal(s: str) -> str:
return "".join(p.capitalize() for p in s.split("-"))
if fw.startswith("hanzo"):
for comp in comps:
lines.append(f'import {{ {pascal(comp)} }} from "@hanzo/ui/{comp}"')
elif fw in ("shadcn", "react"):
for comp in comps:
lines.append(f'import {{ {pascal(comp)} }} from "@/components/ui/{comp}"')
lines.append("")
lines.append(f"export function {name}() {{")
lines.append(f" return (")
lines.append(f' <div className="container mx-auto p-6">')
for comp in comps:
lines.append(f" <{pascal(comp)} />")
lines.append(f" </div>")
lines.append(f" )")
lines.append(f"}}")
code = "\n".join(lines) + "\n"
return {
"framework": self._fw_name(fw),
"name": name,
"code": code,
"components": comps,
}
# Inherits call() and register() from BaseTool — action routing is automatic
# Singleton reference
ui_tool = UiTool
pkg/hanzo-tools-ui/pyproject.toml
+65
View File
@@ -0,0 +1,65 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "hanzo-tools-ui"
version = "0.1.0"
description = "UI component registry tools for Hanzo AI (HIP-0300)"
readme = "README.md"
license = "MIT"
requires-python = ">=3.12"
authors = [
{ name = "Hanzo AI Team", email = "[email protected]" },
]
keywords = [
"hanzo",
"mcp",
"tools",
"ui",
"components",
"shadcn",
"registry",
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.12",
]
dependencies = [
"hanzo-tools-core>=0.1.0",
"mcp>=1.0.0",
"httpx>=0.25.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.23.0",
"ruff>=0.1.0",
]
[project.entry-points."hanzo.tools"]
ui = "hanzo_tools.ui:TOOLS"
[project.urls]
Homepage = "https://github.com/hanzoai/python-sdk"
Documentation = "https://docs.hanzo.ai"
Repository = "https://github.com/hanzoai/python-sdk"
[tool.hatch.build.targets.wheel]
packages = ["hanzo_tools"]
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "UP"]
ignore = ["E501"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pkg/hanzo-tools-vcs/README.md
View File
pkg/hanzo-tools-vcs/hanzo_tools/vcs/__init__.py
+8 -2
View File
@@ -27,12 +27,18 @@ Usage:
from hanzo_tools.core import BaseTool, ToolRegistry
from .vcs_tool import VcsTool, vcs_tool
from .git_tool import GitTool, git_tool
# Backward compat
VcsTool = GitTool
vcs_tool = git_tool
# Export list for tool discovery - HIP-0300 unified tool
TOOLS = [VcsTool]
TOOLS = [GitTool]
__all__ = [
"GitTool",
"git_tool",
"VcsTool",
"vcs_tool",
"register_tools",
pkg/hanzo-tools-vcs/hanzo_tools/vcs/vcs_tool.py → pkg/hanzo-tools-vcs/hanzo_tools/vcs/git_tool.py
+4 -4
View File
@@ -1,6 +1,6 @@
"""Unified version control tool for HIP-0300 architecture.
This module provides a single unified 'vcs' tool that handles all version control operations:
This module provides a single unified 'git' tool that handles all version control operations:
- status: Working tree status
- diff: Show differences
- apply: Apply patch
@@ -28,7 +28,7 @@ from hanzo_tools.core import (
)
class VcsTool(BaseTool):
class GitTool(BaseTool):
"""Unified version control tool (HIP-0300).
Handles all VCS operations on a single axis:
@@ -43,7 +43,7 @@ class VcsTool(BaseTool):
Outputs diffs in unified patch format.
"""
name: ClassVar[str] = "vcs"
name: ClassVar[str] = "git"
VERSION: ClassVar[str] = "0.12.0"
def __init__(self, cwd: str | None = None):
@@ -463,4 +463,4 @@ Outputs diffs in unified patch format for use with fs.apply_patch.
# Backward compatibility
vcs_tool = VcsTool
git_tool = GitTool
pkg/hanzo-tools/hanzo_tools/core/__init__.py
+29 -6
View File
@@ -1,14 +1,12 @@
"""Core infrastructure for Hanzo tool packages.
Provides:
- BaseTool: Abstract base class for all tools
- ToolRegistry: Tool registration and discovery
- ToolContext: Execution context utilities
- Decorators for timeouts, error handling, etc.
Exports both:
- HIP-0300 unified tool surface (`BaseTool`, `ToolError`, `Paging`, etc.)
- Low-level base abstractions (`BaseToolABC`, `ToolRegistry`, `FileSystemTool`)
"""
from hanzo_tools.core.base import (
BaseTool,
BaseTool as BaseToolABC,
ToolRegistry,
FileSystemTool,
with_error_logging,
@@ -19,12 +17,37 @@ from hanzo_tools.core.context import ToolContext, create_tool_context
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
from hanzo_tools.core.unified import (
BaseTool, # Unified HIP-0300 base class
ActionHandler,
ErrorCode,
ToolError,
ConflictError,
NotFoundError,
InvalidParamsError,
Paging,
Range,
content_hash,
file_uri,
)
__all__ = [
# Base classes
"BaseTool",
"BaseToolABC",
"FileSystemTool",
"ToolRegistry",
# HIP-0300 unified helpers
"ActionHandler",
"ErrorCode",
"ToolError",
"ConflictError",
"NotFoundError",
"InvalidParamsError",
"Paging",
"Range",
"content_hash",
"file_uri",
# Context
"ToolContext",
"create_tool_context",
pkg/hanzo-tools/hanzo_tools/core/unified.py
+449
View File
@@ -0,0 +1,449 @@
"""Base tool class for HIP-0300 architecture.
Provides the foundation for orthogonal, composable tools following Unix philosophy:
- Permissive input, strict output
- Action routing with alias resolution
- Unified response envelope with ok/data/error/meta
- Built-in help and schema actions
- Typed error codes with Unix exit code semantics
- Paging/cursor support for large results
- Composable via stable identifiers (uri, hash, range, ref)
Design principles:
- Input parsing = permissive (accept many spellings/forms)
- Output + semantics = strict (one canonical shape)
- Aliases never appear in output; only canonical action names
Reference: HIP-0300 Unified MCP Tools Architecture
"""
import json
import hashlib
import inspect
from abc import abstractmethod
from typing import Any, Literal, TypeVar, Callable, ClassVar, get_type_hints
from dataclasses import field, dataclass
from collections.abc import Awaitable
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from .base import BaseTool as _BaseToolABC
# Error codes for structured error handling
ErrorCode = Literal[
"UNKNOWN_ACTION", # Action not found
"INVALID_PARAMS", # Invalid parameters
"NOT_FOUND", # Resource not found
"CONFLICT", # Precondition failed (e.g., base_hash mismatch)
"PERMISSION_DENIED", # Access denied
"TIMEOUT", # Operation timed out
"INTERNAL_ERROR", # Unexpected error
]
@dataclass
class Paging:
"""Pagination info for large results."""
cursor: str | None = None
more: bool = False
total: int | None = None
@dataclass
class ToolError(Exception):
"""Structured error for tool operations."""
code: ErrorCode
message: str
details: dict[str, Any] = field(default_factory=dict)
class ConflictError(ToolError):
"""Precondition failed (e.g., base_hash mismatch)."""
def __init__(
self, message: str, expected: str | None = None, actual: str | None = None
):
details = {}
if expected:
details["expected"] = expected
if actual:
details["actual"] = actual
super().__init__(code="CONFLICT", message=message, details=details)
class NotFoundError(ToolError):
"""Resource not found."""
def __init__(self, message: str, uri: str | None = None):
details = {"uri": uri} if uri else {}
super().__init__(code="NOT_FOUND", message=message, details=details)
class InvalidParamsError(ToolError):
"""Invalid parameters."""
def __init__(
self, message: str, param: str | None = None, expected: str | None = None
):
details = {}
if param:
details["param"] = param
if expected:
details["expected"] = expected
super().__init__(code="INVALID_PARAMS", message=message, details=details)
@dataclass
class ActionHandler:
"""Metadata for a registered action handler."""
name: str
handler: Callable[..., Awaitable[Any]]
description: str
schema: dict[str, Any] | None = None
examples: list[str] = field(default_factory=list)
class BaseTool(_BaseToolABC):
"""Base class for HIP-0300 unified tools with action routing.
Design principles:
- Permissive input, strict output
- Action routing with alias resolution
- Unified response envelope: {ok, data, error, meta}
- Aliases never appear in output; only canonical action names
Usage:
class FsTool(BaseTool):
name = "fs"
description = "Filesystem operations"
def __init__(self):
super().__init__()
self._register_actions()
def _register_actions(self):
@self.action("read", "Read file contents")
async def read(ctx, uri: str, range: dict | None = None) -> dict:
content = await read_file(uri)
return {"text": content, "hash": hash_content(content)}
@self.action("apply_patch", "Edit file with precondition")
async def apply_patch(ctx, uri: str, patch: str, base_hash: str) -> dict:
if get_hash(uri) != base_hash:
raise ConflictError("base_hash mismatch", expected=base_hash)
# Apply patch...
return {"uri": uri, "hash": new_hash}
"""
# Subclasses must define these
name: ClassVar[str]
# Version for meta envelope
VERSION: ClassVar[str] = "0.12.0"
def __init__(self):
self._handlers: dict[str, ActionHandler] = {}
self._register_builtin_actions()
def _register_builtin_actions(self):
"""Register built-in help and schema actions."""
@self.action("help", "List available actions with descriptions")
async def help_action(ctx: MCPContext) -> dict:
actions = {}
for name, handler in self._handlers.items():
if name not in ("help", "schema", "status"):
actions[name] = {
"description": handler.description,
"examples": handler.examples,
}
return {"actions": actions, "tool": self.name}
@self.action("schema", "Get JSON Schema for action parameters")
async def schema_action(
ctx: MCPContext, action_name: str | None = None
) -> dict:
if action_name:
handler = self._handlers.get(action_name)
if not handler:
raise ToolError(
code="UNKNOWN_ACTION",
message=f"Unknown action: {action_name}",
details={"available": list(self._handlers.keys())},
)
return {"action": action_name, "schema": handler.schema or {}}
schemas = {}
for name, handler in self._handlers.items():
if name not in ("help", "schema", "status"):
schemas[name] = handler.schema or {}
return {"schemas": schemas}
@self.action("status", "Get tool status and version")
async def status_action(ctx: MCPContext) -> dict:
return {
"tool": self.name,
"version": self.VERSION,
"enabled": True,
"actions": list(self._handlers.keys()),
}
def action(
self,
name: str,
description: str = "",
schema: dict[str, Any] | None = None,
examples: list[str] | None = None,
) -> Callable:
"""Decorator to register an action handler.
Args:
name: Action name (e.g., "read", "apply_patch")
description: Human-readable description
schema: Optional JSON Schema for parameters
examples: Optional list of example usages
Returns:
Decorator function
"""
def decorator(
fn: Callable[..., Awaitable[Any]],
) -> Callable[..., Awaitable[Any]]:
# Auto-generate schema from type hints if not provided
auto_schema = schema
if auto_schema is None:
auto_schema = self._generate_schema(fn)
self._handlers[name] = ActionHandler(
name=name,
handler=fn,
description=description or fn.__doc__ or "",
schema=auto_schema,
examples=examples or [],
)
return fn
return decorator
def _generate_schema(self, fn: Callable) -> dict[str, Any]:
"""Generate JSON Schema from function signature and type hints."""
try:
hints = get_type_hints(fn)
except Exception:
hints = {}
sig = inspect.signature(fn)
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name in ("self", "ctx"):
continue
param_schema: dict[str, Any] = {}
hint = hints.get(param_name)
if hint is str:
param_schema["type"] = "string"
elif hint is int:
param_schema["type"] = "integer"
elif hint is float:
param_schema["type"] = "number"
elif hint is bool:
param_schema["type"] = "boolean"
elif hint is dict or (
hasattr(hint, "__origin__") and hint.__origin__ is dict
):
param_schema["type"] = "object"
elif hint is list or (
hasattr(hint, "__origin__") and hint.__origin__ is list
):
param_schema["type"] = "array"
else:
param_schema["type"] = "string" # Default to string
properties[param_name] = param_schema
if param.default is inspect.Parameter.empty:
required.append(param_name)
return {
"type": "object",
"properties": properties,
"required": required,
}
def _envelope(
self,
data: Any,
action: str | None = None,
paging: Paging | None = None,
backend: str | None = None,
trace_id: str | None = None,
) -> dict[str, Any]:
"""Wrap result in unified response envelope."""
meta: dict[str, Any] = {
"tool": self.name,
"version": self.VERSION,
}
if action:
meta["action"] = action
if backend:
meta["backend"] = backend
if trace_id:
meta["trace_id"] = trace_id
if paging:
meta["paging"] = {
"cursor": paging.cursor,
"more": paging.more,
}
if paging.total is not None:
meta["paging"]["total"] = paging.total
else:
meta["paging"] = {"cursor": None, "more": False}
return {
"ok": True,
"data": data,
"error": None,
"meta": meta,
}
def _error(
self,
code: ErrorCode,
message: str,
**details: Any,
) -> dict[str, Any]:
"""Create error response envelope."""
error_dict: dict[str, Any] = {
"code": code,
"message": message,
}
if details:
error_dict.update(details)
return {
"ok": False,
"data": None,
"error": error_dict,
"meta": {"tool": self.name, "version": self.VERSION},
}
@property
@abstractmethod
def description(self) -> str:
"""Tool description for MCP registration."""
pass
async def call(
self, ctx: MCPContext, action: str = "help", **kwargs: Any
) -> dict[str, Any]:
"""Execute tool with action routing.
Args:
ctx: MCP context
action: Action to execute (default: "help")
**kwargs: Action parameters
Returns:
Unified response envelope
"""
if action not in self._handlers:
return self._error(
"UNKNOWN_ACTION",
f"Unknown action: {action}",
available=list(self._handlers.keys()),
)
handler = self._handlers[action]
try:
result = await handler.handler(ctx, **kwargs)
return self._envelope(result, action=action)
except ToolError as e:
return self._error(e.code, e.message, **e.details)
except Exception as e:
return self._error("INTERNAL_ERROR", str(e))
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server.
Creates a single MCP tool with action parameter for routing.
"""
tool_name = self.name
tool_description = (
f"{self.description}\n\nActions: {', '.join(self._handlers.keys())}"
)
@mcp_server.tool(name=tool_name, description=tool_description)
async def handler(
ctx: MCPContext,
action: str = "help",
**kwargs: Any,
) -> str:
result = await self.call(ctx, action=action, **kwargs)
return json.dumps(result, indent=2, default=str)
# Utility functions for composability
def content_hash(content: str | bytes, algorithm: str = "sha256") -> str:
"""Generate content hash for file identity."""
if isinstance(content, str):
content = content.encode("utf-8")
h = hashlib.new(algorithm)
h.update(content)
return f"{algorithm}:{h.hexdigest()}"
def file_uri(path: str) -> str:
"""Convert path to file:// URI."""
from pathlib import Path
abs_path = Path(path).resolve()
return f"file://{abs_path}"
@dataclass
class Range:
"""Text range for composability.
0-based line and column numbers.
"""
start_line: int
start_col: int = 0
end_line: int | None = None
end_col: int | None = None
def to_dict(self) -> dict[str, Any]:
"""Convert to dict format for serialization."""
result = {
"start": {"line": self.start_line, "col": self.start_col},
}
if self.end_line is not None:
result["end"] = {
"line": self.end_line,
"col": self.end_col or 0,
}
return result
@classmethod
def from_dict(cls, d: dict[str, Any]) -> "Range":
"""Create Range from dict."""
start = d.get("start", {})
end = d.get("end")
return cls(
start_line=start.get("line", 0),
start_col=start.get("col", 0),
end_line=end.get("line") if end else None,
end_col=end.get("col") if end else None,
)
pkg/hanzo/src/hanzo/commands/auth.py
+3 -3
View File
@@ -230,7 +230,7 @@ def _login_browser_oauth(auth_mgr: AuthManager):
"scope": "openid profile email",
"state": state,
}
authorize_url = f"{iam_url}/login/oauth/authorize?{urlencode(params)}"
authorize_url = f"{iam_url}/oauth/authorize?{urlencode(params)}"
console.print("Your browser has been opened to visit:\n")
console.print(f" {authorize_url}\n")
@@ -250,7 +250,7 @@ def _login_browser_oauth(auth_mgr: AuthManager):
return
# Exchange authorization code for tokens
token_url = f"{iam_url}/api/login/oauth/access_token"
token_url = f"{iam_url}/oauth/token"
token_data = urlencode(
{
"client_id": HANZO_CLIENT_ID,
@@ -364,7 +364,7 @@ def _login_device_code(auth_mgr: AuthManager, headless: bool):
).encode()
poll_req = urllib.request.Request( # noqa: S310
f"{iam_url}/api/login/oauth/access_token",
f"{iam_url}/oauth/token",
data=poll_data,
headers={"Content-Type": "application/json"},
method="POST",
pkg/hanzo/src/hanzo/commands/iam.py
+2 -2
View File
@@ -1198,7 +1198,7 @@ def iam_login(username: str, password: str, org: str, app: str):
try:
with httpx.Client(timeout=30.0) as http:
token_resp = http.post(
f"{url}/api/login/oauth/access_token",
f"{url}/oauth/token",
data={
"grant_type": "password",
"client_id": client_id,
@@ -1346,7 +1346,7 @@ def tokens_exchange(code: str, redirect_uri: str):
try:
with httpx.Client(timeout=30.0) as http:
resp = http.post(
f"{url}/api/login/oauth/access_token",
f"{url}/oauth/token",
data={
"grant_type": "authorization_code",
"client_id": client_id,
pkg/hanzoai/auth.py
+2 -2
View File
@@ -224,7 +224,7 @@ class HanzoAuth:
"state": state,
}
sso_url = f"{self.base_url}/login/oauth/authorize?{urlencode(params)}"
sso_url = f"{self.base_url}/oauth/authorize?{urlencode(params)}"
# Open browser
webbrowser.open(sso_url)
@@ -265,7 +265,7 @@ class HanzoAuth:
# Exchange code for token
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/api/login/oauth/access_token",
f"{self.base_url}/oauth/token",
json={
"client_id": "app-hanzo",
"client_secret": "hanzo-cli-secret", # Public client
pkg/hanzoai/llm_client.py
+2 -2
View File
@@ -21,7 +21,7 @@ from ._client import Hanzo
# Hanzo AI specific configuration
HANZO_API_BASE = "https://api.hanzo.ai/v1"
HANZO_IAM_BASE = "https://iam.hanzo.ai"
HANZO_IAM_BASE = "https://hanzo.id"
# Check for Hanzo API key
HANZO_API_KEY = os.getenv("HANZO_API_KEY")
@@ -99,7 +99,7 @@ class SimpleLLMClient:
"state": "hanzoai-login",
}
auth_url = (
f"{HANZO_IAM_BASE}/login/oauth/authorize?{urlencode(auth_params)}"
f"{HANZO_IAM_BASE}/oauth/authorize?{urlencode(auth_params)}"
)
print(f"Opening browser for Hanzo AI login...")
pyproject.toml
+2 -2
View File
@@ -48,7 +48,7 @@ dev-dependencies = [
"mypy",
"respx",
"pytest",
"pytest-asyncio",
"pytest-asyncio>=0.26.0,<1.0.0",
"ruff",
"time-machine",
"nox",
@@ -117,7 +117,7 @@ dev = [
"httpx>=0.28.1",
"polars>=1.36.1",
"pytest>=8.4.2",
"pytest-asyncio>=1.2.0",
"pytest-asyncio>=0.26.0,<1.0.0",
]
[tool.hatch.build]
uv.lock
Generated
+8 -8
View File
@@ -886,6 +886,7 @@ dependencies = [
{ name = "tiktoken" },
{ name = "tokenizers" },
]
sdist = { url = "https://files.pythonhosted.org/packages/03/33/7464f7ce2376374ac3dc080cf38ded49f53e4e8e3d5786038faa745349ab/hanzo_llm-1.81.15.tar.gz", hash = "sha256:f9a6c1c2c83f0cfd67856a61f8b1e22825a6d02c60e7a364fd3ce1c193798366", size = 16520951, upload-time = "2026-03-02T15:15:04.283Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/37/15/3cd25b2908b6ec404d073ca57bbe1580e332a0ee094d0a33c36cd32d2e22/hanzo_llm-1.81.15-py3-none-any.whl", hash = "sha256:be3849567dc6a59b711a34f85bff1a74ad417df322cc38f3aebd16f9fa29823c", size = 14677430, upload-time = "2026-02-26T04:25:46.164Z" },
]
@@ -1051,7 +1052,7 @@ dev = [
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "polars", specifier = ">=1.36.1" },
{ name = "pytest", specifier = ">=8.4.2" },
{ name = "pytest-asyncio", specifier = ">=1.2.0" },
{ name = "pytest-asyncio", specifier = ">=0.26.0,<1.0.0" },
]
[[package]]
@@ -2683,7 +2684,7 @@ wheels = [
[[package]]
name = "pytest"
version = "9.0.2"
version = "8.4.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
@@ -2692,22 +2693,21 @@ dependencies = [
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" }
sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618, upload-time = "2025-09-04T14:34:22.711Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" },
{ url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750, upload-time = "2025-09-04T14:34:20.226Z" },
]
[[package]]
name = "pytest-asyncio"
version = "1.3.0"
version = "0.26.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" }
sdist = { url = "https://files.pythonhosted.org/packages/8e/c4/453c52c659521066969523e87d85d54139bbd17b78f09532fb8eb8cdb58e/pytest_asyncio-0.26.0.tar.gz", hash = "sha256:c4df2a697648241ff39e7f0e4a73050b03f123f760673956cf0d72a4990e312f", size = 54156, upload-time = "2025-03-25T06:22:28.883Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
{ url = "https://files.pythonhosted.org/packages/20/7f/338843f449ace853647ace35870874f69a764d251872ed1b4de9f234822c/pytest_asyncio-0.26.0-py3-none-any.whl", hash = "sha256:7b51ed894f4fbea1340262bdae5135797ebbe21d8638978e35d31c6d19f72fb0", size = 19694, upload-time = "2025-03-25T06:22:27.807Z" },
]
[[package]]