mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(mcp): add comprehensive error logging and response truncation
- Add centralized error logging to ~/.hanzo/mcp/logs/ - Implement MCPErrorLogger with structured logging (JSON, daily, tool-specific logs) - Add parameter sanitization to redact sensitive data - Create @with_error_logging decorator for automatic error capture - Add response truncation (25k tokens) to shell tool outputs - Apply error logging to ReadTool as example - Add comprehensive test suite for error logging Fixes call signature errors and token limit issues. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
@@ -6,6 +6,7 @@ behavior and provide a foundation for tool registration and management.
|
||||
"""
|
||||
|
||||
import functools
|
||||
import inspect
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, Callable, final
|
||||
from collections.abc import Awaitable
|
||||
@@ -20,6 +21,60 @@ from hanzo_mcp.tools.common.validation import (
|
||||
validate_path_parameter,
|
||||
)
|
||||
from hanzo_mcp.tools.common.permissions import PermissionManager
|
||||
from hanzo_mcp.tools.common.error_logger import log_tool_error, log_call_signature_error
|
||||
|
||||
|
||||
def with_error_logging(tool_name: str) -> Callable:
|
||||
"""Decorator to add comprehensive error logging to tool functions.
|
||||
|
||||
Args:
|
||||
tool_name: Name of the tool for logging purposes
|
||||
|
||||
Returns:
|
||||
Decorator function
|
||||
"""
|
||||
def decorator(func: Callable[..., Awaitable[str]]) -> Callable[..., Awaitable[str]]:
|
||||
@functools.wraps(func)
|
||||
async def wrapper(*args: Any, **kwargs: Any) -> str:
|
||||
try:
|
||||
return await func(*args, **kwargs)
|
||||
except TypeError as e:
|
||||
# This often indicates a call signature mismatch
|
||||
error_msg = str(e)
|
||||
if "takes" in error_msg and "positional argument" in error_msg:
|
||||
# Log call signature error
|
||||
sig = inspect.signature(func)
|
||||
expected = f"{func.__name__}{sig}"
|
||||
actual = f"{func.__name__}(*args={args}, **kwargs={kwargs})"
|
||||
log_call_signature_error(tool_name, expected, actual, e)
|
||||
|
||||
# Log the error
|
||||
log_tool_error(
|
||||
tool_name,
|
||||
e,
|
||||
params=kwargs,
|
||||
context=f"Call signature mismatch or type error"
|
||||
)
|
||||
|
||||
# Return user-friendly error message
|
||||
return (
|
||||
f"Error executing tool '{tool_name}': {error_msg}\n\n"
|
||||
f"This error has been logged to ~/.hanzo/mcp/logs/ for debugging.\n"
|
||||
f"Check ~/.hanzo/mcp/logs/{tool_name}-errors.log for details."
|
||||
)
|
||||
except Exception as e:
|
||||
# Log all other errors
|
||||
log_tool_error(tool_name, e, params=kwargs)
|
||||
|
||||
# Return error message
|
||||
return (
|
||||
f"Error executing tool '{tool_name}': {str(e)}\n\n"
|
||||
f"This error has been logged to ~/.hanzo/mcp/logs/ for debugging.\n"
|
||||
f"Check ~/.hanzo/mcp/logs/{tool_name}-errors.log for details."
|
||||
)
|
||||
|
||||
return wrapper
|
||||
return decorator
|
||||
|
||||
|
||||
def handle_connection_errors(
|
||||
|
||||
@@ -0,0 +1,289 @@
|
||||
"""Centralized error logging for MCP tools.
|
||||
|
||||
This module provides comprehensive error logging for all MCP tool operations,
|
||||
writing errors to ~/.hanzo/mcp/logs/ for debugging and analysis.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import traceback
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
import os
|
||||
|
||||
|
||||
class MCPErrorLogger:
|
||||
"""Centralized error logger for MCP tools."""
|
||||
|
||||
def __init__(self, log_dir: Optional[Path] = None):
|
||||
"""Initialize the error logger.
|
||||
|
||||
Args:
|
||||
log_dir: Directory for log files (default: ~/.hanzo/mcp/logs/)
|
||||
"""
|
||||
if log_dir is None:
|
||||
log_dir = Path.home() / ".hanzo" / "mcp" / "logs"
|
||||
|
||||
self.log_dir = log_dir
|
||||
self.log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create daily log file
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
self.log_file = self.log_dir / f"mcp-errors-{today}.log"
|
||||
|
||||
# Also create a general errors file
|
||||
self.general_log_file = self.log_dir / "errors.log"
|
||||
|
||||
# Set up Python logging
|
||||
self._setup_logging()
|
||||
|
||||
def _setup_logging(self):
|
||||
"""Set up Python logging infrastructure."""
|
||||
# Create logger
|
||||
self.logger = logging.getLogger("hanzo_mcp.errors")
|
||||
self.logger.setLevel(logging.DEBUG)
|
||||
|
||||
# Prevent duplicate handlers
|
||||
if self.logger.handlers:
|
||||
return
|
||||
|
||||
# File handler for daily logs
|
||||
file_handler = logging.FileHandler(self.log_file)
|
||||
file_handler.setLevel(logging.ERROR)
|
||||
|
||||
# File handler for general log
|
||||
general_handler = logging.FileHandler(self.general_log_file)
|
||||
general_handler.setLevel(logging.ERROR)
|
||||
|
||||
# Formatter
|
||||
formatter = logging.Formatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S'
|
||||
)
|
||||
file_handler.setFormatter(formatter)
|
||||
general_handler.setFormatter(formatter)
|
||||
|
||||
# Add handlers
|
||||
self.logger.addHandler(file_handler)
|
||||
self.logger.addHandler(general_handler)
|
||||
|
||||
def log_tool_error(
|
||||
self,
|
||||
tool_name: str,
|
||||
error: Exception,
|
||||
params: Optional[dict[str, Any]] = None,
|
||||
context: Optional[str] = None,
|
||||
):
|
||||
"""Log a tool execution error.
|
||||
|
||||
Args:
|
||||
tool_name: Name of the tool that errored
|
||||
error: The exception that was raised
|
||||
params: Tool parameters (will be sanitized)
|
||||
context: Additional context about the error
|
||||
"""
|
||||
error_data = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"tool_name": tool_name,
|
||||
"error_type": type(error).__name__,
|
||||
"error_message": str(error),
|
||||
"traceback": traceback.format_exc(),
|
||||
"params": self._sanitize_params(params) if params else None,
|
||||
"context": context,
|
||||
}
|
||||
|
||||
# Write to JSON log for structured parsing
|
||||
json_log_file = self.log_dir / f"tool-errors-{datetime.now().strftime('%Y-%m-%d')}.jsonl"
|
||||
try:
|
||||
with open(json_log_file, "a") as f:
|
||||
json.dump(error_data, f)
|
||||
f.write("\n")
|
||||
except Exception as e:
|
||||
# If JSON logging fails, at least log that
|
||||
self.logger.error(f"Failed to write JSON log: {e}")
|
||||
|
||||
# Also log to standard logger
|
||||
self.logger.error(
|
||||
f"Tool '{tool_name}' error: {type(error).__name__}: {str(error)}",
|
||||
extra={"tool": tool_name, "params": error_data.get("params")}
|
||||
)
|
||||
|
||||
# Write detailed error to tool-specific file
|
||||
tool_log_file = self.log_dir / f"{tool_name}-errors.log"
|
||||
try:
|
||||
with open(tool_log_file, "a") as f:
|
||||
f.write(f"\n{'='*80}\n")
|
||||
f.write(f"ERROR at {error_data['timestamp']}\n")
|
||||
f.write(f"{'='*80}\n")
|
||||
f.write(f"Tool: {tool_name}\n")
|
||||
f.write(f"Error Type: {error_data['error_type']}\n")
|
||||
f.write(f"Error Message: {error_data['error_message']}\n")
|
||||
if context:
|
||||
f.write(f"Context: {context}\n")
|
||||
if params:
|
||||
f.write(f"\nParameters:\n{json.dumps(error_data['params'], indent=2)}\n")
|
||||
f.write(f"\nTraceback:\n{error_data['traceback']}\n")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to write tool-specific log: {e}")
|
||||
|
||||
def log_call_signature_error(
|
||||
self,
|
||||
tool_name: str,
|
||||
expected_signature: str,
|
||||
actual_call: str,
|
||||
error: Exception,
|
||||
):
|
||||
"""Log an error related to incorrect tool call signature.
|
||||
|
||||
Args:
|
||||
tool_name: Name of the tool
|
||||
expected_signature: Expected function signature
|
||||
actual_call: How the tool was actually called
|
||||
error: The exception that was raised
|
||||
"""
|
||||
error_data = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"tool_name": tool_name,
|
||||
"error_type": "CallSignatureError",
|
||||
"expected_signature": expected_signature,
|
||||
"actual_call": actual_call,
|
||||
"error_message": str(error),
|
||||
"traceback": traceback.format_exc(),
|
||||
}
|
||||
|
||||
# Write to signature errors log
|
||||
sig_log_file = self.log_dir / "signature-errors.log"
|
||||
try:
|
||||
with open(sig_log_file, "a") as f:
|
||||
f.write(f"\n{'='*80}\n")
|
||||
f.write(f"CALL SIGNATURE ERROR at {error_data['timestamp']}\n")
|
||||
f.write(f"{'='*80}\n")
|
||||
f.write(f"Tool: {tool_name}\n")
|
||||
f.write(f"Expected: {expected_signature}\n")
|
||||
f.write(f"Actual: {actual_call}\n")
|
||||
f.write(f"Error: {error_data['error_message']}\n")
|
||||
f.write(f"\nTraceback:\n{error_data['traceback']}\n")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to write signature error log: {e}")
|
||||
|
||||
# Also log as regular tool error
|
||||
self.log_tool_error(
|
||||
tool_name,
|
||||
error,
|
||||
context=f"Call signature mismatch - Expected: {expected_signature}, Got: {actual_call}"
|
||||
)
|
||||
|
||||
def _sanitize_params(self, params: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Sanitize parameters to remove sensitive data.
|
||||
|
||||
Args:
|
||||
params: Parameters to sanitize
|
||||
|
||||
Returns:
|
||||
Sanitized parameters
|
||||
"""
|
||||
sanitized = {}
|
||||
sensitive_keys = {
|
||||
"password", "token", "key", "secret", "api_key", "auth",
|
||||
"credential", "private", "ssh_key", "passphrase"
|
||||
}
|
||||
|
||||
for key, value in params.items():
|
||||
# Check if key contains sensitive terms
|
||||
if any(sensitive in key.lower() for sensitive in sensitive_keys):
|
||||
sanitized[key] = "[REDACTED]"
|
||||
# Recursively sanitize nested dicts
|
||||
elif isinstance(value, dict):
|
||||
sanitized[key] = self._sanitize_params(value)
|
||||
# Convert non-serializable types to strings
|
||||
elif not isinstance(value, (str, int, float, bool, list, dict, type(None))):
|
||||
sanitized[key] = str(value)
|
||||
else:
|
||||
sanitized[key] = value
|
||||
|
||||
return sanitized
|
||||
|
||||
def get_recent_errors(self, tool_name: Optional[str] = None, limit: int = 10) -> list[dict]:
|
||||
"""Get recent errors from the JSON log.
|
||||
|
||||
Args:
|
||||
tool_name: Filter by tool name (optional)
|
||||
limit: Maximum number of errors to return
|
||||
|
||||
Returns:
|
||||
List of error dictionaries
|
||||
"""
|
||||
today = datetime.now().strftime('%Y-%m-%d')
|
||||
json_log_file = self.log_dir / f"tool-errors-{today}.jsonl"
|
||||
|
||||
if not json_log_file.exists():
|
||||
return []
|
||||
|
||||
errors = []
|
||||
try:
|
||||
with open(json_log_file, "r") as f:
|
||||
for line in f:
|
||||
try:
|
||||
error = json.loads(line)
|
||||
if tool_name is None or error.get("tool_name") == tool_name:
|
||||
errors.append(error)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to read error log: {e}")
|
||||
|
||||
# Return most recent errors
|
||||
return errors[-limit:]
|
||||
|
||||
|
||||
# Global error logger instance
|
||||
_global_error_logger: Optional[MCPErrorLogger] = None
|
||||
|
||||
|
||||
def get_error_logger() -> MCPErrorLogger:
|
||||
"""Get the global error logger instance.
|
||||
|
||||
Returns:
|
||||
Global error logger
|
||||
"""
|
||||
global _global_error_logger
|
||||
if _global_error_logger is None:
|
||||
_global_error_logger = MCPErrorLogger()
|
||||
return _global_error_logger
|
||||
|
||||
|
||||
def log_tool_error(
|
||||
tool_name: str,
|
||||
error: Exception,
|
||||
params: Optional[dict[str, Any]] = None,
|
||||
context: Optional[str] = None,
|
||||
):
|
||||
"""Convenience function to log a tool error using the global logger.
|
||||
|
||||
Args:
|
||||
tool_name: Name of the tool that errored
|
||||
error: The exception that was raised
|
||||
params: Tool parameters
|
||||
context: Additional context
|
||||
"""
|
||||
logger = get_error_logger()
|
||||
logger.log_tool_error(tool_name, error, params, context)
|
||||
|
||||
|
||||
def log_call_signature_error(
|
||||
tool_name: str,
|
||||
expected_signature: str,
|
||||
actual_call: str,
|
||||
error: Exception,
|
||||
):
|
||||
"""Convenience function to log a call signature error.
|
||||
|
||||
Args:
|
||||
tool_name: Name of the tool
|
||||
expected_signature: Expected function signature
|
||||
actual_call: How the tool was actually called
|
||||
error: The exception that was raised
|
||||
"""
|
||||
logger = get_error_logger()
|
||||
logger.log_call_signature_error(tool_name, expected_signature, actual_call, error)
|
||||
@@ -14,6 +14,7 @@ from mcp.server.fastmcp import Context as MCPContext
|
||||
|
||||
from hanzo_mcp.tools.common.truncate import truncate_response
|
||||
from hanzo_mcp.tools.filesystem.base import FilesystemBaseTool
|
||||
from hanzo_mcp.tools.common.base import with_error_logging
|
||||
|
||||
FilePath = Annotated[
|
||||
str,
|
||||
@@ -252,6 +253,7 @@ Usage:
|
||||
tool_self = self
|
||||
|
||||
@mcp_server.tool(name=self.name, description=self.description)
|
||||
@with_error_logging(self.name)
|
||||
async def read(
|
||||
ctx: MCPContext,
|
||||
file_path: FilePath,
|
||||
|
||||
@@ -12,6 +12,7 @@ from mcp.server.fastmcp import Context as MCPContext
|
||||
|
||||
from hanzo_mcp.tools.common.base import BaseTool
|
||||
from hanzo_mcp.tools.common.permissions import PermissionManager
|
||||
from hanzo_mcp.tools.common.truncate import truncate_response
|
||||
|
||||
|
||||
class BashCommandStatus(Enum):
|
||||
@@ -95,14 +96,15 @@ class CommandResult:
|
||||
return f"Command `{self.command}` failed: {self.error_message}"
|
||||
return f"Command `{self.command}` executed with exit code {self.return_code}."
|
||||
|
||||
def format_output(self, include_exit_code: bool = True) -> str:
|
||||
def format_output(self, include_exit_code: bool = True, max_tokens: int = 25000) -> str:
|
||||
"""Format the command output as a string.
|
||||
|
||||
Args:
|
||||
include_exit_code: Whether to include the exit code in the output
|
||||
max_tokens: Maximum tokens allowed in the response (default: 25000)
|
||||
|
||||
Returns:
|
||||
Formatted output string
|
||||
Formatted output string, truncated if necessary
|
||||
"""
|
||||
result_parts: list[str] = []
|
||||
|
||||
@@ -131,10 +133,24 @@ class CommandResult:
|
||||
result_parts.append(f"STDERR:\n{self.stderr}")
|
||||
|
||||
# Join with newlines
|
||||
return "\n\n".join(result_parts)
|
||||
result = "\n\n".join(result_parts)
|
||||
|
||||
def to_agent_observation(self) -> str:
|
||||
"""Format the result for agent consumption."""
|
||||
# Truncate if necessary
|
||||
return truncate_response(
|
||||
result,
|
||||
max_tokens=max_tokens,
|
||||
truncation_message="\n\n[Shell output truncated due to token limit. Use pagination, filtering, or limit parameters to reduce output size.]"
|
||||
)
|
||||
|
||||
def to_agent_observation(self, max_tokens: int = 25000) -> str:
|
||||
"""Format the result for agent consumption.
|
||||
|
||||
Args:
|
||||
max_tokens: Maximum tokens allowed in the response (default: 25000)
|
||||
|
||||
Returns:
|
||||
Formatted output, truncated if necessary
|
||||
"""
|
||||
content = self.stdout
|
||||
|
||||
additional_info: list[str] = []
|
||||
@@ -144,7 +160,12 @@ class CommandResult:
|
||||
if additional_info:
|
||||
content += "\n" + "\n".join(additional_info)
|
||||
|
||||
return content
|
||||
# Truncate if necessary
|
||||
return truncate_response(
|
||||
content,
|
||||
max_tokens=max_tokens,
|
||||
truncation_message="\n\n[Shell output truncated due to token limit. Use pagination, filtering, or limit parameters to reduce output size.]"
|
||||
)
|
||||
|
||||
|
||||
class ShellBaseTool(BaseTool, ABC):
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Tests for MCP error logging functionality."""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
from hanzo_mcp.tools.common.error_logger import (
|
||||
MCPErrorLogger,
|
||||
get_error_logger,
|
||||
log_tool_error,
|
||||
log_call_signature_error,
|
||||
)
|
||||
|
||||
|
||||
class TestMCPErrorLogger:
|
||||
"""Test suite for MCPErrorLogger."""
|
||||
|
||||
def test_logger_initialization(self, tmp_path):
|
||||
"""Test that logger initializes correctly."""
|
||||
logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
assert logger.log_dir == tmp_path
|
||||
assert logger.log_dir.exists()
|
||||
|
||||
# Check that log files are created with today's date
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
assert logger.log_file == tmp_path / f"mcp-errors-{today}.log"
|
||||
assert logger.general_log_file == tmp_path / "errors.log"
|
||||
|
||||
def test_log_tool_error(self, tmp_path):
|
||||
"""Test logging a tool error."""
|
||||
logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
# Create a test error
|
||||
test_error = ValueError("Test error message")
|
||||
params = {"file_path": "/test/path", "limit": 100}
|
||||
|
||||
# Log the error
|
||||
logger.log_tool_error(
|
||||
tool_name="read",
|
||||
error=test_error,
|
||||
params=params,
|
||||
context="Testing error logging"
|
||||
)
|
||||
|
||||
# Check that log files were created
|
||||
tool_log = tmp_path / "read-errors.log"
|
||||
assert tool_log.exists()
|
||||
|
||||
# Read the log and verify content
|
||||
content = tool_log.read_text()
|
||||
assert "Test error message" in content
|
||||
assert "ValueError" in content
|
||||
assert "Testing error logging" in content
|
||||
|
||||
# Check JSON log
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
json_log = tmp_path / f"tool-errors-{today}.jsonl"
|
||||
assert json_log.exists()
|
||||
|
||||
def test_log_call_signature_error(self, tmp_path):
|
||||
"""Test logging a call signature error."""
|
||||
logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
# Create a test error
|
||||
test_error = TypeError("takes 1 positional argument but 2 were given")
|
||||
|
||||
# Log the error
|
||||
logger.log_call_signature_error(
|
||||
tool_name="read",
|
||||
expected_signature="read(ctx, file_path: str)",
|
||||
actual_call="read(ctx, '/path/to/file')",
|
||||
error=test_error
|
||||
)
|
||||
|
||||
# Check that signature errors log was created
|
||||
sig_log = tmp_path / "signature-errors.log"
|
||||
assert sig_log.exists()
|
||||
|
||||
# Read the log and verify content
|
||||
content = sig_log.read_text()
|
||||
assert "CALL SIGNATURE ERROR" in content
|
||||
assert "read(ctx, file_path: str)" in content
|
||||
assert "read(ctx, '/path/to/file')" in content
|
||||
|
||||
def test_sanitize_params(self, tmp_path):
|
||||
"""Test that sensitive parameters are sanitized."""
|
||||
logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
# Create params with sensitive data
|
||||
params = {
|
||||
"file_path": "/test/path",
|
||||
"api_key": "secret-key-123",
|
||||
"password": "super-secret",
|
||||
"token": "bearer-token",
|
||||
"normal_param": "normal-value"
|
||||
}
|
||||
|
||||
sanitized = logger._sanitize_params(params)
|
||||
|
||||
# Check that sensitive keys are redacted
|
||||
assert sanitized["api_key"] == "[REDACTED]"
|
||||
assert sanitized["password"] == "[REDACTED]"
|
||||
assert sanitized["token"] == "[REDACTED]"
|
||||
|
||||
# Check that normal params are preserved
|
||||
assert sanitized["file_path"] == "/test/path"
|
||||
assert sanitized["normal_param"] == "normal-value"
|
||||
|
||||
def test_get_recent_errors(self, tmp_path):
|
||||
"""Test retrieving recent errors."""
|
||||
logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
# Log multiple errors
|
||||
for i in range(5):
|
||||
error = ValueError(f"Error {i}")
|
||||
logger.log_tool_error(
|
||||
tool_name=f"tool_{i}",
|
||||
error=error,
|
||||
params={"index": i}
|
||||
)
|
||||
|
||||
# Get recent errors
|
||||
recent = logger.get_recent_errors(limit=3)
|
||||
assert len(recent) == 3
|
||||
|
||||
# Check they're the most recent ones
|
||||
assert "Error 4" in recent[-1]["error_message"]
|
||||
assert "Error 3" in recent[-2]["error_message"]
|
||||
|
||||
def test_global_logger(self):
|
||||
"""Test global logger singleton."""
|
||||
logger1 = get_error_logger()
|
||||
logger2 = get_error_logger()
|
||||
|
||||
# Should be the same instance
|
||||
assert logger1 is logger2
|
||||
|
||||
def test_convenience_functions(self, tmp_path):
|
||||
"""Test convenience logging functions."""
|
||||
# Override global logger for testing
|
||||
from hanzo_mcp.tools.common import error_logger as el_module
|
||||
el_module._global_error_logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
# Test log_tool_error
|
||||
error = RuntimeError("Test runtime error")
|
||||
log_tool_error("test_tool", error, params={"test": "param"})
|
||||
|
||||
tool_log = tmp_path / "test_tool-errors.log"
|
||||
assert tool_log.exists()
|
||||
assert "Test runtime error" in tool_log.read_text()
|
||||
|
||||
# Test log_call_signature_error
|
||||
sig_error = TypeError("signature mismatch")
|
||||
log_call_signature_error(
|
||||
"test_tool",
|
||||
"expected signature",
|
||||
"actual call",
|
||||
sig_error
|
||||
)
|
||||
|
||||
sig_log = tmp_path / "signature-errors.log"
|
||||
assert sig_log.exists()
|
||||
|
||||
|
||||
class TestErrorLoggingIntegration:
|
||||
"""Integration tests for error logging with tools."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_read_tool_with_error_logging(self, tmp_path):
|
||||
"""Test that ReadTool logs errors correctly."""
|
||||
from hanzo_mcp.tools.filesystem.read import ReadTool
|
||||
from hanzo_mcp.tools.common.permissions import PermissionManager
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
from hanzo_mcp.tools.common import error_logger as el_module
|
||||
|
||||
# Override global logger
|
||||
el_module._global_error_logger = MCPErrorLogger(log_dir=tmp_path)
|
||||
|
||||
# Create tool
|
||||
perm_mgr = PermissionManager()
|
||||
tool = ReadTool(perm_mgr)
|
||||
|
||||
# Create mock context
|
||||
class MockContext:
|
||||
async def info(self, msg): pass
|
||||
async def error(self, msg): pass
|
||||
async def warning(self, msg): pass
|
||||
|
||||
ctx = MockContext()
|
||||
|
||||
# Try to read non-existent file (should log error if call signature is wrong)
|
||||
# This simulates the original error where LLM called with positional args
|
||||
try:
|
||||
# This would cause an error if called incorrectly
|
||||
result = await tool.call(ctx, file_path="/nonexistent/file.txt")
|
||||
# Should return error message, not raise
|
||||
assert "Error" in result or "does not exist" in result
|
||||
except TypeError as e:
|
||||
# If there's a type error, it should be logged
|
||||
log_tool_error("read", e, context="Integration test")
|
||||
|
||||
# Verify error was logged
|
||||
read_log = tmp_path / "read-errors.log"
|
||||
assert read_log.exists()
|
||||
|
||||
|
||||
def test_error_logger_creates_directory():
|
||||
"""Test that error logger creates ~/.hanzo/mcp/logs/ directory."""
|
||||
logger = get_error_logger()
|
||||
|
||||
expected_dir = Path.home() / ".hanzo" / "mcp" / "logs"
|
||||
assert logger.log_dir == expected_dir
|
||||
assert expected_dir.exists()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user