Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
7 changes: 7 additions & 0 deletions docs/client/transports.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,13 @@ Two things to notice:

Keep the `timeout=`. It is the one the SDK's own client uses (30 seconds, 300 for reads); an `httpx2.AsyncClient` built without one gets `httpx2`'s 5-second default, and a tool call that runs longer than that fails with a read timeout.

An HTTP read timeout while sending a request POST or reading its JSON response
raises `MCPError` with code `REQUEST_TIMEOUT` (-32001) for that request. Other
in-flight requests and subsequent calls on the same client remain usable. The
transport does not retry the request; a timeout does not prove that the server
stopped executing it. Timeouts during the initial `initialize` or `server/discover`
handshake still propagate as transport errors.

One TLS note: `httpx2` verifies certificates against the operating system trust store (via
[`truststore`](https://pypi.org/project/truststore/)), not a bundled CA list. In an environment with
no usable system CA store (some minimal containers), set the standard `SSL_CERT_FILE`/`SSL_CERT_DIR`
Expand Down
19 changes: 15 additions & 4 deletions src/mcp/client/streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
INVALID_REQUEST,
METHOD_NOT_FOUND,
PARSE_ERROR,
REQUEST_TIMEOUT,
ErrorData,
JSONRPCError,
JSONRPCMessage,
Expand Down Expand Up @@ -656,10 +657,20 @@ async def _handle_message(session_message: SessionMessage) -> None:
)

async def handle_request_async():
if is_resumption:
await self._handle_resumption_request(ctx)
else:
await self._handle_post_request(ctx)
try:
if is_resumption:
await self._handle_resumption_request(ctx)
else:
await self._handle_post_request(ctx)
except httpx2.ReadTimeout:
if not isinstance(message, JSONRPCRequest):
raise
if message.method in ("initialize", "server/discover"):
raise
# One HTTP response timing out must not cancel sibling requests.
await self._resolve_abandoned_request(
ctx.read_stream_writer, message.id, "HTTP response timed out", code=REQUEST_TIMEOUT
)

# If this is a request, start a new task to handle it
if isinstance(message, JSONRPCRequest):
Expand Down
150 changes: 150 additions & 0 deletions tests/client/test_streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
INVALID_REQUEST,
METHOD_NOT_FOUND,
PROTOCOL_VERSION_META_KEY,
REQUEST_TIMEOUT,
CallToolRequestParams,
CallToolResult,
JSONRPCError,
Expand Down Expand Up @@ -1279,3 +1280,152 @@ 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
@pytest.mark.parametrize("mode", ["legacy", LATEST_MODERN_VERSION])
@pytest.mark.parametrize("during_body", [False, True], ids=["response-headers", "json-body"])
async def test_http_read_timeout_fails_only_its_call_and_keeps_the_client_usable(mode: str, during_body: bool) -> None:
"""SDK-defined: a POST read timeout leaves sibling and subsequent calls usable.

The HTTP seam injects an error the in-process server cannot produce itself.
"""
sibling_started = anyio.Event()
release_sibling = anyio.Event()
body_closed = anyio.Event()
calls: list[str] = []

async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
assert params.name in ("sibling", "after")
return CallToolResult(content=[TextContent(text=params.name)])

async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(
tools=[Tool(name=name, input_schema={"type": "object"}) for name in ("slow", "sibling", "after")]
)

manager = StreamableHTTPSessionManager(
app=Server("timeout-isolation", on_call_tool=call_tool, on_list_tools=list_tools)
)
app = Starlette(routes=[Mount("/mcp", app=manager.handle_request)])
transport = StreamingASGITransport(app)

class TimeoutBody(httpx2.AsyncByteStream):
async def __aiter__(self) -> AsyncIterator[bytes]:
yield b'{"jsonrpc":'
raise httpx2.ReadTimeout("sensitive upstream detail")

async def aclose(self) -> None:
body_closed.set()

async def handle(request: httpx2.Request) -> httpx2.Response:
if request.method == "POST":
body = json.loads(request.content)
if body.get("method") == "tools/call":
name = body["params"]["name"]
calls.append(name)
if name == "slow":
await sibling_started.wait()
if during_body:
return httpx2.Response(200, headers={"content-type": "application/json"}, stream=TimeoutBody())
raise httpx2.ReadTimeout("sensitive upstream detail", request=request)
if name == "sibling":
sibling_started.set()
await release_sibling.wait()
return await transport.handle_async_request(request)

with anyio.fail_after(5):
async with (
manager.run(),
transport,
httpx2.AsyncClient(transport=httpx2.MockTransport(handle)) as http,
Client(streamable_http_client("http://localhost/mcp/", http_client=http), mode=mode) as client,
):

async def slow_call() -> None:
with pytest.raises(MCPError) as exc_info:
await client.call_tool("slow", {})
assert exc_info.value.error.code == REQUEST_TIMEOUT
assert exc_info.value.error.message == snapshot("HTTP response timed out")
release_sibling.set()

async with anyio.create_task_group() as group:
group.start_soon(slow_call)
sibling = await client.call_tool("sibling", {})
assert sibling.content == [TextContent(text="sibling")]
after = await client.call_tool("after", {})

assert after.content == [TextContent(text="after")]
assert sorted(calls) == ["after", "sibling", "slow"]
assert body_closed.is_set() == during_body


@pytest.mark.anyio
async def test_http_read_timeout_on_resumption_resolves_the_original_request() -> None:
"""SDK-defined: a failed resumption GET resolves the original request without a POST retry.

The raw transport API supplies the resumption token hidden by Client.
"""
seen: list[tuple[str, str | None]] = []

def handle(request: httpx2.Request) -> httpx2.Response:
seen.append((request.method, request.headers.get("last-event-id")))
raise httpx2.ReadTimeout("upstream detail", request=request)

with anyio.fail_after(5):
async with (
httpx2.AsyncClient(transport=httpx2.MockTransport(handle)) as http,
streamable_http_client("http://test/mcp", http_client=http) as (read, write),
):
await write.send(
SessionMessage(
JSONRPCRequest(jsonrpc="2.0", id="resume-1", method="tools/call", params={}),
metadata=ClientMessageMetadata(resumption_token="evt-41"),
)
)
reply = await read.receive()
assert isinstance(reply, SessionMessage)
assert isinstance(reply.message, JSONRPCError)
assert reply.message.id == "resume-1"
assert reply.message.error.code == REQUEST_TIMEOUT
assert reply.message.error.message == snapshot("HTTP response timed out")
assert seen == [("GET", "evt-41")]


@pytest.mark.anyio
async def test_notification_read_timeout_still_ends_the_post_writer(caplog: pytest.LogCaptureFixture) -> None:
"""Pinned gap: notification timeouts still end the writer; request isolation does not cover them.

The raw transport API exposes the writer closure, which Client hides behind session errors.
"""

def handle(request: httpx2.Request) -> httpx2.Response:
raise httpx2.ReadTimeout("notification failed", request=request)

with anyio.fail_after(5):
async with (
httpx2.AsyncClient(transport=httpx2.MockTransport(handle)) as http,
streamable_http_client("http://test/mcp", http_client=http) as (_read, write),
):
await write.send(SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/message")))
await anyio.wait_all_tasks_blocked()
with pytest.raises(anyio.BrokenResourceError):
await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=2, method="ping")))
assert "Error in post_writer" in caplog.text


@pytest.mark.anyio
@pytest.mark.parametrize(("mode", "method"), [("auto", "server/discover"), ("legacy", "initialize")])
async def test_connect_read_timeout_still_propagates_without_a_fallback(mode: str, method: str) -> None:
"""SDK-defined: connect-time read timeouts retain their raw error and do not start another handshake."""
methods: list[str] = []

def handle(request: httpx2.Request) -> httpx2.Response:
methods.append(json.loads(request.content)["method"])
raise httpx2.ReadTimeout("handshake timed out", request=request)

with anyio.fail_after(5):
async with httpx2.AsyncClient(transport=httpx2.MockTransport(handle)) as http:
with pytest.RaisesGroup(httpx2.ReadTimeout, flatten_subgroups=True):
await Client(streamable_http_client("http://test/mcp", http_client=http), mode=mode).__aenter__()
assert methods == [method]
Loading