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

feat(hanzo-mcp): v0.10.0 - minimal orthogonal tool set

Browse Source 
BREAKING: Shell tools consolidated
- dag: Replaces bash, zsh, shell, exec, batch (one way to run commands)
- ps: Replaces process (UNIX-style naming)
- zsh-only by default

30 tools registered (down from ~50+):
- Shell: dag, ps, npx, uvx, open
- Files: read, write, edit, multi_edit
- Search: search, find, ast, tree
- LSP: lsp, refactor
- Thinking: think, critic
- Memory: recall/create/update/delete/manage memories
- Knowledge: recall/store facts, manage knowledge bases
- Meta: rules, mode, todo, version

Philosophy: "One and only one way to do everything"
This commit is contained in:
zeekay
2025-12-25 17:21:06 -08:00
parent 168eabfd99
commit c8576649e3
12 changed files with 1157 additions and 966 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-mcp/hanzo_mcp/__init__.py
+1 -1
View File
@@ -26,4 +26,4 @@ if os.environ.get("HANZO_MCP_TRANSPORT") == "stdio":
except ImportError:
pass
__version__ = "0.9.23"
__version__ = "0.9.24"
pkg/hanzo-mcp/hanzo_mcp/cli.py
+11 -3
View File
@@ -64,11 +64,15 @@ def main() -> None:
)
# Redirect stderr to devnull for stdio transport to prevent any output
sys.stderr = open(os.devnull, "w")
# Store handles for proper cleanup to avoid resource leaks
_stderr_devnull = open(os.devnull, "w")
sys.stderr = _stderr_devnull
# Suppress stdout during potentially noisy imports unless user requested help/version
_stdout_devnull = None
if not any(flag in sys.argv for flag in ("--version", "-h", "--help")):
sys.stdout = open(os.devnull, "w")
_stdout_devnull = open(os.devnull, "w")
sys.stdout = _stdout_devnull
suppress_stdout = True
# Import the server only AFTER transport/logging have been configured to avoid import-time noise
@@ -290,11 +294,15 @@ def main() -> None:
# Restore stdout after parsing, before any explicit output or server start
if suppress_stdout:
try:
sys.stdout.close() # Close devnull handle
if _stdout_devnull is not None:
_stdout_devnull.close() # Close devnull handle properly
except Exception:
pass
sys.stdout = original_stdout
# Note: We intentionally keep stderr redirected to devnull for stdio transport
# to prevent any library noise from corrupting the protocol
# Parse timeout arguments with human-readable format support
command_timeout = _parse_timeout_arg(str(args.command_timeout))
pkg/hanzo-mcp/hanzo_mcp/server.py
+25 -21
View File
@@ -199,10 +199,11 @@ class HanzoMCPServer:
self.enable_agent_tool = enable_agent_tool
self.command_timeout = command_timeout
# Initialize cleanup tracking
# Initialize cleanup tracking with thread-safe lock
self._cleanup_thread: threading.Thread | None = None
self._shutdown_event = threading.Event()
self._cleanup_registered = False
self._cleanup_lock = threading.Lock()
# Apply disabled_tools to enabled_tools
final_enabled_tools = self.enabled_tools.copy()
@@ -234,32 +235,35 @@ class HanzoMCPServer:
def _setup_cleanup_handlers(self) -> None:
"""Set up signal handlers and background cleanup thread."""
if self._cleanup_registered:
return
# Use lock to prevent race condition in concurrent calls
with self._cleanup_lock:
if self._cleanup_registered:
return
# Register cleanup on normal exit
atexit.register(self._cleanup_sessions)
# Mark as registered first to prevent re-entry
self._cleanup_registered = True
# Register signal handlers for graceful shutdown
def signal_handler(signum, frame):
import sys
# Register cleanup on normal exit
atexit.register(self._cleanup_sessions)
# Only log if not stdio transport
if hasattr(self, "_transport") and self._transport != "stdio":
logger = logging.getLogger(__name__)
logger.info("\nShutting down gracefully...")
self._cleanup_sessions()
self._shutdown_event.set()
sys.exit(0)
# Register signal handlers for graceful shutdown
def signal_handler(signum, frame):
import sys
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
# Only log if not stdio transport
if hasattr(self, "_transport") and self._transport != "stdio":
logger = logging.getLogger(__name__)
logger.info("\nShutting down gracefully...")
self._cleanup_sessions()
self._shutdown_event.set()
sys.exit(0)
# Start background cleanup thread for periodic cleanup
self._cleanup_thread = threading.Thread(target=self._background_cleanup, daemon=True)
self._cleanup_thread.start()
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
self._cleanup_registered = True
# Start background cleanup thread for periodic cleanup
self._cleanup_thread = threading.Thread(target=self._background_cleanup, daemon=True)
self._cleanup_thread.start()
def _background_cleanup(self) -> None:
"""Background thread for periodic session cleanup."""
pkg/hanzo-mcp/hanzo_mcp/tools/__init__.py
+6 -5
View File
@@ -249,9 +249,9 @@ def register_all_tools(
for tool in jupyter_tools:
all_tools[tool.name] = tool
# Register shell tools if enabled
if is_tool_enabled("run_command", True):
shell_tools = register_shell_tools(mcp_server, permission_manager)
# Register shell tools if enabled (dag, ps, npx, uvx, open)
if is_tool_enabled("dag", True) or is_tool_enabled("run_command", True):
shell_tools = register_shell_tools(mcp_server, permission_manager, all_tools=all_tools)
for tool in shell_tools:
all_tools[tool.name] = tool
@@ -316,8 +316,9 @@ def register_all_tools(
for tool in vector_tools:
all_tools[tool.name] = tool
# Register batch tool if enabled (batch tool is typically always enabled)
if is_tool_enabled("batch", True):
# Register batch tool if enabled (DEPRECATED - use dag instead)
# dag tool can do everything batch does: dag([{"tool": "x", "input": {...}}], parallel=True)
if is_tool_enabled("batch", False): # Disabled by default in 1.0.0
register_batch_tool(mcp_server, all_tools)
# Register database tools if enabled
pkg/hanzo-mcp/hanzo_mcp/tools/common/__init__.py
+32 -23
View File
@@ -1,45 +1,54 @@
"""Common utilities for Hanzo AI tools."""
"""Common utilities for Hanzo AI tools.
Core thinking tools:
- think: Structured reasoning
- critic: Critical analysis
Note: batch tool is DEPRECATED in 1.0.0 - use `dag` from shell tools instead.
"""
import warnings
from mcp.server import FastMCP
from hanzo_mcp.tools.common.base import BaseTool, ToolRegistry
from hanzo_mcp.tools.common.batch_tool import BatchTool
from hanzo_mcp.tools.common.critic_tool import CriticTool
from hanzo_mcp.tools.common.thinking_tool import ThinkingTool
def register_thinking_tool(
mcp_server: FastMCP,
) -> list[BaseTool]:
"""Register thinking tools with the MCP server.
Args:
mcp_server: The FastMCP server instance
"""
def register_thinking_tool(mcp_server: FastMCP) -> list[BaseTool]:
"""Register thinking tool with the MCP server."""
thinking_tool = ThinkingTool()
ToolRegistry.register_tool(mcp_server, thinking_tool)
return [thinking_tool]
def register_critic_tool(
mcp_server: FastMCP,
) -> list[BaseTool]:
"""Register critic tool with the MCP server.
Args:
mcp_server: The FastMCP server instance
"""
def register_critic_tool(mcp_server: FastMCP) -> list[BaseTool]:
"""Register critic tool with the MCP server."""
critic_tool = CriticTool()
ToolRegistry.register_tool(mcp_server, critic_tool)
return [critic_tool]
def register_batch_tool(mcp_server: FastMCP, tools: dict[str, BaseTool]) -> None:
"""Register batch tool with the MCP server.
Args:
mcp_server: The FastMCP server instance
tools: Dictionary mapping tool names to tool instances
"""DEPRECATED: Use `dag` tool from shell package instead.
The dag tool provides DAG execution with serial, parallel, and mixed modes.
It can invoke any tool, not just shell commands.
Migration:
# Old (batch)
batch(invocations=[{"tool_name": "read", "input": {...}}])
# New (dag)
dag([{"tool": "read", "input": {...}}], parallel=True)
"""
warnings.warn(
"batch tool is deprecated in 1.0.0. Use 'dag' from shell tools instead. "
"dag([{\"tool\": \"name\", \"input\": {...}}], parallel=True)",
DeprecationWarning,
stacklevel=2,
)
# Still register for backward compatibility
from hanzo_mcp.tools.common.batch_tool import BatchTool
batch_tool = BatchTool(tools)
ToolRegistry.register_tool(mcp_server, batch_tool)
pkg/hanzo-mcp/hanzo_mcp/tools/common/permissions.py
+5 -1
View File
@@ -3,11 +3,14 @@
import os
import sys
import json
import logging
import tempfile
from typing import Any, TypeVar, final
from pathlib import Path
from collections.abc import Callable, Awaitable
logger = logging.getLogger(__name__)
# Define type variables for better type annotations
T = TypeVar("T")
P = TypeVar("P")
@@ -136,7 +139,8 @@ class PermissionManager:
# Additional check: is the resolved path still under allowed paths?
pass # Continue to normal checks
except (OSError, RuntimeError) as e:
# Path resolution failed, deny access
# Path resolution failed, deny access - log for debugging
logger.debug(f"Path resolution failed for '{path}': {e}")
return False
# Check exclusions first
pkg/hanzo-mcp/hanzo_mcp/tools/search/search_tool.py
+265 -864
View File
File diff suppressed because it is too large Load Diff
pkg/hanzo-mcp/hanzo_mcp/tools/shell/__init__.py
+38 -30
View File
@@ -1,25 +1,36 @@
"""Shell tools package for Hanzo AI.
This package provides tools for executing shell commands and scripts.
Minimal, orthogonal shell execution.
Core tools:
- dag: Execute commands/tools with DAG flow control (serial, parallel, graph)
- ps: Process management (list, kill, logs)
Convenience tools (optional):
- npx: Node package execution with auto-backgrounding
- uvx: Python package execution with auto-backgrounding
- open: Open files/URLs in system apps
"""
from mcp.server import FastMCP
from hanzo_mcp.tools.shell.open import open_tool
from hanzo_mcp.tools.common.base import BaseTool, ToolRegistry
from hanzo_mcp.tools.common.permissions import PermissionManager
# Core tools
from hanzo_mcp.tools.shell.dag_tool import DagTool, create_dag_tool
from hanzo_mcp.tools.shell.ps_tool import PsTool, ps_tool
# Convenience tools
from hanzo_mcp.tools.shell.open import open_tool
from hanzo_mcp.tools.shell.npx_tool import npx_tool
from hanzo_mcp.tools.shell.uvx_tool import uvx_tool
from hanzo_mcp.tools.shell.zsh_tool import zsh_tool, shell_tool
# Import tools
from hanzo_mcp.tools.shell.bash_tool import bash_tool
from hanzo_mcp.tools.common.permissions import PermissionManager
from hanzo_mcp.tools.shell.process_tool import process_tool
# from hanzo_mcp.tools.shell.streaming_command import StreamingCommandTool
# Export all tool classes
__all__ = [
"DagTool",
"PsTool",
"create_dag_tool",
"ps_tool",
"get_shell_tools",
"register_shell_tools",
]
@@ -27,51 +38,48 @@ __all__ = [
def get_shell_tools(
permission_manager: PermissionManager,
all_tools: dict[str, BaseTool] | None = None,
) -> list[BaseTool]:
"""Create instances of all shell tools.
"""Create instances of shell tools.
Args:
permission_manager: Permission manager for access control
all_tools: Dict of all registered tools (for dag tool invocations)
Returns:
List of shell tool instances
"""
# Set permission manager for tools that need it
bash_tool.permission_manager = permission_manager
zsh_tool.permission_manager = permission_manager
shell_tool.permission_manager = permission_manager
# Create dag tool with access to other tools
dag = create_dag_tool(tools=all_tools or {})
# Set permission manager for convenience tools
npx_tool.permission_manager = permission_manager
uvx_tool.permission_manager = permission_manager
# Note: StreamingCommandTool is abstract and shouldn't be instantiated directly
# It's used as a base class for other streaming tools
# Return shell_tool first (smart default), then specific shells
return [
shell_tool, # Smart shell (prefers zsh if available)
zsh_tool, # Explicit zsh
bash_tool, # Explicit bash
npx_tool,
uvx_tool,
process_tool,
open_tool,
# streaming_command_tool, # Removed as it's abstract
dag, # DAG execution (replaces bash, zsh, shell, exec, batch)
ps_tool, # Process management (replaces process)
npx_tool, # Node packages
uvx_tool, # Python packages
open_tool, # Open files/URLs
]
def register_shell_tools(
mcp_server: FastMCP,
permission_manager: PermissionManager,
all_tools: dict[str, BaseTool] | None = None,
) -> list[BaseTool]:
"""Register all shell tools with the MCP server.
"""Register shell tools with the MCP server.
Args:
mcp_server: The FastMCP server instance
permission_manager: Permission manager for access control
all_tools: Dict of all registered tools (for dag tool invocations)
Returns:
List of registered tools
"""
tools = get_shell_tools(permission_manager)
tools = get_shell_tools(permission_manager, all_tools)
ToolRegistry.register_tools(mcp_server, tools)
return tools
pkg/hanzo-mcp/hanzo_mcp/tools/shell/dag_tool.py
+529
View File
@@ -0,0 +1,529 @@
"""DAG execution tool - directed acyclic graph for command execution.
Run commands/tools with proper dependency ordering using DAG semantics.
Supports serial (default), parallel, and complex mixed execution graphs.
"""
import asyncio
import os
import shutil
from typing import Any, Union, List, Dict, Optional, Annotated, override
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_mcp.tools.common.base import BaseTool
from hanzo_mcp.tools.common.context import create_tool_context
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
class NodeStatus(Enum):
"""Execution status of a DAG node."""
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class DagResult:
"""Result from a single DAG node execution."""
node_id: str
command: str
stdout: str
stderr: str
status: NodeStatus
exit_code: int
duration_ms: int = 0
node_type: str = "shell" # shell, tool, parallel
@dataclass
class DagNode:
"""A node in the execution DAG."""
id: str
command: Union[str, Dict[str, Any]]
depends_on: List[str] = field(default_factory=list)
status: NodeStatus = NodeStatus.PENDING
result: Optional[DagResult] = None
# Type alias
Command = Union[str, Dict[str, Any], List[Any]]
class DagTool(BaseTool):
"""DAG-based execution with proper dependency ordering.
Execute commands (shell or tools) with directed acyclic graph semantics.
Supports serial, parallel, and complex mixed execution patterns.
USAGE PATTERNS:
1. Serial (default) - sequential execution:
dag(["ls", "pwd", "git status"])
2. Parallel - concurrent execution:
dag(["npm install", "cargo build"], parallel=True)
3. DAG - mixed serial and parallel blocks:
dag([
"mkdir -p dist", # Step 1
{"parallel": [ # Step 2 (all concurrent)
"cp manifest.json dist/",
"cp -rf assets/ dist/",
"cp -rf lib/ dist/",
]},
"zip -r package.zip dist/" # Step 3 (after all of step 2)
])
4. Tool invocations - not just shell:
dag([
{"tool": "read", "input": {"file_path": "config.json"}},
{"tool": "search", "input": {"pattern": "TODO"}},
])
5. Named nodes with explicit dependencies:
dag([
{"id": "setup", "run": "mkdir -p dist"},
{"id": "copy", "run": "cp *.txt dist/", "after": ["setup"]},
{"id": "test", "run": "pytest", "after": ["setup"]},
{"id": "package", "run": "tar -czf out.tar.gz dist/", "after": ["copy", "test"]},
])
Uses zsh for shell execution.
"""
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None, default_shell: str = "zsh"):
"""Initialize DAG execution tool.
Args:
tools: Dict of available MCP tools for tool invocations
default_shell: Default shell (zsh, bash, sh)
"""
super().__init__()
self.tools = tools or {}
self.default_shell = self._resolve_shell(default_shell)
def _resolve_shell(self, preferred: str) -> str:
"""Resolve shell - zsh only."""
for path in ["zsh", "/bin/zsh", "/usr/bin/zsh"]:
if shutil.which(path):
return path
return "sh" # Bare minimum fallback
@property
@override
def name(self) -> str:
return "dag"
@property
@override
def description(self) -> str:
return """DAG execution - run commands with dependency ordering.
Execute shell commands or tools with DAG (directed acyclic graph) semantics.
Supports serial, parallel, and complex mixed execution patterns.
MODES:
Serial (default): dag(["ls", "pwd"])
Commands run in sequence.
Parallel: dag(["npm install", "cargo build"], parallel=True)
Commands run concurrently.
Mixed DAG:
dag([
"mkdir -p dist",
{"parallel": ["cp a.txt dist/", "cp b.txt dist/"]},
"zip -r out.zip dist/"
])
Tool invocations:
dag([{"tool": "search", "input": {"pattern": "TODO"}}])
Named with dependencies:
dag([
{"id": "build", "run": "make build"},
{"id": "test", "run": "make test", "after": ["build"]},
])
Uses zsh for shell execution."""
async def _run_shell(
self,
cmd: str,
shell: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 120,
) -> DagResult:
"""Run a shell command asynchronously."""
start_time = datetime.now()
node_id = f"shell_{id(cmd)}"
try:
# Merge environment
run_env = os.environ.copy()
if env:
run_env.update(env)
proc = await asyncio.create_subprocess_exec(
shell, "-c", cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
env=run_env,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout
)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
exit_code = proc.returncode or 0
return DagResult(
node_id=node_id,
command=cmd,
stdout=stdout.decode("utf-8", errors="replace"),
stderr=stderr.decode("utf-8", errors="replace"),
status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
exit_code=exit_code,
duration_ms=duration,
node_type="shell",
)
except asyncio.TimeoutError:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
return DagResult(
node_id=node_id,
command=cmd,
stdout="",
stderr=f"Command timed out after {timeout}s",
status=NodeStatus.FAILED,
exit_code=124,
duration_ms=duration,
node_type="shell",
)
except Exception as e:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
return DagResult(
node_id=node_id,
command=cmd,
stdout="",
stderr=str(e),
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=duration,
node_type="shell",
)
async def _run_tool(
self,
tool_name: str,
tool_input: Dict[str, Any],
ctx: MCPContext
) -> DagResult:
"""Run an MCP tool invocation."""
start_time = datetime.now()
node_id = f"tool_{tool_name}"
if tool_name not in self.tools:
return DagResult(
node_id=node_id,
command=f"tool:{tool_name}",
stdout="",
stderr=f"Tool '{tool_name}' not found. Available: {list(self.tools.keys())}",
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=0,
node_type="tool",
)
try:
tool = self.tools[tool_name]
result = await tool.call(ctx, **tool_input)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
return DagResult(
node_id=node_id,
command=f"tool:{tool_name}",
stdout=str(result) if result else "",
stderr="",
status=NodeStatus.SUCCESS,
exit_code=0,
duration_ms=duration,
node_type="tool",
)
except Exception as e:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
return DagResult(
node_id=node_id,
command=f"tool:{tool_name}",
stdout="",
stderr=str(e),
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=duration,
node_type="tool",
)
async def _execute_node(
self,
cmd: Command,
ctx: MCPContext,
shell: str,
cwd: Optional[str],
env: Optional[Dict[str, str]],
timeout: int,
) -> DagResult:
"""Execute a single DAG node."""
# String command = shell execution
if isinstance(cmd, str):
return await self._run_shell(cmd, shell, cwd, env, timeout)
# Dict command - could be tool, parallel, or named node
if isinstance(cmd, dict):
# Tool invocation: {"tool": "name", "input": {...}}
if "tool" in cmd:
return await self._run_tool(
cmd["tool"],
cmd.get("input", {}),
ctx
)
# Parallel block: {"parallel": [...]}
if "parallel" in cmd:
parallel_cmds = cmd["parallel"]
tasks = [
self._execute_node(c, ctx, shell, cwd, env, timeout)
for c in parallel_cmds
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Combine parallel results
combined_stdout = []
combined_stderr = []
all_success = True
total_duration = 0
for i, r in enumerate(results):
if isinstance(r, Exception):
combined_stderr.append(f"[{i}] Error: {r}")
all_success = False
else:
if r.stdout:
combined_stdout.append(f"[{i}] {r.stdout.rstrip()}")
if r.stderr:
combined_stderr.append(f"[{i}] {r.stderr.rstrip()}")
if r.status != NodeStatus.SUCCESS:
all_success = False
total_duration = max(total_duration, r.duration_ms)
return DagResult(
node_id=f"parallel_{len(parallel_cmds)}",
command=f"parallel[{len(parallel_cmds)} tasks]",
stdout="\n".join(combined_stdout),
stderr="\n".join(combined_stderr),
status=NodeStatus.SUCCESS if all_success else NodeStatus.FAILED,
exit_code=0 if all_success else 1,
duration_ms=total_duration,
node_type="parallel",
)
# Named node: {"id": "x", "run": "...", "after": [...]}
if "run" in cmd:
run_cmd = cmd["run"]
return await self._execute_node(run_cmd, ctx, shell, cwd, env, timeout)
# Unknown dict format
return DagResult(
node_id="unknown",
command=str(cmd),
stdout="",
stderr=f"Unknown command format: {cmd}",
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=0,
node_type="unknown",
)
# Unknown type
return DagResult(
node_id="unknown",
command=str(cmd),
stdout="",
stderr=f"Unknown command type: {type(cmd).__name__}",
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=0,
node_type="unknown",
)
@override
@auto_timeout("dag")
async def call(
self,
ctx: MCPContext,
commands: List[Command],
parallel: bool = False,
shell: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 120,
strict: bool = False,
quiet: bool = False,
**kwargs,
) -> str:
"""Execute commands with DAG semantics.
Args:
ctx: MCP context
commands: List of commands (strings, tool dicts, or parallel blocks)
parallel: Run all top-level commands in parallel
shell: Shell to use (default: zsh)
cwd: Working directory
env: Additional environment variables
timeout: Timeout per command in seconds
strict: Stop on first error
quiet: Suppress stdout in output
Returns:
Combined execution output
"""
tool_ctx = create_tool_context(ctx)
tool_ctx.set_tool_info(self.name)
shell = shell or self.default_shell
results: List[DagResult] = []
if parallel:
# Run all commands in parallel
tasks = [
self._execute_node(cmd, ctx, shell, cwd, env, timeout)
for cmd in commands
]
results = await asyncio.gather(*tasks, return_exceptions=True)
results = [
r if not isinstance(r, Exception)
else DagResult(
node_id=f"error_{i}",
command=str(commands[i]),
stdout="",
stderr=str(r),
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=0,
node_type="error",
)
for i, r in enumerate(results)
]
else:
# Serial execution with DAG support (nested parallel blocks)
for cmd in commands:
result = await self._execute_node(cmd, ctx, shell, cwd, env, timeout)
results.append(result)
# Stop on error if strict
if strict and result.status == NodeStatus.FAILED:
break
# Format output
return self._format_output(results, quiet)
def _format_output(self, results: List[DagResult], quiet: bool = False) -> str:
"""Format DAG execution results."""
output_parts = []
total_duration = 0
failed_count = 0
for r in results:
total_duration += r.duration_ms
if r.status == NodeStatus.FAILED:
failed_count += 1
# Add stdout unless quiet
if r.stdout and not quiet:
output_parts.append(r.stdout.rstrip())
# Always show stderr
if r.stderr:
output_parts.append(f"[stderr] {r.stderr.rstrip()}")
# Add summary for multi-command execution
if len(results) > 1:
status = "✓" if failed_count == 0 else f"✗ ({failed_count} failed)"
output_parts.append(f"\n[dag] {len(results)} nodes, {total_duration}ms, {status}")
return "\n".join(output_parts) if output_parts else "(no output)"
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register DAG tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def dag_handler(
commands: Annotated[
List[Any],
Field(description="Commands to execute (strings, tool dicts, or parallel blocks)")
],
parallel: Annotated[
bool,
Field(description="Run all commands in parallel", default=False)
] = False,
shell: Annotated[
Optional[str],
Field(description="Shell to use (default: zsh)", default=None)
] = None,
cwd: Annotated[
Optional[str],
Field(description="Working directory", default=None)
] = None,
env: Annotated[
Optional[Dict[str, str]],
Field(description="Environment variables", default=None)
] = None,
timeout: Annotated[
int,
Field(description="Timeout per command (seconds)", default=120)
] = 120,
strict: Annotated[
bool,
Field(description="Stop on first error", default=False)
] = False,
quiet: Annotated[
bool,
Field(description="Suppress stdout", default=False)
] = False,
ctx: MCPContext = None,
) -> str:
return await tool_self.call(
ctx,
commands=commands,
parallel=parallel,
shell=shell,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
def create_dag_tool(
tools: Optional[Dict[str, BaseTool]] = None,
default_shell: str = "zsh"
) -> DagTool:
"""Factory to create DAG execution tool."""
return DagTool(tools, default_shell)
pkg/hanzo-mcp/hanzo_mcp/tools/shell/ps_tool.py
+227
View File
@@ -0,0 +1,227 @@
"""Process management tool (ps).
List, monitor, and control background processes.
Clean UNIX-style API.
"""
import signal
import asyncio
from typing import Optional, List, Dict, Any, Annotated, override
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import aiofiles
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_mcp.tools.common.base import BaseTool
from hanzo_mcp.tools.shell.base_process import ProcessManager
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
@dataclass
class ProcessInfo:
"""Process information."""
id: str
pid: int
cmd: str
running: bool
exit_code: Optional[int] = None
started: Optional[datetime] = None
log_file: Optional[Path] = None
class PsTool(BaseTool):
"""Process management - list, kill, logs.
USAGE:
ps() # List all processes
ps(id="abc123") # Get specific process info
ps(kill="abc123") # Kill process (SIGTERM)
ps(kill="abc123", signal="KILL") # Kill with SIGKILL
ps(logs="abc123") # Get process logs
ps(logs="abc123", n=50) # Last 50 lines
"""
_instance: Optional["PsTool"] = None
_process_manager: Optional[ProcessManager] = None
def __new__(cls):
"""Singleton - share process manager across instances."""
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._process_manager = ProcessManager()
return cls._instance
def __init__(self):
super().__init__()
self.process_manager = self._process_manager
@property
@override
def name(self) -> str:
return "ps"
@property
@override
def description(self) -> str:
return """Process management - list, kill, and read logs.
USAGE:
ps List all background processes
ps --id ID Get specific process info
ps --kill ID Kill process (SIGTERM)
ps --kill ID --sig 9 Kill with specific signal
ps --logs ID Get process stdout/stderr
ps --logs ID --n 50 Last N lines of logs"""
def _list_processes(self) -> List[ProcessInfo]:
"""Get all tracked processes."""
processes = []
for proc_id, info in self.process_manager.list_processes().items():
processes.append(ProcessInfo(
id=proc_id,
pid=info.get("pid", 0),
cmd=info.get("cmd", ""),
running=info.get("running", False),
exit_code=info.get("return_code"),
log_file=Path(info["log_file"]) if info.get("log_file") else None,
))
return processes
def _get_process(self, proc_id: str) -> Optional[ProcessInfo]:
"""Get specific process info."""
for p in self._list_processes():
if p.id == proc_id or str(p.pid) == proc_id:
return p
return None
async def _kill_process(self, proc_id: str, sig: int = signal.SIGTERM) -> str:
"""Kill a process."""
process = self.process_manager.get_process(proc_id)
if not process:
return f"Process not found: {proc_id}"
try:
process.send_signal(sig)
sig_name = signal.Signals(sig).name
return f"Sent {sig_name} to {proc_id} (PID {process.pid})"
except ProcessLookupError:
return f"Process {proc_id} already terminated"
except Exception as e:
return f"Failed to kill {proc_id}: {e}"
async def _get_logs(self, proc_id: str, n: int = 100) -> str:
"""Get process logs."""
log_file = self.process_manager.get_log_file(proc_id)
if not log_file or not log_file.exists():
return f"No logs for process: {proc_id}"
try:
async with aiofiles.open(log_file, "r") as f:
content = await f.read()
lines = content.splitlines()
if len(lines) > n:
lines = lines[-n:]
return "\n".join(lines)
except Exception as e:
return f"Error reading logs: {e}"
def _format_list(self, processes: List[ProcessInfo]) -> str:
"""Format process list for display."""
if not processes:
return "No background processes"
lines = ["PID ID STATUS CMD"]
lines.append("-" * 60)
for p in processes:
status = "running" if p.running else f"exit({p.exit_code})"
cmd = p.cmd[:30] + "..." if len(p.cmd) > 30 else p.cmd
lines.append(f"{p.pid:<7} {p.id:<15} {status:<9} {cmd}")
return "\n".join(lines)
@override
@auto_timeout("ps")
async def call(
self,
ctx: MCPContext,
id: Optional[str] = None,
kill: Optional[str] = None,
logs: Optional[str] = None,
sig: int = 15, # SIGTERM
n: int = 100,
**kwargs,
) -> str:
"""Process management.
Args:
ctx: MCP context
id: Get specific process info
kill: Kill process by ID
logs: Get logs for process
sig: Signal number for kill (default: 15/SIGTERM)
n: Number of log lines to show
Returns:
Process information or action result
"""
# Kill action
if kill:
return await self._kill_process(kill, sig)
# Logs action
if logs:
return await self._get_logs(logs, n)
# Get specific process
if id:
proc = self._get_process(id)
if not proc:
return f"Process not found: {id}"
status = "running" if proc.running else f"exited ({proc.exit_code})"
return f"ID: {proc.id}\nPID: {proc.pid}\nStatus: {status}\nCommand: {proc.cmd}"
# Default: list all
return self._format_list(self._list_processes())
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register ps tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def ps_handler(
id: Annotated[
Optional[str],
Field(description="Get specific process by ID", default=None)
] = None,
kill: Annotated[
Optional[str],
Field(description="Kill process by ID", default=None)
] = None,
logs: Annotated[
Optional[str],
Field(description="Get logs for process ID", default=None)
] = None,
sig: Annotated[
int,
Field(description="Signal for kill (default: 15/SIGTERM)", default=15)
] = 15,
n: Annotated[
int,
Field(description="Number of log lines", default=100)
] = 100,
ctx: MCPContext = None,
) -> str:
return await tool_self.call(ctx, id=id, kill=kill, logs=logs, sig=sig, n=n)
# Singleton instance
ps_tool = PsTool()
pkg/hanzo-mcp/pyproject.toml
+9 -9
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-mcp"
version = "0.9.23"
version = "0.10.0"
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"
@@ -18,20 +18,20 @@ classifiers = [
keywords = ["mcp", "claude", "hanzo", "code", "agent"]
dependencies = [
"aiofiles>=24.1.0",
"mcp>=1.9.4",
"fastmcp>=2.9.2",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"httpx>=0.28.1",
"uvicorn>=0.34.0",
"openai>=1.62.0",
"openai>=1.82.0",
"python-dotenv>=1.0.1",
"litellm>=1.73.2",
"litellm>=1.80.11",
"grep-ast>=0.8.1",
"bashlex>=0.18",
"libtmux>=0.39.0",
"nbformat>=5.10.4",
"psutil>=6.0.0",
"pydantic>=2.9.2",
"pydantic-settings>=2.7.0",
"pydantic>=2.12.5",
"pydantic-settings>=2.12.0",
"typing-extensions>=4.13.0",
"watchdog>=6.0.0",
"keyring>=24.0.0",
@@ -47,7 +47,7 @@ dependencies = [
dev = [
"pytest>=7.0.0",
"pytest-cov>=4.1.0",
"ruff>=0.13.0",
"ruff>=0.14.0",
"black>=23.3.0",
"sphinx>=8.0.0",
"sphinx-rtd-theme>=3.0.0",
@@ -64,7 +64,7 @@ test = [
"pytest>=7.0.0",
"pytest-cov>=4.1.0",
"pytest-mock>=3.10.0",
"pytest-asyncio>=0.25.3",
"pytest-asyncio>=0.26.0",
"twisted",
]
agents = ["hanzo-agents>=0.1.0"]
pkg/hanzo-mcp/uv.lock
Generated
+9 -9
View File
@@ -1232,7 +1232,7 @@ wheels = [
[[package]]
name = "hanzo-mcp"
version = "0.9.23"
version = "0.10.0"
source = { editable = "." }
dependencies = [
{ name = "aiofiles" },
@@ -1308,7 +1308,7 @@ requires-dist = [
{ name = "bashlex", specifier = ">=0.18" },
{ name = "black", marker = "extra == 'dev'", specifier = ">=23.3.0" },
{ name = "build", marker = "extra == 'publish'", specifier = ">=1.0.3" },
{ name = "fastmcp", specifier = ">=2.9.2" },
{ name = "fastmcp", specifier = ">=2.14.1" },
{ name = "ffind", specifier = ">=1.3.0" },
{ name = "grep-ast", specifier = ">=0.8.1" },
{ name = "hanzo-agents", marker = "extra == 'agents'", specifier = ">=0.1.0" },
@@ -1316,26 +1316,26 @@ requires-dist = [
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "keyring", specifier = ">=24.0.0" },
{ name = "libtmux", specifier = ">=0.39.0" },
{ name = "litellm", specifier = ">=1.73.2" },
{ name = "mcp", specifier = ">=1.9.4" },
{ name = "litellm", specifier = ">=1.80.11" },
{ name = "mcp", specifier = ">=1.25.0" },
{ name = "mypy", marker = "extra == 'dev'", specifier = ">=1.10.0" },
{ name = "myst-parser", marker = "extra == 'dev'", specifier = ">=4.0.0" },
{ name = "myst-parser", marker = "extra == 'docs'", specifier = ">=4.0.0" },
{ name = "nbformat", specifier = ">=5.10.4" },
{ name = "openai", specifier = ">=1.62.0" },
{ name = "openai", specifier = ">=1.82.0" },
{ name = "orjson", marker = "extra == 'performance'", specifier = ">=3.9.0" },
{ name = "posthog", marker = "extra == 'analytics'", specifier = ">=3.0.0" },
{ name = "psutil", specifier = ">=6.0.0" },
{ name = "pydantic", specifier = ">=2.9.2" },
{ name = "pydantic-settings", specifier = ">=2.7.0" },
{ name = "pydantic", specifier = ">=2.12.5" },
{ 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.25.3" },
{ name = "pytest-asyncio", marker = "extra == 'test'", specifier = ">=0.26.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" },
{ name = "python-dotenv", specifier = ">=1.0.1" },
{ name = "ruff", marker = "extra == 'dev'", specifier = ">=0.13.0" },
{ name = "ruff", marker = "extra == 'dev'", specifier = ">=0.14.0" },
{ name = "sphinx", marker = "extra == 'dev'", specifier = ">=8.0.0" },
{ name = "sphinx", marker = "extra == 'docs'", specifier = ">=8.0.0" },
{ name = "sphinx-copybutton", marker = "extra == 'dev'", specifier = ">=0.5.0" },