MCP Module¶

def route_table_to_mcp_server(
    route_table: "RouteTable",
    name: str = "ToolRegistry-Server",
) -> "Server":
    """Create an MCP low-level Server from a RouteTable.

    Registers list_tools and call_tool handlers that read directly
    from the route table, ensuring enable/disable state is always
    in sync (no drift).

    Args:
        route_table: The RouteTable instance to expose as MCP tools.
        name: Server name for MCP identification.

    Returns:
        A configured mcp.server.lowlevel.Server instance.

    Raises:
        ImportError: If MCP SDK is not installed.
    """
    try:
        from mcp.server.lowlevel import Server
        from mcp.shared.exceptions import McpError
        from mcp.types import INTERNAL_ERROR, ErrorData, TextContent
        from mcp.types import Tool as MCPTool
    except ImportError as e:
        raise ImportError(
            "MCP SDK is required for MCP support. "
            "Install with: pip install toolregistry-server[mcp]"
        ) from e

    server = Server(name)
    session_mgr = SessionManager()

    @server.list_tools()
    async def handle_list_tools() -> list[MCPTool]:
        """Return MCP tool definitions for non-deferred enabled tools.

        Deferred tools are excluded from the initial listing so that LLMs
        discover them via discover_tools.
        """
        tools: list[MCPTool] = []
        for route in route_table.list_routes(enabled_only=True, include_deferred=False):
            tools.append(
                MCPTool(
                    name=route.tool_name,
                    description=route.description or "",
                    inputSchema=normalize_parameters_schema(route.parameters_schema),
                )
            )
        logger.debug(f"list_tools: returning {len(tools)} enabled tools")
        return tools

    @server.call_tool(validate_input=False)
    async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
        """Execute a tool by name with the given arguments.

        Args:
            name: The tool name to invoke.
            arguments: The input arguments for the tool.

        Returns:
            A list containing a single TextContent with the result.

        Raises:
            McpError: If the tool is disabled or not found.
        """
        # Get the route entry
        route = route_table.get_route(name)

        # Check if tool exists
        if route is None:
            raise McpError(
                ErrorData(
                    code=INTERNAL_ERROR,
                    message=f"Tool '{name}' not found",
                )
            )

        # Check if tool is disabled
        if not route.enabled:
            reason = route.disable_reason or "unknown reason"
            raise McpError(
                ErrorData(
                    code=INTERNAL_ERROR,
                    message=f"Tool '{name}' is disabled: {reason}",
                )
            )

        # --- Session context ---
        session_ctx = _get_session_context(session_mgr)
        token = None
        if session_ctx is not None:
            token = session_context_var.set(session_ctx)

        try:
            result = await _execute_tool(route, arguments, session_ctx, session_mgr)
            text = _serialize_result(result)
            logger.debug(f"call_tool '{name}': success")
            return [TextContent(type="text", text=text)]

        except McpError:
            raise
        except Exception as e:
            logger.warning(f"call_tool '{name}': error - {e}")
            raise McpError(
                ErrorData(
                    code=INTERNAL_ERROR,
                    message=str(e),
                )
            ) from e
        finally:
            if token is not None:
                session_context_var.reset(token)

    logger.info(
        f"MCP server '{name}' created with {len(route_table.list_routes())} "
        f"enabled tool(s) out of {len(route_table.list_routes(enabled_only=False))} total"
    )
    return server

