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

Replace httpx and httpx-sse with httpx2 (#2972)
Some checks failed
CI / checks (push) Has been cancelled
Details
CI / all-green (push) Has been cancelled
Details

Browse source 
Co-authored-by: Max Isbey <224885523+maxisbey@users.noreply.github.com>
This commit is contained in:
Marcelo Trylesinski 2026-07-14 18:05:08 +02:00 • committed by GitHub
commit 2713b53b12
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 1029 additions and 948 deletions
Download patch file Download diff file Expand all files Collapse all files

18
tests/docs_src/test_authorization.py
View file

@ -1,6 +1,6 @@
"""`docs/run/authorization.md`: every claim the page makes, proved against the real SDK."""
import httpx
import httpx2
import pytest
from inline_snapshot import snapshot
from mcp_types import TextContent
@ -40,8 +40,8 @@ async def test_the_app_grows_a_protected_resource_metadata_route() -> None:
async def test_the_metadata_document_is_built_from_auth_settings() -> None:
"""tutorial001: `GET` on the well-known route returns the Protected Resource Metadata the page shows."""
transport = httpx.ASGITransport(app=tutorial001.mcp.streamable_http_app())
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
transport = httpx2.ASGITransport(app=tutorial001.mcp.streamable_http_app())
async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
response = await http_client.get("/.well-known/oauth-protected-resource/mcp")
assert response.status_code == 200
assert response.json() == snapshot(
@ -56,8 +56,8 @@ async def test_the_metadata_document_is_built_from_auth_settings() -> None:
async def test_a_request_without_a_token_never_reaches_the_protocol() -> None:
"""The `!!! check`: no `Authorization` header means a 401 that points at the metadata document."""
transport = httpx.ASGITransport(app=tutorial001.mcp.streamable_http_app())
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
transport = httpx2.ASGITransport(app=tutorial001.mcp.streamable_http_app())
async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
response = await http_client.post("/mcp", json={})
assert response.status_code == 401
assert response.json() == {"error": "invalid_token", "error_description": "Authentication required"}
@ -69,8 +69,8 @@ async def test_a_request_without_a_token_never_reaches_the_protocol() -> None:
async def test_a_token_the_verifier_rejects_gets_the_same_401() -> None:
"""tutorial001: `verify_token` returning `None` and a missing header are indistinguishable to the caller."""
transport = httpx.ASGITransport(app=tutorial001.mcp.streamable_http_app())
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
transport = httpx2.ASGITransport(app=tutorial001.mcp.streamable_http_app())
async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
response = await http_client.post("/mcp", json={}, headers={"Authorization": "Bearer not-a-real-token"})
assert response.status_code == 401
assert response.json() == {"error": "invalid_token", "error_description": "Authentication required"}
@ -86,11 +86,11 @@ async def test_get_access_token_is_none_outside_an_authenticated_request() -> No
async def test_get_access_token_is_the_callers_access_token() -> None:
"""tutorial002: over Streamable HTTP a valid bearer token reaches the tool as an `AccessToken`."""
url = "http://127.0.0.1:8000/mcp"
transport = httpx.ASGITransport(app=tutorial002.mcp.streamable_http_app())
transport = httpx2.ASGITransport(app=tutorial002.mcp.streamable_http_app())
headers = {"Authorization": "Bearer alice-token"}
async with tutorial002.mcp.session_manager.run():
async with (
httpx.AsyncClient(transport=transport, base_url=url, headers=headers) as http_client,
httpx2.AsyncClient(transport=transport, base_url=url, headers=headers) as http_client,
Client(streamable_http_client(url, http_client=http_client)) as client,
):
result = await client.call_tool("whoami", {})