Skip to content
Open
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
17 changes: 12 additions & 5 deletions src/mcp/client/auth/oauth2.py
Original file line number Diff line number Diff line change
Expand Up @@ -598,9 +598,13 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
if self.context.is_token_valid():
self._add_auth_header(request)

response = yield request
# Released before the request goes out: the lock serialises token acquisition, and
# holding it for the lifetime of the response would stall every other request on
# this provider until the response ends - unbounded for the standalone GET SSE stream.
response = yield request

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: Concurrent requests can make a 401/403 re-authorization use another request’s MCP-Protocol-Version, which can flip resource parameter inclusion and build incorrect OAuth requests. This comes from dropping the lock before yield request while keeping protocol_version as shared mutable context; a per-request version should be carried through re-auth.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/client/auth/oauth2.py, line 604:

<comment>Concurrent requests can make a 401/403 re-authorization use another request’s `MCP-Protocol-Version`, which can flip `resource` parameter inclusion and build incorrect OAuth requests. This comes from dropping the lock before `yield request` while keeping `protocol_version` as shared mutable context; a per-request version should be carried through re-auth.</comment>

<file context>
@@ -598,9 +598,13 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
+        # Released before the request goes out: the lock serialises token acquisition, and
+        # holding it for the lifetime of the response would stall every other request on
+        # this provider until the response ends - unbounded for the standalone GET SSE stream.
+        response = yield request
 
-            if response.status_code == 401:
</file context>


if response.status_code == 401:
if response.status_code == 401:
async with self.context.lock:
# Perform full OAuth flow
try:
# OAuth flow must be inline due to generator constraints
Expand Down Expand Up @@ -751,8 +755,10 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx

# Retry with new tokens
self._add_auth_header(request)
yield request
elif response.status_code == 403:

yield request
elif response.status_code == 403:
async with self.context.lock:
# Step 1: Extract error field from WWW-Authenticate header
error = extract_field_from_www_auth(response, "error")

Expand Down Expand Up @@ -782,4 +788,5 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx

# Retry with new tokens
self._add_auth_header(request)
yield request

yield request
42 changes: 42 additions & 0 deletions tests/client/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from unittest import mock
from urllib.parse import parse_qs, quote, unquote, urlparse

import anyio
import httpx2
import pytest
from inline_snapshot import Is, snapshot
Expand Down Expand Up @@ -3253,3 +3254,44 @@ async def echo_callback() -> AuthorizationCodeResult:
await auth_flow.asend(httpx2.Response(200, request=final_req))
except StopAsyncIteration:
pass


@pytest.mark.anyio
async def test_in_flight_request_does_not_block_a_concurrent_request(
oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
):
"""A request still in flight must not hold up the next one on the same provider.

The standalone GET SSE stream lives as long as the server keeps it open, so holding
``context.lock`` until its response arrived stalled the first ``tools/call`` for that
whole time (#3209).
"""
oauth_provider.context.current_tokens = valid_tokens
oauth_provider.context.token_expiry_time = time.time() + 1800
oauth_provider._initialized = True

sse_sent = anyio.Event()
call_done = anyio.Event()

async def get_sse_stream() -> None:
flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
request = await flow.__anext__()
sse_sent.set()
# The server holds the stream open, so the response lands after the call is answered.
await call_done.wait()

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P3: This new concurrency test can hang indefinitely because it waits on call_done without a timeout. If the sibling task fails before setting the event, this task stays blocked and can make failures much harder to diagnose. Wrapping the wait in anyio.fail_after(5) keeps the test deterministic and fail-fast.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/client/test_auth.py, line 3281:

<comment>This new concurrency test can hang indefinitely because it waits on `call_done` without a timeout. If the sibling task fails before setting the event, this task stays blocked and can make failures much harder to diagnose. Wrapping the wait in `anyio.fail_after(5)` keeps the test deterministic and fail-fast.</comment>

<file context>
@@ -3253,3 +3254,44 @@ async def echo_callback() -> AuthorizationCodeResult:
+        request = await flow.__anext__()
+        sse_sent.set()
+        # The server holds the stream open, so the response lands after the call is answered.
+        await call_done.wait()
+        with pytest.raises(StopAsyncIteration):
+            await flow.asend(httpx2.Response(200, request=request))
</file context>
Suggested change
await call_done.wait()
with anyio.fail_after(5):
await call_done.wait()

with pytest.raises(StopAsyncIteration):
await flow.asend(httpx2.Response(200, request=request))

async def call_tool() -> None:
await sse_sent.wait()
flow = oauth_provider.async_auth_flow(httpx2.Request("POST", "https://api.example.com/v1/mcp"))
with anyio.fail_after(5):
request = await flow.__anext__()
assert request.headers["Authorization"] == "Bearer test_access_token"
with pytest.raises(StopAsyncIteration):
await flow.asend(httpx2.Response(200, request=request))
call_done.set()

async with anyio.create_task_group() as tg:
tg.start_soon(get_sse_stream)
tg.start_soon(call_tool)
Loading