dispatcher.py•2.68 kB
"""
Tool dispatcher for MCP server.
Manages registration and execution of tools.
"""
from typing import Any, Callable, Dict, List
from jsonrpc import JSONRPCError, METHOD_NOT_FOUND, INVALID_PARAMS
from logger import log_debug, log_error
class ToolDispatcher:
"""Manages and dispatches tool calls."""
def __init__(self):
"""Initialize the dispatcher with empty tool registry."""
self.tools: Dict[str, Dict[str, Any]] = {}
def register_tool(self, name: str, handler: Callable, description: str,
input_schema: Dict[str, Any]):
"""
Register a tool with the dispatcher.
Args:
name: Tool name
handler: Function to call when tool is invoked
description: Human-readable description
input_schema: JSON Schema for tool parameters
"""
self.tools[name] = {
"handler": handler,
"description": description,
"inputSchema": input_schema
}
log_debug(f"Registered tool: {name}")
def get_tool_list(self) -> List[Dict[str, Any]]:
"""
Get list of available tools for capabilities response.
Returns:
List of tool definitions
"""
tool_list = []
for name, info in self.tools.items():
tool_list.append({
"name": name,
"description": info["description"],
"inputSchema": info["inputSchema"]
})
return tool_list
def dispatch(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""
Dispatch a tool call.
Args:
tool_name: Name of the tool to call
arguments: Arguments to pass to the tool
Returns:
Result from the tool handler
Raises:
JSONRPCError: If tool not found or execution fails
"""
if tool_name not in self.tools:
raise JSONRPCError(METHOD_NOT_FOUND, f"Tool not found: {tool_name}")
tool = self.tools[tool_name]
handler = tool["handler"]
try:
log_debug(f"Executing tool: {tool_name} with args: {arguments}")
result = handler(arguments)
return result
except TypeError as e:
log_error(f"Invalid parameters for {tool_name}: {e}")
raise JSONRPCError(INVALID_PARAMS, f"Invalid parameters: {e}")
except Exception as e:
log_error(f"Tool execution error in {tool_name}: {e}")
raise JSONRPCError(-32000, f"Tool execution failed: {e}")