mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
mcp_proxy reaches running servers over HTTP and SSE
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
This commit is contained in:
@@ -65,6 +65,7 @@ try:
|
||||
MCPServerConfig,
|
||||
MCPProxyRegistry,
|
||||
MCPServerConnection,
|
||||
http_session,
|
||||
call_mcp_tool,
|
||||
enable_mcp_server,
|
||||
)
|
||||
@@ -76,6 +77,7 @@ except ImportError as e:
|
||||
ProxiedTool = None
|
||||
enable_mcp_server = None
|
||||
call_mcp_tool = None
|
||||
http_session = None
|
||||
BUILTIN_SERVERS = {}
|
||||
|
||||
TOOLS = _tools
|
||||
@@ -94,6 +96,7 @@ __all__ = [
|
||||
"ProxiedTool",
|
||||
"enable_mcp_server",
|
||||
"call_mcp_tool",
|
||||
"http_session",
|
||||
"BUILTIN_SERVERS",
|
||||
"register_tools",
|
||||
]
|
||||
|
||||
@@ -18,19 +18,82 @@ import shutil
|
||||
import asyncio
|
||||
import logging
|
||||
import subprocess
|
||||
from typing import Any, Dict, List, Callable, Optional, Awaitable
|
||||
from typing import Any, Dict, List, Callable, Optional, Awaitable, AsyncIterator
|
||||
from pathlib import Path
|
||||
from datetime import timedelta
|
||||
from contextlib import AsyncExitStack, asynccontextmanager
|
||||
from dataclasses import field, dataclass
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
from mcp import ClientSession
|
||||
from mcp.client.sse import sse_client
|
||||
from mcp.client.streamable_http import streamable_http_client
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LOOPBACK = frozenset({"localhost", "127.0.0.1", "::1"})
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def http_session(url: str, timeout: float = 30.0) -> AsyncIterator[ClientSession]:
|
||||
"""An initialized MCP session over HTTP: SSE when the path ends in /sse,
|
||||
streamable HTTP otherwise.
|
||||
|
||||
A loopback URL skips TLS verification: a local dev server's certificate is
|
||||
self-signed, and the bytes never leave the machine.
|
||||
"""
|
||||
verify = urlparse(url).hostname not in LOOPBACK
|
||||
|
||||
def client(
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
timeout: Optional[httpx.Timeout] = None,
|
||||
auth: Optional[httpx.Auth] = None,
|
||||
) -> httpx.AsyncClient:
|
||||
return httpx.AsyncClient(
|
||||
headers=headers,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
verify=verify,
|
||||
follow_redirects=True,
|
||||
)
|
||||
|
||||
async with AsyncExitStack() as stack:
|
||||
if urlparse(url).path.endswith("/sse"):
|
||||
read, write = await stack.enter_async_context(
|
||||
sse_client(url, timeout=timeout, httpx_client_factory=client)
|
||||
)
|
||||
else:
|
||||
http = await stack.enter_async_context(
|
||||
client(timeout=httpx.Timeout(timeout))
|
||||
)
|
||||
read, write, _ = await stack.enter_async_context(
|
||||
streamable_http_client(url, http_client=http)
|
||||
)
|
||||
session = await stack.enter_async_context(
|
||||
ClientSession(read, write, read_timeout_seconds=timedelta(seconds=timeout))
|
||||
)
|
||||
await session.initialize()
|
||||
yield session
|
||||
|
||||
|
||||
def _cause(error: BaseException) -> str:
|
||||
"""The one error inside the task groups the MCP client transports nest."""
|
||||
while isinstance(error, BaseExceptionGroup) and len(error.exceptions) == 1:
|
||||
error = error.exceptions[0]
|
||||
return f"{type(error).__name__}: {error}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class MCPServerConfig:
|
||||
"""Configuration for an external MCP server."""
|
||||
"""Configuration for an external MCP server.
|
||||
|
||||
``command`` spawns a stdio server; ``url`` reaches a running one over HTTP.
|
||||
"""
|
||||
|
||||
name: str
|
||||
command: List[str]
|
||||
command: List[str] = field(default_factory=list)
|
||||
url: Optional[str] = None
|
||||
env: Dict[str, str] = field(default_factory=dict)
|
||||
working_dir: Optional[str] = None
|
||||
description: str = ""
|
||||
@@ -65,9 +128,13 @@ class MCPServerConnection:
|
||||
self._request_id = 0
|
||||
self._pending_requests: Dict[int, asyncio.Future] = {}
|
||||
self._read_task: Optional[asyncio.Task] = None
|
||||
self.error: Optional[str] = None
|
||||
|
||||
async def connect(self) -> bool:
|
||||
"""Connect to the MCP server."""
|
||||
if self.config.url:
|
||||
return await self._connect_http()
|
||||
|
||||
if self.process is not None:
|
||||
return True
|
||||
|
||||
@@ -122,10 +189,34 @@ class MCPServerConnection:
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.error = str(e)
|
||||
logger.error(f"Failed to connect to MCP server '{self.config.name}': {e}")
|
||||
await self.disconnect()
|
||||
return False
|
||||
|
||||
async def _connect_http(self) -> bool:
|
||||
"""List an HTTP server's tools. HTTP holds no connection open, so each
|
||||
request opens its own session."""
|
||||
try:
|
||||
async with http_session(self.config.url) as session:
|
||||
listed = await session.list_tools()
|
||||
except Exception as e:
|
||||
self.error = _cause(e)
|
||||
logger.error(
|
||||
f"Failed to connect to MCP server '{self.config.name}': {self.error}"
|
||||
)
|
||||
return False
|
||||
self.tools = [
|
||||
ProxiedTool(
|
||||
name=tool.name,
|
||||
description=tool.description or "",
|
||||
input_schema=tool.inputSchema,
|
||||
server_name=self.config.name,
|
||||
)
|
||||
for tool in listed.tools
|
||||
]
|
||||
return True
|
||||
|
||||
async def disconnect(self):
|
||||
"""Disconnect from the MCP server."""
|
||||
if self._read_task:
|
||||
@@ -261,7 +352,12 @@ class MCPServerConnection:
|
||||
logger.warning(f"Failed to discover tools from '{self.config.name}': {e}")
|
||||
|
||||
async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
|
||||
"""Call a tool on the MCP server."""
|
||||
"""Call a tool on the MCP server; the result is the JSON-RPC result either way."""
|
||||
if self.config.url:
|
||||
async with http_session(self.config.url) as session:
|
||||
result = await session.call_tool(tool_name, arguments)
|
||||
return result.model_dump(mode="json", by_alias=True, exclude_none=True)
|
||||
|
||||
result = await self._send_request(
|
||||
"tools/call",
|
||||
{
|
||||
|
||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "hanzo-tools-mcp"
|
||||
version = "0.3.0"
|
||||
version = "0.3.1"
|
||||
description = "MCP management tools for Hanzo AI - server management, dynamic proxy for external MCPs"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
"""MCPServerConnection over HTTP, against a real MCP server on loopback.
|
||||
|
||||
The server runs in its own process: the SDK's streamable HTTP server leaves
|
||||
memory streams unclosed, and in-process that would fail this suite's
|
||||
warnings-as-errors on the server's account rather than the client's.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
import socket
|
||||
import subprocess
|
||||
from contextlib import contextmanager
|
||||
|
||||
import pytest
|
||||
from hanzo_tools.mcp_tools.mcp_proxy import MCPServerConfig, MCPServerConnection
|
||||
|
||||
SERVER = """
|
||||
import sys, socket, uvicorn
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP("echo", stateless_http=True)
|
||||
|
||||
@mcp.tool()
|
||||
def echo(text: str) -> str:
|
||||
\"\"\"Say it back.\"\"\"
|
||||
return text
|
||||
|
||||
app = mcp.sse_app() if sys.argv[1] == "sse" else mcp.streamable_http_app()
|
||||
sock = socket.socket()
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
print(sock.getsockname()[1], flush=True)
|
||||
uvicorn.Server(uvicorn.Config(app, log_level="error")).run(sockets=[sock])
|
||||
"""
|
||||
|
||||
|
||||
@contextmanager
|
||||
def serving(transport: str):
|
||||
"""An echo MCP server on a free loopback port, for the length of the block."""
|
||||
proc = subprocess.Popen([sys.executable, "-c", SERVER, transport], stdout=subprocess.PIPE, text=True)
|
||||
try:
|
||||
port = int(proc.stdout.readline())
|
||||
deadline = time.monotonic() + 10
|
||||
while True:
|
||||
try:
|
||||
socket.create_connection(("127.0.0.1", port), timeout=0.2).close()
|
||||
break
|
||||
except OSError:
|
||||
if time.monotonic() > deadline:
|
||||
raise
|
||||
time.sleep(0.05)
|
||||
yield port
|
||||
finally:
|
||||
proc.terminate()
|
||||
proc.wait()
|
||||
proc.stdout.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("transport,path", [("streamable", "/mcp"), ("sse", "/sse")])
|
||||
async def test_lists_and_calls_over_http(transport, path):
|
||||
with serving(transport) as port:
|
||||
conn = MCPServerConnection(MCPServerConfig(name="echo", url=f"http://localhost:{port}{path}"))
|
||||
assert await conn.connect(), conn.error
|
||||
assert [t.name for t in conn.tools] == ["echo"]
|
||||
assert conn.tools[0].input_schema["required"] == ["text"]
|
||||
|
||||
result = await conn.call_tool("echo", {"text": "hi"})
|
||||
assert result["content"] == [{"type": "text", "text": "hi"}]
|
||||
assert result["isError"] is False
|
||||
|
||||
|
||||
async def test_unreachable_url_reports_why():
|
||||
sock = socket.socket()
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
port = sock.getsockname()[1]
|
||||
sock.close()
|
||||
conn = MCPServerConnection(MCPServerConfig(name="gone", url=f"http://localhost:{port}/mcp"))
|
||||
assert not await conn.connect()
|
||||
assert conn.error.startswith("ConnectError")
|
||||
Reference in New Issue
Block a user