Skip to content

fix(client): contain HTTP transport errors on request POSTs #3525

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
macrogui wants to merge 1 commit into from
+124 −71
Closed

fix(client): contain HTTP transport errors on request POSTs #3525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
159 changes: 88 additions & 71 deletions src/mcp/client/streamable_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,83 +352,100 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
if ctx.metadata is not None and ctx.metadata.headers is not None:
headers.update(ctx.metadata.headers)

async with stream_within_origin(
ctx.client,
"POST",
self.url,
json=message.model_dump(by_alias=True, mode="json", exclude_unset=True),
headers=headers,
) as response:
if response.status_code == 202:
logger.debug("Received 202 Accepted")
if isinstance(message, JSONRPCRequest):
# A request's response arrives on this POST's body; 202 says
# none will follow. Resolve rather than park the caller forever.
await self._resolve_abandoned_request(
ctx.read_stream_writer,
message.id,
"server answered a request with 202 Accepted",
code=INVALID_REQUEST,
)
return
try:
async with stream_within_origin(
ctx.client,
"POST",
self.url,
json=message.model_dump(by_alias=True, mode="json", exclude_unset=True),
headers=headers,
) as response:
if response.status_code == 202:
logger.debug("Received 202 Accepted")
if isinstance(message, JSONRPCRequest):
# A request's response arrives on this POST's body; 202 says
# none will follow. Resolve rather than park the caller forever.
await self._resolve_abandoned_request(
ctx.read_stream_writer,
message.id,
"server answered a request with 202 Accepted",
code=INVALID_REQUEST,
)
return

if (redirect := _unfollowed_redirect(response)) is not None:
logger.warning(redirect)
if isinstance(message, JSONRPCRequest):
await self._resolve_abandoned_request(
ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST
)
return
if (redirect := _unfollowed_redirect(response)) is not None:
logger.warning(redirect)
if isinstance(message, JSONRPCRequest):
await self._resolve_abandoned_request(
ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST
)
return

if response.status_code >= 400:
if isinstance(message, JSONRPCRequest):
# A spec-correct server may return the JSON-RPC error in the
# body at a non-2xx status (e.g. 400 for INVALID_PARAMS, 404
# for METHOD_NOT_FOUND). Surface that error rather than the
# status-derived stand-in below.
if response.headers.get("content-type", "").lower().startswith("application/json"):
try:
body = await response.aread()
parsed = jsonrpc_message_adapter.validate_json(body, by_name=False)
if isinstance(parsed, JSONRPCError):
# The server may have set `id: null` (request rejected before its
# id was parsed); use this request's id so correlation works.
reply = JSONRPCError(jsonrpc="2.0", id=message.id, error=parsed.error)
await ctx.read_stream_writer.send(SessionMessage(reply))
return
except (httpx2.StreamError, ValidationError):
pass
logger.debug("Non-2xx body was not a JSON-RPC error; using fallback")
if response.status_code == 404:
if self.session_id is None:
# No session yet → 404 is the HTTP-level spelling of
# METHOD_NOT_FOUND (gateway / legacy server doesn't know
# this method); "Session terminated" would be a lie here.
error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found")
if response.status_code >= 400:
if isinstance(message, JSONRPCRequest):
# A spec-correct server may return the JSON-RPC error in the
# body at a non-2xx status (e.g. 400 for INVALID_PARAMS, 404
# for METHOD_NOT_FOUND). Surface that error rather than the
# status-derived stand-in below.
if response.headers.get("content-type", "").lower().startswith("application/json"):
try:
body = await response.aread()
parsed = jsonrpc_message_adapter.validate_json(body, by_name=False)
if isinstance(parsed, JSONRPCError):
# The server may have set `id: null` (request rejected before its
# id was parsed); use this request's id so correlation works.
reply = JSONRPCError(jsonrpc="2.0", id=message.id, error=parsed.error)
await ctx.read_stream_writer.send(SessionMessage(reply))
return
except (httpx2.StreamError, ValidationError):
pass
logger.debug("Non-2xx body was not a JSON-RPC error; using fallback")
if response.status_code == 404:
if self.session_id is None:
# No session yet → 404 is the HTTP-level spelling of
# METHOD_NOT_FOUND (gateway / legacy server doesn't know
# this method); "Session terminated" would be a lie here.
error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found")
else:
error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
else:
error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
else:
error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
await ctx.read_stream_writer.send(session_message)
return
error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
await ctx.read_stream_writer.send(session_message)
return

