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

feat: add hanzo-async unified async I/O with uvloop support

Browse Source 
New package: hanzo-async
- Automatic uvloop detection and configuration
- Async file I/O (read_file, write_file, append_file, read_json, write_json)
- Async path operations (path_exists, is_file, is_dir, mkdir, etc.)
- Async subprocess execution (run_command, run_shell)
- Graceful fallback to asyncio on Windows or when uvloop not installed

Updated packages to use hanzo-async:
- hanzo-tools-shell: base_process.py, ps_tool.py, shell_tools.py
- hanzo-tools-agent: agent_tool.py, dataset.py
- hanzo-tools-fs: search.py
- hanzo-mcp: cli.py, auto_timeout.py, event_loop.py, version_tool.py

Shell improvements:
- Renamed zsh_tool.py to shell_tools.py (better organization)
- Added DashTool for Debian/Ubuntu's fast POSIX shell
- All shell tools now thin shims over CmdTool

Version info now shows async backend:
- CLI: hanzo-mcp 0.10.27 (async: uvloop 0.22.1)
- Tool: async: uvloop 0.22.1
This commit is contained in:
zeekay
2026-01-03 17:39:48 -08:00
parent af7272fba4
commit 98c079f4cb
28 changed files with 2746 additions and 1003 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-async/README.md
+128
View File
@@ -0,0 +1,128 @@
# hanzo-async
Unified async I/O for Hanzo AI - high-performance with uvloop fallback.
## Features
- **Automatic uvloop configuration** - Falls back to asyncio on Windows
- **Async file I/O** - Read, write, append with aiofiles
- **Async path operations** - exists, is_file, mkdir, etc.
- **Async subprocess execution** - run_command, run_shell
- **Consistent patterns** - Same API across all Hanzo packages
## Installation
```bash
pip install hanzo-async
# With uvloop support (recommended for macOS/Linux)
pip install hanzo-async[uvloop]
```
## Quick Start
```python
from hanzo_async import (
# Loop configuration
configure_loop,
using_uvloop,
# File operations
read_file,
write_file,
read_json,
write_json,
# Path operations
path_exists,
is_file,
is_dir,
mkdir,
# Process operations
run_command,
run_shell,
)
# Check if using uvloop
if using_uvloop():
print("Using uvloop for high-performance async")
# Async file operations
async def example():
# Read/write files
content = await read_file("/path/to/file.txt")
await write_file("/path/to/output.txt", content)
# JSON handling
data = await read_json("/path/to/config.json")
data["updated"] = True
await write_json("/path/to/config.json", data)
# Path operations
if await path_exists("/path/to/file"):
if await is_file("/path/to/file"):
await unlink("/path/to/file")
await mkdir("/path/to/new/dir", parents=True, exist_ok=True)
# Run commands
stdout, stderr, code = await run_command("ls", "-la")
stdout, stderr, code = await run_shell("echo $HOME && pwd")
```
## API Reference
### Loop Configuration
- `configure_loop()` - Configure uvloop (auto-called on import)
- `using_uvloop()` - Check if uvloop is active
- `get_loop()` - Get current event loop
### File Operations
- `read_file(path)` - Read file content
- `write_file(path, content)` - Write content to file
- `append_file(path, content)` - Append to file
- `read_json(path)` - Read and parse JSON
- `write_json(path, data)` - Write JSON to file
- `read_lines(path)` - Read file as lines
- `write_lines(path, lines)` - Write lines to file
- `read_bytes(path)` - Read file as bytes
- `write_bytes(path, content)` - Write bytes to file
### Path Operations
- `path_exists(path)` - Check if path exists
- `is_file(path)` - Check if path is file
- `is_dir(path)` - Check if path is directory
- `mkdir(path, parents=False, exist_ok=False)` - Create directory
- `rmdir(path)` - Remove empty directory
- `unlink(path, missing_ok=False)` - Remove file
- `stat(path)` - Get file/directory stats
- `listdir(path)` - List directory contents
- `glob(pattern, root_dir=None, recursive=False)` - Find files
- `rename(src, dst)` - Rename/move file
- `copy(src, dst)` - Copy file
- `copytree(src, dst)` - Copy directory tree
- `rmtree(path)` - Remove directory tree
### Process Operations
- `run_command(*args)` - Run command with args
- `run_shell(command)` - Run shell command
- `check_command(command)` - Check if command exists
- `which(command)` - Find command path
- `start_background(*args)` - Start background process
## Why hanzo-async?
1. **Consistent async patterns** - All Hanzo packages use the same I/O patterns
2. **High performance** - uvloop provides 2-4x faster async on Unix
3. **Automatic fallback** - Works on Windows with standard asyncio
4. **No event loop blocking** - All operations are truly async
5. **Easy to use** - Simple function-based API
## License
MIT
pkg/hanzo-async/hanzo_async/__init__.py
+161
View File
@@ -0,0 +1,161 @@
"""Hanzo Async - Unified async I/O for Hanzo AI.
High-performance async operations with uvloop backend and asyncio fallback.
Features:
- Automatic uvloop configuration (falls back to asyncio on Windows)
- Async file I/O (read, write, append, exists, mkdir)
- Async path operations (exists, is_file, is_dir, stat)
- Async subprocess execution
- Consistent patterns across all Hanzo packages
Usage:
from hanzo_async import read_file, write_file, path_exists, run_command
from hanzo_async import configure_loop, using_uvloop
# Check if using uvloop
if using_uvloop():
print("Using uvloop for high-performance async")
# Async file operations
content = await read_file("/path/to/file")
await write_file("/path/to/file", "content")
# Async path operations
if await path_exists("/path/to/file"):
...
# Async subprocess
stdout, stderr, code = await run_command("ls", "-la")
"""
import sys
import asyncio
from functools import lru_cache
# Track uvloop configuration state
_uvloop_configured = False
_using_uvloop = False
def configure_loop() -> bool:
"""Configure the event loop for high performance.
Uses uvloop on macOS/Linux, falls back to asyncio on Windows.
Safe to call multiple times - only configures once.
Returns:
True if uvloop is active, False if using asyncio fallback
"""
global _uvloop_configured, _using_uvloop
if _uvloop_configured:
return _using_uvloop
_uvloop_configured = True
# Windows doesn't support uvloop
if sys.platform == "win32":
_using_uvloop = False
return False
try:
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
_using_uvloop = True
return True
except ImportError:
_using_uvloop = False
return False
def using_uvloop() -> bool:
"""Check if uvloop is active.
Returns:
True if using uvloop, False if using asyncio
"""
if not _uvloop_configured:
configure_loop()
return _using_uvloop
def get_loop() -> asyncio.AbstractEventLoop:
"""Get the current event loop, creating if necessary.
Handles Python 3.10+ deprecation of get_event_loop() in non-async contexts.
Returns:
The current event loop
"""
try:
return asyncio.get_running_loop()
except RuntimeError:
# Not in async context, create new loop
if not _uvloop_configured:
configure_loop()
return asyncio.new_event_loop()
# Auto-configure on import
configure_loop()
# Export file operations
from hanzo_async.files import (
read_file,
read_json,
write_file,
write_json,
append_file,
read_lines,
write_lines,
)
# Export path operations
from hanzo_async.paths import (
path_exists,
is_file,
is_dir,
mkdir,
rmdir,
unlink,
stat,
listdir,
glob,
)
# Export process operations
from hanzo_async.process import (
run_command,
run_shell,
check_command,
)
__all__ = [
# Loop configuration
"configure_loop",
"using_uvloop",
"get_loop",
# File operations
"read_file",
"read_json",
"write_file",
"write_json",
"append_file",
"read_lines",
"write_lines",
# Path operations
"path_exists",
"is_file",
"is_dir",
"mkdir",
"rmdir",
"unlink",
"stat",
"listdir",
"glob",
# Process operations
"run_command",
"run_shell",
"check_command",
]
pkg/hanzo-async/hanzo_async/files.py
+218
View File
@@ -0,0 +1,218 @@
"""Async file operations.
Provides non-blocking file I/O using aiofiles.
All operations are fully async and safe for use in event loops.
"""
import json
from typing import Any, List, Optional, Union
from pathlib import Path
import aiofiles
import aiofiles.os
async def read_file(
path: Union[str, Path],
encoding: str = "utf-8",
errors: str = "replace",
) -> str:
"""Read entire file content asynchronously.
Args:
path: Path to file
encoding: Text encoding (default: utf-8)
errors: Error handling mode (default: replace)
Returns:
File content as string
Raises:
FileNotFoundError: If file doesn't exist
PermissionError: If file can't be read
"""
async with aiofiles.open(path, "r", encoding=encoding, errors=errors) as f:
return await f.read()
async def read_json(
path: Union[str, Path],
encoding: str = "utf-8",
) -> Any:
"""Read and parse JSON file asynchronously.
Args:
path: Path to JSON file
encoding: Text encoding (default: utf-8)
Returns:
Parsed JSON data
Raises:
FileNotFoundError: If file doesn't exist
json.JSONDecodeError: If JSON is invalid
"""
content = await read_file(path, encoding=encoding)
return json.loads(content)
async def write_file(
path: Union[str, Path],
content: str,
encoding: str = "utf-8",
mkdir_parents: bool = True,
) -> None:
"""Write content to file asynchronously.
Args:
path: Path to file
content: Content to write
encoding: Text encoding (default: utf-8)
mkdir_parents: Create parent directories if needed (default: True)
Raises:
PermissionError: If file can't be written
"""
path = Path(path)
if mkdir_parents:
from hanzo_async.paths import mkdir
await mkdir(path.parent, parents=True, exist_ok=True)
async with aiofiles.open(path, "w", encoding=encoding) as f:
await f.write(content)
async def write_json(
path: Union[str, Path],
data: Any,
encoding: str = "utf-8",
indent: int = 2,
mkdir_parents: bool = True,
) -> None:
"""Write data as JSON file asynchronously.
Args:
path: Path to file
data: Data to serialize as JSON
encoding: Text encoding (default: utf-8)
indent: JSON indentation (default: 2)
mkdir_parents: Create parent directories if needed (default: True)
Raises:
TypeError: If data is not JSON serializable
PermissionError: If file can't be written
"""
content = json.dumps(data, indent=indent, ensure_ascii=False)
await write_file(path, content, encoding=encoding, mkdir_parents=mkdir_parents)
async def append_file(
path: Union[str, Path],
content: str,
encoding: str = "utf-8",
mkdir_parents: bool = True,
) -> None:
"""Append content to file asynchronously.
Args:
path: Path to file
content: Content to append
encoding: Text encoding (default: utf-8)
mkdir_parents: Create parent directories if needed (default: True)
Raises:
PermissionError: If file can't be written
"""
path = Path(path)
if mkdir_parents:
from hanzo_async.paths import mkdir
await mkdir(path.parent, parents=True, exist_ok=True)
async with aiofiles.open(path, "a", encoding=encoding) as f:
await f.write(content)
async def read_lines(
path: Union[str, Path],
encoding: str = "utf-8",
strip: bool = True,
) -> List[str]:
"""Read file lines asynchronously.
Args:
path: Path to file
encoding: Text encoding (default: utf-8)
strip: Strip whitespace from lines (default: True)
Returns:
List of lines
Raises:
FileNotFoundError: If file doesn't exist
"""
content = await read_file(path, encoding=encoding)
lines = content.splitlines()
if strip:
lines = [line.strip() for line in lines]
return lines
async def write_lines(
path: Union[str, Path],
lines: List[str],
encoding: str = "utf-8",
mkdir_parents: bool = True,
) -> None:
"""Write lines to file asynchronously.
Args:
path: Path to file
lines: Lines to write
encoding: Text encoding (default: utf-8)
mkdir_parents: Create parent directories if needed (default: True)
Raises:
PermissionError: If file can't be written
"""
content = "\n".join(lines) + "\n"
await write_file(path, content, encoding=encoding, mkdir_parents=mkdir_parents)
async def read_bytes(path: Union[str, Path]) -> bytes:
"""Read file as bytes asynchronously.
Args:
path: Path to file
Returns:
File content as bytes
Raises:
FileNotFoundError: If file doesn't exist
"""
async with aiofiles.open(path, "rb") as f:
return await f.read()
async def write_bytes(
path: Union[str, Path],
content: bytes,
mkdir_parents: bool = True,
) -> None:
"""Write bytes to file asynchronously.
Args:
path: Path to file
content: Bytes to write
mkdir_parents: Create parent directories if needed (default: True)
Raises:
PermissionError: If file can't be written
"""
path = Path(path)
if mkdir_parents:
from hanzo_async.paths import mkdir
await mkdir(path.parent, parents=True, exist_ok=True)
async with aiofiles.open(path, "wb") as f:
await f.write(content)
pkg/hanzo-async/hanzo_async/paths.py
+246
View File
@@ -0,0 +1,246 @@
"""Async path operations.
Provides non-blocking path operations using run_in_executor.
All operations are fully async and safe for use in event loops.
"""
import os
import asyncio
from glob import glob as sync_glob
from typing import List, Optional, Union
from pathlib import Path
async def _run_in_executor(func, *args):
"""Run a sync function in executor."""
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, func, *args)
async def path_exists(path: Union[str, Path]) -> bool:
"""Check if path exists asynchronously.
Args:
path: Path to check
Returns:
True if path exists
"""
path = Path(path)
return await _run_in_executor(path.exists)
async def is_file(path: Union[str, Path]) -> bool:
"""Check if path is a file asynchronously.
Args:
path: Path to check
Returns:
True if path is a file
"""
path = Path(path)
return await _run_in_executor(path.is_file)
async def is_dir(path: Union[str, Path]) -> bool:
"""Check if path is a directory asynchronously.
Args:
path: Path to check
Returns:
True if path is a directory
"""
path = Path(path)
return await _run_in_executor(path.is_dir)
async def mkdir(
path: Union[str, Path],
parents: bool = False,
exist_ok: bool = False,
) -> None:
"""Create directory asynchronously.
Args:
path: Directory path to create
parents: Create parent directories (default: False)
exist_ok: Don't error if exists (default: False)
Raises:
FileExistsError: If directory exists and exist_ok=False
PermissionError: If permission denied
"""
path = Path(path)
def _mkdir():
path.mkdir(parents=parents, exist_ok=exist_ok)
await _run_in_executor(_mkdir)
async def rmdir(path: Union[str, Path]) -> None:
"""Remove empty directory asynchronously.
Args:
path: Directory to remove
Raises:
OSError: If directory not empty
FileNotFoundError: If directory doesn't exist
"""
path = Path(path)
await _run_in_executor(path.rmdir)
async def unlink(path: Union[str, Path], missing_ok: bool = False) -> None:
"""Remove file asynchronously.
Args:
path: File to remove
missing_ok: Don't error if file missing (default: False)
Raises:
FileNotFoundError: If file doesn't exist and missing_ok=False
"""
path = Path(path)
def _unlink():
path.unlink(missing_ok=missing_ok)
await _run_in_executor(_unlink)
async def stat(path: Union[str, Path]) -> os.stat_result:
"""Get file/directory stats asynchronously.
Args:
path: Path to stat
Returns:
os.stat_result with file info
Raises:
FileNotFoundError: If path doesn't exist
"""
path = Path(path)
return await _run_in_executor(path.stat)
async def listdir(path: Union[str, Path] = ".") -> List[str]:
"""List directory contents asynchronously.
Args:
path: Directory to list (default: current directory)
Returns:
List of entry names
Raises:
FileNotFoundError: If directory doesn't exist
NotADirectoryError: If path is not a directory
"""
return await _run_in_executor(os.listdir, str(path))
async def glob(
pattern: str,
root_dir: Optional[Union[str, Path]] = None,
recursive: bool = False,
) -> List[str]:
"""Find files matching pattern asynchronously.
Args:
pattern: Glob pattern (e.g., "*.py", "**/*.txt")
root_dir: Root directory for search (default: current directory)
recursive: Enable ** pattern (default: False)
Returns:
List of matching paths
"""
def _glob():
if root_dir:
old_cwd = os.getcwd()
os.chdir(str(root_dir))
try:
return sync_glob(pattern, recursive=recursive)
finally:
os.chdir(old_cwd)
else:
return sync_glob(pattern, recursive=recursive)
return await _run_in_executor(_glob)
async def rename(src: Union[str, Path], dst: Union[str, Path]) -> None:
"""Rename/move file or directory asynchronously.
Args:
src: Source path
dst: Destination path
Raises:
FileNotFoundError: If source doesn't exist
FileExistsError: If destination exists
"""
src = Path(src)
dst = Path(dst)
await _run_in_executor(src.rename, dst)
async def copy(src: Union[str, Path], dst: Union[str, Path]) -> None:
"""Copy file asynchronously.
Args:
src: Source file path
dst: Destination file path
Raises:
FileNotFoundError: If source doesn't exist
PermissionError: If permission denied
"""
import shutil
await _run_in_executor(shutil.copy2, str(src), str(dst))
async def copytree(
src: Union[str, Path],
dst: Union[str, Path],
dirs_exist_ok: bool = False,
) -> None:
"""Copy directory tree asynchronously.
Args:
src: Source directory path
dst: Destination directory path
dirs_exist_ok: Don't error if dst exists (default: False)
Raises:
FileNotFoundError: If source doesn't exist
FileExistsError: If destination exists and dirs_exist_ok=False
"""
import shutil
def _copytree():
shutil.copytree(str(src), str(dst), dirs_exist_ok=dirs_exist_ok)
await _run_in_executor(_copytree)
async def rmtree(path: Union[str, Path], ignore_errors: bool = False) -> None:
"""Remove directory tree asynchronously.
Args:
path: Directory to remove
ignore_errors: Ignore errors during removal (default: False)
Raises:
FileNotFoundError: If directory doesn't exist and ignore_errors=False
"""
import shutil
def _rmtree():
shutil.rmtree(str(path), ignore_errors=ignore_errors)
await _run_in_executor(_rmtree)
pkg/hanzo-async/hanzo_async/process.py
+236
View File
@@ -0,0 +1,236 @@
"""Async process operations.
Provides non-blocking subprocess execution.
All operations are fully async and safe for use in event loops.
"""
import asyncio
import shutil
from typing import Dict, List, Optional, Tuple, Union
async def run_command(
*args: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[float] = None,
capture: bool = True,
) -> Tuple[str, str, int]:
"""Run command asynchronously.
Args:
*args: Command and arguments
cwd: Working directory (default: current)
env: Environment variables to add
timeout: Timeout in seconds (default: None = no timeout)
capture: Capture stdout/stderr (default: True)
Returns:
Tuple of (stdout, stderr, exit_code)
Raises:
asyncio.TimeoutError: If timeout exceeded
FileNotFoundError: If command not found
Examples:
stdout, stderr, code = await run_command("ls", "-la")
stdout, stderr, code = await run_command("git", "status", cwd="/repo")
"""
import os
# Merge environment
full_env = None
if env:
full_env = os.environ.copy()
full_env.update(env)
# Create subprocess
proc = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE if capture else asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.PIPE if capture else asyncio.subprocess.DEVNULL,
cwd=cwd,
env=full_env,
)
try:
if timeout:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
else:
stdout, stderr = await proc.communicate()
stdout_str = stdout.decode("utf-8", errors="replace") if stdout else ""
stderr_str = stderr.decode("utf-8", errors="replace") if stderr else ""
return stdout_str, stderr_str, proc.returncode or 0
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise
async def run_shell(
command: str,
shell: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[float] = None,
capture: bool = True,
) -> Tuple[str, str, int]:
"""Run shell command asynchronously.
Args:
command: Shell command string
shell: Shell to use (default: auto-detect zsh > bash > sh)
cwd: Working directory (default: current)
env: Environment variables to add
timeout: Timeout in seconds (default: None = no timeout)
capture: Capture stdout/stderr (default: True)
Returns:
Tuple of (stdout, stderr, exit_code)
Raises:
asyncio.TimeoutError: If timeout exceeded
FileNotFoundError: If shell not found
Examples:
stdout, stderr, code = await run_shell("ls -la && pwd")
stdout, stderr, code = await run_shell("echo $PATH", shell="bash")
"""
import os
# Auto-detect shell
if not shell:
shell = _detect_shell()
# Merge environment
full_env = None
if env:
full_env = os.environ.copy()
full_env.update(env)
# Create subprocess with shell
proc = await asyncio.create_subprocess_exec(
shell, "-c", command,
stdout=asyncio.subprocess.PIPE if capture else asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.PIPE if capture else asyncio.subprocess.DEVNULL,
cwd=cwd,
env=full_env,
)
try:
if timeout:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
else:
stdout, stderr = await proc.communicate()
stdout_str = stdout.decode("utf-8", errors="replace") if stdout else ""
stderr_str = stderr.decode("utf-8", errors="replace") if stderr else ""
return stdout_str, stderr_str, proc.returncode or 0
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise
async def check_command(command: str) -> bool:
"""Check if command is available.
Args:
command: Command name to check
Returns:
True if command is available
"""
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, shutil.which, command)
return result is not None
def _detect_shell() -> str:
"""Detect the best available shell.
Returns:
Path to shell (zsh > bash > fish > sh)
"""
import os
# Check environment variable override
force_shell = os.environ.get("HANZO_SHELL")
if force_shell:
return force_shell
# Shell priority
shells = ["zsh", "bash", "fish", "sh"]
search_paths = [
"/opt/homebrew/bin",
"/usr/local/bin",
"/bin",
"/usr/bin",
]
for shell in shells:
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"
async def which(command: str) -> Optional[str]:
"""Find command path asynchronously.
Args:
command: Command name to find
Returns:
Full path to command, or None if not found
"""
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, shutil.which, command)
async def start_background(
*args: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
) -> asyncio.subprocess.Process:
"""Start a background process.
Args:
*args: Command and arguments
cwd: Working directory (default: current)
env: Environment variables to add
Returns:
asyncio.subprocess.Process handle
Examples:
proc = await start_background("python", "server.py")
# Later...
proc.terminate()
await proc.wait()
"""
import os
# Merge environment
full_env = None
if env:
full_env = os.environ.copy()
full_env.update(env)
return await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
env=full_env,
)
pkg/hanzo-async/pyproject.toml
+38
View File
@@ -0,0 +1,38 @@
[build-system]
requires = ["setuptools>=61.0.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "hanzo-async"
version = "0.1.0"
description = "Unified async I/O for Hanzo AI - high-performance with uvloop fallback"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
]
keywords = ["hanzo", "async", "uvloop", "aiofiles", "high-performance"]
dependencies = [
"aiofiles>=23.2.1",
]
[project.optional-dependencies]
uvloop = ["uvloop>=0.21.0; sys_platform != 'win32'"]
full = [
"uvloop>=0.21.0; sys_platform != 'win32'",
]
dev = ["pytest>=7.0.0", "pytest-asyncio>=0.21.0", "ruff>=0.14.0"]
[project.urls]
"Homepage" = "https://github.com/hanzoai/python-sdk"
[tool.setuptools.packages.find]
where = ["."]
include = ["hanzo_async*"]
[tool.setuptools.package-data]
hanzo_async = ["py.typed"]
pkg/hanzo-mcp/hanzo_mcp/cli.py
+19 -3
View File
@@ -95,9 +95,25 @@ def main() -> None:
except Exception:
_version = "unknown"
# Get async backend info
try:
from hanzo_async import using_uvloop
if using_uvloop():
try:
import uvloop
_async_backend = f"uvloop {uvloop.__version__}"
except ImportError:
_async_backend = "uvloop"
else:
_async_backend = "asyncio"
except ImportError:
_async_backend = "asyncio"
_version_info = f"hanzo-mcp {_version} (async: {_async_backend})"
parser = argparse.ArgumentParser(description="MCP server implementing Hanzo AI capabilities")
parser.add_argument("--version", action="version", version=f"hanzo-mcp {_version}")
parser.add_argument("--version", action="version", version=_version_info)
_ = parser.add_argument(
"--transport",
@@ -179,8 +195,8 @@ def main() -> None:
"--command-timeout",
dest="command_timeout",
type=str,
default="120s",
help="Default timeout for command execution (default: 120s). Supports: 2min, 5m, 120s, 30sec, 1.5h",
default="45s",
help="Default timeout for command execution (default: 45s). Supports: 2min, 5m, 120s, 30sec, 1.5h",
)
_ = parser.add_argument(
pkg/hanzo-mcp/hanzo_mcp/tools/common/auto_timeout.py
+20 -17
View File
@@ -14,7 +14,7 @@ from typing import Any, Tuple, Callable, Optional
from pathlib import Path
from collections.abc import Awaitable
import aiofiles
from hanzo_async import append_file
from mcp.server.fastmcp import Context as MCPContext
from .timeout_parser import parse_timeout, format_timeout
@@ -94,33 +94,36 @@ class MCPToolTimeoutManager:
"""
try:
# Log start (async)
async with aiofiles.open(log_file, "a") as f:
await f.write(f"=== Background execution started for {tool_name} ===\n")
await f.write(f"Parameters: {json.dumps(params, indent=2, default=str)}\n")
await f.write(f"Started at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n")
await append_file(log_file,
f"=== Background execution started for {tool_name} ===\n"
f"Parameters: {json.dumps(params, indent=2, default=str)}\n"
f"Started at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n"
)
# Execute the tool
result = await tool_func(ctx, **params)
# Log completion (async)
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n\n=== Tool execution completed ===\n")
await f.write(f"Completed at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
await f.write(f"Result length: {len(str(result))} characters\n")
await f.write("\n=== RESULT ===\n")
await f.write(str(result))
await f.write("\n=== END RESULT ===\n")
await append_file(log_file,
f"\n\n=== Tool execution completed ===\n"
f"Completed at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Result length: {len(str(result))} characters\n"
f"\n=== RESULT ===\n"
f"{str(result)}\n"
f"=== END RESULT ===\n"
)
# Mark as completed
self.process_manager.mark_completed(process_id, 0)
except Exception as e:
# Log error (async)
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n\n=== Tool execution failed ===\n")
await f.write(f"Failed at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
await f.write(f"Error: {str(e)}\n")
await f.write(f"Error type: {type(e).__name__}\n")
await append_file(log_file,
f"\n\n=== Tool execution failed ===\n"
f"Failed at: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Error: {str(e)}\n"
f"Error type: {type(e).__name__}\n"
)
self.process_manager.mark_completed(process_id, 1)
pkg/hanzo-mcp/hanzo_mcp/tools/common/tool.py
+27 -20
View File
@@ -3,7 +3,6 @@
Combines install, uninstall, upgrade, enable, disable, and list into a single tool.
"""
import json
from typing import Any, Unpack, Literal, Optional, Annotated, TypedDict, final, override
from pathlib import Path
@@ -14,6 +13,9 @@ from hanzo_mcp.tools.common.base import BaseTool
from hanzo_mcp.tools.common.context import create_tool_context
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
# Import async I/O utilities
from hanzo_async import read_json, write_json, path_exists, mkdir
Action = Annotated[
Literal[
"install", # Install a tool package
@@ -58,35 +60,37 @@ class UnifiedToolTool(BaseTool):
def __init__(self):
"""Initialize the tool."""
if not UnifiedToolTool._initialized:
self._load_states()
UnifiedToolTool._initialized = True
# Note: Async initialization happens on first use via _ensure_initialized
pass
@classmethod
def _load_states(cls):
"""Load tool states from config file."""
if cls._config_file.exists():
async def _ensure_initialized(cls):
"""Ensure states are loaded (async)."""
if not cls._initialized:
await cls._load_states_async()
cls._initialized = True
@classmethod
async def _load_states_async(cls):
"""Load tool states from config file (async)."""
if await path_exists(cls._config_file):
try:
with open(cls._config_file, "r") as f:
cls._tool_states = json.load(f)
cls._tool_states = await read_json(cls._config_file)
except Exception:
cls._tool_states = {}
else:
cls._tool_states = {}
@classmethod
def _save_states(cls):
"""Save tool states to config file."""
cls._config_file.parent.mkdir(parents=True, exist_ok=True)
with open(cls._config_file, "w") as f:
json.dump(cls._tool_states, f, indent=2)
async def _save_states_async(cls):
"""Save tool states to config file (async)."""
await mkdir(cls._config_file.parent, parents=True, exist_ok=True)
await write_json(cls._config_file, cls._tool_states)
@classmethod
def is_tool_enabled(cls, tool_name: str) -> bool:
"""Check if a tool is enabled."""
if not cls._initialized:
cls._load_states()
cls._initialized = True
"""Check if a tool is enabled (sync, uses cached state)."""
# Uses cached state - call _ensure_initialized first in async contexts
return cls._tool_states.get(tool_name, True)
@property
@@ -130,6 +134,9 @@ Examples:
**params: Unpack[ToolParams],
) -> str:
"""Execute tool management action."""
# Ensure async initialization
await self._ensure_initialized()
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
@@ -251,7 +258,7 @@ Description: {config.description}"""
self._tool_states[name] = True
if persist:
self._save_states()
await self._save_states_async()
return f"✓ Enabled tool '{name}'" + ("" if persist else " (temporary)")
@@ -270,7 +277,7 @@ Description: {config.description}"""
self._tool_states[name] = False
if persist:
self._save_states()
await self._save_states_async()
return f"○ Disabled tool '{name}'" + ("" if persist else " (temporary)")
pkg/hanzo-mcp/hanzo_mcp/tools/common/tool_enable.py
+28 -28
View File
@@ -1,6 +1,5 @@
"""Enable tools dynamically."""
import json
from typing import Unpack, Annotated, TypedDict, final, override
from pathlib import Path
@@ -11,6 +10,9 @@ from hanzo_mcp.tools.common.base import BaseTool
from hanzo_mcp.tools.common.context import create_tool_context
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
# Import async I/O utilities
from hanzo_async import read_json, write_json, path_exists, mkdir
ToolName = Annotated[
str,
Field(
@@ -46,33 +48,36 @@ class ToolEnableTool(BaseTool):
def __init__(self):
"""Initialize the tool enable tool."""
if not ToolEnableTool._initialized:
self._load_states()
ToolEnableTool._initialized = True
# Async initialization happens on first use
pass
@classmethod
def _load_states(cls):
"""Load tool states from config file."""
if cls._config_file.exists():
async def _ensure_initialized(cls):
"""Ensure states are loaded (async)."""
if not cls._initialized:
await cls._load_states_async()
cls._initialized = True
@classmethod
async def _load_states_async(cls):
"""Load tool states from config file (async)."""
if await path_exists(cls._config_file):
try:
with open(cls._config_file, "r") as f:
cls._tool_states = json.load(f)
cls._tool_states = await read_json(cls._config_file)
except Exception:
cls._tool_states = {}
else:
# Default all tools to enabled
cls._tool_states = {}
@classmethod
def _save_states(cls):
"""Save tool states to config file."""
cls._config_file.parent.mkdir(parents=True, exist_ok=True)
with open(cls._config_file, "w") as f:
json.dump(cls._tool_states, f, indent=2)
async def _save_states_async(cls):
"""Save tool states to config file (async)."""
await mkdir(cls._config_file.parent, parents=True, exist_ok=True)
await write_json(cls._config_file, cls._tool_states)
@classmethod
def is_tool_enabled(cls, tool_name: str) -> bool:
"""Check if a tool is enabled.
"""Check if a tool is enabled (sync, uses cached state).
Args:
tool_name: Name of the tool
@@ -80,20 +85,12 @@ class ToolEnableTool(BaseTool):
Returns:
True if enabled (default), False if explicitly disabled
"""
# Load states if not initialized
if not cls._initialized:
cls._load_states()
cls._initialized = True
# Default to enabled if not in states
# Uses cached state - call _ensure_initialized first in async contexts
return cls._tool_states.get(tool_name, True)
@classmethod
def get_all_states(cls) -> dict:
"""Get all tool states."""
if not cls._initialized:
cls._load_states()
cls._initialized = True
"""Get all tool states (sync, uses cached state)."""
return cls._tool_states.copy()
@property
@@ -135,6 +132,9 @@ Use 'tool_list' to see all available tools and their status.
Returns:
Result of enabling the tool
"""
# Ensure async initialization
await self._ensure_initialized()
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
@@ -154,9 +154,9 @@ Use 'tool_list' to see all available tools and their status.
# Enable the tool
self._tool_states[tool_name] = True
# Persist if requested
# Persist if requested (async)
if persist:
self._save_states()
await self._save_states_async()
await tool_ctx.info(f"Enabled tool '{tool_name}' (persisted)")
else:
await tool_ctx.info(f"Enabled tool '{tool_name}' (temporary)")
pkg/hanzo-mcp/hanzo_mcp/tools/common/version_tool.py
+13
View File
@@ -6,6 +6,7 @@ from typing import Any
from mcp.server.fastmcp import FastMCP
import hanzo_mcp
from hanzo_async import using_uvloop
def register_version_tool(mcp: FastMCP) -> None:
@@ -19,14 +20,25 @@ def register_version_tool(mcp: FastMCP) -> None:
- hanzo-mcp version
- Python version
- Platform info
- Async backend (uvloop or asyncio)
"""
import platform
# Get uvloop version if active
async_backend = "asyncio"
if using_uvloop():
try:
import uvloop
async_backend = f"uvloop {uvloop.__version__}"
except ImportError:
async_backend = "uvloop"
info = {
"hanzo_mcp": hanzo_mcp.__version__,
"python": sys.version.split()[0],
"platform": platform.system(),
"arch": platform.machine(),
"async": async_backend,
}
# Return formatted string for clean display
@@ -34,5 +46,6 @@ def register_version_tool(mcp: FastMCP) -> None:
f"hanzo-mcp: v{info['hanzo_mcp']}",
f"python: {info['python']}",
f"platform: {info['platform']} ({info['arch']})",
f"async: {info['async']}",
]
return "\n".join(lines)
pkg/hanzo-mcp/hanzo_mcp/utils/event_loop.py
+20 -21
View File
@@ -2,12 +2,16 @@
This module provides utilities for configuring the asyncio event loop
with optional uvloop support for improved performance on Linux/macOS.
Uses hanzo_async for unified async I/O configuration.
"""
import sys
import asyncio
from typing import Optional
from hanzo_async import configure_loop, using_uvloop
def configure_event_loop(*, quiet: bool = False) -> Optional[str]:
"""Configure the event loop with uvloop if available.
@@ -21,27 +25,24 @@ def configure_event_loop(*, quiet: bool = False) -> Optional[str]:
Returns:
The name of the event loop policy being used, or None if default
"""
# uvloop is not available on Windows
if sys.platform == "win32":
return None
try:
import uvloop
# Install uvloop as the default event loop policy
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# Use hanzo_async for unified configuration
if configure_loop():
if not quiet:
import logging
try:
import uvloop
logger = logging.getLogger(__name__)
logger.debug(f"Using uvloop {uvloop.__version__} for event loop")
except ImportError:
pass
logger = logging.getLogger(__name__)
logger.debug(f"Using uvloop {uvloop.__version__} for event loop")
try:
import uvloop
return f"uvloop-{uvloop.__version__}"
except ImportError:
return None
return f"uvloop-{uvloop.__version__}"
except ImportError:
# uvloop not installed, use default asyncio event loop
return None
return None
def get_event_loop_info() -> dict:
@@ -59,17 +60,15 @@ def get_event_loop_info() -> dict:
"loop_class": type(loop).__name__ if loop else None,
"loop_module": type(loop).__module__ if loop else None,
"platform": sys.platform,
"using_uvloop": using_uvloop(),
}
# Check if uvloop is being used
# Check if uvloop is available
try:
import uvloop
info["uvloop_available"] = True
info["uvloop_version"] = uvloop.__version__
info["using_uvloop"] = isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy)
except ImportError:
info["uvloop_available"] = False
info["using_uvloop"] = False
return info
pkg/hanzo-mcp/pyproject.toml
+4 -3
View File
@@ -23,11 +23,12 @@ dependencies = [
"pydantic>=2.12.5",
"pydantic-settings>=2.12.0",
"typing-extensions>=4.13.0",
"aiofiles>=23.2.1",
# Unified async I/O (includes aiofiles + uvloop)
"hanzo-async>=0.1.0",
# Hanzo tool infrastructure (all tools)
"hanzo-tools>=0.3.0",
"hanzo-tools-fs>=0.1.0",
"hanzo-tools-shell>=0.2.0",
"hanzo-tools-shell>=0.5.0",
"hanzo-tools-browser>=0.1.0",
"hanzo-tools-memory>=0.1.0",
"hanzo-tools-todo>=0.1.0",
@@ -76,7 +77,7 @@ test = [
"pytest-mock>=3.10.0",
"pytest-asyncio>=0.26.0",
]
performance = ["ujson>=5.7.0", "orjson>=3.9.0", "uvloop>=0.21.0; sys_platform != 'win32'"]
performance = ["ujson>=5.7.0", "orjson>=3.9.0"] # uvloop now standard
publish = ["twine>=4.0.2", "build>=1.0.3"]
[project.scripts]
pkg/hanzo-tools-agent/hanzo_tools/agent/agent_tool.py
+10 -14
View File
@@ -20,7 +20,6 @@ from pathlib import Path
from contextlib import suppress
from dataclasses import field, dataclass
import aiofiles
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
@@ -28,13 +27,10 @@ from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
from hanzo_tools.shell import ProcessManager
# Optional uvloop for high-performance async
try:
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
HAS_UVLOOP = True
except ImportError:
HAS_UVLOOP = False
# Unified async I/O with uvloop support
from hanzo_async import append_file, configure_loop, using_uvloop
configure_loop() # Auto-configure uvloop if available
HAS_UVLOOP = using_uvloop()
# Optional httpx for API mode
try:
@@ -665,8 +661,7 @@ Consensus: https://github.com/luxfi/consensus
async for line in proc.stdout:
line_str = line.decode("utf-8", errors="replace")
output_lines.append(line_str)
async with aiofiles.open(log_file, "a") as f:
await f.write(line_str)
await append_file(log_file, line_str)
read_task = asyncio.create_task(read_output())
wait_task = asyncio.create_task(proc.wait())
@@ -699,10 +694,11 @@ Consensus: https://github.com/luxfi/consensus
partial = "".join(output_lines)
# Write backgrounding message to log
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n[agent] Backgrounded after {timeout}s timeout\n")
await f.write(f"[agent] Process ID: {process_id}\n")
await f.write(f"[agent] PID: {proc.pid}\n")
await append_file(log_file,
f"\n[agent] Backgrounded after {timeout}s timeout\n"
f"[agent] Process ID: {process_id}\n"
f"[agent] PID: {proc.pid}\n"
)
bg_msg = (
f"[backgrounded] Agent {agent} running in background.\n"
pkg/hanzo-tools-agent/hanzo_tools/agent/dataset.py
+311
View File
@@ -0,0 +1,311 @@
"""Dataset collection for agent telemetry.
Captures prompts, responses, and full telemetry data for training dataset annotation.
Usage:
from hanzo_tools.agent.dataset import DatasetCollector
collector = DatasetCollector("my_dataset.jsonl")
# Collect single sample
await collector.collect("claude", "Explain quicksort")
# Collect batch
await collector.collect_batch([
("claude", "What is recursion?"),
("gemini", "Explain binary search"),
])
# Save dataset
collector.save()
"""
import json
import time
import asyncio
from pathlib import Path
from datetime import datetime
from typing import Optional, List, Tuple, Any, Dict
from dataclasses import dataclass, asdict, field
from hanzo_async import append_file, write_file
from .agent_tool import AgentTool, Result, Telemetry
@dataclass
class DatasetSample:
"""Single dataset sample with full telemetry."""
# Core fields
id: str
timestamp: str
agent: str
prompt: str
response: str
# Status
ok: bool
error: Optional[str] = None
# Timing
latency_ms: int = 0
# Token usage
input_tokens: int = 0
output_tokens: int = 0
cache_read_tokens: int = 0
cache_creation_tokens: int = 0
# Cost
cost_usd: float = 0.0
# Model info
model: Optional[str] = None
service_name: Optional[str] = None
service_version: Optional[str] = None
# Full telemetry for annotation
telemetry_raw: Optional[Dict[str, Any]] = None
# Metadata
tags: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
class DatasetCollector:
"""Collect dataset samples from agent runs."""
def __init__(self, output_path: str = "dataset.jsonl", auto_save: bool = True):
"""Initialize collector.
Args:
output_path: Path to output JSONL file
auto_save: Auto-save after each sample
"""
self.output_path = Path(output_path)
self.auto_save = auto_save
self.samples: List[DatasetSample] = []
self.tool = AgentTool()
self._sample_count = 0
# Load existing samples if file exists
if self.output_path.exists():
self._load_existing()
def _load_existing(self):
"""Load existing samples from file."""
try:
with open(self.output_path, "r") as f:
for line in f:
if line.strip():
data = json.loads(line)
self.samples.append(DatasetSample(**data))
self._sample_count = len(self.samples)
print(f"Loaded {len(self.samples)} existing samples from {self.output_path}")
except Exception as e:
print(f"Could not load existing samples: {e}")
def _generate_id(self) -> str:
"""Generate unique sample ID."""
self._sample_count += 1
return f"sample_{self._sample_count:06d}_{int(time.time())}"
async def collect(
self,
agent: str,
prompt: str,
timeout: int = 60,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> DatasetSample:
"""Collect a single dataset sample.
Args:
agent: Agent to use (claude, gemini, etc.)
prompt: Prompt to send
timeout: Timeout in seconds
tags: Optional tags for categorization
metadata: Optional additional metadata
Returns:
DatasetSample with full telemetry
"""
result = await self.tool._exec(agent, prompt, None, timeout)
sample = DatasetSample(
id=self._generate_id(),
timestamp=datetime.utcnow().isoformat() + "Z",
agent=result.agent,
prompt=prompt,
response=result.output,
ok=result.ok,
error=result.error,
latency_ms=result.ms,
tags=tags or [],
metadata=metadata or {},
)
# Add telemetry if available
if result.telemetry:
t = result.telemetry
sample.input_tokens = t.input_tokens
sample.output_tokens = t.output_tokens
sample.cache_read_tokens = t.cache_read_input_tokens
sample.cache_creation_tokens = t.cache_creation_input_tokens
sample.cost_usd = t.cost_usd
sample.model = t.model
sample.service_name = t.service_name
sample.service_version = t.service_version
sample.telemetry_raw = t.raw
self.samples.append(sample)
if self.auto_save:
await self._append_sample_async(sample)
return sample
async def collect_batch(
self,
prompts: List[Tuple[str, str]],
timeout: int = 60,
max_concurrent: int = 5,
tags: Optional[List[str]] = None,
) -> List[DatasetSample]:
"""Collect batch of samples with concurrency control.
Args:
prompts: List of (agent, prompt) tuples
timeout: Timeout per sample
max_concurrent: Max concurrent requests
tags: Tags to apply to all samples
Returns:
List of DatasetSamples
"""
sem = asyncio.Semaphore(max_concurrent)
async def collect_one(agent: str, prompt: str, idx: int) -> DatasetSample:
async with sem:
sample = await self.collect(
agent, prompt, timeout,
tags=tags,
metadata={"batch_index": idx}
)
print(f"[{idx+1}/{len(prompts)}] {agent}: {len(sample.response)} chars, ${sample.cost_usd:.4f}")
return sample
tasks = [
collect_one(agent, prompt, i)
for i, (agent, prompt) in enumerate(prompts)
]
return await asyncio.gather(*tasks)
async def _append_sample_async(self, sample: DatasetSample):
"""Append single sample to file (non-blocking)."""
await append_file(self.output_path, json.dumps(asdict(sample)) + "\n")
def _append_sample(self, sample: DatasetSample):
"""Append single sample to file (sync version for compatibility)."""
with open(self.output_path, "a") as f:
f.write(json.dumps(asdict(sample)) + "\n")
async def save_async(self):
"""Save all samples to file asynchronously (overwrites)."""
content = "\n".join(json.dumps(asdict(sample)) for sample in self.samples)
if content:
content += "\n"
await write_file(self.output_path, content)
print(f"Saved {len(self.samples)} samples to {self.output_path}")
def save(self):
"""Save all samples to file (sync version, overwrites)."""
with open(self.output_path, "w") as f:
for sample in self.samples:
f.write(json.dumps(asdict(sample)) + "\n")
print(f"Saved {len(self.samples)} samples to {self.output_path}")
def stats(self) -> Dict[str, Any]:
"""Get dataset statistics."""
if not self.samples:
return {"count": 0}
total_cost = sum(s.cost_usd for s in self.samples)
total_input = sum(s.input_tokens for s in self.samples)
total_output = sum(s.output_tokens for s in self.samples)
total_cache = sum(s.cache_read_tokens for s in self.samples)
avg_latency = sum(s.latency_ms for s in self.samples) / len(self.samples)
agents = {}
models = {}
for s in self.samples:
agents[s.agent] = agents.get(s.agent, 0) + 1
if s.model:
models[s.model] = models.get(s.model, 0) + 1
return {
"count": len(self.samples),
"total_cost_usd": total_cost,
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"total_cache_read_tokens": total_cache,
"avg_latency_ms": avg_latency,
"success_rate": sum(1 for s in self.samples if s.ok) / len(self.samples),
"agents": agents,
"models": models,
}
async def build_dataset(
prompts: List[str],
agents: List[str] = ["claude"],
output: str = "dataset.jsonl",
max_concurrent: int = 3,
timeout: int = 60,
) -> DatasetCollector:
"""Build a dataset from prompts.
Args:
prompts: List of prompts to collect
agents: Agents to use (will cycle through)
output: Output JSONL path
max_concurrent: Max concurrent requests
timeout: Timeout per request
Returns:
DatasetCollector with collected samples
"""
collector = DatasetCollector(output, auto_save=True)
# Create prompt-agent pairs, cycling through agents
pairs = [
(agents[i % len(agents)], prompt)
for i, prompt in enumerate(prompts)
]
print(f"Collecting {len(prompts)} samples across {len(agents)} agents...")
start = time.time()
await collector.collect_batch(pairs, timeout=timeout, max_concurrent=max_concurrent)
elapsed = time.time() - start
stats = collector.stats()
print(f"\n=== Dataset Stats ===")
print(f"Samples: {stats['count']}")
print(f"Total cost: ${stats['total_cost_usd']:.4f}")
print(f"Total tokens: {stats['total_input_tokens']}→{stats['total_output_tokens']}")
print(f"Cache reads: {stats['total_cache_read_tokens']}")
print(f"Avg latency: {stats['avg_latency_ms']:.0f}ms")
print(f"Success rate: {stats['success_rate']*100:.1f}%")
print(f"Time: {elapsed:.1f}s")
print(f"Agents: {stats['agents']}")
print(f"Models: {stats['models']}")
print(f"\nSaved to: {output}")
return collector
# Export
__all__ = ["DatasetCollector", "DatasetSample", "build_dataset"]
pkg/hanzo-tools-agent/pyproject.toml
+1 -3
View File
@@ -19,22 +19,20 @@ keywords = ["hanzo", "tools", "agent", "swarm", "mcp", "ai"]
dependencies = [
"hanzo-tools>=0.3.0",
"hanzo-tools-shell>=0.2.0",
"hanzo-async>=0.1.0", # Unified async I/O with uvloop
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",
"aiofiles>=24.0.0",
]
[project.optional-dependencies]
consensus = ["hanzo-consensus>=0.1.0"]
api = ["httpx>=0.28.0"] # Direct API mode
perf = ["uvloop>=0.21.0"] # High-performance async
full = [
"openai>=1.0.0",
"anthropic>=0.40.0",
"hanzo-consensus>=0.1.0",
"httpx>=0.28.0",
"uvloop>=0.21.0",
]
dev = ["pytest>=7.0.0", "ruff>=0.14.0"]
pkg/hanzo-tools-fs/hanzo_tools/fs/search.py
+3 -4
View File
@@ -5,7 +5,7 @@ import asyncio
from typing import Optional, Annotated
from pathlib import Path
import aiofiles
from hanzo_async import read_file
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
@@ -142,9 +142,8 @@ Returns:
continue
try:
async with aiofiles.open(file_path, "r", encoding="utf-8", errors="ignore") as f:
content = await f.read()
lines = content.splitlines()
content = await read_file(file_path, encoding="utf-8", errors="ignore")
lines = content.splitlines()
for i, line in enumerate(lines, 1):
if regex.search(line):
pkg/hanzo-tools-fs/pyproject.toml
+1 -1
View File
@@ -17,7 +17,7 @@ classifiers = [
]
keywords = ["hanzo", "tools", "filesystem", "mcp", "ai"]
dependencies = [
"aiofiles>=24.1.0",
"hanzo-async>=0.1.0", # Unified async I/O with uvloop
"grep-ast>=0.8.1",
"ffind>=1.3.0",
"watchdog>=6.0.0",
pkg/hanzo-tools-shell/hanzo_tools/shell/__init__.py
+61 -29
View File
@@ -1,12 +1,14 @@
"""Shell tools package for Hanzo AI.
Minimal, orthogonal shell execution.
Minimal, orthogonal shell execution with uvloop for high performance.
Core tools:
- dag: DAG execution (parallel, serial, graph) - RECOMMENDED for complex workflows
- zsh: Primary shell with shellflow DSL support
- shell: Smart shell alias (zsh > bash fallback)
- bash: Explicit bash for bash-specific scripts
- cmd: Unified execution graph (DAG) - RECOMMENDED
- zsh: Zsh shell (thin wrapper over cmd)
- bash: Bash shell (thin wrapper over cmd)
- fish: Fish shell (thin wrapper over cmd)
- dash: Dash shell (fast POSIX, Ubuntu's /bin/sh)
- shell: Smart shell alias (zsh > bash > fish > dash > sh fallback)
- ps: Process management (list, kill, logs)
HTTP/Data tools:
@@ -19,16 +21,26 @@ Convenience tools:
- uvx: Python package execution with auto-backgrounding
- open: Open files/URLs in system apps
Use 'dag' for:
- Parallel execution: dag(["npm install", "cargo build"], parallel=True)
- Mixed DAG: dag(["mkdir dist", {"parallel": ["cp a dist/", "cp b dist/"]}, "zip out"])
- Tool invocations: dag([{"tool": "search", "input": {"pattern": "TODO"}}])
Use 'cmd' for:
- Simple commands: cmd("ls -la")
- Parallel execution: cmd(["npm install", "cargo build"], parallel=True)
- Mixed DAG: cmd(["mkdir dist", {"parallel": ["cp a dist/", "cp b dist/"]}, "zip out"])
- Tool invocations: cmd([{"tool": "search", "input": {"pattern": "TODO"}}])
Use 'zsh/shell' for:
- Simple commands: zsh("ls -la")
- Shellflow DSL: zsh("cmd1 ; { cmd2 & cmd3 } ; cmd4")
Auto-backgrounding: Commands exceeding 45s automatically background.
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 configure_loop, using_uvloop
# Auto-configure uvloop on import
configure_loop()
_using_uvloop = using_uvloop()
from mcp.server import FastMCP
from hanzo_tools.core import BaseTool, ToolRegistry, PermissionManager
@@ -36,11 +48,11 @@ from hanzo_tools.shell.jq_tool import JqTool
from hanzo_tools.shell.ps_tool import PsTool, ps_tool
# Core tools
from hanzo_tools.shell.dag_tool import DagTool, create_dag_tool
from hanzo_tools.shell.cmd_tool import CmdTool, CmdResult, CmdNode, NodeStatus, create_cmd_tool, cmd_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
from hanzo_tools.shell.zsh_tool import ZshTool, BashTool, ShellTool, zsh_tool, bash_tool, shell_tool
from hanzo_tools.shell.shell_tools import ZshTool, BashTool, FishTool, DashTool, ShellTool, zsh_tool, bash_tool, fish_tool, dash_tool, shell_tool
# HTTP/Data tools (no shell escaping issues)
from hanzo_tools.shell.curl_tool import CurlTool
@@ -57,19 +69,26 @@ from hanzo_tools.shell.base_process import (
ProcessManager,
BaseProcessTool,
AutoBackgroundExecutor,
ShellExecutor,
get_shell_executor,
)
# Backwards compatibility - DagTool is now CmdTool
from hanzo_tools.shell.dag_tool import DagTool, DagResult, DagNode, create_dag_tool
# Tools list for entry point discovery
# - DagTool: Semantic DAG execution (parallel, serial, graph)
# - ZshTool: Primary shell with shellflow DSL support
# - ShellTool: Smart shell alias (zsh > bash fallback)
# - BashTool: Explicit bash for bash-specific scripts
TOOLS = [DagTool, ZshTool, BashTool, ShellTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
# - CmdTool: Primary command execution with DAG support
# - ZshTool, BashTool, ShellTool: Shell wrappers (thin shims over CmdTool)
TOOLS = [CmdTool, ZshTool, BashTool, FishTool, DashTool, ShellTool, PsTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
__all__ = [
# uvloop status
"_using_uvloop",
# Base classes
"ProcessManager",
"AutoBackgroundExecutor",
"ShellExecutor",
"get_shell_executor",
"BaseProcessTool",
"BaseBinaryTool",
"BaseScriptTool",
@@ -78,8 +97,12 @@ __all__ = [
"truncate_lines",
"estimate_tokens",
# Core tools
"DagTool",
"create_dag_tool",
"CmdTool",
"CmdResult",
"CmdNode",
"NodeStatus",
"cmd_tool",
"create_cmd_tool",
"PsTool",
"ps_tool",
"ZshTool",
@@ -88,6 +111,15 @@ __all__ = [
"shell_tool",
"BashTool",
"bash_tool",
"FishTool",
"fish_tool",
"DashTool",
"dash_tool",
# Backwards compatibility
"DagTool",
"DagResult",
"DagNode",
"create_dag_tool",
# Convenience tools
"OpenTool",
"open_tool",
@@ -114,15 +146,15 @@ def get_shell_tools(
Args:
permission_manager: Permission manager for access control
all_tools: Dict of all registered tools (for DAG tool invocations)
all_tools: Dict of all registered tools (for cmd tool invocations)
Returns:
List of shell tool instances
"""
# Create DAG tool with access to other tools (for tool invocations)
dag = DagTool(tools=all_tools or {})
# Create zsh tool with access to other tools
# Create cmd tool with access to other tools (for tool invocations)
cmd = CmdTool(tools=all_tools or {})
# Create shell wrappers with access to other tools
zsh = ZshTool(tools=all_tools or {})
# Set permission manager for convenience tools
@@ -130,8 +162,8 @@ def get_shell_tools(
uvx_tool.permission_manager = permission_manager
return [
dag, # DAG execution (parallel, serial, graph)
zsh, # Primary shell with shellflow DSL
cmd, # Primary command execution with DAG support
zsh, # Zsh shell wrapper
ps_tool, # Process management
npx_tool, # Node packages
uvx_tool, # Python packages
@@ -149,7 +181,7 @@ def register_shell_tools(
Args:
mcp_server: The FastMCP server instance
permission_manager: Permission manager for access control
all_tools: Dict of all registered tools (for dag tool invocations)
all_tools: Dict of all registered tools (for cmd tool invocations)
Returns:
List of registered tools
pkg/hanzo-tools-shell/hanzo_tools/shell/base_process.py
+201 -36
View File
@@ -1,7 +1,7 @@
"""Base classes for process execution tools.
All process execution uses asyncio.subprocess for consistency.
All file I/O uses aiofiles for non-blocking operations.
All file I/O uses hanzo_async for non-blocking operations with uvloop support.
"""
import os
@@ -13,8 +13,7 @@ from abc import abstractmethod
from typing import Any, Dict, List, Tuple, Optional, override
from pathlib import Path
import aiofiles
import aiofiles.os
from hanzo_async import mkdir, write_file, append_file
from hanzo_tools.core import BaseTool, PermissionManager
from hanzo_tools.shell.truncate import truncate_response
@@ -40,7 +39,7 @@ class ProcessManager:
async def _ensure_log_dir(self) -> None:
"""Ensure log directory exists (async-safe)."""
if not self._initialized:
await aiofiles.os.makedirs(self._log_dir, exist_ok=True)
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:
@@ -93,8 +92,7 @@ class ProcessManager:
"""Create a log file for a process (async-safe)."""
await self._ensure_log_dir()
log_file = self._log_dir / f"{process_id}.log"
async with aiofiles.open(log_file, "w") as f:
pass
await write_file(log_file, "")
return log_file
def mark_completed(self, process_id: str, return_code: int) -> None:
@@ -103,9 +101,15 @@ class ProcessManager:
class AutoBackgroundExecutor:
"""Executor that automatically backgrounds long-running processes."""
"""Executor that automatically backgrounds long-running processes.
DEFAULT_TIMEOUT = 60.0 # 60 seconds - background faster for efficiency
IMPORTANT: Always backgrounds after MAX_FOREGROUND_TIMEOUT (45s) to keep
the agent loop responsive. Longer timeouts only affect the background
process lifetime, not the foreground wait time.
"""
DEFAULT_TIMEOUT = 45.0 # Default timeout for backgrounding
MAX_FOREGROUND_TIMEOUT = 45.0 # ALWAYS background after 45s, even if longer timeout given
def __init__(self, process_manager: ProcessManager, timeout: float = DEFAULT_TIMEOUT):
"""Initialize the auto-background executor."""
@@ -124,29 +128,38 @@ class AutoBackgroundExecutor:
Returns:
Tuple of (output/status, was_backgrounded, process_id)
Note: Always backgrounds after MAX_FOREGROUND_TIMEOUT (60s) to keep
the agent loop responsive. The passed timeout is stored for reference
but doesn't extend foreground wait time.
"""
# Use passed timeout or default
effective_timeout = timeout if timeout is not None else self.default_timeout
# 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
# Fast path for tests - still async but with shorter timeout
if os.getenv("HANZO_MCP_FAST_TESTS") == "1":
import subprocess
try:
proc = subprocess.run(
cmd_args,
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,
capture_output=True,
text=True,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
if proc.returncode != 0:
return (
f"Command failed with exit code {proc.returncode}:\n{proc.stdout}{proc.stderr}",
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 proc.stdout, 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
@@ -179,8 +192,7 @@ class AutoBackgroundExecutor:
async for line in process.stdout:
line_str = line.decode("utf-8", errors="replace")
output_lines.append(line_str)
async with aiofiles.open(log_file, "a") as f:
await f.write(line_str)
await append_file(log_file, line_str)
async def wait_for_process():
return await process.wait()
@@ -248,18 +260,15 @@ class AutoBackgroundExecutor:
try:
if process.stdout:
async for line in process.stdout:
async with aiofiles.open(log_file, "a") as f:
await f.write(line.decode("utf-8", errors="replace"))
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)
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n\n=== Process completed with exit code {return_code} ===\n")
await append_file(log_file, f"\n\n=== Process completed with exit code {return_code} ===\n")
except Exception as e:
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n\n=== Background reader error: {str(e)} ===\n")
await append_file(log_file, f"\n\n=== Background reader error: {str(e)} ===\n")
self.process_manager.mark_completed(process_id, -1)
@@ -364,22 +373,20 @@ class BaseProcessTool(BaseTool):
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:
async with aiofiles.open(log_file, "w") as f:
if process.stdout:
async for line in process.stdout:
await f.write(line.decode("utf-8", errors="replace"))
await f.flush()
# 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()
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n=== Process completed with exit code {return_code} ===\n")
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:
async with aiofiles.open(log_file, "a") as f:
await f.write(f"\n=== Error: {str(e)} ===\n")
await append_file(log_file, f"\n=== Error: {str(e)} ===\n")
self.process_manager.mark_completed(process_id, -1)
@@ -440,3 +447,161 @@ class BaseScriptTool(BaseProcessTool):
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.
"""
DEFAULT_TIMEOUT = 45.0 # Auto-background after 45 seconds
_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: 45s)
tool_name: Tool name for process ID prefix
Returns:
Tuple of (stdout, stderr, exit_code, was_backgrounded, process_id)
"""
# Cap timeout at 45s 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,
)
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=effective_timeout
)
exit_code = proc.returncode or 0
return (
stdout.decode("utf-8", errors="replace"),
stderr.decode("utf-8", errors="replace"),
exit_code,
False, # Not backgrounded
None, # No process_id (completed)
)
except asyncio.TimeoutError:
# Background the process
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"
)
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:
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/hanzo_tools/shell/cmd_tool.py
+514
View File
@@ -0,0 +1,514 @@
"""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 = 45,
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 = 45,
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=45)] = 45,
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/hanzo_tools/shell/dag_tool.py
+33 -99
View File
@@ -14,13 +14,12 @@ from pathlib import Path
from datetime import datetime
from dataclasses import field, dataclass
import aiofiles
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 ProcessManager
from hanzo_tools.shell.base_process import get_shell_executor
class NodeStatus(Enum):
@@ -174,114 +173,49 @@ backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
shell: str,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 120,
timeout: int = 45,
) -> DagResult:
"""Run a shell command with auto-backgrounding on timeout."""
"""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)}"
process_manager = ProcessManager()
executor = get_shell_executor()
# Defensive check - should never happen but provides better error message
if process_manager is None:
raise RuntimeError("ProcessManager() returned None - singleton initialization failed")
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",
)
try:
run_env = os.environ.copy()
if env:
run_env.update(env)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
process_id = f"dag_{uuid.uuid4().hex[:8]}"
log_file = await process_manager.create_log_file(process_id)
proc = await asyncio.create_subprocess_exec(
shell,
"-c",
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
env=run_env,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
exit_code = proc.returncode or 0
return DagResult(
node_id=node_id,
command=cmd,
stdout=stdout.decode("utf-8", errors="replace"),
stderr=stderr.decode("utf-8", errors="replace"),
status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
exit_code=exit_code,
duration_ms=duration,
node_type="shell",
)
except asyncio.TimeoutError:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
async with aiofiles.open(log_file, "w") as f:
await f.write(f"[dag] Command backgrounded after {timeout}s timeout\n")
await f.write(f"[dag] Command: {cmd}\n")
await f.write(f"[dag] PID: {proc.pid}\n")
await f.write(f"[dag] Started: {start_time.isoformat()}\n")
await f.write("-" * 40 + "\n")
process_manager.add_process(process_id, proc, str(log_file))
asyncio.create_task(self._capture_background_output(proc, log_file))
return DagResult(
node_id=node_id,
command=cmd,
stdout=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",
stderr="",
status=NodeStatus.SUCCESS,
exit_code=0,
duration_ms=duration,
node_type="shell",
)
except Exception as e:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
if was_backgrounded:
return DagResult(
node_id=node_id,
command=cmd,
stdout="",
stderr=str(e),
status=NodeStatus.FAILED,
exit_code=1,
stdout=stdout,
stderr=stderr,
status=NodeStatus.SUCCESS, # Backgrounded = success from caller's POV
exit_code=0,
duration_ms=duration,
node_type="shell",
)
async def _capture_background_output(self, proc: asyncio.subprocess.Process, log_file: Path) -> None:
"""Capture output from backgrounded process to log file."""
try:
async with aiofiles.open(log_file, "a") as f:
async def read_stream(stream, prefix: str):
if stream:
while True:
line = await stream.readline()
if not line:
break
await f.write(f"{prefix}{line.decode('utf-8', errors='replace')}")
await f.flush()
await asyncio.gather(
read_stream(proc.stdout, ""),
read_stream(proc.stderr, "[stderr] "),
)
await proc.wait()
await f.write(f"\n[dag] Process exited with code {proc.returncode}\n")
except Exception:
pass
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."""
@@ -450,7 +384,7 @@ backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
shell: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: int = 120,
timeout: int = 45,
strict: bool = False,
quiet: bool = False,
**kwargs,
@@ -528,7 +462,7 @@ backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
shell: Annotated[Optional[str], Field(description="Shell to use (default: zsh)", default=None)] = None,
cwd: Annotated[Optional[str], Field(description="Working directory", default=None)] = None,
env: Annotated[Optional[Dict[str, str]], Field(description="Environment variables", default=None)] = None,
timeout: Annotated[int, Field(description="Timeout per command (seconds)", default=120)] = 120,
timeout: Annotated[int, Field(description="Timeout per command (seconds)", default=45)] = 45,
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,
pkg/hanzo-tools-shell/hanzo_tools/shell/open_tool.py
+10 -4
View File
@@ -66,14 +66,19 @@ open /path/to/image.png"""
if is_url:
try:
webbrowser.open(path)
# Run webbrowser.open in executor to avoid blocking
loop = asyncio.get_event_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()
if not file_path.exists():
# Check file exists asynchronously
loop = asyncio.get_event_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()
@@ -94,8 +99,9 @@ open /path/to/image.png"""
raise RuntimeError("No suitable file opener found on Linux")
elif system == "windows":
import os
os.startfile(str(file_path))
# Run os.startfile in executor to avoid blocking
loop = asyncio.get_event_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}")
pkg/hanzo-tools-shell/hanzo_tools/shell/ps_tool.py
+10 -5
View File
@@ -11,7 +11,7 @@ from pathlib import Path
from datetime import datetime
from dataclasses import dataclass
import aiofiles
from hanzo_async import read_file, path_exists
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
@@ -115,14 +115,19 @@ USAGE:
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 or not log_file.exists():
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:
async with aiofiles.open(log_file, "r") as f:
content = await f.read()
lines = content.splitlines()
content = await read_file(log_file)
lines = content.splitlines()
if len(lines) > n:
lines = lines[-n:]
pkg/hanzo-tools-shell/hanzo_tools/shell/shell_tools.py
+421
View File
@@ -0,0 +1,421 @@
"""Shell tool shims - zsh, bash, fish, dash, 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)
- shell: Auto-selects best available (zsh > bash > fish > dash > 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=45)] = 45,
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=45)] = 45,
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=45)] = 45,
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=45)] = 45,
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 ShellTool(CmdTool):
"""Smart shell - auto-selects best available shell (zsh > bash > fish > dash > 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=45)] = 45,
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()
shell_tool = ShellTool()
pkg/hanzo-tools-shell/hanzo_tools/shell/wget_tool.py
+9 -1
View File
@@ -144,11 +144,19 @@ Returns status and download summary.
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=output if (output and os.path.isdir(output)) else None,
cwd=cwd,
)
stdout, stderr = await asyncio.wait_for(
pkg/hanzo-tools-shell/hanzo_tools/shell/zsh_tool.py
-712
View File
@@ -1,712 +0,0 @@
"""Unified Zsh shell tool with DAG execution support.
Combines simple shell execution with DAG (directed acyclic graph) semantics.
Supports serial, parallel, and complex mixed execution patterns.
Usage:
zsh("ls -la") # Single command
zsh(["ls", "pwd", "git status"]) # Serial execution
zsh(["npm i", "cargo build"], parallel=True) # Parallel
zsh([
"mkdir dist",
{"parallel": ["cp a.txt dist/", "cp b.txt dist/"]},
"zip -r out.zip dist/"
]) # Mixed DAG
"""
import os
import uuid
import shutil
import asyncio
import platform
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
import aiofiles
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.shellflow import parse as parse_shellflow, to_commands as shellflow_to_commands
from hanzo_tools.shell.base_process import ProcessManager
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 ZshTool(BaseTool):
"""Zsh shell with full DAG execution support.
The primary shell tool for hanzo-mcp. Supports:
- Simple command execution: zsh("ls -la")
- Serial execution: zsh(["cmd1", "cmd2"])
- Parallel execution: zsh(["cmd1", "cmd2"], parallel=True)
- Mixed DAG patterns: zsh(["setup", {"parallel": ["a", "b"]}, "finish"])
- Tool invocations: zsh([{"tool": "search", "input": {"pattern": "TODO"}}])
- Named dependencies: zsh([{"id": "a", "run": "...", "after": ["b"]}])
Uses zsh by default, falls back to bash.
Auto-backgrounds long-running processes.
"""
name = "zsh"
def __init__(self, tools: Optional[Dict[str, BaseTool]] = None):
"""Initialize Zsh tool with optional tool registry for DAG tool invocations."""
super().__init__()
self.tools = tools or {}
self._shell = self._resolve_shell()
def _resolve_shell(self) -> str:
"""Resolve shell - prefer zsh, fallback to bash."""
# Allow override
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
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:
shell_name = os.path.basename(self._shell)
return f"""Zsh shell with DAG execution support (using: {shell_name}).
SHELLFLOW SYNTAX (inline DAG):
zsh("A ; B ; C") # Sequential
zsh("{{ A & B & C }}") # Parallel
zsh("setup ; {{ a & b & c }} ; done") # Mixed
ARRAY SYNTAX:
zsh(["a", "b", "c"]) # Sequential
zsh(["a", ["b", "c", "d"], "e"]) # Nested = parallel
zsh(["a", "b"], parallel=True) # All parallel
EXAMPLE:
zsh("mkdir -p dist ; {{ cp a dist/ & cp b dist/ }} ; zip -r out.zip dist/")
OPTIONS:
shell: Use different shell (bash, sh, fish)
parallel: Run ALL top-level commands concurrently
strict: Stop on first error
quiet: Suppress stdout
timeout: Per-command timeout (default: 120s)
cwd: Working directory
env: Environment variables
AUTO-BACKGROUNDING: Commands exceeding timeout 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 = 120,
shell: Optional[str] = None,
) -> DagResult:
"""Run a shell command with auto-backgrounding on timeout."""
start_time = datetime.now()
node_id = f"shell_{id(cmd)}"
process_manager = ProcessManager()
# Use specified shell or default
use_shell = shell or self._shell
try:
run_env = os.environ.copy()
if env:
run_env.update(env)
shell_name = os.path.basename(use_shell)
process_id = f"{shell_name}_{uuid.uuid4().hex[:8]}"
log_file = await process_manager.create_log_file(process_id)
proc = await asyncio.create_subprocess_exec(
use_shell,
"-c",
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
env=run_env,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
duration = int((datetime.now() - start_time).total_seconds() * 1000)
exit_code = proc.returncode or 0
return DagResult(
node_id=node_id,
command=cmd,
stdout=stdout.decode("utf-8", errors="replace"),
stderr=stderr.decode("utf-8", errors="replace"),
status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
exit_code=exit_code,
duration_ms=duration,
node_type="shell",
)
except asyncio.TimeoutError:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
async with aiofiles.open(log_file, "w") as f:
await f.write(f"[{shell_name}] Command backgrounded after {timeout}s timeout\n")
await f.write(f"[{shell_name}] Command: {cmd}\n")
await f.write(f"[{shell_name}] PID: {proc.pid}\n")
await f.write(f"[{shell_name}] Started: {start_time.isoformat()}\n")
await f.write("-" * 40 + "\n")
process_manager.add_process(process_id, proc, str(log_file))
asyncio.create_task(self._capture_background_output(proc, log_file))
return DagResult(
node_id=node_id,
command=cmd,
stdout=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",
stderr="",
status=NodeStatus.SUCCESS,
exit_code=0,
duration_ms=duration,
node_type="shell",
)
except Exception as e:
duration = int((datetime.now() - start_time).total_seconds() * 1000)
return DagResult(
node_id=node_id,
command=cmd,
stdout="",
stderr=str(e),
status=NodeStatus.FAILED,
exit_code=1,
duration_ms=duration,
node_type="shell",
)
async def _capture_background_output(self, proc: asyncio.subprocess.Process, log_file: Path) -> None:
"""Capture output from backgrounded process to log file."""
try:
async with aiofiles.open(log_file, "a") as f:
async def read_stream(stream, prefix: str):
if stream:
while True:
line = await stream.readline()
if not line:
break
await f.write(f"{prefix}{line.decode('utf-8', errors='replace')}")
await f.flush()
await asyncio.gather(
read_stream(proc.stdout, ""),
read_stream(proc.stderr, "[stderr] "),
)
await proc.wait()
await f.write(f"\n[shell] Process exited with code {proc.returncode}\n")
except Exception:
pass
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,
cwd: Optional[str],
env: Optional[Dict[str, str]],
timeout: int,
shell: Optional[str] = None,
) -> DagResult:
"""Execute a single DAG 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 DagResult(
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 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, 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",
)
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[zsh] {len(results)} nodes, {total_duration}ms, {status}")
return "\n".join(output_parts) if output_parts else "(no output)"
@override
@auto_timeout("zsh")
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 = 120,
strict: bool = False,
quiet: bool = False,
**kwargs,
) -> str:
"""Execute shell 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, options: bash, sh, fish, etc.)
cwd: Working directory
env: Additional environment variables
timeout: Per-command timeout in seconds
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:
# Check for shellflow syntax (contains ; or { & })
if ';' in command or ('&' in command and '{' in command):
# Parse as shellflow
commands = shellflow_to_commands(parse_shellflow(command))
else:
# Simple command
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[DagResult] = []
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 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, 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 zsh tool with MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def zsh(
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 to use: zsh, bash, sh, fish (default: zsh)", default=None)
] = None,
cwd: Annotated[Optional[str], Field(description="Working directory", default=None)] = None,
env: Annotated[Optional[Dict[str, str]], Field(description="Environment variables", default=None)] = None,
timeout: Annotated[int, Field(description="Timeout per command (seconds)", default=120)] = 120,
strict: Annotated[bool, Field(description="Stop on first error", default=False)] = False,
quiet: Annotated[bool, Field(description="Suppress stdout", default=False)] = False,
ctx: MCPContext = None,
) -> str:
return await tool_self.call(
ctx,
command=command,
commands=commands,
parallel=parallel,
shell=shell,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class ShellTool(ZshTool):
"""Alias for ZshTool - smart shell that uses zsh > bash."""
name = "shell"
@property
@override
def description(self) -> str:
shell_name = os.path.basename(self._shell)
return f"""Run shell commands using the best available shell (currently: {shell_name}).
Automatically selects:
- Zsh if available (with .zshrc)
- User's preferred shell ($SHELL)
- Bash as fallback
Full Shellflow + DAG support.
SHELLFLOW:
shell("A ; B ; C") # Sequential
shell("{{ A & B & C }}") # Parallel
shell("setup ; {{ a & b }} ; done") # Mixed
ARRAYS:
shell(["a", "b", "c"]) # Sequential
shell(["a", ["b", "c"], "d"]) # Nested = parallel"""
@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(
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 to use: zsh, bash, sh, fish", default=None)
] = None,
cwd: Annotated[Optional[str], Field(description="Working directory", default=None)] = None,
env: Annotated[Optional[Dict[str, str]], Field(description="Environment variables", default=None)] = None,
timeout: Annotated[int, Field(description="Timeout per command (seconds)", default=120)] = 120,
strict: Annotated[bool, Field(description="Stop on first error", default=False)] = False,
quiet: Annotated[bool, Field(description="Suppress stdout", default=False)] = False,
ctx: MCPContext = None,
) -> str:
return await tool_self.call(
ctx,
command=command,
commands=commands,
parallel=parallel,
shell=shell,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
class BashTool(ZshTool):
"""Bash shell with DAG support - alias for ZshTool with bash default."""
name = "bash"
def _resolve_shell(self) -> str:
"""Use bash as default shell."""
force_shell = os.environ.get("HANZO_MCP_FORCE_SHELL")
if force_shell:
return force_shell
# Prefer bash
bash_path = shutil.which("bash")
if bash_path:
return bash_path
return "sh"
@property
@override
def description(self) -> str:
return """Bash shell with DAG execution support.
Same as zsh but defaults to bash. Full Shellflow + DAG support.
SHELLFLOW:
bash("A ; B ; C") # Sequential
bash("{ A & B & C }") # Parallel
bash("setup ; { a & b } ; done") # Mixed
ARRAYS:
bash(["a", "b", "c"]) # Sequential
bash(["a", ["b", "c"], "d"]) # Nested = parallel"""
@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(
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=120)] = 120,
strict: Annotated[bool, Field(description="Stop on first error", default=False)] = False,
quiet: Annotated[bool, Field(description="Suppress stdout", default=False)] = False,
ctx: MCPContext = None,
) -> str:
return await tool_self.call(
ctx,
command=command,
commands=commands,
parallel=parallel,
cwd=cwd,
env=env,
timeout=timeout,
strict=strict,
quiet=quiet,
)
# Create tool instances
zsh_tool = ZshTool()
shell_tool = ShellTool()
bash_tool = BashTool()
pkg/hanzo-tools-shell/pyproject.toml
+3 -3
View File
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools-shell"
version = "0.4.1"
description = "Shell/command execution tools - dag, zsh (with shellflow DSL), ps, npx, uvx, open, curl, jq, wget"
version = "0.5.0"
description = "Shell/command execution tools - cmd (unified execution graph), zsh, bash, shell, ps, npx, uvx, open, curl, jq, wget"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
@@ -13,9 +13,9 @@ authors = [{ name = "Hanzo Industries Inc", email = "[email protected]" }]
keywords = ["hanzo", "tools", "shell", "mcp", "ai", "dag", "zsh"]
dependencies = [
"hanzo-tools>=0.3.0",
"hanzo-async>=0.1.0", # Unified async I/O with uvloop
"mcp>=1.25.0",
"pydantic>=2.12.5",
"aiofiles>=24.0.0",
"tiktoken>=0.8.0",
]