File size: 2,595 Bytes
a2ec7b6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | """
Minimal MCP HTTP Server Implementation
=======================================
Provides HTTP-based MCP server communication for services like GitHub.
"""
import asyncio
from contextlib import AsyncExitStack
from typing import Any, Dict, List, Optional
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
class MCPHttpServer:
"""
HTTP-based MCP client using the official MCP Python SDK
(Streamable HTTP transport).
"""
def __init__(
self,
url: str,
headers: Optional[Dict[str, str]] = None,
timeout: int = 30,
):
self.url = url.rstrip("/")
self.headers = headers or {}
self.timeout = timeout
self._stack: Optional[AsyncExitStack] = None
self.session: Optional[ClientSession] = None
self._tools_cache: Optional[List[Dict[str, Any]]] = None
async def __aenter__(self):
await self.start()
return self
async def __aexit__(self, exc_type, exc, tb):
await self.stop()
async def start(self):
"""Open Streamable HTTP transport and initialize MCP session."""
self._stack = AsyncExitStack()
read_stream, write_stream, _ = await self._stack.enter_async_context(
streamablehttp_client(self.url, headers=self.headers)
)
self.session = await self._stack.enter_async_context(ClientSession(read_stream, write_stream))
await asyncio.wait_for(self.session.initialize(), timeout=self.timeout)
async def stop(self):
"""Close the session/transport cleanly."""
if self._stack:
await self._stack.aclose()
self._stack = None
self.session = None
self._tools_cache = None
async def list_tools(self) -> List[Dict[str, Any]]:
"""Return tool definitions (cached)."""
if self._tools_cache is not None:
return self._tools_cache
if not self.session:
raise RuntimeError("MCP HTTP client not started")
resp = await asyncio.wait_for(self.session.list_tools(), timeout=self.timeout)
self._tools_cache = [t.model_dump() for t in resp.tools]
return self._tools_cache
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
"""Invoke a remote tool and return the structured result."""
if not self.session:
raise RuntimeError("MCP HTTP client not started")
result = await asyncio.wait_for(self.session.call_tool(name, arguments), timeout=self.timeout)
return result.model_dump()
|