Endpoint for receiving messages from the client. Uses a buffer framing collector to ensure the full JSON-RPC payload is received before processing. This prevents crashes caused by large responses being split across SSE line boundaries.
(request: Request)
| 107 | |
| 108 | |
| 109 | async def handle_messages(request: Request): |
| 110 | """Endpoint for receiving messages from the client. |
| 111 | |
| 112 | Uses a buffer framing collector to ensure the full JSON-RPC payload |
| 113 | is received before processing. This prevents crashes caused by large |
| 114 | responses being split across SSE line boundaries. |
| 115 | """ |
| 116 | # Buffer the complete request body before any parsing occurs. |
| 117 | # request.body() accumulates all chunks, preventing partial JSON reads. |
| 118 | raw_body = await request.body() |
| 119 | |
| 120 | try: |
| 121 | json.loads(raw_body) |
| 122 | except json.JSONDecodeError as e: |
| 123 | return Response( |
| 124 | content=json.dumps({ |
| 125 | "jsonrpc": "2.0", |
| 126 | "error": { |
| 127 | "code": -32700, |
| 128 | "message": f"Parse error: incomplete or malformed JSON-RPC message: {e}" |
| 129 | }, |
| 130 | "id": None |
| 131 | }), |
| 132 | status_code=400, |
| 133 | media_type="application/json" |
| 134 | ) |
| 135 | |
| 136 | # `handle_post_message` builds its own Request and awaits `.body()` again. |
| 137 | # Starlette caches a body on the Request instance, not in the scope, so |
| 138 | # handing it the raw `receive` would make it wait for a body that has |
| 139 | # already been drained above — the POST hangs until the client gives up and |
| 140 | # the message never reaches the session. Replay the buffered bytes instead. |
| 141 | async def replay_receive() -> dict: |
| 142 | return {"type": "http.request", "body": raw_body, "more_body": False} |
| 143 | |
| 144 | sender = _SendTracker(request._send) |
| 145 | try: |
| 146 | await sse.handle_post_message(request.scope, replay_receive, sender) |
| 147 | except Exception as exc: |
| 148 | logger.debug("Message handler closed: %s", type(exc).__name__) |
| 149 | |
| 150 | if sender.response_started: |
| 151 | return _AlreadySentResponse() |
| 152 | return Response(status_code=202, content="Accepted") |