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

chore: gitignore build/ artifacts, remove tracked build files

Browse Source
This commit is contained in:
zeekay
2026-03-06 12:24:01 -08:00
parent 0cb80b31de
commit a05fef2bf3
19 changed files with 1 additions and 5432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -11,6 +11,7 @@ __pycache__
.venv
# Build
build/
dist
_dev
pkg/hanzo-tools-shell/build/lib/hanzo_tools/__init__.py
-3
View File
@@ -1,3 +0,0 @@
"""Hanzo Tools namespace package."""
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/__init__.py
-362
View File
@@ -1,362 +0,0 @@
"""Shell tools package for Hanzo AI.
Minimal, orthogonal shell execution with uvloop for high performance.
SHELL DETECTION:
By default, only the user's active shell is exposed to MCP. This avoids
cluttering the tool list with shells the user doesn't use.
Detection order:
1. HANZO_MCP_SHELL env var (e.g., "zsh", "bash", "fish")
2. HANZO_MCP_FORCE_SHELL env var (e.g., "/opt/homebrew/bin/zsh")
3. --shell CLI flag (e.g., --shell=/path/to/fish)
4. Invoking shell (the shell that launched this process)
5. Login shell (from passwd/Directory Services)
6. $SHELL environment variable
On macOS with Homebrew, this will prefer /opt/homebrew/bin/zsh over /bin/zsh.
If detection fails, falls back to 'shell' tool (smart auto-detect).
Core tools:
- <detected>: Your active shell (zsh, bash, fish, or dash)
- ps: Process management (list, kill, logs)
HTTP/Data tools:
- curl: HTTP client without shell escaping issues
- jq: JSON processor without shell escaping issues
- wget: File/site downloads with mirroring support
Convenience tools:
- npx: Node package execution with auto-backgrounding
- uvx: Python package execution with auto-backgrounding
- open: Open files/URLs in system apps
Auto-backgrounding: Commands exceeding timeout automatically background.
Configure via: export HANZO_AUTO_BACKGROUND_TIMEOUT=30 (default: 30s)
export HANZO_AUTO_BACKGROUND_TIMEOUT=0 (disabled)
Use ps --logs <id> to view, ps --kill <id> to stop.
Performance: Uses uvloop on macOS/Linux for 2-4x faster async I/O.
Falls back to standard asyncio on Windows.
"""
# Use hanzo-async for unified async I/O and uvloop configuration
from hanzo_async import using_uvloop, configure_loop
# Auto-configure uvloop on import
configure_loop()
_using_uvloop = using_uvloop()
from mcp.server import FastMCP
from hanzo_tools.core import BaseTool, ToolRegistry, PermissionManager
from hanzo_tools.shell.jq_tool import JqTool
from hanzo_tools.shell.ps_tool import PsTool, ps_tool
# Core tools
from hanzo_tools.shell.cmd_tool import CmdNode, CmdTool, CmdResult, NodeStatus, cmd_tool, create_cmd_tool
# Backwards compatibility - DagTool is now CmdTool
from hanzo_tools.shell.dag_tool import DagNode, DagTool, DagResult, create_dag_tool
from hanzo_tools.shell.npx_tool import NpxTool, npx_tool
from hanzo_tools.shell.truncate import truncate_lines, estimate_tokens, truncate_response
from hanzo_tools.shell.uvx_tool import UvxTool, uvx_tool
# HTTP/Data tools (no shell escaping issues)
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
from hanzo_tools.shell.shellflow import parse as parse_shellflow, compile as compile_shellflow
from hanzo_tools.shell.wget_tool import WgetTool
from hanzo_tools.shell.shell_tools import (
CshTool,
KshTool,
ZshTool,
BashTool,
DashTool,
FishTool,
TcshTool,
ShellTool,
csh_tool,
ksh_tool,
zsh_tool,
bash_tool,
dash_tool,
fish_tool,
tcsh_tool,
shell_tool,
)
# Base classes
from hanzo_tools.shell.base_process import (
AUTO_BACKGROUND_TIMEOUT,
ShellExecutor,
BaseBinaryTool,
BaseScriptTool,
ProcessManager,
BaseProcessTool,
AutoBackgroundExecutor,
get_shell_executor,
)
# Shell detection
from hanzo_tools.shell.shell_detect import (
SUPPORTED_SHELLS,
ShellInfo,
detect_shells,
get_active_shell,
clear_shell_cache,
resolve_shell_path,
get_shell_tool_class,
get_cached_active_shell,
)
# Session management
from hanzo_tools.shell.session_storage import (
SessionInfo,
SessionStorage,
clear_all_sessions,
cleanup_expired_sessions,
)
def _get_detected_shell_tools() -> list:
"""
Get shell tools list with only the detected active shell.
This reduces tool clutter by only exposing the user's configured shell.
Override with:
- HANZO_MCP_SHELL=bash (shell name)
- HANZO_MCP_FORCE_SHELL=/path/to/shell (full path)
- HANZO_MCP_ALL_SHELLS=1 (expose all shells)
"""
import os
# Allow exposing all shells via env var (for debugging/testing)
if os.environ.get("HANZO_MCP_ALL_SHELLS", "").lower() in ("1", "true", "yes"):
return [
ZshTool,
BashTool,
FishTool,
DashTool,
KshTool,
TcshTool,
CshTool,
PsTool,
NpxTool,
UvxTool,
OpenTool,
CurlTool,
JqTool,
WgetTool,
]
# Detect active shell
shell_name, shell_path = get_cached_active_shell()
# Get the appropriate shell tool class
shell_tool_class = get_shell_tool_class(shell_name)
# Base tools (always included) - no cmd, just the detected shell
tools = [PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
# Add detected shell tool, or fall back to ShellTool (smart auto-detect)
if shell_tool_class:
tools.insert(0, shell_tool_class) # Put detected shell first
else:
tools.insert(0, ShellTool) # Fallback to smart shell
return tools
# Tools list for entry point discovery
# Only the detected shell is included (not all 4 shell variants)
# Override with HANZO_MCP_ALL_SHELLS=1 to expose all shells
TOOLS = _get_detected_shell_tools()
__all__ = [
# uvloop status
"_using_uvloop",
# HIP-0300: Unified proc tool
"ProcTool",
"proc_tool",
# Base classes
"ProcessManager",
"AutoBackgroundExecutor",
"ShellExecutor",
"get_shell_executor",
"BaseProcessTool",
"BaseBinaryTool",
"BaseScriptTool",
# Session management
"SessionStorage",
"SessionInfo",
"cleanup_expired_sessions",
"clear_all_sessions",
# Utilities
"truncate_response",
"truncate_lines",
"estimate_tokens",
# Shell detection
"ShellInfo",
"detect_shells",
"get_active_shell",
"get_cached_active_shell",
"clear_shell_cache",
"get_shell_tool_class",
"resolve_shell_path",
"SUPPORTED_SHELLS",
# Legacy tools
"CmdTool",
"CmdResult",
"CmdNode",
"NodeStatus",
"cmd_tool",
"create_cmd_tool",
"PsTool",
"ps_tool",
"ZshTool",
"zsh_tool",
"ShellTool",
"shell_tool",
"BashTool",
"bash_tool",
"FishTool",
"fish_tool",
"DashTool",
"dash_tool",
"KshTool",
"ksh_tool",
"TcshTool",
"tcsh_tool",
"CshTool",
"csh_tool",
# Backwards compatibility
"DagTool",
"DagResult",
"DagNode",
"create_dag_tool",
# Convenience tools
"OpenTool",
"open_tool",
"NpxTool",
"npx_tool",
"UvxTool",
"uvx_tool",
# HTTP/Data tools
"CurlTool",
"JqTool",
"WgetTool",
# Registration
"TOOLS",
"get_shell_tools",
"register_shell_tools",
]
def get_shell_tools(
permission_manager: PermissionManager,
all_tools: dict[str, BaseTool] | None = None,
shell_override: str | None = None,
) -> list[BaseTool]:
"""Create instances of shell tools.
Only the user's detected/configured shell is included to reduce tool clutter.
Args:
permission_manager: Permission manager for access control
all_tools: Dict of all registered tools (for tool invocations)
shell_override: Override shell (name like "zsh" or path like "/path/to/fish")
Returns:
List of shell tool instances
"""
import os
# Set permission manager for convenience tools
npx_tool.permission_manager = permission_manager
uvx_tool.permission_manager = permission_manager
# Base tools list (no cmd - just the detected shell and utilities)
tools = [
ps_tool, # Process management
npx_tool, # Node packages
uvx_tool, # Python packages
open_tool, # Open files/URLs
]
# Check for all-shells mode
if os.environ.get("HANZO_MCP_ALL_SHELLS", "").lower() in ("1", "true", "yes"):
tools.insert(0, ZshTool(tools=all_tools or {}))
tools.insert(1, BashTool(tools=all_tools or {}))
tools.insert(2, FishTool(tools=all_tools or {}))
tools.insert(3, DashTool(tools=all_tools or {}))
tools.insert(4, KshTool(tools=all_tools or {}))
tools.insert(5, TcshTool(tools=all_tools or {}))
tools.insert(6, CshTool(tools=all_tools or {}))
return tools
# Determine which shell to expose
if shell_override:
# Handle path override (e.g., --shell=/opt/homebrew/bin/fish)
if "/" in shell_override:
shell_name = os.path.basename(shell_override)
os.environ["HANZO_MCP_FORCE_SHELL"] = shell_override
else:
shell_name = shell_override.lower()
else:
shell_name, _ = get_cached_active_shell()
# Create and add detected shell tool, or fall back to ShellTool
shell_tool_class = get_shell_tool_class(shell_name)
if shell_tool_class:
shell_instance = shell_tool_class(tools=all_tools or {})
tools.insert(0, shell_instance) # Put detected shell first
else:
tools.insert(0, ShellTool(tools=all_tools or {})) # Fallback to smart shell
return tools
def register_shell_tools(
mcp_server: FastMCP,
permission_manager: PermissionManager,
all_tools: dict[str, BaseTool] | None = None,
shell_override: str | None = None,
) -> list[BaseTool]:
"""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 cmd tool invocations)
shell_override: Override shell (name like "zsh" or path like "/path/to/fish")
Returns:
List of registered tools
"""
tools = get_shell_tools(permission_manager, all_tools, shell_override)
ToolRegistry.register_tools(mcp_server, tools)
return tools
def register_tools(mcp_server: FastMCP, **kwargs) -> list[BaseTool]:
"""Register all shell tools with the MCP server.
This is the standard entry point called by the tool discovery system.
Supports:
- shell_override: Override detected shell (name or path)
- permission_manager: Permission manager instance
- all_tools: Dict of all registered tools
"""
from hanzo_tools.core import PermissionManager
permission_manager = kwargs.get("permission_manager") or PermissionManager()
all_tools = kwargs.get("all_tools")
shell_override = kwargs.get("shell_override") or kwargs.get("shell")
return register_shell_tools(mcp_server, permission_manager, all_tools, shell_override)
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/base_process.py
-663
View File
@@ -1,663 +0,0 @@
"""Base classes for process execution tools.
All process execution uses asyncio.subprocess for consistency.
All file I/O uses hanzo_async for non-blocking operations with uvloop support.
Auto-backgrounding timeout can be configured via:
export HANZO_AUTO_BACKGROUND_TIMEOUT=30 # seconds (default: 30)
export HANZO_AUTO_BACKGROUND_TIMEOUT=0 # disable auto-backgrounding
"""
import os
import time
import uuid
import asyncio
import tempfile
from abc import abstractmethod
from typing import Any, Dict, List, Tuple, Optional, override
from pathlib import Path
from hanzo_async import mkdir, write_file, append_file
from hanzo_tools.core import BaseTool, PermissionManager
from hanzo_tools.shell.truncate import truncate_response
# Configurable auto-background timeout (seconds)
# Set via HANZO_AUTO_BACKGROUND_TIMEOUT env var
# 0 or negative = disabled (never auto-background, use 24h timeout)
_raw_timeout = float(os.getenv("HANZO_AUTO_BACKGROUND_TIMEOUT", "30"))
AUTO_BACKGROUND_TIMEOUT = _raw_timeout if _raw_timeout > 0 else 86400.0 # 24 hours if disabled
class ProcessManager:
"""Singleton manager for background processes.
All processes are asyncio.subprocess.Process instances.
"""
_instance = None
_processes: Dict[str, asyncio.subprocess.Process] = {}
_logs: Dict[str, str] = {}
_log_dir = Path(tempfile.gettempdir()) / "hanzo_mcp_logs"
_initialized: bool = False
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
async def _ensure_log_dir(self) -> None:
"""Ensure log directory exists (async-safe)."""
if not self._initialized:
await mkdir(self._log_dir, parents=True, exist_ok=True)
self._initialized = True
def add_process(self, process_id: str, process: asyncio.subprocess.Process, log_file: str) -> None:
"""Add a process to track."""
self._processes[process_id] = process
self._logs[process_id] = log_file
def get_process(self, process_id: str) -> Optional[asyncio.subprocess.Process]:
"""Get a tracked process."""
return self._processes.get(process_id)
def remove_process(self, process_id: str) -> None:
"""Remove a process from tracking."""
self._processes.pop(process_id, None)
self._logs.pop(process_id, None)
def list_processes(self) -> Dict[str, Dict[str, Any]]:
"""List all tracked processes.
Note: Uses list(items()) to create snapshot before cleanup,
preventing RuntimeError from dict modification during iteration.
Completed processes are cleaned up after being reported.
"""
result = {}
for pid, proc in list(self._processes.items()):
is_running = proc.returncode is None
if is_running:
result[pid] = {
"pid": proc.pid,
"running": True,
"log_file": self._logs.get(pid),
}
else:
result[pid] = {
"pid": proc.pid,
"running": False,
"return_code": proc.returncode,
"log_file": self._logs.get(pid),
}
self.remove_process(pid)
return result
def get_log_file(self, process_id: str) -> Optional[Path]:
"""Get log file path for a process."""
log_path = self._logs.get(process_id)
return Path(log_path) if log_path else None
@property
def log_dir(self) -> Path:
"""Get the log directory."""
return self._log_dir
async def create_log_file(self, process_id: str) -> Path:
"""Create a log file for a process (async-safe)."""
await self._ensure_log_dir()
log_file = self._log_dir / f"{process_id}.log"
await write_file(log_file, "")
return log_file
def mark_completed(self, process_id: str, return_code: int) -> None:
"""Mark a process as completed."""
self.remove_process(process_id)
class AutoBackgroundExecutor:
"""Executor that automatically backgrounds long-running processes.
IMPORTANT: Always backgrounds after AUTO_BACKGROUND_TIMEOUT to keep
the agent loop responsive. Longer timeouts only affect the background
process lifetime, not the foreground wait time.
Configure via:
export HANZO_AUTO_BACKGROUND_TIMEOUT=30 # 30s timeout (default)
export HANZO_AUTO_BACKGROUND_TIMEOUT=0 # disabled (never auto-background)
"""
DEFAULT_TIMEOUT = AUTO_BACKGROUND_TIMEOUT
MAX_FOREGROUND_TIMEOUT = AUTO_BACKGROUND_TIMEOUT
def __init__(self, process_manager: ProcessManager, timeout: float = DEFAULT_TIMEOUT):
"""Initialize the auto-background executor."""
self.process_manager = process_manager
self.default_timeout = timeout
async def execute_with_auto_background(
self,
cmd_args: list[str],
tool_name: str,
cwd: Optional[Path] = None,
env: Optional[dict[str, str]] = None,
timeout: Optional[float] = None,
) -> Tuple[str, bool, Optional[str]]:
"""Execute a command with automatic backgrounding if it takes too long.
Returns:
Tuple of (output/status, was_backgrounded, process_id)
Note: Always backgrounds after MAX_FOREGROUND_TIMEOUT (30s) to keep
the agent loop responsive. The passed timeout is stored for reference
but doesn't extend foreground wait time.
"""
# ALWAYS cap at MAX_FOREGROUND_TIMEOUT to keep agent loop responsive
# Longer timeouts only affect background process, not foreground wait
requested_timeout = timeout if timeout is not None else self.default_timeout
effective_timeout = min(requested_timeout, self.MAX_FOREGROUND_TIMEOUT)
# Fast path for tests - still async but with shorter timeout
if os.getenv("HANZO_MCP_FAST_TESTS") == "1":
try:
proc = await asyncio.create_subprocess_exec(
*cmd_args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=str(cwd) if cwd else None,
env=env,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
if proc.returncode != 0:
return (
f"Command failed with exit code {proc.returncode}:\n"
f"{stdout.decode('utf-8', errors='replace')}"
f"{stderr.decode('utf-8', errors='replace')}",
False,
None,
)
return stdout.decode("utf-8", errors="replace"), False, None
except asyncio.TimeoutError:
return "Command timed out in test mode", False, None
except Exception as e:
return f"Error executing command: {e}", False, None
# Generate process ID
process_id = f"{tool_name}_{uuid.uuid4().hex[:8]}"
# Create log file
log_file = await self.process_manager.create_log_file(process_id)
# Start the process
process = await asyncio.create_subprocess_exec(
*cmd_args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
cwd=cwd,
env=env,
)
# Track in process manager
self.process_manager.add_process(process_id, process, str(log_file))
# Try to wait for completion with timeout
start_time = time.time()
output_lines = []
try:
async def read_output():
if process.stdout:
async for line in process.stdout:
line_str = line.decode("utf-8", errors="replace")
output_lines.append(line_str)
await append_file(log_file, line_str)
async def wait_for_process():
return await process.wait()
read_task = asyncio.create_task(read_output())
wait_task = asyncio.create_task(wait_for_process())
done, pending = await asyncio.wait(
[read_task, wait_task],
timeout=effective_timeout,
return_when=asyncio.FIRST_COMPLETED,
)
if wait_task in done:
return_code = await wait_task
try:
await asyncio.wait_for(read_task, timeout=0.5)
except asyncio.TimeoutError:
read_task.cancel()
try:
await read_task
except asyncio.CancelledError:
pass
self.process_manager.mark_completed(process_id, return_code)
output = "".join(output_lines)
if return_code != 0:
return (
f"Command failed with exit code {return_code}:\n{output}",
False,
None,
)
else:
return output, False, None
else:
# Timeout - background the process
for task in pending:
task.cancel()
asyncio.create_task(self._background_reader(process, process_id, log_file))
elapsed = time.time() - start_time
partial_output = "".join(output_lines[-50:])
return (
f"Process automatically backgrounded after {elapsed:.1f}s\n"
f"Process ID: {process_id}\n"
f"Log file: {log_file}\n\n"
f"Use 'ps --logs {process_id}' to view full output\n"
f"Use 'ps --kill {process_id}' to stop the process\n\n"
f"=== Last output ===\n{partial_output}",
True,
process_id,
)
except Exception as e:
self.process_manager.mark_completed(process_id, -1)
return f"Error executing command: {str(e)}", False, None
async def _background_reader(self, process, process_id: str, log_file: Path):
"""Continue reading output from a backgrounded process."""
try:
if process.stdout:
async for line in process.stdout:
await append_file(log_file, line.decode("utf-8", errors="replace"))
return_code = await process.wait()
self.process_manager.mark_completed(process_id, return_code)
await append_file(log_file, f"\n\n=== Process completed with exit code {return_code} ===\n")
except Exception as e:
await append_file(log_file, f"\n\n=== Background reader error: {str(e)} ===\n")
self.process_manager.mark_completed(process_id, -1)
class BaseProcessTool(BaseTool):
"""Base class for all process execution tools."""
def __init__(self, permission_manager: Optional[PermissionManager] = None):
"""Initialize the process tool."""
super().__init__()
self.permission_manager = permission_manager
self.process_manager = ProcessManager()
self.auto_background_executor = AutoBackgroundExecutor(self.process_manager)
@abstractmethod
def get_command_args(self, command: str, **kwargs) -> List[str]:
"""Get the command arguments for subprocess."""
pass
@abstractmethod
def get_tool_name(self) -> str:
"""Get the name of the tool being used."""
pass
async def execute_sync(
self,
command: str,
cwd: Optional[Path] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[int] = None,
**kwargs,
) -> str:
"""Execute a command with auto-backgrounding after 2 minutes."""
if self.permission_manager and cwd:
if not self.permission_manager.is_path_allowed(str(cwd)):
raise PermissionError(f"Access denied to path: {cwd}")
cmd_args = self.get_command_args(command, **kwargs)
process_env = os.environ.copy()
if env:
process_env.update(env)
output, was_backgrounded, process_id = await self.auto_background_executor.execute_with_auto_background(
cmd_args=cmd_args,
tool_name=self.get_tool_name(),
cwd=cwd,
env=process_env,
timeout=float(timeout) if timeout is not None else None,
)
if was_backgrounded:
return output
else:
if output.startswith("Command failed"):
raise RuntimeError(output)
max_tokens = int(os.environ.get("HANZO_MCP_MAX_RESPONSE_TOKENS", "25000"))
return truncate_response(
output,
max_tokens=max_tokens,
truncation_message=f"\n\n[Command output truncated due to {max_tokens} token limit.]",
)
async def execute_background(
self,
command: str,
cwd: Optional[Path] = None,
env: Optional[Dict[str, str]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Execute a command in the background."""
if self.permission_manager and cwd:
if not self.permission_manager.is_path_allowed(str(cwd)):
raise PermissionError(f"Access denied to path: {cwd}")
process_id = f"{self.get_tool_name()}_{uuid.uuid4().hex[:8]}"
log_file = self.process_manager.log_dir / f"{process_id}.log"
cmd_args = self.get_command_args(command, **kwargs)
process_env = os.environ.copy()
if env:
process_env.update(env)
process = await asyncio.create_subprocess_exec(
*cmd_args,
cwd=cwd,
env=process_env,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
self.process_manager.add_process(process_id, process, str(log_file))
asyncio.create_task(self._write_output_to_log(process, log_file, process_id))
return {
"process_id": process_id,
"pid": process.pid,
"log_file": str(log_file),
"status": "started",
}
async def _write_output_to_log(self, process: asyncio.subprocess.Process, log_file: Path, process_id: str) -> None:
"""Write process output to log file in background."""
try:
# Clear log file
await write_file(log_file, "")
if process.stdout:
async for line in process.stdout:
await append_file(log_file, line.decode("utf-8", errors="replace"))
return_code = await process.wait()
await append_file(log_file, f"\n=== Process completed with exit code {return_code} ===\n")
self.process_manager.mark_completed(process_id, return_code)
except Exception as e:
await append_file(log_file, f"\n=== Error: {str(e)} ===\n")
self.process_manager.mark_completed(process_id, -1)
class BaseBinaryTool(BaseProcessTool):
"""Base class for binary execution tools (like npx, uvx)."""
@abstractmethod
def get_binary_name(self) -> str:
"""Get the name of the binary to execute."""
pass
@override
def get_command_args(self, command: str, **kwargs) -> List[str]:
"""Get command arguments for binary execution."""
cmd_args = [self.get_binary_name()]
if "flags" in kwargs:
cmd_args.extend(kwargs["flags"])
cmd_args.append(command)
if "args" in kwargs:
if isinstance(kwargs["args"], str):
cmd_args.extend(kwargs["args"].split())
else:
cmd_args.extend(kwargs["args"])
return cmd_args
@override
def get_tool_name(self) -> str:
"""Get the tool name (same as binary name by default)."""
return self.get_binary_name()
class BaseScriptTool(BaseProcessTool):
"""Base class for script execution tools (like bash, python)."""
@abstractmethod
def get_interpreter(self) -> str:
"""Get the interpreter to use."""
pass
@abstractmethod
def get_script_flags(self) -> List[str]:
"""Get default flags for the interpreter."""
pass
@override
def get_command_args(self, command: str, **kwargs) -> List[str]:
"""Get command arguments for script execution."""
cmd_args = [self.get_interpreter()]
cmd_args.extend(self.get_script_flags())
cmd_args.append(command)
return cmd_args
@override
def get_tool_name(self) -> str:
"""Get the tool name (interpreter name by default)."""
return self.get_interpreter()
# Shared singleton for shell execution
_shell_executor: Optional["ShellExecutor"] = None
class ShellExecutor:
"""Shared async shell executor for all shell tools.
Ensures consistent auto-backgrounding behavior across dag, zsh, shell, bash tools.
Uses a singleton pattern to share process management state.
Configure via:
export HANZO_AUTO_BACKGROUND_TIMEOUT=30 # 30s timeout (default)
export HANZO_AUTO_BACKGROUND_TIMEOUT=0 # disabled (never auto-background)
"""
DEFAULT_TIMEOUT = AUTO_BACKGROUND_TIMEOUT
_instance: Optional["ShellExecutor"] = None
def __new__(cls) -> "ShellExecutor":
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self) -> None:
if self._initialized:
return
self._initialized = True
self._process_manager = ProcessManager()
@property
def process_manager(self) -> ProcessManager:
return self._process_manager
async def run_shell(
self,
command: str,
shell: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: float = DEFAULT_TIMEOUT,
tool_name: str = "shell",
) -> Tuple[str, str, int, bool, Optional[str]]:
"""Run a shell command with auto-backgrounding on timeout.
Args:
command: Shell command to execute
shell: Shell binary path (e.g., /bin/zsh)
cwd: Working directory
env: Environment variables
timeout: Timeout before auto-backgrounding (default: 30s)
tool_name: Tool name for process ID prefix
Returns:
Tuple of (stdout, stderr, exit_code, was_backgrounded, process_id)
"""
# Cap timeout at 30s for foreground wait - keep agent loop responsive
effective_timeout = min(timeout, self.DEFAULT_TIMEOUT)
run_env = os.environ.copy()
if env:
run_env.update(env)
shell_name = os.path.basename(shell)
process_id = f"{tool_name}_{uuid.uuid4().hex[:8]}"
log_file = await self._process_manager.create_log_file(process_id)
try:
proc = await asyncio.create_subprocess_exec(
shell,
"-c",
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
env=run_env,
# Prevent zombie processes - use start_new_session on Unix
start_new_session=True,
)
stdout_chunks: list[bytes] = []
stderr_chunks: list[bytes] = []
async def read_stdout():
if proc.stdout:
while True:
chunk = await proc.stdout.read(8192)
if not chunk:
break
stdout_chunks.append(chunk)
async def read_stderr():
if proc.stderr:
while True:
chunk = await proc.stderr.read(8192)
if not chunk:
break
stderr_chunks.append(chunk)
try:
# Read streams and wait for process with timeout
await asyncio.wait_for(
asyncio.gather(read_stdout(), read_stderr(), proc.wait()), timeout=effective_timeout
)
exit_code = proc.returncode or 0
return (
b"".join(stdout_chunks).decode("utf-8", errors="replace"),
b"".join(stderr_chunks).decode("utf-8", errors="replace"),
exit_code,
False, # Not backgrounded
None, # No process_id (completed)
)
except asyncio.TimeoutError:
# Background the process - don't kill it
partial_stdout = b"".join(stdout_chunks).decode("utf-8", errors="replace")
partial_stderr = b"".join(stderr_chunks).decode("utf-8", errors="replace")
await write_file(
log_file,
f"[{shell_name}] Command backgrounded after {effective_timeout}s timeout\n"
f"[{shell_name}] Command: {command}\n"
f"[{shell_name}] PID: {proc.pid}\n" + "-" * 40 + "\n"
f"{partial_stdout}{partial_stderr}",
)
self._process_manager.add_process(process_id, proc, str(log_file))
asyncio.create_task(self._capture_background_output(proc, log_file, process_id))
return (
f"[backgrounded] Process {process_id} (PID {proc.pid}) running in background.\n"
f"Use: ps --logs {process_id} # view output\n"
f"Use: ps --kill {process_id} # stop process",
"",
0,
True, # Was backgrounded
process_id,
)
except Exception as e:
# Clean up on error - kill process if it exists
try:
if proc and proc.returncode is None:
proc.kill()
await proc.wait()
except Exception:
pass
return (
"",
str(e),
1,
False,
None,
)
async def _capture_background_output(
self,
proc: asyncio.subprocess.Process,
log_file: Path,
process_id: str,
) -> None:
"""Capture output from backgrounded process to log file."""
try:
async def read_stream(stream, prefix: str) -> None:
if stream:
while True:
line = await stream.readline()
if not line:
break
await append_file(log_file, f"{prefix}{line.decode('utf-8', errors='replace')}")
await asyncio.gather(
read_stream(proc.stdout, ""),
read_stream(proc.stderr, "[stderr] "),
)
await proc.wait()
await append_file(log_file, f"\n[shell] Process exited with code {proc.returncode}\n")
self._process_manager.mark_completed(process_id, proc.returncode or 0)
except Exception:
self._process_manager.mark_completed(process_id, -1)
def get_shell_executor() -> ShellExecutor:
"""Get the shared shell executor singleton."""
global _shell_executor
if _shell_executor is None:
_shell_executor = ShellExecutor()
return _shell_executor
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/cmd_tool.py
-512
View File
@@ -1,512 +0,0 @@
"""Cmd tool - unified execution graph (DAG) for command execution.
The primary command execution tool for hanzo-mcp. Run commands with:
- Serial execution (default)
- Parallel execution
- Mixed execution graphs
- Tool invocations
- Auto-backgrounding at 45s
"""
import os
import shutil
import asyncio
from enum import Enum
from typing import Any, Dict, List, Union, Optional, Annotated, override
from datetime import datetime
from dataclasses import field, dataclass
from pydantic import Field
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.shell.base_process import get_shell_executor
class NodeStatus(Enum):
"""Execution status of a command node."""
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class CmdResult:
"""Result from a single command execution."""
node_id: str
command: str
stdout: str
stderr: str
status: NodeStatus
exit_code: int
duration_ms: int = 0
node_type: str = "shell"
@dataclass
class CmdNode:
"""A node in the execution graph."""
id: str
command: Union[str, Dict[str, Any]]
depends_on: List[str] = field(default_factory=list)
status: NodeStatus = NodeStatus.PENDING
result: Optional[CmdResult] = None
Command = Union[str, Dict[str, Any], List[Any]]
class CmdTool(BaseTool):
"""Unified command execution with DAG support.
The primary tool for running shell commands. Supports:
- Simple commands: cmd("ls -la")
- Serial execution: cmd(["ls", "pwd", "git status"])
- Parallel execution: cmd(["npm i", "cargo build"], parallel=True)
- Mixed DAG: cmd(["mkdir dist", {"parallel": ["cp a", "cp b"]}, "zip out"])
- Tool invocations: cmd([{"tool": "search", "input": {"pattern": "TODO"}}])
- Named deps: cmd([{"id": "a", "run": "...", "after": ["b"]}])
Auto-backgrounds commands after 45s. Uses zsh by default.
"""
name = "cmd"
def __init__(
self,
tools: Optional[Dict[str, BaseTool]] = None,
default_shell: str = "zsh",
):
"""Initialize command execution tool."""
super().__init__()
self.tools = tools or {}
self.default_shell = self._resolve_shell(default_shell)
def _resolve_shell(self, preferred: str) -> str:
"""Resolve shell - prefer zsh, fallback to bash/fish/sh.
Supports: zsh, bash, fish, sh, dash, ksh, tcsh, csh
"""
# Allow override via environment
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
# Shell priority: modern shells first
shell_priority = ["zsh", "bash", "fish", "dash", "sh"]
search_paths = [
"/opt/homebrew/bin",
"/usr/local/bin",
"/bin",
"/usr/bin",
"/usr/local/fish/bin", # Common fish location
]
for shell in shell_priority:
for prefix in search_paths:
full_path = f"{prefix}/{shell}"
if os.path.isfile(full_path) and os.access(full_path, os.X_OK):
return full_path
found = shutil.which(shell)
if found:
return found
return "sh"
@property
@override
def description(self) -> str:
shell_name = os.path.basename(self.default_shell)
return f"""Execute commands with DAG support (shell: {shell_name}).
SIMPLE:
cmd("ls -la") # Single command
cmd("A ; B ; C") # Sequential via shell
ARRAYS:
cmd(["a", "b", "c"]) # Sequential
cmd(["a", "b"], parallel=True) # All parallel
cmd(["a", ["b", "c"], "d"]) # Nested = parallel
DAG:
cmd([
"mkdir dist",
{{"parallel": ["cp a dist/", "cp b dist/"]}},
"zip -r out.zip dist/"
])
TOOL INVOCATION:
cmd([{{"tool": "search", "input": {{"pattern": "TODO"}}}}])
OPTIONS:
parallel: Run ALL top-level commands concurrently
shell: Use different shell (zsh, bash, fish, dash, sh)
strict: Stop on first error
quiet: Suppress stdout
timeout: Per-command timeout (default: 45s)
cwd: Working directory
env: Environment variables
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background.
Use ps --logs <id> to view, ps --kill <id> to stop."""
async def _run_shell(
self,
cmd: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 30,
shell: Optional[str] = None,
) -> CmdResult:
"""Run a shell command with auto-backgrounding on timeout.
Uses the shared ShellExecutor for consistent 45s auto-backgrounding.
"""
start_time = datetime.now()
node_id = f"shell_{id(cmd)}"
executor = get_shell_executor()
use_shell = shell or self.default_shell
stdout, stderr, exit_code, was_backgrounded, process_id = await executor.run_shell(
command=cmd,
shell=use_shell,
cwd=cwd,
env=env,
timeout=float(timeout),
tool_name="cmd",
)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
if was_backgrounded:
return CmdResult(
node_id=node_id,
command=cmd,
stdout=stdout,
stderr=stderr,
status=NodeStatus.SUCCESS, # Backgrounded = success from caller's POV
exit_code=0,
duration_ms=duration,
node_type="shell",
)
return CmdResult(
node_id=node_id,
command=cmd,
stdout=stdout,
stderr=stderr,
status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
exit_code=exit_code,
duration_ms=duration,
node_type="shell",
)
async def _run_tool(self, tool_name: str, tool_input: Dict[str, Any], ctx: MCPContext) -> CmdResult:
"""Run an MCP tool invocation."""
start_time = datetime.now()
node_id = f"tool_{tool_name}"
if tool_name not in self.tools:
return CmdResult(
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 CmdResult(
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 CmdResult(
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,
cwd: Optional[str],
env: Optional[Dict[str, str]],
timeout: int,
shell: Optional[str] = None,
) -> CmdResult:
"""Execute a single command node.
Supports:
- str: Single command
- list: Nested array = parallel execution
- dict with "parallel": Explicit parallel block
- dict with "tool": Tool invocation
- dict with "run": Command wrapper
"""
if isinstance(cmd, str):
return await self._run_shell(cmd, cwd, env, timeout, shell)
# Nested array = parallel execution
if isinstance(cmd, list):
tasks = [self._execute_node(c, ctx, cwd, env, timeout, shell) for c in cmd]
results = await asyncio.gather(*tasks, return_exceptions=True)
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 CmdResult(
node_id=f"parallel_{len(cmd)}",
command=f"parallel[{len(cmd)} 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",
)
if isinstance(cmd, dict):
if "tool" in cmd:
return await self._run_tool(cmd["tool"], cmd.get("input", {}), ctx)
if "parallel" in cmd:
parallel_cmds = cmd["parallel"]
tasks = [self._execute_node(c, ctx, cwd, env, timeout, shell) for c in parallel_cmds]
results = await asyncio.gather(*tasks, return_exceptions=True)
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 CmdResult(
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",
)
if "run" in cmd:
run_cmd = cmd["run"]
return await self._execute_node(run_cmd, ctx, cwd, env, timeout, shell)
return CmdResult(
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",
)
return CmdResult(
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",
)
def _format_output(self, results: List[CmdResult], quiet: bool = False) -> str:
"""Format command 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
if r.stdout and not quiet:
output_parts.append(r.stdout.rstrip())
if r.stderr:
output_parts.append(f"[stderr] {r.stderr.rstrip()}")
if len(results) > 1:
status = "✓" if failed_count == 0 else f"✗ ({failed_count} failed)"
output_parts.append(f"\n[cmd] {len(results)} nodes, {total_duration}ms, {status}")
return "\n".join(output_parts) if output_parts else "(no output)"
@override
@auto_timeout("cmd")
async def call(
self,
ctx: MCPContext,
command: Optional[str] = None,
commands: Optional[List[Command]] = None,
parallel: bool = False,
shell: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 30,
strict: bool = False,
quiet: bool = False,
**kwargs,
) -> str:
"""Execute commands with optional DAG semantics.
Args:
command: Single command string (simple mode)
commands: List of commands for DAG mode
parallel: Run all commands concurrently
shell: Shell to use (default: zsh)
cwd: Working directory
env: Additional environment variables
timeout: Per-command timeout in seconds (default: 45)
strict: Stop on first error
quiet: Suppress stdout
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
# Single command mode
if command and not commands:
result = await self._run_shell(command, cwd, env, timeout, shell)
if result.status == NodeStatus.SUCCESS:
return result.stdout if result.stdout else "(no output)"
return f"{result.stdout}\n[stderr] {result.stderr}" if result.stderr else result.stdout
# DAG mode
if not commands:
return "Error: Provide either 'command' (string) or 'commands' (list)"
results: List[CmdResult] = []
if parallel:
tasks = [self._execute_node(cmd, ctx, cwd, env, timeout, shell) for cmd in commands]
results = await asyncio.gather(*tasks, return_exceptions=True)
results = [
r
if not isinstance(r, Exception)
else CmdResult(
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:
for cmd in commands:
result = await self._execute_node(cmd, ctx, cwd, env, timeout, shell)
results.append(result)
if strict and result.status == NodeStatus.FAILED:
break
return self._format_output(results, quiet)
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register cmd tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def cmd_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[
Optional[List[Any]], Field(description="List of commands for DAG execution", default=None)
] = None,
parallel: Annotated[bool, Field(description="Run all commands in parallel", default=False)] = False,
shell: Annotated[Optional[str], Field(description="Shell: zsh, bash, sh", 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=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=shell,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
def create_cmd_tool(tools: Optional[Dict[str, BaseTool]] = None, default_shell: str = "zsh") -> CmdTool:
"""Factory to create command execution tool."""
return CmdTool(tools, default_shell)
# Singleton instance
cmd_tool = CmdTool()
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/curl_tool.py
-189
View File
@@ -1,189 +0,0 @@
"""Curl tool - HTTP client without shell escaping issues.
Provides a clean interface for HTTP requests, avoiding shell quoting problems.
"""
import json
import asyncio
from typing import Literal, Optional, Annotated, final, override
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
Method = Annotated[
Literal["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
Field(description="HTTP method"),
]
@final
class CurlTool(BaseTool):
"""HTTP client tool - curl without shell escaping nightmares.
Clean interface for HTTP requests with proper JSON handling.
"""
name = "curl"
@property
@override
def description(self) -> str:
return """HTTP client - make requests without shell escaping issues.
Examples:
curl --url "https://api.example.com/health"
curl --url "https://api.example.com/data" --method POST --json '{"key": "value"}'
curl --url "https://api.example.com/auth" --headers '{"Authorization": "Bearer token"}'
Parameters:
url: Request URL (required)
method: GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS (default: GET)
json: JSON body (will be serialized properly)
data: Raw body data
headers: Additional headers as JSON object
timeout: Request timeout in seconds (default: 30)
follow_redirects: Follow redirects (default: true)
verbose: Include response headers (default: false)
Returns formatted response with status, headers (if verbose), and body.
"""
@override
@auto_timeout("curl")
async def call(
self,
ctx: MCPContext,
url: str,
method: str = "GET",
json_body: Optional[str] = None,
data: Optional[str] = None,
headers: Optional[str] = None,
timeout: int = 30,
follow_redirects: bool = True,
verbose: bool = False,
**kwargs,
) -> str:
"""Make HTTP request."""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
# Build curl command
cmd = ["curl", "-s", "-S"] # Silent but show errors
# Method
if method != "GET":
cmd.extend(["-X", method])
# Follow redirects
if follow_redirects:
cmd.append("-L")
# Timeout
cmd.extend(["--max-time", str(timeout)])
# Include headers in output if verbose
if verbose:
cmd.append("-i")
# Headers
default_headers = {}
if headers:
try:
parsed_headers = json.loads(headers)
if isinstance(parsed_headers, dict):
default_headers.update(parsed_headers)
except json.JSONDecodeError:
return f"Error: Invalid headers JSON: {headers}"
# JSON body
if json_body:
default_headers.setdefault("Content-Type", "application/json")
# Parse and re-serialize to ensure valid JSON
try:
parsed_json = json.loads(json_body)
serialized = json.dumps(parsed_json)
cmd.extend(["-d", serialized])
except json.JSONDecodeError:
return f"Error: Invalid JSON body: {json_body}"
elif data:
cmd.extend(["-d", data])
# Add all headers
for key, value in default_headers.items():
cmd.extend(["-H", f"{key}: {value}"])
# URL (must be last)
cmd.append(url)
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout + 5, # Give curl time to timeout first
)
output = stdout.decode("utf-8", errors="replace")
errors = stderr.decode("utf-8", errors="replace")
if proc.returncode != 0:
return f"curl failed (exit {proc.returncode}):\n{errors}\n{output}"
# Try to pretty-print JSON responses
if not verbose: # Don't try to parse if headers included
try:
parsed = json.loads(output)
return json.dumps(parsed, indent=2)
except json.JSONDecodeError:
pass
return output
except asyncio.TimeoutError:
return f"Request timed out after {timeout}s"
except FileNotFoundError:
return "Error: curl not found. Install curl to use this tool."
except Exception as e:
return f"Error: {e}"
def register(self, mcp_server: FastMCP) -> None:
"""Register with MCP server."""
tool_instance = self
@mcp_server.tool()
async def curl(
url: Annotated[str, Field(description="Request URL")],
method: Method = "GET",
json_body: Annotated[
Optional[str], Field(description="JSON body (will be properly escaped)", alias="json")
] = None,
data: Annotated[Optional[str], Field(description="Raw body data")] = None,
headers: Annotated[Optional[str], Field(description="Headers as JSON object")] = None,
timeout: Annotated[int, Field(description="Timeout in seconds")] = 30,
follow_redirects: Annotated[bool, Field(description="Follow redirects")] = True,
verbose: Annotated[bool, Field(description="Include response headers")] = False,
ctx: MCPContext = None,
) -> str:
"""HTTP client - curl without shell escaping issues.
Make HTTP requests with proper JSON handling.
No more shell quoting nightmares!
"""
return await tool_instance.call(
ctx,
url=url,
method=method,
json_body=json_body,
data=data,
headers=headers,
timeout=timeout,
follow_redirects=follow_redirects,
verbose=verbose,
)
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/dag_tool.py
-485
View File
@@ -1,485 +0,0 @@
"""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 os
import uuid
import shutil
import asyncio
from enum import Enum
from typing import Any, Dict, List, Union, Optional, Annotated, override
from pathlib import Path
from datetime import datetime
from dataclasses import field, dataclass
from pydantic import Field
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.shell.base_process import get_shell_executor
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"
@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
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",
{"parallel": [
"cp manifest.json dist/",
"cp -rf assets/ dist/",
]},
"zip -r package.zip dist/"
])
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.
"""
name = "dag"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None, default_shell: str = "zsh"):
"""Initialize DAG execution tool."""
super().__init__()
self.tools = tools or {}
self.default_shell = self._resolve_shell(default_shell)
def _resolve_shell(self, preferred: str) -> str:
"""Resolve shell - prefer zsh, fallback to bash."""
shell_priority = ["zsh", "bash"]
search_paths = [
"/opt/homebrew/bin",
"/usr/local/bin",
"/bin",
"/usr/bin",
]
for shell in shell_priority:
for prefix in search_paths:
full_path = f"{prefix}/{shell}"
if os.path.isfile(full_path) and os.access(full_path, os.X_OK):
return full_path
found = shutil.which(shell)
if found:
return found
return "sh"
@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.
AUTO-BACKGROUNDING: Commands that exceed timeout are automatically
backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
async def _run_shell(
self,
cmd: str,
shell: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 30,
) -> DagResult:
"""Run a shell command with auto-backgrounding on timeout.
Uses the shared ShellExecutor for consistent 45s auto-backgrounding.
"""
start_time = datetime.now()
node_id = f"shell_{id(cmd)}"
executor = get_shell_executor()
stdout, stderr, exit_code, was_backgrounded, process_id = await executor.run_shell(
command=cmd,
shell=shell,
cwd=cwd,
env=env,
timeout=float(timeout),
tool_name="dag",
)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
if was_backgrounded:
return DagResult(
node_id=node_id,
command=cmd,
stdout=stdout,
stderr=stderr,
status=NodeStatus.SUCCESS, # Backgrounded = success from caller's POV
exit_code=0,
duration_ms=duration,
node_type="shell",
)
return DagResult(
node_id=node_id,
command=cmd,
stdout=stdout,
stderr=stderr,
status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
exit_code=exit_code,
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."""
if isinstance(cmd, str):
return await self._run_shell(cmd, shell, cwd, env, timeout)
# Nested array = auto-parallel (e.g., ["a", ["b", "c"], "d"] runs b,c in parallel)
if isinstance(cmd, list):
tasks = [self._execute_node(c, ctx, shell, cwd, env, timeout) for c in cmd]
results = await asyncio.gather(*tasks, return_exceptions=True)
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"auto_parallel_{len(cmd)}",
command=f"parallel[{len(cmd)} 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",
)
if isinstance(cmd, dict):
if "tool" in cmd:
return await self._run_tool(cmd["tool"], cmd.get("input", {}), ctx)
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)
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",
)
if "run" in cmd:
run_cmd = cmd["run"]
return await self._execute_node(run_cmd, ctx, shell, cwd, env, timeout)
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",
)
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 = 30,
strict: bool = False,
quiet: bool = False,
**kwargs,
) -> str:
"""Execute commands with DAG semantics."""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
shell = shell or self.default_shell
results: List[DagResult] = []
if 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:
for cmd in commands:
result = await self._execute_node(cmd, ctx, shell, cwd, env, timeout)
results.append(result)
if strict and result.status == NodeStatus.FAILED:
break
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
if r.stdout and not quiet:
output_parts.append(r.stdout.rstrip())
if r.stderr:
output_parts.append(f"[stderr] {r.stderr.rstrip()}")
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=30)] = 30,
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-tools-shell/build/lib/hanzo_tools/shell/jq_tool.py
-167
View File
@@ -1,167 +0,0 @@
"""JQ tool - JSON processing without shell escaping issues.
Provides a clean interface for jq queries, avoiding shell quoting problems.
"""
import json
import asyncio
from typing import Optional, Annotated, final, override
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
@final
class JqTool(BaseTool):
"""JQ tool - JSON processing without shell escaping nightmares.
Clean interface for jq queries with proper handling.
"""
name = "jq"
@property
@override
def description(self) -> str:
return """JSON processor - jq without shell escaping issues.
Examples:
jq --filter ".result.data" --input '{"result": {"data": [1,2,3]}}'
jq --filter ".[] | select(.active)" --file data.json
jq --filter "keys" --input '{"a": 1, "b": 2}'
jq --filter '.checks | to_entries[] | select(.value.error != null)' --file health.json
Parameters:
filter: jq filter expression (required)
input: JSON input as string
file: Path to JSON file (alternative to input)
raw: Output raw strings without quotes (default: false)
compact: Compact output (default: false)
slurp: Read entire input as single array (default: false)
sort_keys: Sort object keys (default: false)
The filter is passed directly to jq without shell interpretation,
so you don't need to escape special characters like ! or |
"""
@override
@auto_timeout("jq")
async def call(
self,
ctx: MCPContext,
filter: str,
input: Optional[str] = None,
file: Optional[str] = None,
raw: bool = False,
compact: bool = False,
slurp: bool = False,
sort_keys: bool = False,
**kwargs,
) -> str:
"""Process JSON with jq."""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
if not input and not file:
return "Error: Either 'input' or 'file' is required"
# Build jq command
cmd = ["jq"]
# Options
if raw:
cmd.append("-r")
if compact:
cmd.append("-c")
if slurp:
cmd.append("-s")
if sort_keys:
cmd.append("-S")
# Filter (passed as argument, not through shell)
cmd.append(filter)
# File input
if file:
cmd.append(file)
try:
if input:
# Validate input is valid JSON first
try:
json.loads(input)
except json.JSONDecodeError as e:
return f"Error: Invalid JSON input: {e}"
proc = await asyncio.create_subprocess_exec(
*cmd,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(input=input.encode("utf-8")),
timeout=30,
)
else:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=30,
)
output = stdout.decode("utf-8", errors="replace")
errors = stderr.decode("utf-8", errors="replace")
if proc.returncode != 0:
# Provide helpful error message
if "syntax error" in errors.lower():
return f"jq syntax error in filter:\n {filter}\n\nError: {errors}"
return f"jq failed (exit {proc.returncode}):\n{errors}"
return output.rstrip()
except asyncio.TimeoutError:
return "jq timed out after 30s"
except FileNotFoundError:
return "Error: jq not found. Install jq: brew install jq"
except Exception as e:
return f"Error: {e}"
def register(self, mcp_server: FastMCP) -> None:
"""Register with MCP server."""
tool_instance = self
@mcp_server.tool()
async def jq(
filter: Annotated[str, Field(description="jq filter expression")],
input: Annotated[Optional[str], Field(description="JSON input string")] = None,
file: Annotated[Optional[str], Field(description="Path to JSON file")] = None,
raw: Annotated[bool, Field(description="Output raw strings")] = False,
compact: Annotated[bool, Field(description="Compact output")] = False,
slurp: Annotated[bool, Field(description="Read as single array")] = False,
sort_keys: Annotated[bool, Field(description="Sort object keys")] = False,
ctx: MCPContext = None,
) -> str:
"""JSON processor - jq without shell escaping issues.
Process JSON with jq filters. No shell quoting nightmares!
Supports all standard jq operations.
"""
return await tool_instance.call(
ctx,
filter=filter,
input=input,
file=file,
raw=raw,
compact=compact,
slurp=slurp,
sort_keys=sort_keys,
)
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/npx_tool.py
-90
View File
@@ -1,90 +0,0 @@
"""NPX tool for both sync and background execution."""
from typing import Optional, override
from pathlib import Path
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import auto_timeout
from hanzo_tools.shell.base_process import BaseBinaryTool
class NpxTool(BaseBinaryTool):
"""Tool for running npx commands."""
name = "npx"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Run npx packages with automatic backgrounding for long-running processes.
Commands that run for more than 2 minutes will automatically continue in the background.
Usage:
npx create-react-app my-app
npx http-server -p 8080 # Auto-backgrounds after 2 minutes
npx prettier --write "**/*.js"
npx json-server db.json # Auto-backgrounds if needed"""
@override
def get_binary_name(self) -> str:
"""Get the binary name."""
return "npx"
@override
async def run(
self,
ctx: MCPContext,
package: str,
args: str = "",
cwd: Optional[str] = None,
yes: bool = True,
) -> str:
"""Run an npx command with auto-backgrounding."""
work_dir = Path(cwd).resolve() if cwd else Path.cwd()
flags = []
if yes:
flags.append("-y")
full_args = args.split() if args else []
return await self.execute_sync(
package,
cwd=work_dir,
flags=flags,
args=full_args,
timeout=None,
)
def register(self, server: FastMCP) -> None:
"""Register the tool with the MCP server."""
tool_self = self
@server.tool(name=self.name, description=self.description)
async def npx(
ctx: MCPContext,
package: str,
args: str = "",
cwd: Optional[str] = None,
yes: bool = True,
) -> str:
return await tool_self.run(ctx, package=package, args=args, cwd=cwd, yes=yes)
@auto_timeout("npx")
async def call(self, ctx: MCPContext, **params) -> str:
"""Call the tool with arguments."""
return await self.run(
ctx,
package=params["package"],
args=params.get("args", ""),
cwd=params.get("cwd"),
yes=params.get("yes", True),
)
# Create tool instance
npx_tool = NpxTool()
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/open_tool.py
-117
View File
@@ -1,117 +0,0 @@
"""Open files or URLs in the default application."""
import asyncio
import platform
import webbrowser
from typing import override
from pathlib import Path
from urllib.parse import urlparse
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout
class OpenTool(BaseTool):
"""Tool for opening files or URLs in the default application."""
name = "open"
def register(self, server: FastMCP) -> None:
"""Register the tool with the MCP server."""
tool_self = self
@server.tool(name=self.name, description=self.description)
async def open(path: str, ctx: MCPContext) -> str:
return await tool_self.run(ctx, path)
@auto_timeout("open")
async def call(self, ctx: MCPContext, **params) -> str:
"""Call the tool with arguments."""
return await self.run(ctx, params["path"])
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Open files or URLs. Platform-aware.
Usage:
open https://example.com
open ./document.pdf
open /path/to/image.png"""
async def _run_opener(self, cmd: list[str], file_path: str) -> bool:
"""Run an opener command asynchronously (fire-and-forget pattern)."""
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
try:
await asyncio.wait_for(proc.wait(), timeout=5)
return proc.returncode == 0
except asyncio.TimeoutError:
# Process is still running, which is fine for opener commands
return True
except Exception:
return False
async def run(self, ctx: MCPContext, path: str) -> str:
"""Open a file or URL in the default application."""
parsed = urlparse(path)
is_url = parsed.scheme in ("http", "https", "ftp", "file")
if is_url:
try:
# Run webbrowser.open in executor to avoid blocking
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, webbrowser.open, path)
return f"Opened URL in browser: {path}"
except Exception as e:
raise RuntimeError(f"Failed to open URL: {e}")
file_path = Path(path).expanduser().resolve()
# Check file exists asynchronously
loop = asyncio.get_running_loop()
exists = await loop.run_in_executor(None, file_path.exists)
if not exists:
raise RuntimeError(f"File not found: {file_path}")
system = platform.system().lower()
try:
if system == "darwin": # macOS
if await self._run_opener(["open", str(file_path)], str(file_path)):
return f"Opened file: {file_path}"
raise RuntimeError("Failed to open file with 'open' command")
elif system == "linux":
# Try xdg-open first
if await self._run_opener(["xdg-open", str(file_path)], str(file_path)):
return f"Opened file: {file_path}"
# Fallback to other openers
for opener in ["gnome-open", "kde-open", "exo-open"]:
if await self._run_opener([opener, str(file_path)], str(file_path)):
return f"Opened file: {file_path}"
raise RuntimeError("No suitable file opener found on Linux")
elif system == "windows":
import os
# Run os.startfile in executor to avoid blocking
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, os.startfile, str(file_path))
return f"Opened file: {file_path}"
else:
raise RuntimeError(f"Unsupported platform: {system}")
except RuntimeError:
raise
except Exception as e:
raise RuntimeError(f"Error opening file: {e}")
# Create tool instance
open_tool = OpenTool()
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/proc_tool.py
-370
View File
@@ -1,370 +0,0 @@
"""Unified process tool for HIP-0300 architecture.
This module provides a single unified 'proc' tool that handles all process operations:
- exec: Execute commands (the ONE execution primitive)
- ps: List processes
- kill: Kill processes
- logs: Get process logs
Following Unix philosophy: one tool for the Execution axis.
All command execution goes through proc.exec.
"""
import os
import json
import signal
import asyncio
from typing import Any, Dict, ClassVar, Optional
from pathlib import Path
from datetime import datetime
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import (
Paging,
BaseTool,
ToolError,
NotFoundError,
InvalidParamsError,
)
from hanzo_tools.shell.base_process import (
AUTO_BACKGROUND_TIMEOUT,
ProcessManager,
get_shell_executor,
)
class ProcTool(BaseTool):
"""Unified process execution tool (HIP-0300).
Handles all process operations on a single axis:
- exec: Execute commands
- ps: List processes
- kill: Kill processes
- logs: Get process logs
CRITICAL: exec is the ONE execution primitive.
All command execution goes through proc.exec.
"""
name: ClassVar[str] = "proc"
VERSION: ClassVar[str] = "0.12.0"
def __init__(self, tools: Optional[Dict[str, Any]] = None):
super().__init__()
self.tools = tools or {}
self.process_manager = ProcessManager()
self._shell = self._resolve_shell()
self._register_proc_actions()
def _resolve_shell(self) -> str:
"""Resolve shell - prefer zsh, fallback to bash/sh."""
import shutil
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
for shell in ["zsh", "bash", "fish", "dash", "sh"]:
for prefix in ["/opt/homebrew/bin", "/usr/local/bin", "/bin", "/usr/bin"]:
full_path = f"{prefix}/{shell}"
if os.path.isfile(full_path) and os.access(full_path, os.X_OK):
return full_path
found = shutil.which(shell)
if found:
return found
return "sh"
@property
def description(self) -> str:
shell_name = os.path.basename(self._shell)
return f"""Unified process execution tool (HIP-0300).
Actions:
- exec: Execute command (shell: {shell_name})
- ps: List processes
- kill: Kill process
- logs: Get process logs
Returns: {{proc_id, exit_code, stdout_ref, stderr_ref}}
Auto-backgrounds commands after {AUTO_BACKGROUND_TIMEOUT}s.
"""
def _register_proc_actions(self):
"""Register all process actions."""
@self.action("exec", "Execute command")
async def exec_cmd(
ctx: MCPContext,
command: str,
cwd: str | None = None,
env: dict | None = None,
timeout: int | None = None,
shell: str | None = None,
) -> dict:
"""Execute a command.
This is the ONE execution primitive per HIP-0300.
Args:
command: Command to execute
cwd: Working directory
env: Environment variables
timeout: Timeout in seconds (default: auto-background at 45s)
shell: Shell to use (default: detected shell)
Returns:
proc_id, exit_code, stdout_ref, stderr_ref
"""
if not command:
raise InvalidParamsError("Command is required", param="command")
# Resolve shell
shell_path = shell or self._shell
# Build environment
process_env = os.environ.copy()
if env:
process_env.update(env)
# Get shell executor
executor = get_shell_executor()
try:
# Execute with auto-background support
result = await executor.execute(
command,
shell=shell_path,
cwd=cwd,
env=process_env,
timeout=timeout or AUTO_BACKGROUND_TIMEOUT,
)
# Check if backgrounded
if result.get("backgrounded"):
proc_id = result.get("proc_id", "unknown")
return {
"proc_id": proc_id,
"exit_code": None,
"stdout_ref": f"proc:{proc_id}:stdout",
"stderr_ref": f"proc:{proc_id}:stderr",
"status": "running",
"message": f"Command backgrounded after {timeout or AUTO_BACKGROUND_TIMEOUT}s. Use proc(action='logs', proc_id='{proc_id}') to view output.",
}
# Command completed
exit_code = result.get("return_code", result.get("exit_code", 0))
stdout = result.get("stdout", "")
stderr = result.get("stderr", "")
return {
"proc_id": result.get("proc_id"),
"exit_code": exit_code,
"stdout": stdout,
"stderr": stderr,
"duration_ms": result.get("duration_ms"),
"status": "success" if exit_code == 0 else "failed",
}
except asyncio.TimeoutError:
raise ToolError(
code="TIMEOUT",
message=f"Command timed out after {timeout}s",
)
except Exception as e:
raise ToolError(
code="INTERNAL_ERROR",
message=f"Execution failed: {e}",
)
@self.action("ps", "List processes")
async def ps(
ctx: MCPContext,
proc_id: str | None = None,
filter: str | None = None,
) -> dict:
"""List tracked processes.
Args:
proc_id: Get specific process
filter: Filter by command pattern
Returns:
List of processes with status
"""
processes = []
for pid, info in self.process_manager.list_processes().items():
# Filter by proc_id
if proc_id and pid != proc_id:
continue
# Filter by command pattern
cmd = info.get("cmd", "")
if filter and filter.lower() not in cmd.lower():
continue
processes.append(
{
"proc_id": pid,
"pid": info.get("pid"),
"command": cmd,
"running": info.get("running", False),
"exit_code": info.get("return_code"),
"started": info.get("started"),
"log_file": info.get("log_file"),
}
)
if proc_id and not processes:
raise NotFoundError(f"Process not found: {proc_id}")
return {
"processes": processes,
"total": len(processes),
}
@self.action("kill", "Kill process")
async def kill(
ctx: MCPContext,
proc_id: str,
signal: str | int = "TERM",
) -> dict:
"""Kill a process.
Args:
proc_id: Process ID to kill
signal: Signal to send (default: TERM)
Returns:
Success status
"""
if not proc_id:
raise InvalidParamsError("proc_id is required", param="proc_id")
# Find process
procs = self.process_manager.list_processes()
if proc_id not in procs:
raise NotFoundError(f"Process not found: {proc_id}")
info = procs[proc_id]
pid = info.get("pid")
if not pid:
raise ToolError(
code="INTERNAL_ERROR",
message="Process has no PID",
)
# Resolve signal
if isinstance(signal, str):
signal_map = {
"TERM": 15,
"KILL": 9,
"INT": 2,
"HUP": 1,
"QUIT": 3,
}
sig_num = signal_map.get(signal.upper(), 15)
else:
sig_num = signal
try:
os.kill(pid, sig_num)
return {
"proc_id": proc_id,
"pid": pid,
"signal": sig_num,
"killed": True,
}
except ProcessLookupError:
return {
"proc_id": proc_id,
"pid": pid,
"signal": sig_num,
"killed": False,
"message": "Process already terminated",
}
except PermissionError:
raise ToolError(
code="PERMISSION_DENIED",
message=f"Cannot kill process {pid}: permission denied",
)
@self.action("logs", "Get process logs")
async def logs(
ctx: MCPContext,
proc_id: str,
tail: int = 100,
since: str | None = None,
) -> dict:
"""Get process stdout/stderr.
Args:
proc_id: Process ID
tail: Number of lines (default: 100)
since: ISO timestamp to filter from
Returns:
stdout and stderr content
"""
if not proc_id:
raise InvalidParamsError("proc_id is required", param="proc_id")
procs = self.process_manager.list_processes()
if proc_id not in procs:
raise NotFoundError(f"Process not found: {proc_id}")
info = procs[proc_id]
log_file = info.get("log_file")
if not log_file or not Path(log_file).exists():
return {
"proc_id": proc_id,
"stdout": "",
"stderr": "",
"message": "No log file available",
}
try:
from hanzo_async import read_file
content = await read_file(log_file, encoding="utf-8", errors="replace")
# Apply tail
lines = content.splitlines()
if tail and len(lines) > tail:
lines = lines[-tail:]
return {
"proc_id": proc_id,
"output": "\n".join(lines),
"running": info.get("running", False),
"exit_code": info.get("return_code"),
"total_lines": len(content.splitlines()),
}
except Exception as e:
raise ToolError(
code="INTERNAL_ERROR",
message=f"Failed to read logs: {e}",
)
def register(self, mcp_server: FastMCP) -> None:
"""Register as 'proc' tool with MCP server."""
tool_name = self.name
tool_description = self.description
@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)
# Backward compatibility
proc_tool = ProcTool
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/ps_tool.py
-212
View File
@@ -1,212 +0,0 @@
"""Process management tool (ps).
List, monitor, and control background processes.
Clean UNIX-style API.
"""
import signal
import asyncio
from typing import Any, Dict, List, Optional, Annotated, override
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass
from pydantic import Field
from mcp.server import FastMCP
from hanzo_async import read_file, path_exists
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout
from hanzo_tools.shell.base_process import ProcessManager
@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
"""
name = "ps"
_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 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."""
import asyncio
log_file = self.process_manager.get_log_file(proc_id)
if not log_file:
return f"No logs for process: {proc_id}"
# Check file exists asynchronously
if not await path_exists(log_file):
return f"No logs for process: {proc_id}"
try:
content = await read_file(log_file)
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
"""
if kill:
return await self._kill_process(kill, sig)
if logs:
return await self._get_logs(logs, n)
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}"
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-tools-shell/build/lib/hanzo_tools/shell/session_storage.py
-281
View File
@@ -1,281 +0,0 @@
"""Session storage for managing shell sessions.
This module provides session management with cleanup capabilities
for shell processes and background tasks.
Usage:
from hanzo_tools.shell.session_storage import SessionStorage
# Clean up expired sessions (older than max_age_seconds)
SessionStorage.cleanup_expired_sessions(max_age_seconds=300)
# Clear all active sessions
cleared = SessionStorage.clear_all_sessions()
"""
from __future__ import annotations
import os
import time
import signal
import threading
from typing import Dict, ClassVar, Optional
from pathlib import Path
from dataclasses import field, dataclass
from hanzo_tools.shell.base_process import ProcessManager
@dataclass
class SessionInfo:
"""Information about a shell session."""
session_id: str
pid: int
command: str
created_at: float = field(default_factory=time.time)
last_activity: float = field(default_factory=time.time)
log_file: Optional[Path] = None
class SessionStorage:
"""Singleton storage for managing shell sessions.
Provides centralized tracking and cleanup of shell processes
to prevent resource leaks and orphaned processes.
"""
_instance: ClassVar[Optional["SessionStorage"]] = None
_lock: ClassVar[threading.Lock] = threading.Lock()
_sessions: ClassVar[Dict[str, SessionInfo]] = {}
_process_manager: ClassVar[Optional[ProcessManager]] = None
def __new__(cls) -> "SessionStorage":
"""Singleton pattern for session storage."""
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._process_manager = ProcessManager()
return cls._instance
@classmethod
def add_session(
cls,
session_id: str,
pid: int,
command: str,
log_file: Optional[Path] = None,
) -> SessionInfo:
"""Add a new session to storage.
Args:
session_id: Unique identifier for the session
pid: Process ID of the session
command: Command that started the session
log_file: Optional path to session log file
Returns:
SessionInfo for the new session
"""
with cls._lock:
session = SessionInfo(
session_id=session_id,
pid=pid,
command=command,
log_file=log_file,
)
cls._sessions[session_id] = session
return session
@classmethod
def remove_session(cls, session_id: str) -> Optional[SessionInfo]:
"""Remove a session from storage.
Args:
session_id: Session identifier to remove
Returns:
Removed SessionInfo or None if not found
"""
with cls._lock:
return cls._sessions.pop(session_id, None)
@classmethod
def get_session(cls, session_id: str) -> Optional[SessionInfo]:
"""Get a session by ID.
Args:
session_id: Session identifier
Returns:
SessionInfo or None if not found
"""
with cls._lock:
return cls._sessions.get(session_id)
@classmethod
def list_sessions(cls) -> Dict[str, SessionInfo]:
"""List all active sessions.
Returns:
Dictionary of session_id -> SessionInfo
"""
with cls._lock:
return cls._sessions.copy()
@classmethod
def update_activity(cls, session_id: str) -> bool:
"""Update last activity timestamp for a session.
Args:
session_id: Session to update
Returns:
True if session exists and was updated
"""
with cls._lock:
if session_id in cls._sessions:
cls._sessions[session_id].last_activity = time.time()
return True
return False
@classmethod
def cleanup_expired_sessions(cls, max_age_seconds: int = 300) -> int:
"""Clean up sessions older than max_age_seconds.
This method terminates processes for expired sessions and
removes them from storage.
Args:
max_age_seconds: Maximum session age in seconds (default: 5 minutes)
Returns:
Number of sessions cleaned up
"""
current_time = time.time()
expired_ids = []
with cls._lock:
for session_id, session in cls._sessions.items():
age = current_time - session.last_activity
if age > max_age_seconds:
expired_ids.append(session_id)
cleaned = 0
for session_id in expired_ids:
if cls._terminate_session(session_id):
cleaned += 1
return cleaned
@classmethod
def clear_all_sessions(cls) -> int:
"""Clear all active sessions.
Terminates all tracked processes and clears storage.
Returns:
Number of sessions cleared
"""
with cls._lock:
session_ids = list(cls._sessions.keys())
cleared = 0
for session_id in session_ids:
if cls._terminate_session(session_id):
cleared += 1
# Also clean up any processes tracked by ProcessManager
if cls._process_manager:
for proc_id in list(cls._process_manager.list_processes().keys()):
proc = cls._process_manager.get_process(proc_id)
if proc and proc.returncode is None:
try:
proc.terminate()
cleared += 1
except Exception:
pass
return cleared
@classmethod
def _terminate_session(cls, session_id: str) -> bool:
"""Terminate a session's process and remove from storage.
Args:
session_id: Session to terminate
Returns:
True if session was terminated successfully
"""
session = cls.remove_session(session_id)
if session is None:
return False
try:
# Try to terminate the process gracefully
os.kill(session.pid, signal.SIGTERM)
# Give it a moment to terminate
time.sleep(0.1)
# Force kill if still running
try:
os.kill(session.pid, 0) # Check if process exists
os.kill(session.pid, signal.SIGKILL)
except OSError:
pass # Process already terminated
return True
except OSError:
# Process doesn't exist or permission denied
return True
except Exception:
return False
@classmethod
def get_stats(cls) -> dict:
"""Get session storage statistics.
Returns:
Dictionary with stats about sessions
"""
with cls._lock:
sessions = list(cls._sessions.values())
if not sessions:
return {
"total_sessions": 0,
"oldest_session_age": 0,
"newest_session_age": 0,
}
current_time = time.time()
ages = [current_time - s.last_activity for s in sessions]
return {
"total_sessions": len(sessions),
"oldest_session_age": max(ages) if ages else 0,
"newest_session_age": min(ages) if ages else 0,
}
# Module-level convenience functions
def cleanup_expired_sessions(max_age_seconds: int = 300) -> int:
"""Clean up expired sessions."""
return SessionStorage.cleanup_expired_sessions(max_age_seconds)
def clear_all_sessions() -> int:
"""Clear all sessions."""
return SessionStorage.clear_all_sessions()
__all__ = [
"SessionStorage",
"SessionInfo",
"cleanup_expired_sessions",
"clear_all_sessions",
]
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/shell_detect.py
-410
View File
@@ -1,410 +0,0 @@
"""Shell detection for hanzo-mcp.
Detects the user's active shell and login shell to expose only
the relevant shell tool via MCP. On Mac with Homebrew, this will
detect and use /opt/homebrew/bin/zsh if that's what the user has
configured.
Environment variables for override:
- HANZO_MCP_SHELL: Force a specific shell (e.g., "zsh", "bash")
- HANZO_MCP_FORCE_SHELL: Force a specific shell path (e.g., "/opt/homebrew/bin/zsh")
"""
from __future__ import annotations
import os
import shlex
import shutil
import platform
import subprocess
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
# Common shells (names you might see from `ps -o comm=`) and typical full paths.
KNOWN_SHELL_NAMES = {
"zsh",
"bash",
"fish",
"sh",
"ksh",
"tcsh",
"csh",
"dash",
"nu",
"xonsh",
"pwsh",
"powershell",
}
KNOWN_SHELL_PATH_BASENAMES = KNOWN_SHELL_NAMES | {"busybox"} # sometimes sh is busybox
# Shells we support with dedicated tools
SUPPORTED_SHELLS = {"zsh", "bash", "fish", "dash", "ksh", "tcsh", "csh"}
@dataclass(frozen=True)
class ShellInfo:
"""Information about detected shells."""
# "Login shell" as configured on the user account (what ssh/login uses).
login_shell: Optional[str]
# The shell that appears to have invoked this process (best-effort).
invoking_shell: Optional[str]
# $SHELL (often, but not always, matches login shell)
env_shell: Optional[str]
# Debug evidence for logging / telemetry.
evidence: Dict[str, str]
def _run(cmd: List[str]) -> Tuple[int, str, str]:
"""Run a command and return (returncode, stdout, stderr)."""
try:
p = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
return p.returncode, p.stdout.strip(), p.stderr.strip()
except Exception as e:
return 127, "", f"{type(e).__name__}: {e}"
def _which(cmd: str) -> Optional[str]:
"""Find command in PATH."""
paths = os.environ.get("PATH", "").split(os.pathsep)
for d in paths:
p = os.path.join(d, cmd)
if os.path.isfile(p) and os.access(p, os.X_OK):
return p
return None
def _parse_passwd_line(line: str) -> Optional[str]:
"""Parse passwd format: name:pw:uid:gid:gecos:dir:shell"""
parts = line.split(":")
if len(parts) >= 7:
shell = parts[6].strip()
return shell or None
return None
def get_login_shell() -> Tuple[Optional[str], Dict[str, str]]:
"""
Tries multiple sources to determine the *account login shell*.
Order of preference:
1) Python's pwd database (POSIX)
2) macOS: `id -P user` (NSS/passwd view)
3) Linux: `getent passwd user`
"""
evidence: Dict[str, str] = {}
user = os.environ.get("USER") or os.environ.get("LOGNAME") or ""
# POSIX: pwd module uses NSS; on macOS it generally reflects Directory Services.
if os.name == "posix":
try:
import pwd
if user:
shell = pwd.getpwnam(user).pw_shell
evidence["pwd.getpwnam"] = shell
if shell:
return shell, evidence
except Exception as e:
evidence["pwd.getpwnam_error"] = f"{type(e).__name__}: {e}"
system = platform.system().lower()
# macOS: `id -P user` prints passwd-style record.
if system == "darwin" and user:
rc, out, err = _run(["id", "-P", user])
evidence["id_-P_rc"] = str(rc)
if err:
evidence["id_-P_err"] = err
if rc == 0 and out:
evidence["id_-P_out"] = out
# id -P output is colon-separated, shell is the last field.
shell = out.split(":")[-1].strip() or None
if shell:
return shell, evidence
# Linux/BSD: getent is the most robust CLI view into NSS.
if user and _which("getent"):
rc, out, err = _run(["getent", "passwd", user])
evidence["getent_rc"] = str(rc)
if err:
evidence["getent_err"] = err
if rc == 0 and out:
evidence["getent_out"] = out
shell = _parse_passwd_line(out)
if shell:
return shell, evidence
# As a last resort: $SHELL (not authoritative for login shell)
env_shell = os.environ.get("SHELL")
if env_shell:
evidence["fallback_env_SHELL"] = env_shell
return env_shell, evidence
def _ps_comm(pid: int) -> Optional[str]:
"""Get executable name for a process."""
rc, out, _ = _run(["ps", "-p", str(pid), "-o", "comm="])
if rc == 0 and out:
return out.strip()
return None
def _ps_args(pid: int) -> Optional[str]:
"""Get full command line for a process."""
rc, out, _ = _run(["ps", "-p", str(pid), "-o", "args="])
if rc == 0 and out:
return out.strip()
return None
def get_invoking_shell(max_hops: int = 12) -> Tuple[Optional[str], Dict[str, str]]:
"""
Best-effort guess of the *shell that invoked this process* by walking parent PIDs.
This is useful if hanzo-mcp is launched from a user's interactive shell.
If your process is daemonized / launched by a service manager, this may return None.
"""
evidence: Dict[str, str] = {}
if os.name != "posix":
return None, {"note": "invoking shell detection is POSIX-only"}
pid = os.getpid()
ppid = os.getppid()
evidence["self_pid"] = str(pid)
evidence["self_ppid"] = str(ppid)
# Walk parent chain looking for a known shell.
cur = ppid
for hop in range(max_hops):
comm = _ps_comm(cur) or ""
args = _ps_args(cur) or ""
evidence[f"hop_{hop}_pid"] = str(cur)
if comm:
evidence[f"hop_{hop}_comm"] = comm
if args:
evidence[f"hop_{hop}_args"] = args
base = os.path.basename(comm).strip()
if base in KNOWN_SHELL_PATH_BASENAMES:
# Try to return a plausible full path if args contains one.
if os.path.isabs(comm):
return comm, evidence
# Pull first token from args if it looks like a path to shell.
try:
argv0 = shlex.split(args)[0]
except Exception:
argv0 = ""
if argv0 and (os.path.isabs(argv0) or os.path.basename(argv0) in KNOWN_SHELL_NAMES):
return argv0, evidence
return comm, evidence
# Next parent: use `ps -o ppid=` to get parent of parent.
rc, out, _ = _run(["ps", "-p", str(cur), "-o", "ppid="])
if rc != 0 or not out.strip():
break
next_ppid = int(out.strip())
if next_ppid <= 1 or next_ppid == cur:
break
cur = next_ppid
return None, evidence
def normalize_shell(shell: Optional[str]) -> Optional[str]:
"""Normalize shell path - return as-is if valid."""
if not shell:
return None
shell = shell.strip()
return shell or None
def shell_basename(shell: Optional[str]) -> Optional[str]:
"""Get the basename of a shell path (e.g., '/opt/homebrew/bin/zsh' -> 'zsh')."""
if not shell:
return None
return os.path.basename(shell).strip() or None
def choose_default_shell(info: ShellInfo) -> str:
"""
Policy: prefer invoking shell if detected; else login shell; else $SHELL; else /bin/sh.
"""
for candidate in (info.invoking_shell, info.login_shell, info.env_shell):
c = normalize_shell(candidate)
if c:
return c
# Very conservative fallback
return "C:\\Windows\\System32\\cmd.exe" if os.name == "nt" else "/bin/sh"
def detect_shells() -> ShellInfo:
"""Detect all shell information."""
login_shell, login_ev = get_login_shell()
invoking_shell, inv_ev = get_invoking_shell()
env_shell = os.environ.get("SHELL")
evidence: Dict[str, str] = {}
evidence.update({f"login.{k}": v for k, v in login_ev.items()})
evidence.update({f"invoking.{k}": v for k, v in inv_ev.items()})
if env_shell:
evidence["env.SHELL"] = env_shell
return ShellInfo(
login_shell=normalize_shell(login_shell),
invoking_shell=normalize_shell(invoking_shell),
env_shell=normalize_shell(env_shell),
evidence=evidence,
)
def resolve_shell_path(shell_name: str) -> Optional[str]:
"""
Resolve a shell name to its full path, preferring Homebrew on macOS.
Args:
shell_name: Shell name (e.g., 'zsh', 'bash')
Returns:
Full path to shell or None if not found
"""
# Check environment override first
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
if os.path.isfile(force_shell) and os.access(force_shell, os.X_OK):
return force_shell
# Priority paths - Homebrew first on macOS
search_paths = [
f"/opt/homebrew/bin/{shell_name}", # Apple Silicon Homebrew
f"/usr/local/bin/{shell_name}", # Intel Homebrew
f"/bin/{shell_name}", # System
f"/usr/bin/{shell_name}", # System alternative
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
# Fallback to which
return shutil.which(shell_name)
def get_active_shell() -> Tuple[str, str]:
"""
Get the user's active shell name and path.
Respects environment overrides:
- HANZO_MCP_SHELL: Force shell name (e.g., "zsh")
- HANZO_MCP_FORCE_SHELL: Force shell path (e.g., "/opt/homebrew/bin/zsh")
Returns:
Tuple of (shell_name, shell_path)
e.g., ("zsh", "/opt/homebrew/bin/zsh")
"""
# Check for explicit override
override_shell = os.environ.get("HANZO_MCP_SHELL")
if override_shell:
override_shell = override_shell.strip().lower()
if override_shell in SUPPORTED_SHELLS:
path = resolve_shell_path(override_shell)
if path:
return override_shell, path
# Check for explicit path override
force_path = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_path and os.path.isfile(force_path) and os.access(force_path, os.X_OK):
name = shell_basename(force_path)
if name and name in SUPPORTED_SHELLS:
return name, force_path
# If it's an unknown shell, still use it but expose as "shell"
return name or "shell", force_path
# Detect from environment
info = detect_shells()
chosen_path = choose_default_shell(info)
chosen_name = shell_basename(chosen_path)
# Map to supported shell name
if chosen_name in SUPPORTED_SHELLS:
# Resolve to best available path (prefer Homebrew)
best_path = resolve_shell_path(chosen_name)
return chosen_name, best_path or chosen_path
# Fallback: if detected shell isn't supported, default to zsh
for fallback in ["zsh", "bash", "fish", "dash"]:
path = resolve_shell_path(fallback)
if path:
return fallback, path
# Ultimate fallback
return "sh", "/bin/sh"
def get_shell_tool_class(shell_name: str):
"""
Get the appropriate shell tool class for the given shell name.
Args:
shell_name: Shell name (e.g., 'zsh', 'bash')
Returns:
Shell tool class or None if not supported
"""
# Import here to avoid circular imports
from hanzo_tools.shell.shell_tools import CshTool, KshTool, ZshTool, BashTool, DashTool, FishTool, TcshTool
shell_map = {
"zsh": ZshTool,
"bash": BashTool,
"fish": FishTool,
"dash": DashTool,
"ksh": KshTool,
"ksh93": KshTool, # Alias
"pdksh": KshTool, # Alias
"mksh": KshTool, # Alias
"tcsh": TcshTool,
"csh": CshTool,
}
return shell_map.get(shell_name.lower())
# Cache for shell detection (avoid repeated subprocess calls)
_cached_shell: Optional[Tuple[str, str]] = None
def get_cached_active_shell() -> Tuple[str, str]:
"""Get cached active shell info (detects once per process)."""
global _cached_shell
if _cached_shell is None:
_cached_shell = get_active_shell()
return _cached_shell
def clear_shell_cache():
"""Clear the shell detection cache (useful for testing)."""
global _cached_shell
_cached_shell = None
if __name__ == "__main__":
info = detect_shells()
shell_name, shell_path = get_active_shell()
print("=== Shell Detection ===")
print(f"login_shell : {info.login_shell}")
print(f"invoking_shell: {info.invoking_shell}")
print(f"env_shell : {info.env_shell}")
print(f"")
print(f"=== Active Shell ===")
print(f"name : {shell_name}")
print(f"path : {shell_path}")
print(f"")
print(f"=== Environment ===")
print(f"HANZO_MCP_SHELL : {os.environ.get('HANZO_MCP_SHELL', '(not set)')}")
print(f"HANZO_MCP_FORCE_SHELL: {os.environ.get('HANZO_MCP_FORCE_SHELL', '(not set)')}")
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/shell_tools.py
-652
View File
@@ -1,652 +0,0 @@
"""Shell tool shims - zsh, bash, fish, dash, ksh, tcsh, csh, shell wrappers over cmd.
These are thin wrappers that set the shell and delegate to CmdTool.
Use cmd directly for full control.
Supported shells:
- zsh: Z shell (default on macOS, popular on Linux)
- bash: Bourne-Again shell (most common, default on most Linux)
- fish: Friendly Interactive Shell (modern, user-friendly)
- dash: Debian Almquist Shell (fast POSIX shell, Ubuntu's /bin/sh)
- ksh: KornShell (enterprise Unix, efficient scripting)
- tcsh: TENEX C Shell (enhanced csh with command completion)
- csh: C Shell (older, C-like syntax)
- shell: Auto-selects best available (zsh > bash > fish > dash > ksh > sh)
"""
import os
import shutil
from typing import Any, Dict, List, Optional, Annotated, override
from pydantic import Field
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.shell.cmd_tool import CmdTool, Command
class ZshTool(CmdTool):
"""Zsh shell - thin wrapper over cmd with shell=zsh.
Usage:
zsh("ls -la") # Single command
zsh(["ls", "pwd"]) # Sequential
zsh(["a", "b"], parallel=True) # Parallel
"""
name = "zsh"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with zsh as default shell."""
super().__init__(tools=tools, default_shell="zsh")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve zsh shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
search_paths = [
"/opt/homebrew/bin/zsh",
"/usr/local/bin/zsh",
"/bin/zsh",
"/usr/bin/zsh",
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
found = shutil.which("zsh")
if found:
return found
# Fallback to bash if zsh not found
return shutil.which("bash") or "sh"
@property
@override
def description(self) -> str:
shell_name = os.path.basename(self.default_shell)
return f"""Zsh shell (using: {shell_name}).
SIMPLE:
zsh("ls -la") # Single command
zsh("A ; B ; C") # Sequential via shell
ARRAYS:
zsh(["a", "b", "c"]) # Sequential
zsh(["a", "b"], parallel=True) # All parallel
zsh(["a", ["b", "c"], "d"]) # Nested = parallel
OPTIONS:
parallel: Run ALL commands concurrently
strict: Stop on first error
quiet: Suppress stdout
timeout: Per-command timeout (default: 45s)
cwd: Working directory
env: Environment variables
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background.
Use ps --logs <id> to view, ps --kill <id> to stop."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register zsh tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def zsh_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[
Optional[List[Any]], Field(description="List of commands for DAG execution", default=None)
] = None,
parallel: Annotated[bool, Field(description="Run all commands in parallel", default=False)] = False,
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=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None, # Use default (zsh)
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class BashTool(CmdTool):
"""Bash shell - thin wrapper over cmd with shell=bash."""
name = "bash"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with bash as default shell."""
super().__init__(tools=tools, default_shell="bash")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve bash shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
found = shutil.which("bash")
if found:
return found
return "sh"
@property
@override
def description(self) -> str:
return """Bash shell - thin wrapper over cmd.
USAGE:
bash("ls -la") # Single command
bash(["a", "b", "c"]) # Sequential
bash(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register bash tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def bash_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None, # Use default (bash)
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class FishTool(CmdTool):
"""Fish shell - thin wrapper over cmd with shell=fish."""
name = "fish"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with fish as default shell."""
super().__init__(tools=tools, default_shell="fish")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve fish shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
search_paths = [
"/opt/homebrew/bin/fish",
"/usr/local/bin/fish",
"/usr/bin/fish",
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
found = shutil.which("fish")
if found:
return found
# Fallback to zsh/bash if fish not found
return shutil.which("zsh") or shutil.which("bash") or "sh"
@property
@override
def description(self) -> str:
shell_name = os.path.basename(self.default_shell)
return f"""Fish shell (using: {shell_name}).
USAGE:
fish("ls -la") # Single command
fish(["a", "b", "c"]) # Sequential
fish(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register fish tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def fish_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None, # Use default (fish)
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class DashTool(CmdTool):
"""Dash shell - fast POSIX-compliant shell (Ubuntu's /bin/sh)."""
name = "dash"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with dash as default shell."""
super().__init__(tools=tools, default_shell="dash")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve dash shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
# dash is often /bin/sh on Debian/Ubuntu
search_paths = [
"/bin/dash",
"/usr/bin/dash",
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
found = shutil.which("dash")
if found:
return found
# Fallback to sh (which may be dash on Ubuntu)
return "sh"
@property
@override
def description(self) -> str:
return """Dash shell - fast POSIX-compliant shell.
Dash is the Debian Almquist Shell, a POSIX-compliant shell that's
faster than bash for scripts. It's the default /bin/sh on Ubuntu/Debian.
USAGE:
dash("ls -la") # Single command
dash(["a", "b", "c"]) # Sequential
dash(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register dash tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def dash_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None, # Use default (dash)
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class KshTool(CmdTool):
"""KornShell - enterprise Unix shell with efficient scripting."""
name = "ksh"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with ksh as default shell."""
super().__init__(tools=tools, default_shell="ksh")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve ksh shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
search_paths = [
"/bin/ksh",
"/usr/bin/ksh",
"/usr/local/bin/ksh",
"/opt/homebrew/bin/ksh",
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
found = shutil.which("ksh")
if found:
return found
# Also try ksh93 and pdksh
for variant in ["ksh93", "pdksh", "mksh"]:
found = shutil.which(variant)
if found:
return found
return "sh"
@property
@override
def description(self) -> str:
return """KornShell (ksh) - enterprise Unix shell.
KornShell is historically important in enterprise Unix environments.
Efficient for scripting, still used on some commercial Unix systems.
USAGE:
ksh("ls -la") # Single command
ksh(["a", "b", "c"]) # Sequential
ksh(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register ksh tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def ksh_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class TcshTool(CmdTool):
"""TENEX C Shell - enhanced csh with command completion."""
name = "tcsh"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with tcsh as default shell."""
super().__init__(tools=tools, default_shell="tcsh")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve tcsh shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
search_paths = [
"/bin/tcsh",
"/usr/bin/tcsh",
"/usr/local/bin/tcsh",
"/opt/homebrew/bin/tcsh",
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
found = shutil.which("tcsh")
if found:
return found
# Fallback to csh
return shutil.which("csh") or "sh"
@property
@override
def description(self) -> str:
return """TENEX C Shell (tcsh) - enhanced C shell.
Tcsh is an enhanced version of csh with command completion,
command-line editing, and other improvements. Mostly legacy now
but still encountered in some environments.
USAGE:
tcsh("ls -la") # Single command
tcsh(["a", "b", "c"]) # Sequential
tcsh(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register tcsh tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def tcsh_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class CshTool(CmdTool):
"""C Shell - older shell with C-like syntax."""
name = "csh"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with csh as default shell."""
super().__init__(tools=tools, default_shell="csh")
def _resolve_shell(self, preferred: str) -> str:
"""Resolve csh shell path."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
search_paths = [
"/bin/csh",
"/usr/bin/csh",
"/usr/local/bin/csh",
]
for path in search_paths:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
found = shutil.which("csh")
if found:
return found
# Fallback to tcsh (often linked to csh)
return shutil.which("tcsh") or "sh"
@property
@override
def description(self) -> str:
return """C Shell (csh) - shell with C-like syntax.
The C shell is an older shell with syntax reminiscent of C.
Mostly legacy now but still encountered in some environments.
Consider using tcsh (enhanced csh) or bash instead.
USAGE:
csh("ls -la") # Single command
csh(["a", "b", "c"]) # Sequential
csh(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register csh tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def csh_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class ShellTool(CmdTool):
"""Smart shell - auto-selects best available shell (zsh > bash > fish > dash > ksh > sh)."""
name = "shell"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize with best available shell."""
super().__init__(tools=tools, default_shell="zsh") # Will resolve to best available
@property
@override
def description(self) -> str:
shell_name = os.path.basename(self.default_shell)
return f"""Smart shell (auto-selected: {shell_name}).
Automatically uses the best available shell:
zsh > bash > fish > dash > sh
USAGE:
shell("ls -la") # Single command
shell(["a", "b", "c"]) # Sequential
shell(["a", "b"], parallel=True) # Parallel
AUTO-BACKGROUNDING: Commands exceeding 45s auto-background."""
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register shell tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def shell_handler(
command: Annotated[Optional[str], Field(description="Single command to execute", default=None)] = None,
commands: Annotated[Optional[List[Any]], Field(description="List of commands", default=None)] = None,
parallel: Annotated[bool, Field(description="Run in parallel", default=False)] = False,
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 (seconds)", default=30)] = 30,
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,
command=command,
commands=commands,
parallel=parallel,
shell=None, # Use default
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
# Singleton instances
zsh_tool = ZshTool()
bash_tool = BashTool()
fish_tool = FishTool()
dash_tool = DashTool()
ksh_tool = KshTool()
tcsh_tool = TcshTool()
csh_tool = CshTool()
shell_tool = ShellTool()
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/shellflow.py
-498
View File
@@ -1,498 +0,0 @@
"""Shellflow - minimal DSL for DAG execution.
Syntax:
A ; B ; C → sequential (do)
{ A & B & C } → parallel (all)
A ; { B & C } ; D → mixed
Compiles to JSON AST:
{"type": "do", "steps": ["A", {"type": "all", "steps": ["B", "C"]}, "D"]}
Examples:
mkdir -p dist ; { cp a.txt dist/ & cp b.txt dist/ } ; zip -r out.zip dist/
Performance:
Optimized for high throughput with:
- Precompiled regex patterns
- Local variable caching
- Fast-path for simple commands
- LRU cache for repeated patterns
- Full type annotations for mypyc compilation
Compile with mypyc for 2-5x speedup:
mypyc hanzo_tools/shell/shellflow.py
"""
from __future__ import annotations
import re
from typing import Any, Dict, List, Final, Tuple, Union
from functools import lru_cache # noqa: TID251 - not using Stainless SDK
# Type aliases with full annotations
ASTDict = Dict[str, Any]
ASTNode = Union[str, ASTDict]
ASTTuple = Tuple[str, Tuple[Any, ...]]
TokenList = List[str]
CommandList = List[Any]
# Constants
TYPE_DO: Final[str] = "do"
TYPE_ALL: Final[str] = "all"
# Precompiled regex for performance
# Match: quoted strings, braces, semicolon, shell operators, single &
_TOKEN_PATTERN: Final[str] = r"""
(?P<dquote>"(?:[^"\\]|\\.)*") # Double-quoted string
|(?P<squote>'(?:[^'\\]|\\.)*') # Single-quoted string
|(?P<lbrace>\{) # Left brace
|(?P<rbrace>\}) # Right brace
|(?P<semi>;) # Semicolon
|(?P<shell_op>&&|\|\|) # Shell operators (keep together)
|(?P<and>&) # Single & (parallel separator)
|(?P<text>[^{};'"&|]+) # Other text (no & or |)
|(?P<pipe>\|(?!\|)) # Single | (not ||)
"""
_TOKEN_RE: Final = re.compile(_TOKEN_PATTERN, re.VERBOSE)
# Preserved token types (tuple for faster `in` check than frozenset for small n)
_PRESERVED_KINDS: Final[Tuple[str, ...]] = ("dquote", "squote", "shell_op", "pipe")
# Local reference for hot path
_finditer = _TOKEN_RE.finditer
def parse(source: str) -> ASTDict:
"""Parse shellflow source to AST.
Args:
source: Shellflow source string
Returns:
AST in object form: {"type": "do"|"all", "steps": [...]}
"""
# Strip and fast-path empty
source = source.strip()
if not source:
return {"type": TYPE_DO, "steps": []}
# Fast path: single simple command (no operators)
# Using 'in' is faster than regex for simple checks
if ";" not in source and "{" not in source and "&" not in source:
return {"type": TYPE_DO, "steps": [source]}
# Tokenize and parse
tokens: TokenList = _tokenize_fast(source)
ast: ASTDict = _parse_tokens(tokens)
return _normalize(ast)
def _tokenize_fast(source: str) -> TokenList:
"""Tokenize shellflow into commands and operators.
Optimized with local variable caching and type hints for mypyc.
"""
tokens: TokenList = []
tokens_append = tokens.append
current_parts: List[str] = []
parts_append = current_parts.append
depth: int = 0
for match in _finditer(source):
kind: str | None = match.lastgroup
value: str = match.group()
if kind in _PRESERVED_KINDS:
parts_append(value)
elif kind == "lbrace":
if depth == 0 and current_parts:
cmd: str = "".join(current_parts).strip()
if cmd:
tokens_append(cmd)
current_parts = []
parts_append = current_parts.append
depth += 1
parts_append(value)
elif kind == "rbrace":
parts_append(value)
depth -= 1
if depth == 0:
block: str = "".join(current_parts).strip()
if block:
tokens_append(block)
current_parts = []
parts_append = current_parts.append
elif kind == "semi" and depth == 0:
cmd = "".join(current_parts).strip()
if cmd:
tokens_append(cmd)
tokens_append(";")
current_parts = []
parts_append = current_parts.append
elif kind == "and" and depth == 0:
cmd = "".join(current_parts).strip()
if cmd:
tokens_append(cmd)
tokens_append("&")
current_parts = []
parts_append = current_parts.append
else:
parts_append(value)
# Flush remaining
if current_parts:
cmd = "".join(current_parts).strip()
if cmd:
tokens_append(cmd)
return tokens
def _tokenize_parallel_fast(source: str) -> TokenList:
"""Tokenize content inside { } splitting on & (but not &&).
Optimized version with full type hints.
"""
tokens: TokenList = []
tokens_append = tokens.append
current_parts: List[str] = []
parts_append = current_parts.append
depth: int = 0
for match in _finditer(source):
kind: str | None = match.lastgroup
value: str = match.group()
if kind in _PRESERVED_KINDS:
parts_append(value)
elif kind == "lbrace":
depth += 1
parts_append(value)
elif kind == "rbrace":
depth -= 1
parts_append(value)
elif kind == "and" and depth == 0:
cmd: str = "".join(current_parts).strip()
if cmd:
tokens_append(cmd)
current_parts = []
parts_append = current_parts.append
else:
parts_append(value)
if current_parts:
cmd = "".join(current_parts).strip()
if cmd:
tokens_append(cmd)
return tokens
def _parse_tokens(tokens: TokenList) -> ASTDict:
"""Parse token list to AST."""
if not tokens:
return {"type": TYPE_DO, "steps": []}
# Check if this is a single block
if len(tokens) == 1:
t: str = tokens[0]
if t.startswith("{") and t.endswith("}"):
inner: str = t[1:-1].strip()
inner_tokens: TokenList = _tokenize_parallel_fast(inner)
steps: List[ASTNode] = [_parse_single(tok) for tok in inner_tokens if tok and tok != "&"]
return {"type": TYPE_ALL, "steps": steps}
# Parse as sequential
steps = []
steps_append = steps.append
for token in tokens:
if token == ";":
continue
elif token.startswith("{") and token.endswith("}"):
inner = token[1:-1].strip()
inner_tokens = _tokenize_parallel_fast(inner)
inner_steps: List[ASTNode] = [_parse_single(tok) for tok in inner_tokens if tok and tok != "&"]
if len(inner_steps) == 1:
steps_append(inner_steps[0])
else:
steps_append({"type": TYPE_ALL, "steps": inner_steps})
else:
steps_append(token)
if len(steps) == 1:
s: ASTNode = steps[0]
if isinstance(s, dict):
return s
return {"type": TYPE_DO, "steps": steps}
return {"type": TYPE_DO, "steps": steps}
def _parse_single(token: str) -> ASTNode:
"""Parse a single token (command or nested block)."""
token = token.strip()
if token.startswith("{") and token.endswith("}"):
inner: str = token[1:-1].strip()
inner_tokens: TokenList = _tokenize_parallel_fast(inner)
steps: List[ASTNode] = [_parse_single(t) for t in inner_tokens if t]
return {"type": TYPE_ALL, "steps": steps}
return token
def _normalize(ast: ASTNode) -> ASTDict:
"""Normalize AST: flatten nested do/all, drop singletons."""
if isinstance(ast, str):
return {"type": TYPE_DO, "steps": [ast]}
node_type: str = ast.get("type", TYPE_DO)
steps: List[ASTNode] = ast.get("steps", [])
# Recursively normalize children
normalized_steps: List[ASTNode] = []
normalized_append = normalized_steps.append
for step in steps:
if isinstance(step, dict):
step = _normalize(step)
# Flatten same-type nesting
step_type: str = step.get("type", "")
if step_type == node_type:
normalized_steps.extend(step.get("steps", []))
else:
normalized_append(step)
elif step:
normalized_append(step)
# Drop singleton wrappers
if len(normalized_steps) == 1:
ns: ASTNode = normalized_steps[0]
if isinstance(ns, dict):
return ns
return {"type": TYPE_DO, "steps": normalized_steps}
if not normalized_steps:
return {"type": TYPE_DO, "steps": []}
return {"type": node_type, "steps": normalized_steps}
def to_sexp(ast: ASTNode) -> Any:
"""Convert object-form AST to S-expression form.
{"type": "do", "steps": ["A", {"type": "all", "steps": ["B", "C"]}]}
→ ["do", "A", ["all", "B", "C"]]
"""
if isinstance(ast, str):
return ast
node_type: str = ast.get("type", TYPE_DO)
steps: List[ASTNode] = ast.get("steps", [])
return [node_type] + [to_sexp(s) for s in steps]
def from_sexp(sexp: Any) -> ASTNode:
"""Convert S-expression to object-form AST.
["do", "A", ["all", "B", "C"]]
→ {"type": "do", "steps": ["A", {"type": "all", "steps": ["B", "C"]}]}
"""
if isinstance(sexp, str):
return sexp
if not sexp:
return {"type": TYPE_DO, "steps": []}
node_type: str = sexp[0]
steps: List[ASTNode] = [from_sexp(s) for s in sexp[1:]]
return {"type": node_type, "steps": steps}
def to_commands(ast: ASTNode) -> CommandList:
"""Convert AST to commands list for existing DAG executor.
Transforms:
- {"type": "do", "steps": [...]} → [...] (serial)
- {"type": "all", "steps": [...]} → [[...]] (nested = parallel)
- str → [str]
"""
if isinstance(ast, str):
return [ast]
node_type: str = ast.get("type", TYPE_DO)
steps: List[ASTNode] = ast.get("steps", [])
if node_type == TYPE_ALL:
# Parallel: return as nested list
result: CommandList = []
result_append = result.append
for step in steps:
if isinstance(step, str):
result_append(step)
else:
result.extend(to_commands(step))
return [result]
# Sequential (do)
result = []
result_append = result.append
for step in steps:
if isinstance(step, str):
result_append(step)
elif isinstance(step, dict):
step_type: str = step.get("type", "")
if step_type == TYPE_ALL:
parallel_cmds: CommandList = []
p_append = parallel_cmds.append
for s in step.get("steps", []):
if isinstance(s, str):
p_append(s)
else:
parallel_cmds.extend(to_commands(s))
result_append(parallel_cmds)
else:
result.extend(to_commands(step))
return result
def render_ascii(ast: ASTNode, indent: int = 0) -> str:
"""Render AST as ASCII tree."""
prefix: str = " " * indent
if isinstance(ast, str):
return f"{prefix}└─ {ast}\n"
node_type: str = ast.get("type", TYPE_DO)
steps: List[ASTNode] = ast.get("steps", [])
lines: List[str] = [f"{prefix}{'├─' if indent else ''}[{node_type}]\n"]
lines_append = lines.append
num_steps: int = len(steps)
for i, step in enumerate(steps):
is_last: bool = i == num_steps - 1
if isinstance(step, str):
marker: str = "└─" if is_last else "├─"
lines_append(f"{prefix} {marker} {step}\n")
else:
lines_append(render_ascii(step, indent + 1))
return "".join(lines)
# Cached compile for repeated patterns
@lru_cache(maxsize=256)
def _parse_cached(source: str) -> ASTTuple:
"""Cached version of parse that returns a hashable tuple form."""
ast: ASTDict = parse(source)
return _dict_to_tuple(ast)
def _dict_to_tuple(d: ASTNode) -> ASTTuple | str:
"""Convert AST dict to hashable tuple."""
if isinstance(d, str):
return d
node_type: str = d.get("type", TYPE_DO)
steps: List[ASTNode] = d.get("steps", [])
return (node_type, tuple(_dict_to_tuple(s) for s in steps))
def _tuple_to_dict(t: ASTTuple | str) -> ASTNode:
"""Convert tuple back to AST dict."""
if isinstance(t, str):
return t
return {"type": t[0], "steps": [_tuple_to_dict(s) for s in t[1]]}
def compile(source: str, format: str = "commands", cached: bool = False) -> Any:
"""Compile shellflow to various formats.
Args:
source: Shellflow source string
format: Output format - "ast", "sexp", "commands", "ascii"
cached: Use LRU cache for repeated patterns (default: False)
Returns:
Compiled output in requested format
"""
ast: ASTDict
if cached:
ast_tuple: ASTTuple | str = _parse_cached(source)
result: ASTNode = _tuple_to_dict(ast_tuple)
ast = result if isinstance(result, dict) else {"type": TYPE_DO, "steps": [result]}
else:
ast = parse(source)
if format == "ast":
return ast
elif format == "sexp":
return to_sexp(ast)
elif format == "commands":
return to_commands(ast)
elif format == "ascii":
return render_ascii(ast)
else:
raise ValueError(f"Unknown format: {format}")
# ============================================================================
# Inline optimized versions for hot paths
# These use minimal function calls and are designed for mypyc compilation
# ============================================================================
def parse_fast(source: str) -> ASTDict:
"""Ultra-fast parse for simple sequential commands.
Falls back to full parse for complex syntax.
"""
source = source.strip()
if not source:
return {"type": TYPE_DO, "steps": []}
# Fast path: no special characters at all
if ";" not in source and "{" not in source and "&" not in source:
return {"type": TYPE_DO, "steps": [source]}
# Fast path: simple sequential (only semicolons, no braces or ampersands)
if "{" not in source and "&" not in source:
# Split on semicolon, strip each part
parts: List[str] = [p.strip() for p in source.split(";") if p.strip()]
if parts:
return {"type": TYPE_DO, "steps": parts}
return {"type": TYPE_DO, "steps": []}
# Fall back to full parse
return parse(source)
def compile_to_commands_fast(source: str) -> CommandList:
"""Compile directly to commands list, optimized path.
Skips intermediate representations when possible.
"""
source = source.strip()
if not source:
return []
# Fast path: single command
if ";" not in source and "{" not in source and "&" not in source:
return [source]
# Fast path: simple sequential
if "{" not in source and "&" not in source:
return [p.strip() for p in source.split(";") if p.strip()]
# Full compilation
ast: ASTDict = parse(source)
return to_commands(ast)
# Aliases for backwards compatibility
_tokenize = _tokenize_fast
_tokenize_parallel = _tokenize_parallel_fast
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/truncate.py
-98
View File
@@ -1,98 +0,0 @@
"""Response truncation utilities for shell tools.
Ensures shell command output doesn't exceed token limits.
"""
import tiktoken
def estimate_tokens(text: str, model: str = "gpt-4") -> int:
"""Estimate the number of tokens in a text string.
Args:
text: The text to estimate tokens for
model: The model to use for token estimation (default: gpt-4)
Returns:
Estimated number of tokens
"""
try:
encoding = tiktoken.encoding_for_model(model)
except KeyError:
encoding = tiktoken.get_encoding("cl100k_base")
return len(encoding.encode(text))
def truncate_response(
response: str,
max_tokens: int = 20000,
truncation_message: str = "\n\n[Response truncated due to length. Please use pagination, filtering, or limit parameters to see more.]",
) -> str:
"""Truncate a response to fit within token limits.
Args:
response: The response text to truncate
max_tokens: Maximum number of tokens allowed (default: 20000)
truncation_message: Message to append when truncating
Returns:
Truncated response if needed, original response otherwise
"""
# Quick check - if response is short, no need to count tokens
if len(response) < max_tokens * 2: # Rough estimate: 1 token ≈ 2-4 chars
return response
# Estimate tokens
token_count = estimate_tokens(response)
# If within limit, return as-is
if token_count <= max_tokens:
return response
# Need to truncate
left, right = 0, len(response)
truncation_msg_tokens = estimate_tokens(truncation_message)
target_tokens = max_tokens - truncation_msg_tokens
while left < right - 1:
mid = (left + right) // 2
mid_tokens = estimate_tokens(response[:mid])
if mid_tokens <= target_tokens:
left = mid
else:
right = mid
# Find a good break point (newline or space)
truncate_at = left
for i in range(min(100, left), -1, -1):
if response[left - i] in "\n ":
truncate_at = left - i
break
return response[:truncate_at] + truncation_message
def truncate_lines(
response: str,
max_lines: int = 1000,
truncation_message: str = "\n\n[Response truncated to {max_lines} lines. Please use pagination or filtering to see more.]",
) -> str:
"""Truncate a response by number of lines.
Args:
response: The response text to truncate
max_lines: Maximum number of lines allowed (default: 1000)
truncation_message: Message template to append when truncating
Returns:
Truncated response if needed, original response otherwise
"""
lines = response.split("\n")
if len(lines) <= max_lines:
return response
truncated = "\n".join(lines[:max_lines])
return truncated + truncation_message.format(max_lines=max_lines)
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/uvx_tool.py
-90
View File
@@ -1,90 +0,0 @@
"""UVX tool for both sync and background execution."""
from typing import Optional, override
from pathlib import Path
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import auto_timeout
from hanzo_tools.shell.base_process import BaseBinaryTool
class UvxTool(BaseBinaryTool):
"""Tool for running uvx commands."""
name = "uvx"
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Run Python packages with uvx with automatic backgrounding for long-running processes.
Commands that run for more than 2 minutes will automatically continue in the background.
Usage:
uvx ruff check .
uvx mkdocs serve # Auto-backgrounds after 2 minutes
uvx black --check src/
uvx jupyter lab --port 8888 # Auto-backgrounds if needed"""
@override
def get_binary_name(self) -> str:
"""Get the binary name."""
return "uvx"
@override
async def run(
self,
ctx: MCPContext,
package: str,
args: str = "",
cwd: Optional[str] = None,
python: Optional[str] = None,
) -> str:
"""Run a uvx command with auto-backgrounding."""
work_dir = Path(cwd).resolve() if cwd else Path.cwd()
flags = []
if python:
flags.extend(["--python", python])
full_args = args.split() if args else []
return await self.execute_sync(
package,
cwd=work_dir,
flags=flags,
args=full_args,
timeout=None,
)
def register(self, server: FastMCP) -> None:
"""Register the tool with the MCP server."""
tool_self = self
@server.tool(name=self.name, description=self.description)
async def uvx(
ctx: MCPContext,
package: str,
args: str = "",
cwd: Optional[str] = None,
python: Optional[str] = None,
) -> str:
return await tool_self.run(ctx, package=package, args=args, cwd=cwd, python=python)
@auto_timeout("uvx")
async def call(self, ctx: MCPContext, **params) -> str:
"""Call the tool with arguments."""
return await self.run(
ctx,
package=params["package"],
args=params.get("args", ""),
cwd=params.get("cwd"),
python=params.get("python"),
)
# Create tool instance
uvx_tool = UvxTool()
pkg/hanzo-tools-shell/build/lib/hanzo_tools/shell/wget_tool.py
-233
View File
@@ -1,233 +0,0 @@
"""Wget tool - reliable file and site downloads.
Provides a clean interface for wget with proper handling of recursive downloads.
"""
import os
import asyncio
from typing import Optional, Annotated, final, override
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
@final
class WgetTool(BaseTool):
"""Wget tool - reliable downloads without shell escaping issues.
Supports single file downloads and full site mirroring.
"""
name = "wget"
@property
@override
def description(self) -> str:
return """Download files and mirror websites reliably.
Examples:
wget --url "https://example.com/file.zip"
wget --url "https://example.com/file.zip" --output "/tmp/myfile.zip"
wget --url "https://docs.example.com" --mirror true
wget --url "https://docs.example.com" --recursive true --depth 2
wget --url "https://example.com" --recursive true --accept "*.pdf,*.doc"
Parameters:
url: URL to download (required)
output: Output file or directory
mirror: Mirror site for offline viewing (sets recursive + timestamps)
recursive: Download recursively
depth: Maximum recursion depth (default: 5 for recursive)
accept: Accept only files matching pattern (e.g., "*.pdf,*.html")
reject: Reject files matching pattern
domains: Limit to specific domains
no_parent: Don't ascend to parent directory
continue_download: Continue partial downloads
timeout: Timeout in seconds (default: 300 for mirrors, 60 for files)
quiet: Suppress output
Returns status and download summary.
"""
@override
@auto_timeout("wget")
async def call(
self,
ctx: MCPContext,
url: str,
output: Optional[str] = None,
mirror: bool = False,
recursive: bool = False,
depth: Optional[int] = None,
accept: Optional[str] = None,
reject: Optional[str] = None,
domains: Optional[str] = None,
no_parent: bool = True,
continue_download: bool = False,
timeout: Optional[int] = None,
quiet: bool = False,
**kwargs,
) -> str:
"""Download files or mirror websites."""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
# Build wget command
cmd = ["wget"]
# Mirror mode (most common for site downloads)
if mirror:
cmd.extend(
[
"--mirror", # Turn on mirroring
"--convert-links", # Convert links for offline viewing
"--adjust-extension", # Add .html extension
"--page-requisites", # Get all assets (css, js, images)
"--no-host-directories", # Don't create host directory
]
)
recursive = True # Mirror implies recursive
# Recursive download
if recursive:
cmd.append("-r")
if depth is not None:
cmd.extend(["-l", str(depth)])
elif not mirror:
cmd.extend(["-l", "5"]) # Default depth limit
# No parent directory traversal (safety)
if no_parent:
cmd.append("--no-parent")
# Continue partial downloads
if continue_download:
cmd.append("-c")
# Accept/reject patterns
if accept:
cmd.extend(["-A", accept])
if reject:
cmd.extend(["-R", reject])
# Domain restrictions
if domains:
cmd.extend(["-D", domains])
# Output location
if output:
if recursive or mirror:
cmd.extend(["-P", output]) # Directory for recursive
else:
cmd.extend(["-O", output]) # File for single download
# Timeout
actual_timeout = timeout or (300 if (mirror or recursive) else 60)
cmd.extend(["--timeout", str(min(actual_timeout, 30))]) # Per-request timeout
cmd.extend(["--tries", "3"]) # Retry count
cmd.extend(["--waitretry", "1"]) # Wait between retries
# Progress
if quiet:
cmd.append("-q")
else:
cmd.append("--progress=dot:mega") # Show progress for large files
# User agent (avoid blocks)
cmd.extend(["--user-agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"])
# URL (last)
cmd.append(url)
try:
# Check if output is a directory (non-blocking)
cwd = None
if output:
loop = asyncio.get_event_loop()
is_dir = await loop.run_in_executor(None, os.path.isdir, output)
if is_dir:
cwd = output
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=actual_timeout + 30,
)
output_text = stdout.decode("utf-8", errors="replace")
errors = stderr.decode("utf-8", errors="replace")
# Combine stdout and stderr (wget uses stderr for progress)
combined = f"{errors}\n{output_text}".strip()
if proc.returncode != 0:
return f"wget failed (exit {proc.returncode}):\n{combined}"
# Summarize success
lines = combined.split("\n")
summary_lines = [
l
for l in lines
if any(x in l.lower() for x in ["saved", "downloaded", "finished", "total", "retrieved"])
]
if summary_lines:
return "Download complete:\n" + "\n".join(summary_lines[-5:])
return f"Download complete:\n{combined[-500:]}"
except asyncio.TimeoutError:
return f"Download timed out after {actual_timeout}s"
except FileNotFoundError:
return "Error: wget not found. Install wget: brew install wget"
except Exception as e:
return f"Error: {e}"
def register(self, mcp_server: FastMCP) -> None:
"""Register with MCP server."""
tool_instance = self
@mcp_server.tool()
async def wget(
url: Annotated[str, Field(description="URL to download")],
output: Annotated[Optional[str], Field(description="Output file or directory")] = None,
mirror: Annotated[bool, Field(description="Mirror site for offline viewing")] = False,
recursive: Annotated[bool, Field(description="Download recursively")] = False,
depth: Annotated[Optional[int], Field(description="Max recursion depth")] = None,
accept: Annotated[Optional[str], Field(description="Accept pattern (e.g. '*.pdf')")] = None,
reject: Annotated[Optional[str], Field(description="Reject pattern")] = None,
domains: Annotated[Optional[str], Field(description="Limit to domains")] = None,
no_parent: Annotated[bool, Field(description="Don't go to parent directories")] = True,
continue_download: Annotated[bool, Field(description="Continue partial downloads")] = False,
timeout: Annotated[Optional[int], Field(description="Timeout in seconds")] = None,
quiet: Annotated[bool, Field(description="Suppress output")] = False,
ctx: MCPContext = None,
) -> str:
"""Download files and mirror websites reliably.
Supports single file downloads and full site mirroring.
Handles retries, timeouts, and link conversion automatically.
"""
return await tool_instance.call(
ctx,
url=url,
output=output,
mirror=mirror,
recursive=recursive,
depth=depth,
accept=accept,
reject=reject,
domains=domains,
no_parent=no_parent,
continue_download=continue_download,
timeout=timeout,
quiet=quiet,
)