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
2 changes: 1 addition & 1 deletion src/mcp/server/_streamable_http_modern.py
Original file line number Diff line number Diff line change
Expand Up @@ -512,7 +512,7 @@ async def watch_disconnect(cancel_scope: anyio.CancelScope) -> None:
# First notification arrived, or the deferral window elapsed: commit
# `text/event-stream` and start pinging so a proxy idle-read timeout
# cannot close the stream (which on this path cancels the handler).
await send({"type": "http.response.start", "status": _OK_STATUS, "headers": _SSE_HEADERS})
await send({"type": "http.response.start", "status": _OK_STATUS, "headers": list(_SSE_HEADERS)})
while not done:
await send({"type": "http.response.body", "body": event or b": ping\r\n\r\n", "more_body": True})
event = None
Expand Down
36 changes: 36 additions & 0 deletions tests/server/test_streamable_http_modern.py
Original file line number Diff line number Diff line change
Expand Up @@ -500,6 +500,42 @@ async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams |
assert events[2]["result"]["tools"] == []


async def test_sse_mode_gives_each_response_its_own_header_list() -> None:
"""Middleware that appends to `message["headers"]` in place must not leak into later SSE responses."""

async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
await ctx.session.report_progress(1.0, total=3.0)
return ListToolsResult(tools=[], ttl_ms=0, cache_scope="public")

server = Server("test", on_list_tools=list_tools)

async def app(scope: Scope, receive: Receive, send: Send) -> None:
async def appending_send(message: Message) -> None:
if message["type"] == "http.response.start":
message["headers"].append((b"x-correlation-id", b"abc"))
await send(message)

async with server.lifespan(server) as lifespan_state:
await handle_modern_request(server, None, False, lifespan_state, scope, receive, appending_send)

async with httpx2.AsyncClient(
transport=StreamingASGITransport(app),
base_url="http://testserver",
headers={
MCP_PROTOCOL_VERSION_HEADER: LATEST_MODERN_VERSION,
"content-type": "application/json",
"accept": "application/json, text/event-stream",
},
) as http:
with anyio.fail_after(5):
for token in ("tok-1", "tok-2"):
response = await http.post(
"/mcp", json=_list_tools_body_with_token(token), headers={MCP_METHOD_HEADER: "tools/list"}
)
assert response.headers["content-type"].split(";", 1)[0] == "text/event-stream"
assert response.headers.get_list("x-correlation-id") == ["abc"]


@pytest.mark.parametrize(
"anyio_backend",
[pytest.param(("trio", {"clock": MockClock(autojump_threshold=0)}), id="trio-mockclock")],
Expand Down
Loading