René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"get_access_token() returns stale token in stateful streamable-HTTP sessions","articleBody":"## Summary\n\n`get_access_token()` returns the bearer token from the **session-creating request** for the entire lifetime of a stateful streamable-HTTP session, regardless of what `Authorization` header later requests send.\n\n## Repro\n\n\u003cdetails\u003e\u003csummary\u003eOpen for code\u003c/summary\u003e\n\n```python\nimport multiprocessing\nimport socket\nimport time\n\nimport httpx\nimport pytest\nimport uvicorn\nfrom starlette.applications import Starlette\nfrom starlette.middleware import Middleware\nfrom starlette.middleware.authentication import AuthenticationMiddleware\nfrom starlette.routing import Mount\n\nfrom mcp.client.session import ClientSession\nfrom mcp.client.streamable_http import streamable_http_client\nfrom mcp.server import Server, ServerRequestContext\nfrom mcp.server.auth.middleware.auth_context import AuthContextMiddleware, get_access_token\nfrom mcp.server.auth.middleware.bearer_auth import BearerAuthBackend\nfrom mcp.server.auth.provider import AccessToken\nfrom mcp.server.streamable_http_manager import StreamableHTTPSessionManager\nfrom mcp.server.transport_security import TransportSecuritySettings\nfrom mcp.types import (\n    CallToolRequestParams,\n    CallToolResult,\n    ListToolsResult,\n    PaginatedRequestParams,\n    TextContent,\n    Tool,\n)\n\n\nclass _EchoTokenVerifier:\n    \"\"\"Accepts any bearer and echoes it back so we can tell tokens apart.\"\"\"\n\n    async def verify_token(self, token: str) -\u003e AccessToken | None:\n        return AccessToken(token=token, client_id=token, scopes=[], expires_at=int(time.time()) + 3600)\n\n\nasync def _handle_whoami(ctx: ServerRequestContext, params: CallToolRequestParams) -\u003e CallToolResult:\n    # The user-facing contract of get_access_token(): call it from a handler,\n    # get the token for the current request.\n    access = get_access_token()\n    text = access.token if access else \"\u003cnone\u003e\"\n    return CallToolResult(content=[TextContent(type=\"text\", text=text)])\n\n\nasync def _handle_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -\u003e ListToolsResult:\n    return ListToolsResult(tools=[Tool(name=\"whoami\", input_schema={\"type\": \"object\", \"properties\": {}})])\n\n\ndef _run_auth_server(port: int) -\u003e None:\n    server = Server(name=\"auth_test_server\", on_call_tool=_handle_whoami, on_list_tools=_handle_list_tools)\n    security = TransportSecuritySettings(allowed_hosts=[\"127.0.0.1:*\"], allowed_origins=[\"http://127.0.0.1:*\"])\n    session_manager = StreamableHTTPSessionManager(app=server, security_settings=security, stateless=False)\n\n    # Same middleware chain lowlevel Server.streamable_http_app builds when auth is on\n    asgi_app = Starlette(\n        routes=[Mount(\"/mcp\", app=session_manager.handle_request)],\n        middleware=[\n            Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(_EchoTokenVerifier())),\n            Middleware(AuthContextMiddleware),\n        ],\n        lifespan=lambda app: session_manager.run(),\n    )\n    uvicorn.run(asgi_app, host=\"127.0.0.1\", port=port, log_level=\"error\")\n\n\nclass _MutableBearerAuth(httpx.Auth):\n    \"\"\"Reads the bearer from a mutable attribute at send-time so we can swap mid-session.\"\"\"\n\n    def __init__(self, token: str) -\u003e None:\n        self.token = token\n\n    def auth_flow(self, request: httpx.Request):\n        request.headers[\"Authorization\"] = f\"Bearer {self.token}\"\n        yield request\n\n\n@pytest.mark.anyio\nasync def test_get_access_token_reflects_current_request_in_stateful_session() -\u003e None:\n    with socket.socket() as s:\n        s.bind((\"127.0.0.1\", 0))\n        port = s.getsockname()[1]\n\n    proc = multiprocessing.Process(target=_run_auth_server, args=(port,), daemon=True)\n    proc.start()\n    try:\n        # wait for server\n        for _ in range(200):\n            try:\n                with socket.socket() as s:\n                    s.connect((\"127.0.0.1\", port))\n                break\n            except OSError:\n                time.sleep(0.01)\n\n        url = f\"http://127.0.0.1:{port}/mcp\"\n        auth = _MutableBearerAuth(\"token-A\")\n\n        async with httpx.AsyncClient(auth=auth, timeout=httpx.Timeout(30, read=30), follow_redirects=True) as http_client:\n            async with streamable_http_client(url, http_client=http_client) as (read_stream, write_stream):\n                async with ClientSession(read_stream, write_stream) as session:\n                    await session.initialize()\n\n                    # Request 1: session created, _receive_loop spawns with token-A in its context\n                    r1 = await session.call_tool(\"whoami\", {})\n                    assert isinstance(r1.content[0], TextContent)\n                    assert r1.content[0].text == \"token-A\"\n\n                    # Request 2: same session, different bearer — reuses the existing _receive_loop\n                    auth.token = \"token-B\"\n                    r2 = await session.call_tool(\"whoami\", {})\n                    assert isinstance(r2.content[0], TextContent)\n                    # EXPECTED: \"token-B\" — handler should see the token sent with THIS request\n                    # ACTUAL:   \"token-A\" — handler sees the session-creating request's token\n                    assert r2.content[0].text == \"token-B\"\n    finally:\n        proc.kill()\n        proc.join(timeout=2)\n```\n\n**Result on `main`:**\n\n```\nAssertionError: assert 'token-A' == 'token-B'\n  - token-B\n  ?       ^\n  + token-A\n  ?       ^\n```\n\n\u003c/details\u003e\n\n## Root cause\n\n`AuthContextMiddleware` sets `auth_context_var` inside the ASGI request's task. But the tool handler doesn't run in that task — it runs in a task spawned by `Server.run()`'s `tg.start_soon(_handle_message, ...)`, which is itself inside `run_server`, which was spawned at session creation: https://github.com/modelcontextprotocol/python-sdk/blob/528abfab86f1f3c003bd7d54a1f0bbd65d81c59c/src/mcp/server/streamable_http_manager.py#L267\n\n`tg.start()` copies the caller's `contextvars.Context` at call time. So `run_server` (and every task it spawns) carries a snapshot from request 1. Requests 2..N write to the transport's read stream from the new ASGI task, but the reader is `_receive_loop` — still running with the request-1 snapshot. The ContextVar set in request N's ASGI task never reaches the handler.\n\n```\nASGI req 1 (auth_context_var=A)\n  └─ tg.start(run_server)                           ← context copied: A\n       └─ ServerSession.__aenter__\n            └─ tg.start_soon(_receive_loop)         ← inherits A\n       └─ async for msg in session.incoming_messages:\n            └─ tg.start_soon(_handle_message)       ← inherits A\n                 └─ tool handler: get_access_token() → A  ✓\n\nASGI req 2 (auth_context_var=B)\n  └─ transport.handle_request(...)                  ← writes to read_stream\n       ... _receive_loop (still ctx A) reads it ...\n       ... tg.start_soon(_handle_message)           ← inherits A, not B\n            └─ tool handler: get_access_token() → A  ✗\n```\n\nThe existing unit test passes because `MockApp` runs inline in the same task as the middleware — no stream crossing: https://github.com/modelcontextprotocol/python-sdk/blob/528abfab86f1f3c003bd7d54a1f0bbd65d81c59c/tests/server/auth/middleware/test_auth_context.py#L33\n\n## Impact\n\n- **Token refresh mid-session**: refreshed token is invisible to handlers.\n- **Compounds #2100**: if a session ID is hijacked, the attacker's requests execute with the victim's token visible to `get_access_token()`.\n- **Stateless HTTP is unaffected** — new session per request means a fresh context snapshot each time.\n\n## The correct path already exists\n\nThe Starlette `Request` is threaded explicitly through `ServerMessageMetadata.request_context` → `ServerRequestContext.request`. Inside a handler:\n\n```python\nrequest: Request = ctx.request_context.request\nuser = request.user                       # set by AuthenticationMiddleware\ntoken = user.access_token                 # per-request, correct\n```\n\n## Proposed fix\n\nGiven `get_access_token()` has no callers in `src/` or `examples/` and isn't documented: remove `auth_context_var`, `get_access_token()`, and `AuthContextMiddleware`. Expose auth on `Context` as part of #2098 using the explicit `request` threading above.\n\nAlternative (if we want to keep the API): thread the `AuthenticatedUser` alongside `request` on `ServerMessageMetadata` and set the contextvar at the `tg.start_soon` site in `Server.run()`. But that's re-inventing what `request.user` already provides.\n\n## Related\n\n- #2098 — expose session/auth/transport on `Context`\n- #2100 — bind authenticated identity to sessions\n- #1743 — auth rework\n- #1969 — general user-ContextVar propagation through streams (different scope: arbitrary user vars, not just auth)\n\n\u003csub\u003e[AI Disclaimer](https://gist.github.com/maxisbey/6123d132484e4c533eab519a2800693d)\u003c/sub\u003e","author":{"url":"https://github.com/maxisbey","@type":"Person","name":"maxisbey"},"datePublished":"2026-03-04T18:50:27.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":0},"url":"https://github.com/2208/python-sdk/issues/2208"}