From 077f7ad40f40e0cf46b5df73feb70d6cc9a768ee Mon Sep 17 00:00:00 2001 From: ptraaa <122882415+Ptraaah@users.noreply.github.com> Date: Sun, 1 Nov 2026 02:52:00 +0800 Subject: [PATCH 1/3] Apply the current request's auth context while handling a request (#1250) --- src/mcp/server/lowlevel/server.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 25a8fde37c..1d1e3faaae 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -83,6 +83,8 @@ async def main(): from typing_extensions import TypeVar, deprecated import mcp.types as types +from mcp.server.auth.middleware.auth_context import auth_context_var +from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser from mcp.server.experimental.request_context import Experimental from mcp.server.lowlevel.experimental import ExperimentalHandlers from mcp.server.lowlevel.func_inspection import create_call_wrapper @@ -736,6 +738,7 @@ async def _handle_request( logger.debug("Dispatching request of type %s", type(req).__name__) token = None + auth_token: contextvars.Token[AuthenticatedUser | None] | None = None try: # Extract request context and close_sse_stream from message metadata request_data = None @@ -756,6 +759,18 @@ async def _handle_request( task_metadata = None if hasattr(req, "params") and req.params is not None: task_metadata = getattr(req.params, "task", None) + # The HTTP transports attach the current request, and its scope + # carries the user authenticated for this request. Handlers run + # in the session's long-lived task, where the auth_context_var + # set by AuthContextMiddleware still holds the session-opening + # request's user, so a refreshed token never reaches the handler + # (#1250). Apply this request's user for the handler's duration. + if request_data is not None: + scope = getattr(request_data, "scope", None) + if isinstance(scope, dict): + user = cast(dict[str, Any], scope).get("user") + if isinstance(user, AuthenticatedUser): + auth_token = auth_context_var.set(user) token = request_ctx.set( RequestContext( message.request_id, @@ -794,6 +809,8 @@ async def _handle_request( response = types.ErrorData(code=0, message=str(err), data=None) finally: # Reset the global state after we are done + if auth_token is not None: + auth_context_var.reset(auth_token) if token is not None: # pragma: no branch request_ctx.reset(token) else: # pragma: no cover From 1631a8f3086fdaa2a6290a1aa066ac9aded0d9bb Mon Sep 17 00:00:00 2001 From: ptraaa <122882415+Ptraaah@users.noreply.github.com> Date: Sun, 1 Nov 2026 02:52:06 +0800 Subject: [PATCH 2/3] Add regression tests for refreshed access tokens (#1250) --- .../test_1250_refreshed_access_token.py | 231 ++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 tests/issues/test_1250_refreshed_access_token.py diff --git a/tests/issues/test_1250_refreshed_access_token.py b/tests/issues/test_1250_refreshed_access_token.py new file mode 100644 index 0000000000..b4d6adb8d6 --- /dev/null +++ b/tests/issues/test_1250_refreshed_access_token.py @@ -0,0 +1,231 @@ +"""Regression tests for issue #1250 - refreshed access tokens stay invisible to tools. + +A client that refreshes its access token mid-session sends the new token on +the next request's `Authorization` header, but a tool calling +`get_access_token()` kept receiving the token the session was opened with: +the auth context middleware sets the contextvar on each HTTP request's task, +while handlers run in the session's long-lived task. + +These tests drive the server over raw HTTP in process because the property +under test - which credential a *changed* `Authorization` header maps to +inside an already-open session - cannot be produced by the typed client API, +whose auth layer owns the header. Everything else (the session manager, the +middleware stack, the tool handler) is the real stack. +""" + +import json +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager +from typing import Any + +import anyio +import httpx +import pytest +from starlette.applications import Starlette +from starlette.middleware import Middleware +from starlette.middleware.authentication import AuthenticationMiddleware +from starlette.routing import Mount + +from mcp.server.auth.middleware.auth_context import AuthContextMiddleware, get_access_token +from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend +from mcp.server.auth.provider import AccessToken, TokenVerifier +from mcp.server.lowlevel import Server +from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from mcp.types import TextContent, Tool + +pytestmark = pytest.mark.anyio + +PROTOCOL_VERSION = "2025-06-18" +BASE_URL = "http://127.0.0.1:8000" + + +class AcceptAllTokenVerifier(TokenVerifier): + """Verifies any bearer token as the same client; the token string is its identity. + + Models a refresh faithfully: the refreshed token belongs to the same + client (and principal) as the one it replaces, only the token value differs. + """ + + async def verify_token(self, token: str) -> AccessToken | None: + return AccessToken(token=token, client_id="test-client", scopes=["mcp"]) + + +def build_app(server: Server) -> tuple[Starlette, StreamableHTTPSessionManager]: + session_manager = StreamableHTTPSessionManager(app=server, stateless=False, json_response=False) + + @asynccontextmanager + async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: + async with session_manager.run(): + yield + + app = Starlette( + routes=[Mount("/mcp", app=session_manager.handle_request)], + middleware=[ + Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(AcceptAllTokenVerifier())), + Middleware(AuthContextMiddleware), + ], + lifespan=lifespan, + ) + return app, session_manager + + +def make_server(call_tool: Any) -> Server: + server = Server("token-echo") + + @server.list_tools() + async def list_tools() -> list[Tool]: + return [ + Tool( + name="echo_token", + description="Return the token get_access_token() sees", + inputSchema={"type": "object"}, + ) + ] + + server.call_tool()(call_tool) + return server + + +async def post_message( + http: httpx.AsyncClient, + payload: dict[str, Any], + *, + token: str, + session_id: str | None = None, +) -> httpx.Response: + headers = { + "Authorization": f"Bearer {token}", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + if session_id is not None: + headers["mcp-session-id"] = session_id + headers["mcp-protocol-version"] = PROTOCOL_VERSION + return await http.post("/mcp/", headers=headers, json=payload) + + +def response_for(response: httpx.Response, request_id: int) -> dict[str, Any]: + """Extract the JSON-RPC response for `request_id` from an SSE response body.""" + for line in response.text.splitlines(): + if line.startswith("data:"): + message = json.loads(line.removeprefix("data:").strip()) + if message.get("id") == request_id: + return message + raise AssertionError(f"no response for request {request_id} in body: {response.text!r}") + + +async def open_session(http: httpx.AsyncClient, token: str) -> str: + """Run the initialize handshake with `token` and return the session id.""" + response = await post_message( + http, + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "issue-1250-regression", "version": "0"}, + }, + }, + token=token, + ) + assert response.status_code == 200 + session_id = response.headers["mcp-session-id"] + initialized = await post_message( + http, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + token=token, + session_id=session_id, + ) + assert initialized.status_code == 202 + return session_id + + +async def call_echo_token(http: httpx.AsyncClient, session_id: str, token: str, request_id: int) -> str: + """Call the echo tool presenting `token`; return the token the tool observed.""" + response = await post_message( + http, + { + "jsonrpc": "2.0", + "id": request_id, + "method": "tools/call", + "params": {"name": "echo_token", "arguments": {}}, + }, + token=token, + session_id=session_id, + ) + assert response.status_code == 200 + message = response_for(response, request_id) + content = message["result"]["content"] + assert content[0]["type"] == "text" + return content[0]["text"] + + +async def test_a_refreshed_token_is_what_tools_see_in_the_same_session() -> None: + """After a refresh, the next call in the same session observes the new token. + + Steps: + 1. Open a session presenting `token-a`; the tool sees `token-a`. + 2. Present the refreshed `token-b` on the next call in the same session; + the tool must see `token-b`, not the session-opening `token-a`. + """ + + async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: + assert name == "echo_token" + token = get_access_token() + return [TextContent(type="text", text=token.token if token else "no-token")] + + server = make_server(call_tool) + app, session_manager = build_app(server) + + with anyio.fail_after(5): + async with session_manager.run(): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url=BASE_URL) as http: + session_id = await open_session(http, "token-a") + assert await call_echo_token(http, session_id, "token-a", 2) == "token-a" + assert await call_echo_token(http, session_id, "token-b", 3) == "token-b" + + +async def test_parallel_sessions_each_see_their_own_current_token() -> None: + """Interleaved calls in two sessions never observe each other's tokens. + + Both calls are held in flight at the same time (the handler parks on a + barrier until both have read their token), so the scheduler cannot + serialize them into looking correct. + """ + in_flight = 0 + both_in_flight = anyio.Event() + release = anyio.Event() + + async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: + nonlocal in_flight + assert name == "echo_token" + token = get_access_token() + seen = token.token if token else "no-token" + in_flight += 1 + if in_flight == 2: + both_in_flight.set() + await release.wait() + return [TextContent(type="text", text=seen)] + + server = make_server(call_tool) + app, session_manager = build_app(server) + + results: dict[str, str] = {} + + async def call(http: httpx.AsyncClient, session_id: str, token: str, request_id: int) -> None: + results[token] = await call_echo_token(http, session_id, token, request_id) + + with anyio.fail_after(5): + async with session_manager.run(): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url=BASE_URL) as http: + session_one = await open_session(http, "token-a") + session_two = await open_session(http, "token-c") + async with anyio.create_task_group() as tg: + # Session one presents its refreshed token; session two its own. + tg.start_soon(call, http, session_one, "token-b", 2) + tg.start_soon(call, http, session_two, "token-c", 3) + await both_in_flight.wait() + release.set() + assert results == {"token-b": "token-b", "token-c": "token-c"} From db58af36d0716a106fdf1244c6aa13bb7462f8cd Mon Sep 17 00:00:00 2001 From: ptraaa <122882415+Ptraaah@users.noreply.github.com> Date: Sun, 1 Nov 2026 02:52:12 +0800 Subject: [PATCH 3/3] Add dispatch-level tests for auth context from the request scope (#1250) --- .../test_auth_context_from_request.py | 133 ++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 tests/server/lowlevel/test_auth_context_from_request.py diff --git a/tests/server/lowlevel/test_auth_context_from_request.py b/tests/server/lowlevel/test_auth_context_from_request.py new file mode 100644 index 0000000000..54cdd44a4e --- /dev/null +++ b/tests/server/lowlevel/test_auth_context_from_request.py @@ -0,0 +1,133 @@ +"""The lowlevel server applies each request's authenticated user while handling it. + +Regression coverage for issue #1250 at the dispatch layer: handlers run in +the session's long-lived task, so the `auth_context_var` set by +`AuthContextMiddleware` on the HTTP request's task would otherwise still hold +the session-opening request's user. `_handle_request` re-applies the user +from the request the transport attached to the message metadata. +""" + +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest +from starlette.requests import Request +from starlette.types import Scope + +import mcp.types as types +from mcp.server.auth.middleware.auth_context import auth_context_var, get_access_token +from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser +from mcp.server.auth.provider import AccessToken +from mcp.server.lowlevel.server import Server +from mcp.server.session import ServerSession +from mcp.shared.message import ServerMessageMetadata +from mcp.shared.session import RequestResponder + + +def make_server(observed: list[str | None]) -> Server: + server = Server("test-server") + + @server.list_tools() + async def handle_list_tools() -> list[types.Tool]: + return [types.Tool(name="echo_token", description="echo", inputSchema={"type": "object"})] + + @server.call_tool() + async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]: + assert name == "echo_token" + token = get_access_token() + observed.append(token.token if token else None) + return [types.TextContent(type="text", text=token.token if token else "no-token")] + + return server + + +async def dispatch(server: Server, metadata: ServerMessageMetadata | None) -> Mock: + message = Mock(spec=RequestResponder) + message.request_id = "req-1" + message.request_meta = None + message.message_metadata = metadata + message.respond = AsyncMock() + + session = Mock(spec=ServerSession) + session.client_params = None + + request = types.CallToolRequest(params=types.CallToolRequestParams(name="echo_token", arguments={})) + await server._handle_request(message, request, session, {}, raise_exceptions=False) + return message + + +def request_with_scope(scope: Scope) -> Request: + return Request(scope) + + +@pytest.mark.anyio +async def test_handle_request_applies_the_user_from_the_request_scope() -> None: + """A handler sees the user authenticated for its own request, and the var is reset after.""" + observed: list[str | None] = [] + server = make_server(observed) + + verified = AccessToken(token="scope" + "-token", client_id="test-client", scopes=["test"]) + scope: Scope = { + "type": "http", + "method": "POST", + "path": "/mcp", + "headers": [], + "user": AuthenticatedUser(verified), + } + message = await dispatch(server, ServerMessageMetadata(request_context=request_with_scope(scope))) + + assert observed == ["scope-token"] + assert auth_context_var.get() is None + response = message.respond.call_args.args[0] + assert isinstance(response.root, types.CallToolResult) + content = response.root.content[0] + assert isinstance(content, types.TextContent) + assert content.text == "scope-token" + + +@pytest.mark.anyio +async def test_handle_request_without_a_user_in_scope_leaves_the_context_empty() -> None: + """Requests whose scope carries no authenticated user run unauthenticated.""" + observed: list[str | None] = [] + server = make_server(observed) + + scope: Scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": []} + await dispatch(server, ServerMessageMetadata(request_context=request_with_scope(scope))) + + assert observed == [None] + assert auth_context_var.get() is None + + +@pytest.mark.anyio +async def test_handle_request_with_a_non_authenticated_user_in_scope_leaves_the_context_empty() -> None: + """A scope user that is not an `AuthenticatedUser` (e.g. a plain Starlette user) is ignored.""" + observed: list[str | None] = [] + server = make_server(observed) + + scope: Scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": [], "user": object()} + await dispatch(server, ServerMessageMetadata(request_context=request_with_scope(scope))) + + assert observed == [None] + + +@pytest.mark.anyio +async def test_handle_request_with_an_attached_object_without_a_scope_leaves_the_context_empty() -> None: + """A request context that is not a Starlette request has no scope to read a user from.""" + observed: list[str | None] = [] + server = make_server(observed) + + await dispatch(server, ServerMessageMetadata(request_context=object())) + + assert observed == [None] + + +@pytest.mark.anyio +async def test_handle_request_without_metadata_leaves_the_context_empty() -> None: + """Transports that attach no request (e.g. stdio) run unauthenticated, as before.""" + observed: list[str | None] = [] + server = make_server(observed) + + await dispatch(server, None) + + assert observed == [None] + assert auth_context_var.get() is None