fix(hanzo-mcp): eliminate all blocking I/O in shell modules

Comprehensive async fix for all shell/process handling modules:

- command_executor.py: Made _get_shell_by_type, _get_system_shell,
  _get_interpreter_path async with asyncio.to_thread for shutil.which.
  Temp file creation/cleanup now uses aiofiles.os.

- logs.py: Added aiofiles for async file reads. Path.exists(), stat(),
  glob() wrapped in asyncio.to_thread(). Batch async stat operations
  with asyncio.gather().

- processes.py: Changed cpu_percent(interval=0.1) to interval=None
  to avoid 100ms blocking per process. Wrapped sync psutil operations
  in asyncio.to_thread().

- streaming_command.py: Replaced subprocess.run with async subprocess
  in tail(). stream_to_file now uses aiofiles. All file I/O operations
  made async throughout.

- Legacy modules (run_background.py, npx_background.py, uvx_background.py):
  Added deprecation warnings pointing to new async implementations in
  base_process.py and tool-specific modules.

Version bump to 0.9.17.
This commit is contained in:
2025-12-24 17:56:26 -08:00
parent 83e74521a0
commit fa82e6aced
8 changed files with 227 additions and 100 deletions
@@ -2,6 +2,9 @@
This module provides tools for executing shell commands and scripts with
comprehensive error handling, permissions checking, and progress tracking.
All blocking file I/O and system calls use asyncio.to_thread() for
non-blocking execution.
"""
import os
@@ -13,6 +16,9 @@ import shutil
import asyncio
import tempfile
from typing import final
import aiofiles
import aiofiles.os
from collections.abc import Callable, Awaitable
from hanzo_mcp.tools.shell.base import CommandResult
@@ -48,7 +54,7 @@ class CommandExecutor:
"fish": self._handle_fish_script,
}
def _get_shell_by_type(self, shell_type: str) -> tuple[str, str]:
async def _get_shell_by_type(self, shell_type: str) -> tuple[str, str]:
"""Get shell information for a specified shell type.
Args:
@@ -57,11 +63,12 @@ class CommandExecutor:
Returns:
Tuple of (shell_basename, shell_path)
"""
shell_path = shutil.which(shell_type)
# Run shutil.which in thread pool to avoid blocking
shell_path = await asyncio.to_thread(shutil.which, shell_type)
if shell_path is None:
# Shell not found, fall back to default
self._log(f"Requested shell '{shell_type}' not found, using system default")
return self._get_system_shell()
return await self._get_system_shell()
if sys.platform == "win32":
shell_path = shell_path.lower()
@@ -69,7 +76,7 @@ class CommandExecutor:
shell_basename = os.path.basename(shell_path).lower()
return shell_basename, shell_path
def _get_system_shell(self, shell_type: str | None = None) -> tuple[str, str]:
async def _get_system_shell(self, shell_type: str | None = None) -> tuple[str, str]:
"""Get the system's default shell based on the platform.
Args:
@@ -80,7 +87,7 @@ class CommandExecutor:
"""
# If a specific shell is requested, use that
if shell_type is not None:
return self._get_shell_by_type(shell_type)
return await self._get_shell_by_type(shell_type)
# Otherwise use system default
if sys.platform == "win32":
@@ -265,8 +272,8 @@ class CommandExecutor:
command_env.update(env)
try:
# Get shell information
shell_basename, user_shell = self._get_system_shell(shell_type)
# Get shell information (async to avoid blocking on shutil.which)
shell_basename, user_shell = await self._get_system_shell(shell_type)
if sys.platform == "win32":
# On Windows, always use shell execution
@@ -439,8 +446,8 @@ class CommandExecutor:
command_env.update(env)
try:
# Get shell information
shell_basename, user_shell = self._get_system_shell(shell_type)
# Get shell information (async to avoid blocking on shutil.which)
shell_basename, user_shell = await self._get_system_shell(shell_type)
if sys.platform == "win32":
# On Windows, always use shell
@@ -628,8 +635,8 @@ class CommandExecutor:
language_info = language_map[language]
extension = language_info["extension"]
# Get interpreter command with full path if possible
command, language_args = self._get_interpreter_path(language, shell_type)
# Get interpreter command with full path if possible (async to avoid blocking)
command, language_args = await self._get_interpreter_path(language, shell_type)
self._log(f"Interpreter path: {command} :: {language_args}")
@@ -638,10 +645,14 @@ class CommandExecutor:
if env:
command_env.update(env)
# Create a temporary file for the script
with tempfile.NamedTemporaryFile(suffix=extension, mode="w", delete=False) as temp:
temp_path = temp.name
_ = temp.write(script) # Explicitly ignore the return value
# Create a temporary file for the script (async to avoid blocking I/O)
def _create_temp_file_sync() -> str:
"""Create temp file synchronously - called via to_thread."""
with tempfile.NamedTemporaryFile(suffix=extension, mode="w", delete=False) as temp:
temp.write(script)
return temp.name
temp_path = await asyncio.to_thread(_create_temp_file_sync)
try:
# Normalize path for the current OS
@@ -649,8 +660,8 @@ class CommandExecutor:
original_temp_path = temp_path
if sys.platform == "win32":
# Windows always uses shell
shell_basename, user_shell = self._get_system_shell(shell_type)
# Windows always uses shell (async to avoid blocking on shutil.which)
shell_basename, user_shell = await self._get_system_shell(shell_type)
# Convert Windows path to WSL path if using WSL
if shell_basename in ["wsl", "wsl.exe"]:
@@ -688,8 +699,8 @@ class CommandExecutor:
else:
# Unix systems - original logic
if use_login_shell:
# Get the user's login shell
shell_basename, user_shell = self._get_system_shell(shell_type)
# Get the user's login shell (async to avoid blocking on shutil.which)
shell_basename, user_shell = await self._get_system_shell(shell_type)
# Build the command including args
cmd = f"{command} {temp_path}"
@@ -752,9 +763,9 @@ class CommandExecutor:
self._log(f"Script file execution error: {str(e)}")
return CommandResult(return_code=1, error_message=f"Error executing script: {str(e)}")
finally:
# Clean up temporary file
# Clean up temporary file (async to avoid blocking)
try:
os.unlink(original_temp_path)
await aiofiles.os.unlink(original_temp_path)
except Exception as e:
self._log(f"Error cleaning up temporary file: {str(e)}")
@@ -817,12 +828,14 @@ class CommandExecutor:
},
}
def _get_interpreter_path(self, language: str, shell_type: str | None = None) -> tuple[str, list[str]]:
async def _get_interpreter_path(self, language: str, shell_type: str | None = None) -> tuple[str, list[str]]:
"""Get the full path to the interpreter for the given language.
Attempts to find the full path to the interpreter command, but only for
Windows shell types (cmd, powershell). For WSL, just returns the command name.
Uses asyncio.to_thread for shutil.which calls to avoid blocking.
Args:
language: The language name (e.g., "python", "javascript")
shell_type: Optional shell type (e.g., "wsl", "cmd", "powershell")
@@ -856,15 +869,15 @@ class CommandExecutor:
not shell_type or shell_type.lower() in ["cmd", "powershell", "cmd.exe", "powershell.exe"]
):
try:
# Try to find the full path to the command
full_path = shutil.which(command)
# Try to find the full path to the command (async to avoid blocking)
full_path = await asyncio.to_thread(shutil.which, command)
if full_path:
self._log(f"Found full path for {language} interpreter: {full_path}")
return full_path, args
# If primary command not found, try alternatives
for alt_command in alternatives:
alt_path = shutil.which(alt_command)
alt_path = await asyncio.to_thread(shutil.which, alt_command)
if alt_path:
self._log(f"Found alternative path for {language} interpreter: {alt_path}")
return alt_path, args
+31 -15
View File
@@ -1,8 +1,14 @@
"""Tool for viewing process logs."""
"""Tool for viewing process logs.
All blocking file I/O operations use aiofiles for async reads and
asyncio.to_thread() for Path operations.
"""
import asyncio
from typing import Unpack, Optional, Annotated, TypedDict, final, override
from pathlib import Path
import aiofiles
from pydantic import Field
from mcp.server.fastmcp import Context as MCPContext
@@ -163,25 +169,27 @@ Use run_command with 'tail -f' for continuous monitoring.
if not self.permission_manager.has_permission(str(log_path)):
return f"Permission denied: {log_path}"
# Check if file exists
if not log_path.exists():
# Check if file exists (async)
file_exists = await asyncio.to_thread(log_path.exists)
if not file_exists:
return f"Log file not found: {log_path}"
# Note about follow mode
if follow:
await tool_ctx.warning("Follow mode not supported in MCP. Showing latest lines instead.")
# Read log file
# Read log file (async)
await tool_ctx.info(f"Reading log file: {log_path}")
try:
with open(log_path, "r") as f:
async with aiofiles.open(log_path, "r") as f:
if lines == -1:
# Read entire file
content = f.read()
# Read entire file asynchronously
content = await f.read()
else:
# Read last N lines
all_lines = f.readlines()
# Read last N lines asynchronously
full_content = await f.read()
all_lines = full_content.splitlines(keepends=True)
if len(all_lines) <= lines:
content = "".join(all_lines)
else:
@@ -214,17 +222,26 @@ Use run_command with 'tail -f' for continuous monitoring.
"""List all available log files."""
await tool_ctx.info("Listing available log files")
if not self.log_dir.exists():
# Check if log directory exists (async)
dir_exists = await asyncio.to_thread(self.log_dir.exists)
if not dir_exists:
return "No logs directory found."
# Get all log files
log_files = list(self.log_dir.glob("*.log"))
# Get all log files (async)
log_files = await asyncio.to_thread(lambda: list(self.log_dir.glob("*.log")))
if not log_files:
return "No log files found."
# Get stat info for sorting (async - batch all stat calls)
async def get_file_stat(f: Path) -> tuple[Path, float, int]:
stat_info = await asyncio.to_thread(f.stat)
return (f, stat_info.st_mtime, stat_info.st_size)
file_stats = await asyncio.gather(*[get_file_stat(f) for f in log_files])
# Sort by modification time (newest first)
log_files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
file_stats.sort(key=lambda x: x[1], reverse=True)
# Check which logs belong to active processes
active_processes = RunBackgroundTool.get_processes()
@@ -234,8 +251,7 @@ Use run_command with 'tail -f' for continuous monitoring.
output = []
output.append("=== Available Log Files ===\n")
for log_file in log_files[:50]: # Limit to 50 most recent
size = log_file.stat().st_size
for log_file, _, size in file_stats[:50]: # Limit to 50 most recent
size_str = self._format_size(size)
# Check if this belongs to an active process
@@ -1,4 +1,23 @@
"""Run Node.js packages in background with npx."""
"""Run Node.js packages in background with npx.
DEPRECATED: This module uses blocking subprocess.Popen operations.
For new code, use the async implementations in:
- hanzo_mcp.tools.shell.npx_tool.NpxTool (based on BaseBinaryTool)
This provides non-blocking process execution using asyncio.create_subprocess_exec
with automatic backgrounding for long-running processes.
This module is maintained for backwards compatibility only.
"""
import warnings
# Emit deprecation warning on import
warnings.warn(
"npx_background.py uses blocking subprocess.Popen. "
"Consider using npx_tool.NpxTool for async execution.",
DeprecationWarning,
stacklevel=2
)
import uuid
import shutil
@@ -1,5 +1,10 @@
"""Tool for listing running background processes."""
"""Tool for listing running background processes.
All blocking operations (psutil calls, file I/O) are wrapped in
asyncio.to_thread() to avoid blocking the event loop.
"""
import asyncio
from typing import Unpack, Optional, Annotated, TypedDict, final, override
from datetime import datetime
@@ -118,11 +123,17 @@ Examples:
if show_all:
# Show all system processes
await tool_ctx.info("Listing all system processes")
return self._list_system_processes(filter_name, show_details)
# Run in thread pool to avoid blocking event loop
return await asyncio.to_thread(
self._list_system_processes, filter_name, show_details
)
else:
# Show only background processes
await tool_ctx.info("Listing background processes")
return self._list_background_processes(filter_name, show_details)
# Run in thread pool to avoid blocking event loop
return await asyncio.to_thread(
self._list_background_processes, filter_name, show_details
)
except Exception as e:
await tool_ctx.error(f"Failed to list processes: {str(e)}")
@@ -172,8 +183,10 @@ Examples:
if show_details and process.is_running:
try:
# Get process details using psutil
# Use interval=None for non-blocking CPU measurement
# (returns average since last call or process start)
p = psutil.Process(process.pid)
output.append(f"CPU: {p.cpu_percent(interval=0.1):.1f}%")
output.append(f"CPU: {p.cpu_percent(interval=None):.1f}%")
output.append(f"Memory: {p.memory_info().rss / 1024 / 1024:.1f} MB")
output.append(f"Threads: {p.num_threads()}")
except (psutil.NoSuchProcess, psutil.AccessDenied):
@@ -222,7 +235,8 @@ Examples:
}
if show_details:
process_info["cpu"] = proc.cpu_percent(interval=0.1)
# Use interval=None for non-blocking CPU measurement
process_info["cpu"] = proc.cpu_percent(interval=None)
process_info["memory"] = proc.memory_info().rss / 1024 / 1024 # MB
processes.append(process_info)
@@ -1,4 +1,23 @@
"""Background process execution tool."""
"""Background process execution tool.
DEPRECATED: This module uses blocking subprocess.Popen operations.
For new code, use the async implementations in:
- hanzo_mcp.tools.shell.base_process.BaseProcessTool
- hanzo_mcp.tools.shell.auto_background.AutoBackgroundExecutor
These provide non-blocking process execution using asyncio.create_subprocess_exec.
This module is maintained for backwards compatibility only.
"""
import warnings
# Emit deprecation warning on import
warnings.warn(
"run_background.py uses blocking subprocess.Popen. "
"Consider using base_process.BaseProcessTool for async execution.",
DeprecationWarning,
stacklevel=2
)
import os
import uuid
@@ -1,6 +1,9 @@
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
"""Streaming command execution with disk-based logging and session management."""
"""Streaming command execution with disk-based logging and session management.
All file I/O uses aiofiles for non-blocking operations.
"""
import os
import json
@@ -8,11 +11,13 @@ import time
import uuid
import shutil
import asyncio
import subprocess
from typing import Any, Dict, List, Union, Optional
from pathlib import Path
from datetime import datetime, timedelta
import aiofiles
import aiofiles.os
from hanzo_mcp.tools.shell.base_process import BaseProcessTool
@@ -261,7 +266,7 @@ class StreamingCommandTool(BaseProcessTool):
error_file = cmd_dir / "error.log"
metadata_file = cmd_dir / "metadata.json"
# Save metadata
# Save metadata (async to avoid blocking)
metadata = {
"command_id": cmd_id,
"command": command,
@@ -271,8 +276,8 @@ class StreamingCommandTool(BaseProcessTool):
"status": "running",
}
with open(metadata_file, "w") as f:
json.dump(metadata, f, indent=2)
async with aiofiles.open(metadata_file, "w") as f:
await f.write(json.dumps(metadata, indent=2))
# Start process with output redirection
try:
@@ -285,14 +290,14 @@ class StreamingCommandTool(BaseProcessTool):
# Create tasks for streaming stdout and stderr to files
async def stream_to_file(stream, file_path):
"""Stream from async pipe to file."""
with open(file_path, "wb") as f:
"""Stream from async pipe to file using aiofiles."""
async with aiofiles.open(file_path, "wb") as f:
while True:
chunk = await stream.read(8192)
if not chunk:
break
f.write(chunk)
f.flush() # Ensure immediate write
await f.write(chunk)
await f.flush() # Ensure immediate write
# Start streaming tasks
stdout_task = asyncio.create_task(stream_to_file(process.stdout, output_file))
@@ -307,17 +312,23 @@ class StreamingCommandTool(BaseProcessTool):
break
await asyncio.sleep(0.1)
# Read initial chunk
# Read initial chunk (async to avoid blocking)
output_content = ""
error_content = ""
if output_file.exists() and output_file.stat().st_size > 0:
with open(output_file, "r", errors="replace") as f:
output_content = f.read(chunk_size)
output_exists = await asyncio.to_thread(output_file.exists)
if output_exists:
output_size = await asyncio.to_thread(lambda: output_file.stat().st_size)
if output_size > 0:
async with aiofiles.open(output_file, "r", errors="replace") as f:
output_content = await f.read(chunk_size)
if error_file.exists() and error_file.stat().st_size > 0:
with open(error_file, "r", errors="replace") as f:
error_content = f.read(1000) # Just first 1KB of errors
error_exists = await asyncio.to_thread(error_file.exists)
if error_exists:
error_size = await asyncio.to_thread(lambda: error_file.stat().st_size)
if error_size > 0:
async with aiofiles.open(error_file, "r", errors="replace") as f:
error_content = await f.read(1000) # Just first 1KB of errors
# Check if process completed quickly
try:
@@ -328,14 +339,14 @@ class StreamingCommandTool(BaseProcessTool):
exit_code = None
status = "running"
# Update metadata
# Update metadata (async to avoid blocking)
metadata["status"] = status
if exit_code is not None:
metadata["exit_code"] = exit_code
metadata["end_time"] = datetime.now().isoformat()
with open(metadata_file, "w") as f:
json.dump(metadata, f, indent=2)
async with aiofiles.open(metadata_file, "w") as f:
await f.write(json.dumps(metadata, indent=2))
# Build response
result = {
@@ -354,8 +365,8 @@ class StreamingCommandTool(BaseProcessTool):
if exit_code is not None:
result["exit_code"] = exit_code
# Add continuation info if more output available
total_size = output_file.stat().st_size
# Add continuation info if more output available (async stat)
total_size = await asyncio.to_thread(lambda: output_file.stat().st_size)
if total_size > len(output_content) or status == "running":
result["has_more"] = True
result["total_bytes"] = total_size
@@ -377,13 +388,13 @@ class StreamingCommandTool(BaseProcessTool):
return result
except Exception as e:
# Update metadata with error
# Update metadata with error (async to avoid blocking)
metadata["status"] = "error"
metadata["error"] = str(e)
metadata["end_time"] = datetime.now().isoformat()
with open(metadata_file, "w") as f:
json.dump(metadata, f, indent=2)
async with aiofiles.open(metadata_file, "w") as f:
await f.write(json.dumps(metadata, indent=2))
return {
"error": str(e),
@@ -410,17 +421,20 @@ class StreamingCommandTool(BaseProcessTool):
}
cmd_dir = self.commands_dir / cmd_id
if not cmd_dir.exists():
cmd_dir_exists = await asyncio.to_thread(cmd_dir.exists)
if not cmd_dir_exists:
return {"error": f"Command directory not found: {cmd_id}"}
# Load metadata
# Load metadata (async to avoid blocking)
metadata_file = cmd_dir / "metadata.json"
with open(metadata_file, "r") as f:
metadata = json.load(f)
async with aiofiles.open(metadata_file, "r") as f:
content = await f.read()
metadata = json.loads(content)
# Determine start position
output_file = cmd_dir / "output.log"
if not output_file.exists():
output_exists = await asyncio.to_thread(output_file.exists)
if not output_exists:
return {"error": "No output file found"}
# If no from_byte specified, read from where we left off
@@ -428,13 +442,14 @@ class StreamingCommandTool(BaseProcessTool):
# Try to determine from previous reads (could track this)
from_byte = 0 # For now, start from beginning if not specified
# Read chunk
# Read chunk (async to avoid blocking)
try:
with open(output_file, "r", errors="replace") as f:
f.seek(from_byte)
content = f.read(chunk_size)
new_position = f.tell()
file_size = output_file.stat().st_size
async with aiofiles.open(output_file, "r", errors="replace") as f:
await f.seek(from_byte)
content = await f.read(chunk_size)
# Calculate new position based on bytes read
new_position = from_byte + len(content.encode("utf-8"))
file_size = await asyncio.to_thread(lambda: output_file.stat().st_size)
# Check if process is still running
status = metadata.get("status", "unknown")
@@ -452,11 +467,14 @@ class StreamingCommandTool(BaseProcessTool):
"total_bytes": file_size,
}
# Add stderr if needed
# Add stderr if needed (async)
error_file = cmd_dir / "error.log"
if error_file.exists() and error_file.stat().st_size > 0:
with open(error_file, "r", errors="replace") as f:
result["stderr"] = f.read(1000)
error_exists = await asyncio.to_thread(error_file.exists)
if error_exists:
error_size = await asyncio.to_thread(lambda: error_file.stat().st_size)
if error_size > 0:
async with aiofiles.open(error_file, "r", errors="replace") as f:
result["stderr"] = await f.read(1000)
# Add continuation info
if new_position < file_size or status == "running":
@@ -477,18 +495,26 @@ class StreamingCommandTool(BaseProcessTool):
return {"error": f"Error reading output: {str(e)}"}
async def _get_recent_commands(self, limit: int = 5) -> List[Dict[str, Any]]:
"""Get list of recent commands for hints."""
"""Get list of recent commands for hints (async to avoid blocking)."""
commands = []
for cmd_dir in sorted(self.commands_dir.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True)[:limit]:
# Get sorted directories asynchronously
def _get_sorted_dirs():
return sorted(self.commands_dir.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True)[:limit]
cmd_dirs = await asyncio.to_thread(_get_sorted_dirs)
for cmd_dir in cmd_dirs:
try:
with open(cmd_dir / "metadata.json", "r") as f:
meta = json.load(f)
async with aiofiles.open(cmd_dir / "metadata.json", "r") as f:
content = await f.read()
meta = json.loads(content)
output_size = 0
output_file = cmd_dir / "output.log"
if output_file.exists():
output_size = output_file.stat().st_size
output_exists = await asyncio.to_thread(output_file.exists)
if output_exists:
output_size = await asyncio.to_thread(lambda: output_file.stat().st_size)
commands.append(
{
@@ -547,16 +573,17 @@ class StreamingCommandTool(BaseProcessTool):
return {"error": "No output file found"}
try:
# Use tail command for efficiency
result = subprocess.run(
["tail", "-n", str(lines or 20), str(output_file)],
capture_output=True,
text=True,
# Use tail command for efficiency (async to avoid blocking)
process = await asyncio.create_subprocess_exec(
"tail", "-n", str(lines or 20), str(output_file),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
return {
"command_id": cmd_id[:8],
"output": result.stdout,
"output": stdout.decode("utf-8", errors="replace"),
"lines": lines,
}
except Exception as e:
@@ -1,4 +1,23 @@
"""Run Python packages in background with uvx."""
"""Run Python packages in background with uvx.
DEPRECATED: This module uses blocking subprocess.Popen and subprocess.run.
For new code, use the async implementations in:
- hanzo_mcp.tools.shell.uvx_tool.UvxTool (based on BaseBinaryTool)
This provides non-blocking process execution using asyncio.create_subprocess_exec
with automatic backgrounding for long-running processes.
This module is maintained for backwards compatibility only.
"""
import warnings
# Emit deprecation warning on import
warnings.warn(
"uvx_background.py uses blocking subprocess. "
"Consider using uvx_tool.UvxTool for async execution.",
DeprecationWarning,
stacklevel=2
)
import uuid
import shutil
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-mcp"
version = "0.9.16"
version = "0.9.17"
description = "The Zen of Hanzo MCP: One server to rule them all. The ultimate MCP that orchestrates all others."
readme = "README.md"
requires-python = ">=3.12"