Mount `server` on a Starlette app exposing the SSE transport at /sse and /messages/.
(server: Server)
| 83 | |
| 84 | |
| 85 | def make_app(server: Server) -> Starlette: |
| 86 | """Mount `server` on a Starlette app exposing the SSE transport at /sse and /messages/.""" |
| 87 | # DNS-rebinding protection validates Host/Origin headers against a network attack that cannot |
| 88 | # exist for an in-process app; the transport security behaviour itself is pinned by |
| 89 | # tests/server/test_sse_security.py. |
| 90 | sse = SseServerTransport( |
| 91 | "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False) |
| 92 | ) |
| 93 | |
| 94 | async def handle_sse(request: Request) -> Response: |
| 95 | async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): |
| 96 | await server.run(read_stream, write_stream, server.create_initialization_options()) |
| 97 | return Response() |
| 98 | |
| 99 | return Starlette( |
| 100 | routes=[ |
| 101 | Route("/sse", endpoint=handle_sse), |
| 102 | Mount("/messages/", app=sse.handle_post_message), |
| 103 | ] |
| 104 | ) |
| 105 | |
| 106 | |
| 107 | def make_server_app() -> Starlette: |
no test coverage detected