Function handle_sse

tests/server/test_sse_security.py:48–57

(request: Request)

    sse_transport = SseServerTransport("/messages/", security_settings)

    async def handle_sse(request: Request) -> Response:
        try:
            async with sse_transport.connect_sse(request.scope, request.receive, request._send) as (read, write):
                await server.run(read, write, server.create_initialization_options())
        except ValueError as e:
            # Validation error was already handled inside connect_sse, which sent the rejection
            # response itself; its non-empty body checkpoints, so the test reads the rejection
            # status before the trailing Response() below sends a second response start.
            logger.debug(f"SSE connection failed validation: {e}")
        return Response()

    app = Starlette(
        routes=[