test(mcp): stay connected until the SSE reply ends; drop cross-context reset

The test client reported http.disconnect as soon as the request body was
read, and the SDK's SSE response cancels its stream on disconnect, so
tools/list came back 200 with an empty body. Model a real client: wait
for the response (with a 15s deadline), then disconnect.

The caller fixture reset its ContextVar from sync teardown, a different
context from the async test that set it; each async test already runs in
its own context copy, so the reset is dropped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/mcp/_http.py

@@ -5,6 +5,7 @@ pipeline), so MCP tests call app.asgi_app directly.
 """
 from __future__ import annotations
 
+import asyncio
 import json
 
 
@@ -21,14 +22,23 @@ async def asgi_request(app, method, path, headers=None, body=None):
     }
     pending = [{"type": "http.request", "body": body or b"", "more_body": False}]
     sent = []
+    responded = asyncio.Event()
 
+    # Like a real client, stay connected until the response is complete. An
+    # SSE response listens for http.disconnect and cancels its stream on it,
+    # so reporting a disconnect as soon as the body is read empties the reply.
     async def receive():
-        return pending.pop(0) if pending else {"type": "http.disconnect"}
+        if pending:
+            return pending.pop(0)
+        await responded.wait()
+        return {"type": "http.disconnect"}
 
     async def send(message):
         sent.append(message)
+        if message["type"] == "http.response.body" and not message.get("more_body"):
+            responded.set()
 
-    await app.asgi_app(scope, receive, send)
+    await asyncio.wait_for(app.asgi_app(scope, receive, send), timeout=15)
     start = next(m for m in sent if m["type"] == "http.response.start")
     data = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
     return start["status"], dict(start.get("headers", [])), data
@@ -40,6 +50,8 @@ def rpc_result(data: bytes) -> dict:
     if text.lstrip().startswith("{"):
         return json.loads(text)
     lines = [ln[5:].strip() for ln in text.splitlines() if ln.startswith("data:")]
+    if not lines:
+        raise AssertionError(f"no JSON-RPC message in response body: {text!r}")
     return json.loads(lines[-1])
 
 

tests/mcp/test_server.py

@@ -26,13 +26,12 @@ OPERATOR = Caller(user_id="u1", username="ana", role=UserRole.operator, token_id
 
 @pytest.fixture
 def as_caller():
-    tokens = []
+    # Each async test runs in its own task, i.e. a copy of the context, so a
-
+    # caller set here can't leak into another test; resetting from the sync
+    # teardown would fail ("created in a different Context").
     def _set(caller):
-        tokens.append(_context._caller_ctx.set(caller))
+        _context._caller_ctx.set(caller)
-    yield _set
+    return _set
-    for t in reversed(tokens):
-        _context._caller_ctx.reset(t)
 
 
 def _server(plugins_cfg=None):