(mcp_server: Server, *, debug: bool = False)
| 58 | |
| 59 | |
| 60 | def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette: |
| 61 | sse = SseServerTransport("/messages/") |
| 62 | |
| 63 | async def handle_sse(request: Request) -> None: |
| 64 | async with sse.connect_sse(request.scope, request.receive, request._send) as ( |
| 65 | read_stream, |
| 66 | write_stream, |
| 67 | ): |
| 68 | await mcp_server.run( |
| 69 | read_stream, write_stream, mcp_server.create_initialization_options() |
| 70 | ) |
| 71 | |
| 72 | return Starlette( |
| 73 | debug=debug, |
| 74 | routes=[ |
| 75 | Route("/sse", endpoint=handle_sse), |
| 76 | Mount("/messages/", app=sse.handle_post_message), |
| 77 | ], |
| 78 | ) |
| 79 | |
| 80 | |
| 81 | if __name__ == "__main__": |
no outgoing calls
no test coverage detected