mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zeekay
351b1a2b1b
MCPServerConnection only spawned stdio servers, so an endpoint a process already serves - Next 16's /_next/mcp, vite-plugin-mcp's /__mcp/sse - had no client in this repo. MCPServerConfig now takes a url. connect lists the tools and call_tool calls one through the SDK's streamable HTTP or SSE client, a session per request, and returns the same JSON-RPC result the stdio path returns. A loopback URL skips TLS verification, since a local dev certificate is self-signed. A failed connect says why in .error. Claude-Session: https://claude.ai/code/session_01Dc8DXc3fEQCRHUa6XWK4Ku
125 lines
2.8 KiB
Python
125 lines
2.8 KiB
Python
"""MCP management tools for Hanzo AI.
|
|
|
|
Tools:
|
|
- mcp: MCP server management
|
|
- mcp_add: Add MCP servers
|
|
- mcp_remove: Remove MCP servers
|
|
- mcp_stats: MCP statistics
|
|
- proxy: Dynamic MCP proxy for external servers (platform, github, cloudflare, etc.)
|
|
|
|
Install:
|
|
pip install hanzo-tools-mcp
|
|
"""
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_tools = []
|
|
|
|
try:
|
|
from .mcp_tool import MCPTool
|
|
|
|
_tools.append(MCPTool)
|
|
except ImportError as e:
|
|
logger.debug(f"MCPTool not available: {e}")
|
|
MCPTool = None
|
|
|
|
try:
|
|
from .mcp_add import McpAddTool
|
|
|
|
_tools.append(McpAddTool)
|
|
except ImportError as e:
|
|
logger.debug(f"McpAddTool not available: {e}")
|
|
McpAddTool = None
|
|
|
|
try:
|
|
from .mcp_remove import McpRemoveTool
|
|
|
|
_tools.append(McpRemoveTool)
|
|
except ImportError as e:
|
|
logger.debug(f"McpRemoveTool not available: {e}")
|
|
McpRemoveTool = None
|
|
|
|
try:
|
|
from .mcp_stats import McpStatsTool
|
|
|
|
_tools.append(McpStatsTool)
|
|
except ImportError as e:
|
|
logger.debug(f"McpStatsTool not available: {e}")
|
|
McpStatsTool = None
|
|
|
|
try:
|
|
from .proxy_tool import ProxyTool
|
|
|
|
_tools.append(ProxyTool)
|
|
except ImportError as e:
|
|
logger.debug(f"ProxyTool not available: {e}")
|
|
ProxyTool = None
|
|
|
|
# Proxy utilities for programmatic use
|
|
try:
|
|
from .mcp_proxy import (
|
|
BUILTIN_SERVERS,
|
|
ProxiedTool,
|
|
MCPServerConfig,
|
|
MCPProxyRegistry,
|
|
MCPServerConnection,
|
|
http_session,
|
|
call_mcp_tool,
|
|
enable_mcp_server,
|
|
)
|
|
except ImportError as e:
|
|
logger.debug(f"MCP proxy utilities not available: {e}")
|
|
MCPProxyRegistry = None
|
|
MCPServerConfig = None
|
|
MCPServerConnection = None
|
|
ProxiedTool = None
|
|
enable_mcp_server = None
|
|
call_mcp_tool = None
|
|
http_session = None
|
|
BUILTIN_SERVERS = {}
|
|
|
|
TOOLS = _tools
|
|
|
|
__all__ = [
|
|
"TOOLS",
|
|
"MCPTool",
|
|
"McpAddTool",
|
|
"McpRemoveTool",
|
|
"McpStatsTool",
|
|
"ProxyTool",
|
|
# Proxy utilities
|
|
"MCPProxyRegistry",
|
|
"MCPServerConfig",
|
|
"MCPServerConnection",
|
|
"ProxiedTool",
|
|
"enable_mcp_server",
|
|
"call_mcp_tool",
|
|
"http_session",
|
|
"BUILTIN_SERVERS",
|
|
"register_tools",
|
|
]
|
|
|
|
|
|
def register_tools(mcp_server, enabled_tools: dict[str, bool] | None = None):
|
|
"""Register MCP tools with MCP server."""
|
|
from hanzo_tools.core import ToolRegistry
|
|
|
|
enabled = enabled_tools or {}
|
|
registered = []
|
|
|
|
for tool_class in TOOLS:
|
|
if tool_class is None:
|
|
continue
|
|
tool_name = getattr(tool_class, "name", tool_class.__name__.lower())
|
|
if enabled.get(tool_name, True):
|
|
try:
|
|
tool = tool_class()
|
|
ToolRegistry.register_tool(mcp_server, tool)
|
|
registered.append(tool)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to register {tool_name}: {e}")
|
|
|
|
return registered
|