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

refactor: merge hanzo-tools-core into hanzo-tools

Browse Source 
- hanzo-tools now contains core infrastructure (BaseTool, ToolRegistry, etc.)
- hanzo-tools-core becomes a thin backwards-compat wrapper
- All tool packages now depend on hanzo-tools>=0.3.0
- Reduces package count and simplifies dependency tree
This commit is contained in:
zeekay
2025-12-26 16:53:58 -08:00
parent d59642ca98
commit efbac9ba1f
28 changed files with 866 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-mcp/pyproject.toml
+3 -3
View File
@@ -25,7 +25,7 @@ dependencies = [
"typing-extensions>=4.13.0",
"aiofiles>=23.2.1",
# Hanzo tool infrastructure (all tools)
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"hanzo-tools-filesystem>=0.1.0",
"hanzo-tools-shell>=0.1.0",
"hanzo-tools-browser>=0.1.0",
@@ -42,7 +42,7 @@ dependencies = [
]
# For local development, install hanzo-tools-* packages from pkg/
# pip install -e ../hanzo-tools-core ../hanzo-tools-filesystem etc.
# pip install -e ../hanzo-tools ../hanzo-tools-filesystem etc.
# Or when published, add them back as dependencies
[project.urls]
@@ -52,7 +52,7 @@ dependencies = [
[project.optional-dependencies]
# NOTE: hanzo-tools-* packages are installed separately from pkg/
# Use: pip install -e ../hanzo-tools-core -e ../hanzo-tools-filesystem etc.
# Use: pip install -e ../hanzo-tools -e ../hanzo-tools-filesystem etc.
# When published to PyPI, these extras will work:
# filesystem = ["hanzo-tools-filesystem>=0.1.0"]
# shell = ["hanzo-tools-shell>=0.1.0"]
pkg/hanzo-tools-agent/pyproject.toml
+1 -1
View File
@@ -17,7 +17,7 @@ classifiers = [
]
keywords = ["hanzo", "tools", "agent", "swarm", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-browser/pyproject.toml
+1 -1
View File
@@ -12,7 +12,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "browser", "playwright", "mcp", "ai", "automation"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"pydantic>=2.12.5",
]
pkg/hanzo-tools-computer/pyproject.toml
+1 -1
View File
@@ -23,7 +23,7 @@ classifiers = [
"Programming Language :: Python :: 3.13",
]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"pyautogui>=0.9.54",
"pillow>=10.0.0",
"mcp>=1.25.0",
pkg/hanzo-tools-config/pyproject.toml
+1 -1
View File
@@ -11,7 +11,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "config", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-core/hanzo_tools/core/__init__.py
+18 -24
View File
@@ -1,42 +1,36 @@
"""Core infrastructure for Hanzo tool packages.
"""DEPRECATED: Use hanzo-tools instead.
Provides:
- BaseTool: Abstract base class for all tools
- ToolRegistry: Tool registration and discovery
- ToolContext: Execution context utilities
- Decorators for timeouts, error handling, etc.
This package re-exports from hanzo_tools.core for backwards compatibility.
"""
import warnings
warnings.warn(
"hanzo-tools-core is deprecated. Use hanzo-tools instead: pip install hanzo-tools",
DeprecationWarning,
stacklevel=2,
)
# Re-export everything from hanzo-tools
from hanzo_tools.core.base import (
BaseTool,
ToolRegistry,
FileSystemTool,
with_error_logging,
handle_connection_errors,
ToolRegistry,
MCPResourceDocument,
)
from hanzo_tools.core.types import MCPResourceDocument
from hanzo_tools.core.permissions import PermissionManager
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.types import ToolCategory
__all__ = [
# Base classes
"BaseTool",
"FileSystemTool",
"ToolRegistry",
# Context
"MCPResourceDocument",
"PermissionManager",
"ToolContext",
"create_tool_context",
# Permissions
"PermissionManager",
# Decorators
"auto_timeout",
"with_error_logging",
"handle_connection_errors",
# Validation
"ValidationResult",
"validate_path_parameter",
# Types
"MCPResourceDocument",
"ToolCategory",
]
pkg/hanzo-tools-core/pyproject.toml
+4 -15
View File
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools-core"
version = "0.2.0"
description = "Core infrastructure for Hanzo tool packages"
version = "0.3.0"
description = "DEPRECATED: Use hanzo-tools instead. This package re-exports from hanzo-tools for backwards compatibility."
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
@@ -16,26 +16,15 @@ classifiers = [
"Operating System :: OS Independent",
]
keywords = ["hanzo", "tools", "mcp", "ai"]
# Just depends on hanzo-tools - re-exports everything
dependencies = [
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
"typing-extensions>=4.13.0",
"hanzo-tools>=0.3.0",
]
[project.urls]
"Homepage" = "https://github.com/hanzoai/python-sdk"
"Bug Tracker" = "https://github.com/hanzoai/python-sdk/issues"
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"ruff>=0.14.0",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["hanzo_tools*"]
[tool.setuptools.package-data]
hanzo_tools = ["py.typed"]
pkg/hanzo-tools-database/pyproject.toml
+1 -1
View File
@@ -17,7 +17,7 @@ classifiers = [
]
keywords = ["hanzo", "tools", "database", "sql", "graph", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-editor/pyproject.toml
+1 -1
View File
@@ -17,7 +17,7 @@ classifiers = [
]
keywords = ["hanzo", "tools", "editor", "neovim", "vscode", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-filesystem/pyproject.toml
+2 -2
View File
@@ -26,8 +26,8 @@ dependencies = [
"pydantic>=2.12.5",
]
# For development, install hanzo-tools-core from local:
# uv pip install -e ../hanzo-tools-core
# For development, install hanzo-tools from local:
# uv pip install -e ../hanzo-tools
[project.urls]
"Homepage" = "https://github.com/hanzoai/python-sdk"
pkg/hanzo-tools-jupyter/pyproject.toml
+1 -1
View File
@@ -17,7 +17,7 @@ classifiers = [
]
keywords = ["hanzo", "tools", "jupyter", "notebook", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-llm/pyproject.toml
+1 -1
View File
@@ -11,7 +11,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "llm", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"hanzo-consensus>=0.1.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
pkg/hanzo-tools-lsp/pyproject.toml
+1 -1
View File
@@ -12,7 +12,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "lsp", "language-server", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"pydantic>=2.12.5",
]
pkg/hanzo-tools-mcp/pyproject.toml
+1 -1
View File
@@ -11,7 +11,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-memory/pyproject.toml
+1 -1
View File
@@ -12,7 +12,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "memory", "knowledge", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"pydantic>=2.12.5",
]
pkg/hanzo-tools-reasoning/pyproject.toml
+1 -1
View File
@@ -12,7 +12,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "reasoning", "thinking", "critic", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"pydantic>=2.12.5",
]
pkg/hanzo-tools-refactor/pyproject.toml
+1 -1
View File
@@ -17,7 +17,7 @@ classifiers = [
]
keywords = ["hanzo", "tools", "refactor", "lsp", "ast", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools-shell/pyproject.toml
+1 -1
View File
@@ -12,7 +12,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "shell", "mcp", "ai", "dag", "zsh"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"pydantic>=2.12.5",
"aiofiles>=24.0.0",
pkg/hanzo-tools-todo/pyproject.toml
+1 -1
View File
@@ -12,7 +12,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "todo", "task", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"pydantic>=2.12.5",
]
pkg/hanzo-tools-vector/pyproject.toml
+1 -1
View File
@@ -11,7 +11,7 @@ license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "vector", "embeddings", "rag", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-tools>=0.3.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
pkg/hanzo-tools/hanzo_tools/core/__init__.py
+42
View File
@@ -0,0 +1,42 @@
"""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.
"""
from hanzo_tools.core.base import (
BaseTool,
ToolRegistry,
FileSystemTool,
with_error_logging,
handle_connection_errors,
)
from hanzo_tools.core.types import MCPResourceDocument
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
__all__ = [
# Base classes
"BaseTool",
"FileSystemTool",
"ToolRegistry",
# Context
"ToolContext",
"create_tool_context",
# Permissions
"PermissionManager",
# Decorators
"auto_timeout",
"with_error_logging",
"handle_connection_errors",
# Validation
"ValidationResult",
"validate_path_parameter",
# Types
"MCPResourceDocument",
]
pkg/hanzo-tools/hanzo_tools/core/base.py
+244
View File
@@ -0,0 +1,244 @@
"""Base classes for Hanzo tool packages.
Provides the foundation for all tool implementations:
- BaseTool: Abstract base class defining the tool interface
- FileSystemTool: Base class for filesystem operations
- ToolRegistry: Central registry for tool management
"""
import inspect
import logging
import functools
from abc import ABC, abstractmethod
from typing import Any, Callable, ClassVar, final
from pathlib import Path
from collections.abc import Awaitable
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
logger = logging.getLogger(__name__)
def with_error_logging(tool_name: str) -> Callable:
"""Decorator to add comprehensive error logging to tool functions.
Args:
tool_name: Name of the tool for logging purposes
Returns:
Decorator function
"""
log_dir = Path.home() / ".hanzo" / "mcp" / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
def decorator(func: Callable[..., Awaitable[str]]) -> Callable[..., Awaitable[str]]:
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> str:
try:
return await func(*args, **kwargs)
except TypeError as e:
error_msg = str(e)
if "takes" in error_msg and "positional argument" in error_msg:
sig = inspect.signature(func)
logger.error(
f"Tool {tool_name} call signature mismatch: "
f"expected {func.__name__}{sig}, got args={args}, kwargs={kwargs}"
)
logger.exception(f"Tool {tool_name} TypeError: {e}")
return (
f"Error executing tool '{tool_name}': {error_msg}\n\nCheck logs at ~/.hanzo/mcp/logs/ for details."
)
except Exception as e:
logger.exception(f"Tool {tool_name} error: {e}")
return f"Error executing tool '{tool_name}': {str(e)}\n\nCheck logs at ~/.hanzo/mcp/logs/ for details."
return wrapper
return decorator
def handle_connection_errors(
func: Callable[..., Awaitable[str]],
) -> Callable[..., Awaitable[str]]:
"""Decorator to handle connection errors gracefully."""
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> str:
try:
return await func(*args, **kwargs)
except Exception as e:
error_name = type(e).__name__
if any(name in error_name for name in ["ClosedResourceError", "ConnectionError", "BrokenPipeError"]):
return f"Client disconnected during operation: {error_name}"
raise
return wrapper
class BaseTool(ABC):
"""Abstract base class for all Hanzo tools.
All tool packages must implement this interface to be compatible
with the hanzo-mcp server and tool registry.
Example:
class MyTool(BaseTool):
@property
def name(self) -> str:
return "my_tool"
@property
def description(self) -> str:
return "Does something useful"
async def call(self, ctx, **params) -> str:
return "Result"
def register(self, mcp_server):
@mcp_server.tool()
async def my_tool(...):
return await self.call(...)
"""
@property
@abstractmethod
def name(self) -> str:
"""Get the tool name as it appears in the MCP server."""
pass
@property
@abstractmethod
def description(self) -> str:
"""Get detailed description of the tool's purpose and usage."""
pass
@abstractmethod
async def call(self, ctx: MCPContext, **params: Any) -> Any:
"""Execute the tool with the given parameters.
Args:
ctx: MCP context for the tool call
**params: Tool parameters provided by the caller
Returns:
Tool execution result
"""
pass
@abstractmethod
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server.
Must create a wrapper function with explicit parameters
that calls this tool's call method.
Args:
mcp_server: The FastMCP server instance
"""
pass
class FileSystemTool(BaseTool, ABC):
"""Base class for filesystem-related tools.
Provides common functionality for working with files and directories,
including permission checking and path validation.
"""
def __init__(self, permission_manager: "PermissionManager | None" = None) -> None:
"""Initialize filesystem tool.
Args:
permission_manager: Permission manager for access control (auto-created if None)
"""
if permission_manager is None:
from hanzo_tools.core.permissions import PermissionManager
permission_manager = PermissionManager()
self.permission_manager = permission_manager
def validate_path(self, path: str, param_name: str = "path") -> "ValidationResult":
"""Validate a path parameter."""
from hanzo_tools.core.validation import validate_path_parameter
return validate_path_parameter(path, param_name)
def is_path_allowed(self, path: str) -> bool:
"""Check if a path is allowed according to permission settings."""
return self.permission_manager.is_path_allowed(path)
@final
class ToolRegistry:
"""Registry for Hanzo tools.
Provides functionality for registering tool implementations
with an MCP server, with support for enable/disable states.
"""
# Class-level storage for tool states
_enabled_tools: ClassVar[dict[str, bool]] = {}
_config_loaded: ClassVar[bool] = False
@classmethod
def _load_config(cls) -> None:
"""Load tool enable/disable states from config."""
if cls._config_loaded:
return
import json
config_file = Path.home() / ".hanzo" / "mcp" / "tool_states.json"
if config_file.exists():
try:
with open(config_file) as f:
cls._enabled_tools = json.load(f)
except Exception:
pass
cls._config_loaded = True
@classmethod
def is_tool_enabled(cls, tool_name: str) -> bool:
"""Check if a tool is enabled."""
cls._load_config()
return cls._enabled_tools.get(tool_name, True) # Enabled by default
@classmethod
def set_tool_enabled(cls, tool_name: str, enabled: bool, persist: bool = True) -> None:
"""Enable or disable a tool."""
import json
cls._load_config()
cls._enabled_tools[tool_name] = enabled
if persist:
config_file = Path.home() / ".hanzo" / "mcp" / "tool_states.json"
config_file.parent.mkdir(parents=True, exist_ok=True)
with open(config_file, "w") as f:
json.dump(cls._enabled_tools, f, indent=2)
@staticmethod
def register_tool(mcp_server: FastMCP, tool: BaseTool) -> None:
"""Register a tool with the MCP server.
Args:
mcp_server: The FastMCP server instance
tool: The tool to register
"""
if ToolRegistry.is_tool_enabled(tool.name):
tool.register(mcp_server)
logger.debug(f"Registered tool: {tool.name}")
else:
logger.debug(f"Skipped disabled tool: {tool.name}")
@staticmethod
def register_tools(mcp_server: FastMCP, tools: list[BaseTool]) -> None:
"""Register multiple tools with the MCP server."""
for tool in tools:
ToolRegistry.register_tool(mcp_server, tool)
# Import PermissionManager type for type hints
from hanzo_tools.core.validation import ValidationResult
from hanzo_tools.core.permissions import PermissionManager
pkg/hanzo-tools/hanzo_tools/core/context.py
+67
View File
@@ -0,0 +1,67 @@
"""Tool execution context utilities."""
import logging
from typing import Any, Optional
from dataclasses import field, dataclass
from mcp.server.fastmcp import Context as MCPContext
logger = logging.getLogger(__name__)
@dataclass
class ToolContext:
"""Extended context for tool execution.
Provides utilities for logging, progress reporting,
and accessing the MCP context.
"""
mcp_ctx: MCPContext
tool_name: Optional[str] = None
metadata: dict[str, Any] = field(default_factory=dict)
async def set_tool_info(self, tool_name: str) -> None:
"""Set the current tool name for logging."""
self.tool_name = tool_name
async def info(self, message: str) -> None:
"""Log an info message."""
logger.info(f"[{self.tool_name or 'tool'}] {message}")
async def warning(self, message: str) -> None:
"""Log a warning message."""
logger.warning(f"[{self.tool_name or 'tool'}] {message}")
async def error(self, message: str) -> None:
"""Log an error message."""
logger.error(f"[{self.tool_name or 'tool'}] {message}")
async def debug(self, message: str) -> None:
"""Log a debug message."""
logger.debug(f"[{self.tool_name or 'tool'}] {message}")
async def progress(self, current: int, total: int, message: str = "") -> None:
"""Report progress."""
pct = (current / total * 100) if total > 0 else 0
await self.info(f"Progress: {current}/{total} ({pct:.1f}%) {message}")
def get(self, key: str, default: Any = None) -> Any:
"""Get a metadata value."""
return self.metadata.get(key, default)
def set(self, key: str, value: Any) -> None:
"""Set a metadata value."""
self.metadata[key] = value
def create_tool_context(mcp_ctx: MCPContext) -> ToolContext:
"""Create a ToolContext from an MCP context.
Args:
mcp_ctx: The MCP context from the tool call
Returns:
Extended ToolContext for tool execution
"""
return ToolContext(mcp_ctx=mcp_ctx)
pkg/hanzo-tools/hanzo_tools/core/decorators.py
+103
View File
@@ -0,0 +1,103 @@
"""Decorators for tool functions."""
import os
import asyncio
import functools
from typing import Any, Callable, Optional
from collections.abc import Awaitable
# Default timeouts per tool (can be overridden via env vars)
DEFAULT_TIMEOUTS: dict[str, int] = {
"read": 30,
"write": 60,
"edit": 60,
"search": 120,
"dag": 600,
"browser": 300,
"default": 120,
}
def get_timeout(tool_name: str) -> int:
"""Get timeout for a tool.
Checks environment variable HANZO_TIMEOUT_{TOOL_NAME} first,
then falls back to defaults.
"""
env_key = f"HANZO_TIMEOUT_{tool_name.upper()}"
if env_val := os.environ.get(env_key):
try:
return int(env_val)
except ValueError:
pass
return DEFAULT_TIMEOUTS.get(tool_name, DEFAULT_TIMEOUTS["default"])
def auto_timeout(
tool_name: str,
timeout: Optional[int] = None,
) -> Callable:
"""Decorator to add automatic timeout to async tool functions.
Args:
tool_name: Name of the tool (for logging and config)
timeout: Override timeout in seconds (optional)
Returns:
Decorator that wraps the function with timeout handling
"""
def decorator(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
effective_timeout = timeout or get_timeout(tool_name)
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=effective_timeout,
)
except asyncio.TimeoutError:
return f"Tool '{tool_name}' timed out after {effective_timeout}s"
return wrapper
return decorator
def retry(
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: tuple = (Exception,),
) -> Callable:
"""Decorator to add retry logic to async functions.
Args:
max_attempts: Maximum number of attempts
delay: Initial delay between retries (seconds)
backoff: Multiplier for delay after each attempt
exceptions: Tuple of exceptions to catch and retry
"""
def decorator(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
current_delay = delay
last_exception = None
for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
await asyncio.sleep(current_delay)
current_delay *= backoff
raise last_exception
return wrapper
return decorator
pkg/hanzo-tools/hanzo_tools/core/permissions.py
+84
View File
@@ -0,0 +1,84 @@
"""Permission management for filesystem tools."""
import os
from typing import Optional
from pathlib import Path
class PermissionManager:
"""Manages filesystem permissions for tools.
Controls which paths tools are allowed to access.
"""
def __init__(
self,
allowed_paths: Optional[list[str | Path]] = None,
deny_patterns: Optional[list[str]] = None,
):
"""Initialize permission manager.
Args:
allowed_paths: List of allowed base paths (defaults to cwd)
deny_patterns: Patterns to deny (e.g., '.git', 'node_modules')
"""
self.allowed_paths: list[Path] = []
if allowed_paths:
for p in allowed_paths:
self.allowed_paths.append(Path(p).resolve())
else:
self.allowed_paths.append(Path.cwd())
self.deny_patterns = deny_patterns or [
".git",
"__pycache__",
".pyc",
"node_modules",
".env",
".secrets",
]
def is_path_allowed(self, path: str | Path) -> bool:
"""Check if a path is allowed.
Args:
path: Path to check
Returns:
True if path is within allowed paths and not denied
"""
try:
resolved = Path(path).resolve()
# Check if path is under any allowed path
is_under_allowed = any(self._is_subpath(resolved, allowed) for allowed in self.allowed_paths)
if not is_under_allowed:
return False
# Check deny patterns
path_str = str(resolved)
for pattern in self.deny_patterns:
if pattern in path_str:
return False
return True
except Exception:
return False
def _is_subpath(self, path: Path, parent: Path) -> bool:
"""Check if path is under parent."""
try:
path.relative_to(parent)
return True
except ValueError:
return False
def add_allowed_path(self, path: str | Path) -> None:
"""Add a path to the allowed list."""
self.allowed_paths.append(Path(path).resolve())
def add_deny_pattern(self, pattern: str) -> None:
"""Add a deny pattern."""
self.deny_patterns.append(pattern)
pkg/hanzo-tools/hanzo_tools/core/types.py
+155
View File
@@ -0,0 +1,155 @@
"""Type definitions for Hanzo tool packages."""
import json
from typing import Any, Dict, List, Optional
from dataclasses import dataclass
@dataclass
class MCPResourceDocument:
"""Resource document returned by MCP tools.
Output format options:
- to_json_string(): Clean JSON format (default for structured data)
- to_readable_string(): Human-readable formatted text for display
- to_dict(): Full dict structure with data/metadata
"""
data: Dict[str, Any]
metadata: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary format with data/metadata structure."""
result = {"data": self.data}
if self.metadata:
result["metadata"] = self.metadata
return result
def to_json_string(self) -> str:
"""Convert to clean JSON string."""
# Return wrapped in "result" for consistency
return json.dumps({"result": self.data}, indent=2)
def to_readable_string(self) -> str:
"""Convert to human-readable formatted string for display.
Optimized for readability in Claude Code output panels.
"""
lines: List[str] = []
if isinstance(self.data, dict):
# Handle search/find results with "results" array
if "results" in self.data:
results = self.data["results"]
stats = self.data.get("stats", {})
pagination = self.data.get("pagination", {})
# Header with stats
if stats:
query = stats.get("query", stats.get("pattern", ""))
total = stats.get("total", len(results))
time_ms = stats.get("time_ms", {})
if time_ms:
if isinstance(time_ms, dict):
total_time = sum(time_ms.values())
else:
total_time = time_ms
lines.append(f"# Search: '{query}' ({total} results, {total_time}ms)")
else:
lines.append(f"# Found {total} results for '{query}'")
else:
lines.append(f"# Found {len(results)} results")
lines.append("")
# Format each result
for i, result in enumerate(results[:50], 1):
if isinstance(result, dict):
# Common patterns for search results
file_path = result.get("file", result.get("path", ""))
line_num = result.get("line", result.get("line_number", ""))
match_text = result.get("match", result.get("text", result.get("content", "")))
result_type = result.get("type", "")
if file_path:
loc = f"{file_path}:{line_num}" if line_num else file_path
lines.append(f"{i}. {loc}")
if match_text:
# Truncate long matches for readability
preview = match_text[:200] + "..." if len(match_text) > 200 else match_text
lines.append(f" {preview}")
if result_type:
lines.append(f" [{result_type}]")
else:
lines.append(f"{i}. {json.dumps(result, default=str)}")
else:
lines.append(f"{i}. {result}")
# Show pagination info
if pagination:
page = pagination.get("page", 1)
total = pagination.get("total", 0)
has_next = pagination.get("has_next", False)
if has_next and total > 0:
total_pages = (total // 50) + 1
lines.append(f"\n... page {page} of {total_pages} ({total} total)")
# Handle command execution results
elif "output" in self.data or "stdout" in self.data or "stderr" in self.data:
# Shell command output
exit_code = self.data.get("exit_code", self.data.get("returncode", 0))
stdout = self.data.get("output", self.data.get("stdout", ""))
stderr = self.data.get("stderr", "")
elapsed = self.data.get("elapsed", self.data.get("time_ms", ""))
if exit_code == 0:
lines.append(f"✓ Command succeeded")
else:
lines.append(f"✗ Command failed (exit {exit_code})")
if elapsed:
lines.append(f"Time: {elapsed}ms" if isinstance(elapsed, (int, float)) else f"Time: {elapsed}")
lines.append("")
if stdout:
lines.append(stdout.rstrip())
if stderr:
lines.append("\n--- stderr ---")
lines.append(stderr.rstrip())
# Handle error results
elif "error" in self.data:
lines.append(f"Error: {self.data['error']}")
if "details" in self.data:
lines.append(f"Details: {self.data['details']}")
# Generic dict - format as key-value pairs
else:
for key, value in self.data.items():
if isinstance(value, (dict, list)):
lines.append(f"{key}:")
lines.append(json.dumps(value, indent=2, default=str))
else:
lines.append(f"{key}: {value}")
elif isinstance(self.data, list):
# List data - format as numbered items
for i, item in enumerate(self.data[:50], 1):
if isinstance(item, dict):
lines.append(f"{i}. {json.dumps(item, default=str)}")
else:
lines.append(f"{i}. {item}")
if len(self.data) > 50:
lines.append(f"\n... {len(self.data) - 50} more items")
else:
# Scalar or other - just convert to string
lines.append(str(self.data))
# Add metadata footer if present
if self.metadata:
lines.append("")
lines.append("---")
for key, value in self.metadata.items():
lines.append(f"{key}: {value}")
return "\n".join(lines)
pkg/hanzo-tools/hanzo_tools/core/validation.py
+115
View File
@@ -0,0 +1,115 @@
"""Validation utilities for tool parameters."""
from typing import Optional
from pathlib import Path
from dataclasses import dataclass
@dataclass
class ValidationResult:
"""Result of a validation check."""
is_valid: bool
error_message: Optional[str] = None
def __bool__(self) -> bool:
return self.is_valid
def validate_path_parameter(
path: str,
param_name: str = "path",
must_exist: bool = False,
must_be_file: bool = False,
must_be_dir: bool = False,
) -> ValidationResult:
"""Validate a path parameter.
Args:
path: Path string to validate
param_name: Name of the parameter for error messages
must_exist: If True, path must exist
must_be_file: If True, path must be a file
must_be_dir: If True, path must be a directory
Returns:
ValidationResult with status and error if any
"""
if not path:
return ValidationResult(
is_valid=False,
error_message=f"{param_name} is required",
)
if not path.strip():
return ValidationResult(
is_valid=False,
error_message=f"{param_name} cannot be empty",
)
try:
p = Path(path)
if must_exist and not p.exists():
return ValidationResult(
is_valid=False,
error_message=f"{param_name} does not exist: {path}",
)
if must_be_file and p.exists() and not p.is_file():
return ValidationResult(
is_valid=False,
error_message=f"{param_name} is not a file: {path}",
)
if must_be_dir and p.exists() and not p.is_dir():
return ValidationResult(
is_valid=False,
error_message=f"{param_name} is not a directory: {path}",
)
return ValidationResult(is_valid=True)
except Exception as e:
return ValidationResult(
is_valid=False,
error_message=f"Invalid {param_name}: {e}",
)
def validate_string_parameter(
value: str,
param_name: str,
min_length: int = 0,
max_length: Optional[int] = None,
pattern: Optional[str] = None,
) -> ValidationResult:
"""Validate a string parameter."""
if not value:
return ValidationResult(
is_valid=False,
error_message=f"{param_name} is required",
)
if len(value) < min_length:
return ValidationResult(
is_valid=False,
error_message=f"{param_name} must be at least {min_length} characters",
)
if max_length and len(value) > max_length:
return ValidationResult(
is_valid=False,
error_message=f"{param_name} must be at most {max_length} characters",
)
if pattern:
import re
if not re.match(pattern, value):
return ValidationResult(
is_valid=False,
error_message=f"{param_name} does not match required pattern",
)
return ValidationResult(is_valid=True)
pkg/hanzo-tools/pyproject.toml
+13 -8
View File
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools"
version = "0.2.0"
description = "All Hanzo AI tools - install everything at once"
version = "0.3.0"
description = "Hanzo AI tools - core infrastructure and tool bundles"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
@@ -17,9 +17,12 @@ classifiers = [
]
keywords = ["hanzo", "tools", "mcp", "ai", "agent"]
# Base dependencies - just the core
# Core dependencies (merged from hanzo-tools-core)
dependencies = [
"hanzo-tools-core>=0.1.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
"typing-extensions>=4.13.0",
]
[project.optional-dependencies]
@@ -40,17 +43,15 @@ todo = ["hanzo-tools-todo>=0.1.0"]
config = ["hanzo-tools-config>=0.1.0"]
mcp_tools = ["hanzo-tools-mcp>=0.1.0"]
reasoning = ["hanzo-tools-reasoning>=0.1.0"]
computer = ["hanzo-tools-computer>=0.1.0"]
# Bundles
core = [
dev = [
"hanzo-tools-filesystem>=0.1.0",
"hanzo-tools-shell>=0.1.0",
"hanzo-tools-todo>=0.1.0",
"hanzo-tools-reasoning>=0.1.0",
"hanzo-tools-config>=0.1.0",
]
dev = [
"hanzo-tools[core]",
"hanzo-tools-editor>=0.1.0",
"hanzo-tools-lsp>=0.1.0",
"hanzo-tools-refactor>=0.1.0",
@@ -77,6 +78,7 @@ all = [
"hanzo-tools-reasoning>=0.1.0",
"hanzo-tools-config>=0.1.0",
"hanzo-tools-mcp>=0.1.0",
"hanzo-tools-computer>=0.1.0",
]
[project.urls]
@@ -86,3 +88,6 @@ all = [
[tool.setuptools.packages.find]
where = ["."]
include = ["hanzo_tools*"]
[tool.setuptools.package-data]
hanzo_tools = ["py.typed"]