Skip to content

Commit 85d6b97

Browse files
author
Kush Zingade
committed
fix(client): surface non-2xx POST failures to the awaiting caller (#2110)
The SSE transport's post_writer caught every exception from the message POST, logged it, and returned. `raise_for_status()` on a 401/403/5xx was therefore swallowed: the request it carried never received a response and its caller waited forever. Fail that request instead, correlated by its own id, so only the caller whose POST was rejected is affected -- a session-wide fan-out would tear down unrelated in-flight requests, and raw `Exception` items on the read stream are also used for non-fatal per-message parse errors. Both transports now also carry the HTTP status on `ErrorData.data`. The JSON-RPC code cannot express it: every non-404 failure maps to INTERNAL_ERROR, which alone cannot tell a caller whether to re-authenticate (401/403) or retry (5xx). Error messages are unchanged. Signed-off-by: Kush Zingade <kushzingade@honorsocietyofcinematicarts.org>
1 parent 56af447 commit 85d6b97

4 files changed

Lines changed: 108 additions & 10 deletions

File tree

src/mcp/client/sse.py

Lines changed: 37 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -115,21 +115,49 @@ async def sse_reader(task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED):
115115
finally:
116116
await read_stream_writer.aclose()
117117

118+
async def _fail_request(message: types.JSONRPCMessage, exc: httpx2.HTTPError) -> None:
119+
"""Resolve the request this failed POST carried, so its caller raises instead of hanging.
120+
121+
Correlated by request id rather than fanned out across the session:
122+
one rejected POST must not tear down unrelated in-flight requests.
123+
A notification has no waiter, so there is nothing to resolve.
124+
"""
125+
if not isinstance(message, types.JSONRPCRequest):
126+
return
127+
status = exc.response.status_code if isinstance(exc, httpx2.HTTPStatusError) else None
128+
error = types.ErrorData(
129+
code=types.INTERNAL_ERROR,
130+
message=f"Sending the request failed: {exc}",
131+
# The HTTP status is the whole diagnosis for 401/403 vs 5xx, and
132+
# the JSON-RPC code cannot carry it.
133+
data={"http_status": status} if status is not None else None,
134+
)
135+
reply = SessionMessage(types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error))
136+
try:
137+
await read_stream_writer.send(reply)
138+
except (anyio.BrokenResourceError, anyio.ClosedResourceError): # pragma: no cover
139+
logger.debug("read stream closed before request %r could be failed", message.id)
140+
118141
async def post_writer(endpoint_url: str):
119142
try:
120143
async with write_stream_reader, write_stream:
121144

122145
async def _send_message(session_message: SessionMessage) -> None:
123146
logger.debug(f"Sending client message: {session_message}")
124-
response = await client.post(
125-
endpoint_url,
126-
json=session_message.message.model_dump(
127-
by_alias=True,
128-
mode="json",
129-
exclude_unset=True,
130-
),
131-
)
132-
response.raise_for_status()
147+
try:
148+
response = await client.post(
149+
endpoint_url,
150+
json=session_message.message.model_dump(
151+
by_alias=True,
152+
mode="json",
153+
exclude_unset=True,
154+
),
155+
)
156+
response.raise_for_status()
157+
except httpx2.HTTPError as exc:
158+
logger.exception("Error sending client message")
159+
await _fail_request(session_message.message, exc)
160+
return
133161
logger.debug(f"Client message sent successfully: {response.status_code}")
134162

135163
async for session_message in write_stream_reader:

src/mcp/client/streamable_http.py

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -367,7 +367,13 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
367367
else:
368368
error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
369369
else:
370-
error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
370+
# The status rides on `data`: INTERNAL_ERROR alone can't tell a
371+
# caller whether to re-authenticate (401/403) or retry (5xx).
372+
error_data = ErrorData(
373+
code=INTERNAL_ERROR,
374+
message="Server returned an error response",
375+
data={"http_status": response.status_code},
376+
)
371377
session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
372378
await ctx.read_stream_writer.send(session_message)
373379
return

tests/client/test_streamable_http.py

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -748,3 +748,28 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain
748748
_abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
749749
)
750750
send.close()
751+
752+
753+
@pytest.mark.anyio
754+
@pytest.mark.parametrize("status_code", [401, 403, 500, 502])
755+
async def test_non_2xx_post_preserves_http_status(status_code: int) -> None:
756+
"""The HTTP status survives on `ErrorData.data` so 401/403 stay distinguishable from 5xx.
757+
758+
The JSON-RPC code cannot carry it: every non-404 failure maps to INTERNAL_ERROR,
759+
which on its own says nothing about whether the caller should re-authenticate.
760+
"""
761+
762+
def handler(request: httpx2.Request) -> httpx2.Response:
763+
return httpx2.Response(status_code)
764+
765+
with anyio.fail_after(5):
766+
async with (
767+
httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http,
768+
streamable_http_client("http://test/mcp", http_client=http) as (read, write),
769+
):
770+
await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})))
771+
reply = await read.receive()
772+
773+
assert isinstance(reply, SessionMessage)
774+
assert isinstance(reply.message, JSONRPCError)
775+
assert reply.message.error.data == {"http_status": status_code}

tests/shared/test_sse.py

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -480,3 +480,42 @@ async def test_sse_session_cleanup_on_disconnect() -> None:
480480
headers={"Content-Type": "application/json"},
481481
)
482482
assert response.status_code == 404
483+
484+
485+
def make_post_rejecting_app(status_code: int) -> Starlette:
486+
"""SSE transport whose message POST endpoint always fails with `status_code`."""
487+
sse = SseServerTransport(
488+
"/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False)
489+
)
490+
server = Server(SERVER_NAME, on_read_resource=_handle_read_resource)
491+
492+
async def handle_sse(request: Request) -> Response:
493+
async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
494+
await server.run(read_stream, write_stream, server.create_initialization_options())
495+
return Response()
496+
497+
async def reject_post(request: Request) -> Response:
498+
return Response(status_code=status_code)
499+
500+
return Starlette(
501+
routes=[
502+
Route("/sse", endpoint=handle_sse),
503+
Route("/messages/", endpoint=reject_post, methods=["POST"]),
504+
]
505+
)
506+
507+
508+
@pytest.mark.anyio
509+
@pytest.mark.parametrize("status_code", [401, 403, 500])
510+
async def test_sse_post_error_reaches_caller(status_code: int) -> None:
511+
"""A non-2xx POST surfaces to the awaiting caller instead of hanging it forever."""
512+
factory = in_process_client_factory(make_post_rejecting_app(status_code))
513+
514+
with anyio.fail_after(5):
515+
async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams:
516+
async with ClientSession(*streams) as session:
517+
with pytest.raises(MCPError) as exc_info:
518+
await session.initialize()
519+
520+
# The HTTP status survives on `data` — 401/403 must stay distinguishable from 5xx.
521+
assert exc_info.value.error.data == {"http_status": status_code}

0 commit comments

Comments
 (0)