if self._is_initialization_request(message):
self._maybe_extract_session_id_from_response(response)
if self._is_initialization_request(message):
self._maybe_extract_session_id_from_response(response)

# Per https://modelcontextprotocol.io/specification/2025-06-18/basic#notifications:
# The server MUST NOT send a response to notifications.
# Per https://modelcontextprotocol.io/specification/2025-06-18/basic#notifications:
# The server MUST NOT send a response to notifications.
if isinstance(message, JSONRPCRequest):
content_type = response.headers.get("content-type", "").lower()
if content_type.startswith("application/json"):
await self._handle_json_response(response, ctx.read_stream_writer, request_id=message.id)
elif content_type.startswith("text/event-stream"):
await self._handle_sse_response(response, ctx)
else:
logger.error(f"Unexpected content type: {content_type}")
error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}")
error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
await ctx.read_stream_writer.send(error_msg)

except httpx2.TransportError as exc:
# The HTTP exchange itself failed (connection dropped before a
# response, connect/read errors, timeouts). This must fail only
# the request that was in flight: resolve its waiter with a
# synthesized error so the session and its transport survive,
# mirroring how non-resumable SSE drops are handled above.
if isinstance(message, JSONRPCRequest):
content_type = response.headers.get("content-type", "").lower()
if content_type.startswith("application/json"):
await self._handle_json_response(response, ctx.read_stream_writer, request_id=message.id)
elif content_type.startswith("text/event-stream"):
await self._handle_sse_response(response, ctx)
else:
logger.error(f"Unexpected content type: {content_type}")
error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}")
error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
await ctx.read_stream_writer.send(error_msg)
logger.warning(f"HTTP transport error on POST for request {message.id}: {exc!r}")
await self._resolve_abandoned_request(
ctx.read_stream_writer,
message.id,
f"HTTP transport error: {exc}",
)
elif isinstance(message, JSONRPCNotification):
logger.warning(f"HTTP transport error on POST for {message.method}: {exc!r}")

async def _handle_json_response(
self,
Expand Down
36 changes: 36 additions & 0 deletions tests/client/test_streamable_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -968,3 +968,39 @@ async def test_https_endpoint_redirected_to_plain_http_elsewhere_never_suggests_
The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust,
often combined with a trailing-slash difference. Try https://backend.lan:8000/mcp/ instead, or fix the proxy settings.\
""")


@pytest.mark.anyio
async def test_a_post_transport_error_fails_only_that_request_and_keeps_the_session() -> None:
"""A POST whose HTTP exchange itself fails (server drops the connection before
responding) must resolve only that request with an error; the session and its
transport stay usable, matching the TypeScript client's recovery behavior."""

def handler(request: httpx2.Request) -> httpx2.Response:
body = json.loads(request.content)
if body.get("id") == "drop-1":
raise httpx2.RemoteProtocolError("Server disconnected without sending a response.")
return httpx2.Response(
200,
headers={"content-type": "application/json"},
json={"jsonrpc": "2.0", "id": body["id"], "result": {}},
)

with anyio.fail_after(5):
async with (
httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http,
streamable_http_client("http://test/mcp", http_client=http) as (read, write),
):
await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="drop-1", method="tools/call", params={})))
await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="after-1", method="ping")))
first = await read.receive()
second = await read.receive()

assert isinstance(first, SessionMessage)
assert isinstance(first.message, JSONRPCError)
assert first.message.id == "drop-1"
assert first.message.error.code == CONNECTION_CLOSED

assert isinstance(second, SessionMessage)
assert isinstance(second.message, JSONRPCResponse)
assert second.message.id == "after-1"
Loading