modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement RFC 7523 JWT flows (#1247)

Browse source 
Co-authored-by: Yann Jouanin <yann.jouanin@valueandco.com>
This commit is contained in:
Luca Chang 2025-10-29 09:48:08 -07:00 • committed by GitHub
commit f161149680
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 463 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

20
tests/client/test_auth.py
View file

@ -391,7 +391,9 @@ class TestOAuthFallback:
)
# Mock the authorization process to minimize unnecessary state in this test
oauth_provider._perform_authorization = mock.AsyncMock(return_value=("test_auth_code", "test_code_verifier"))
oauth_provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# Next request should fall back to legacy behavior and auth with the RS (mocked /authorize, next is /token)
token_request = await auth_flow.asend(oauth_metadata_response_3)
@ -513,7 +515,7 @@ class TestOAuthFallback:
assert request is None
@pytest.mark.anyio
async def test_token_exchange_request(self, oauth_provider: OAuthClientProvider):
async def test_token_exchange_request_authorization_code(self, oauth_provider: OAuthClientProvider):
"""Test token exchange request building."""
# Set up required context
oauth_provider.context.client_info = OAuthClientInformationFull(
@ -522,7 +524,7 @@ class TestOAuthFallback:
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
request = await oauth_provider._exchange_token("test_auth_code", "test_verifier")
request = await oauth_provider._exchange_token_authorization_code("test_auth_code", "test_verifier")
assert request.method == "POST"
assert str(request.url) == "https://api.example.com/token"
@ -576,7 +578,7 @@ class TestProtectedResourceMetadata:
)
# Test in token exchange
request = await oauth_provider._exchange_token("test_code", "test_verifier")
request = await oauth_provider._exchange_token_authorization_code("test_code", "test_verifier")
content = request.content.decode()
assert "resource=" in content
# Check URL-encoded resource parameter
@ -607,7 +609,7 @@ class TestProtectedResourceMetadata:
)
# Test in token exchange
request = await oauth_provider._exchange_token("test_code", "test_verifier")
request = await oauth_provider._exchange_token_authorization_code("test_code", "test_verifier")
content = request.content.decode()
assert "resource=" not in content
@ -637,7 +639,7 @@ class TestProtectedResourceMetadata:
)
# Test in token exchange
request = await oauth_provider._exchange_token("test_code", "test_verifier")
request = await oauth_provider._exchange_token_authorization_code("test_code", "test_verifier")
content = request.content.decode()
assert "resource=" in content
@ -710,7 +712,7 @@ class TestAuthFlow:
pass # Expected
@pytest.mark.anyio
async def test_auth_flow_with_no_tokens(self, oauth_provider: OAuthClientProvider):
async def test_auth_flow_with_no_tokens(self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage):
"""Test auth flow when no tokens are available, triggering the full OAuth flow."""
# Ensure no tokens are stored
oauth_provider.context.current_tokens = None
@ -779,7 +781,9 @@ class TestAuthFlow:
)
# Mock the authorization process
oauth_provider._perform_authorization = mock.AsyncMock(return_value=("test_auth_code", "test_code_verifier"))
oauth_provider._perform_authorization_code_grant = mock.AsyncMock(
return_value=("test_auth_code", "test_code_verifier")
)
# Next request should be to exchange token
token_request = await auth_flow.asend(registration_response)