Function build_sse_app

tests/interaction/_connect.py:323–346

Mount a server on a Starlette app exposing the legacy SSE transport at /sse and /messages/. `MCPServer.sse_app()` exists but does not expose the underlying `SseServerTransport`, which the SSE-specific tests need; building the app explicitly here gives both server flavours the same routi

(server: Server | MCPServer)

def build_sse_app(server: Server | MCPServer) -> tuple[Starlette, SseServerTransport]:
    """Mount a server on a Starlette app exposing the legacy SSE transport at /sse and /messages/.

    `MCPServer.sse_app()` exists but does not expose the underlying `SseServerTransport`, which
    the SSE-specific tests need; building the app explicitly here gives both server flavours the
    same routing while keeping that handle.
    """
    sse = SseServerTransport(
        "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False)
    )
    lowlevel = server._lowlevel_server if isinstance(server, MCPServer) else server

    async def handle_sse(request: Request) -> Response:
        async with sse.connect_sse(request.scope, request.receive, request._send) as (read, write):
            await lowlevel.run(read, write, lowlevel.create_initialization_options())
        return Response()

    app = Starlette(
        routes=[
            Route("/sse", endpoint=handle_sse, methods=["GET"]),
            Mount("/messages/", app=sse.handle_post_message),
        ],
    )
    return app, sse


@asynccontextmanager