async def run_sse(
    server: "Server",
    host: str = "127.0.0.1",
    port: int = 8000,
    path: str = "/sse",
) -> None:
    """Run an MCP server over SSE (Server-Sent Events) transport.

    This transport is suitable for web-based MCP clients that connect
    via HTTP and receive events through SSE.

    Args:
        server: The MCP Server instance to run.
        host: Host address to bind to.
        port: Port number to bind to.
        path: URL path for the SSE endpoint.
    """
    try:
        import uvicorn
        from mcp.server.sse import SseServerTransport
        from starlette.applications import Starlette
        from starlette.requests import Request
        from starlette.responses import Response
        from starlette.routing import Mount, Route
    except ImportError as e:
        raise ImportError(
            "MCP SDK and Starlette are required for SSE transport. "
            "Install with: pip install toolregistry-server[mcp] starlette uvicorn"
        ) from e

    logger.info(f"Starting MCP server with SSE transport on {host}:{port}{path}")

    # Create SSE transport
    sse = SseServerTransport(f"{path}/messages/")

    # SSE endpoint handler - must accept Request and return Response
    # See MCP SDK documentation for the correct pattern
    async def handle_sse(request: Request) -> Response:
        async with sse.connect_sse(
            request.scope, request.receive, request._send
        ) as streams:
            await server.run(
                streams[0],
                streams[1],
                server.create_initialization_options(),
            )
        # Return empty response to avoid NoneType error when client disconnects
        return Response()

    # Create Starlette app
    routes = [
        Route(path, endpoint=handle_sse, methods=["GET"]),
        Mount(f"{path}/messages/", app=sse.handle_post_message),
    ]
    app = Starlette(routes=routes)

    # Run with uvicorn
    config = uvicorn.Config(app, host=host, port=port, log_level="info")
    uvicorn_server = uvicorn.Server(config)

    try:
        await uvicorn_server.serve()
    except KeyboardInterrupt:
        logger.info("MCP SSE server shutdown requested (KeyboardInterrupt)")
    except asyncio.CancelledError:
        logger.info("MCP SSE server shutdown requested (CancelledError)")

async def run_stdio(server: "Server") -> None:
    """Run an MCP server over stdio transport.

    This is the simplest transport, suitable for local tool execution
    where the MCP client spawns the server as a subprocess.

    Args:
        server: The MCP Server instance to run.
    """
    try:
        from mcp.server.stdio import stdio_server
    except ImportError as e:
        raise ImportError(
            "MCP SDK is required for MCP support. "
            "Install with: pip install toolregistry-server[mcp]"
        ) from e

    logger.info("Starting MCP server with stdio transport")
    try:
        async with stdio_server() as (read, write):
            await server.run(
                read,
                write,
                server.create_initialization_options(),
            )
    except KeyboardInterrupt:
        logger.info("MCP stdio server shutdown requested (KeyboardInterrupt)")
    except asyncio.CancelledError:
        logger.info("MCP stdio server shutdown requested (CancelledError)")

async def run_streamable_http(
    server: "Server",
    host: str = "127.0.0.1",
    port: int = 8000,
    path: str = "/mcp",
) -> None:
    """Run an MCP server over streamable HTTP transport.

    This is the recommended HTTP transport for production use,
    supporting bidirectional streaming over HTTP.

    Args:
        server: The MCP Server instance to run.
        host: Host address to bind to.
        port: Port number to bind to.
        path: URL path for the MCP endpoint.
    """
    try:
        import uvicorn
        from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
        from starlette.applications import Starlette
        from starlette.routing import Route
        from starlette.types import Receive, Scope, Send
    except ImportError as e:
        raise ImportError(
            "MCP SDK and Starlette are required for streamable HTTP transport. "
            "Install with: pip install toolregistry-server[mcp] starlette uvicorn"
        ) from e

    logger.info(
        f"Starting MCP server with streamable HTTP transport on {host}:{port}{path}"
    )

    # Create session manager
    session_manager = StreamableHTTPSessionManager(
        app=server,
        json_response=False,
        stateless=False,
    )

    # Create ASGI application class for StreamableHTTP
    # This is necessary because Starlette Route treats ASGI apps differently
    # from regular endpoint functions - ASGI apps receive all HTTP methods
    class StreamableHTTPASGIApp:
        """ASGI application wrapper for StreamableHTTP session manager."""

        def __init__(self, manager: StreamableHTTPSessionManager):
            self.manager = manager

        async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
            await self.manager.handle_request(scope, receive, send)

    streamable_http_app = StreamableHTTPASGIApp(session_manager)

    # Create Starlette app with lifespan
    routes = [Route(path, endpoint=streamable_http_app)]
    app = Starlette(
        routes=routes,
        lifespan=lambda app: session_manager.run(),
    )

    # Run with uvicorn
    config = uvicorn.Config(app, host=host, port=port, log_level="info")
    uvicorn_server = uvicorn.Server(config)

    try:
        await uvicorn_server.serve()
    except KeyboardInterrupt:
        logger.info("MCP streamable HTTP server shutdown requested (KeyboardInterrupt)")
    except asyncio.CancelledError:
        logger.info("MCP streamable HTTP server shutdown requested (CancelledError)")