Add library for serving an sse server proxying a stdio server

pyproject.toml

@@ -26,7 +26,10 @@ classifiers = [
 ]
 version = "0.2.2"
 requires-python = ">=3.11"
-dependencies = ["mcp"]
+dependencies = [
+    "mcp",
+    "uvicorn>=0.34.0",
+]
 
 [build-system]
 requires = ["setuptools"]

src/mcp_proxy/sse_server.py

@@ -0,0 +1,77 @@
+"""Create a local SSE server that proxies requests to a stdio MCP server."""
+
+from dataclasses import dataclass
+from typing import Literal
+
+import uvicorn
+from mcp.client.session import ClientSession
+from mcp.client.stdio import StdioServerParameters, stdio_client
+from mcp.server import Server
+from mcp.server.sse import SseServerTransport
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.routing import Mount, Route
+
+from .proxy_server import create_proxy_server
+
+
+@dataclass
+class SseServerSettings:
+    """Settings for the server."""
+
+    bind_host: str = "127.0.0.1"
+    port: int = 8000
+    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
+
+
+def create_starlette_app(mcp_server: Server, debug: bool | None = None) -> Starlette:
+    """Create a Starlette application that can server the provied mcp server with SSE."""
+    sse = SseServerTransport("/messages/")
+
+    async def handle_sse(request: Request) -> None:
+        async with sse.connect_sse(
+            request.scope,
+            request.receive,
+            request._send,  # noqa: SLF001
+        ) as (read_stream, write_stream):
+            await mcp_server.run(
+                read_stream,
+                write_stream,
+                mcp_server.create_initialization_options(),
+            )
+
+    return Starlette(
+        debug=debug,
+        routes=[
+            Route("/sse", endpoint=handle_sse),
+            Mount("/messages/", app=sse.handle_post_message),
+        ],
+    )
+
+
+async def run_sse_server(
+    stdio_params: StdioServerParameters,
+    sse_settings: SseServerSettings,
+) -> None:
+    """Run the stdio client and expose an SSE server.
+
+    Args:
+        stdio_params: The parameters for the stdio client that spawns a stdio server.
+        sse_settings: The settings for the SSE server that accepts incoming requests.
+
+    """
+    async with stdio_client(stdio_params) as streams, ClientSession(*streams) as session:
+        mcp_server = await create_proxy_server(session)
+
+        # Bind SSE request handling to MCP server
+        starlette_app = await create_starlette_app(mcp_server, sse_settings.log_level == "DEBUG")
+
+        # Configure HTTP server
+        config = uvicorn.Config(
+            starlette_app,
+            host=sse_settings.bind_host,
+            port=sse_settings.port,
+            log_level=sse_settings.log_level.lower(),
+        )
+        http_server = uvicorn.Server(config)
+        await http_server.serve()

tests/test_sse_server.py

@@ -0,0 +1,60 @@
+"""Tests for the sse server."""
+
+import asyncio
+import contextlib
+
+import uvicorn
+from mcp import types
+from mcp.client.session import ClientSession
+from mcp.client.sse import sse_client
+from mcp.server import Server
+
+from mcp_proxy.sse_server import create_starlette_app
+
+
+class BackgroundServer(uvicorn.Server):
+    """A test server that runs in a background thread."""
+
+    def install_signal_handlers(self) -> None:
+        """Do not install signal handlers."""
+
+    @contextlib.asynccontextmanager
+    async def run_in_background(self) -> None:
+        """Run the server in a background thread."""
+        task = asyncio.create_task(self.serve())
+        try:
+            while not self.started:  # noqa: ASYNC110
+                await asyncio.sleep(1e-3)
+            yield
+        finally:
+            task.cancel()
+            self.shutdown()
+
+    @property
+    def url(self) -> str:
+        """Return the url of the started server."""
+        hostport = next(
+            iter([socket.getsockname() for server in self.servers for socket in server.sockets]),
+        )
+        return f"http://{hostport[0]}:{hostport[1]}"
+
+
+async def test_create_starlette_app() -> None:
+    """Test basic glue code for the SSE transport and a fake MCP server."""
+    server = Server("prompt-server")
+
+    @server.list_prompts()
+    async def list_prompts() -> list[types.Prompt]:
+        return [types.Prompt(name="prompt1")]
+
+    app = create_starlette_app(server)
+
+    config = uvicorn.Config(app, port=0, log_level="info")
+    server = BackgroundServer(config)
+    async with server.run_in_background():
+        mcp_url = f"{server.url}/sse"
+        async with sse_client(url=mcp_url) as streams, ClientSession(*streams) as session:
+            await session.initialize()
+            response = await session.list_prompts()
+            assert len(response.prompts) == 1
+            assert response.prompts[0].name == "prompt1"