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

tests/interaction: era-axis machinery for the requirements manifest (#2909)

Browse source
This commit is contained in:
Max 2026-06-19 12:59:39 +01:00 • committed by GitHub
commit 364b762ef6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 1142 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

20
tests/interaction/_connect.py
View file

@ -9,6 +9,7 @@ server's real Starlette app through the in-process streaming bridge, so the full
from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from functools import partial
from typing import Any, Protocol
import httpx
@ -69,6 +70,7 @@ class Connect(Protocol):
message_handler: MessageHandlerFnT | None = None,
client_info: Implementation | None = None,
elicitation_callback: ElicitationFnT | None = None,
protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> AbstractAsyncContextManager[Client]: ...
@ -83,6 +85,7 @@ async def connect_in_memory(
message_handler: MessageHandlerFnT | None = None,
client_info: Implementation | None = None,
elicitation_callback: ElicitationFnT | None = None,
protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> AsyncIterator[Client]:
"""Yield a Client connected to the server over the in-memory transport."""
async with Client(
@ -113,13 +116,15 @@ async def connect_over_streamable_http(
message_handler: MessageHandlerFnT | None = None,
client_info: Implementation | None = None,
elicitation_callback: ElicitationFnT | None = None,
protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> AsyncIterator[Client]:
"""Yield a Client connected to the server's streamable HTTP app, entirely in process.
With the defaults this is the matrix leg (stateful sessions, SSE responses); the
transport-specific tests pass `stateless_http` or `json_response` to select the other
server modes, and the resumability tests pass an `event_store` (with `retry_interval=0` so
the client's reconnection wait is a no-op).
With the defaults this is the matrix leg (stateful sessions, SSE responses); the stateless
matrix arm binds `stateless_http=True` (see `connect_over_streamable_http_stateless`);
transport-specific tests pass `json_response` to select the other server mode, and the
resumability tests pass an `event_store` (with `retry_interval=0` so the client's
reconnection wait is a no-op).
"""
app = server.streamable_http_app(
stateless_http=stateless_http,
@ -145,6 +150,12 @@ async def connect_over_streamable_http(
yield client
connect_over_streamable_http_stateless: Connect = partial(connect_over_streamable_http, stateless_http=True)
"""The streamable-http matrix arm with the server in stateless mode (fresh transport per request,
no session id, no standalone GET stream). The same shared Server instance backs every request --
stateless mode does not require a server factory."""
@asynccontextmanager
async def mounted_app(
server: Server | MCPServer,
@ -326,6 +337,7 @@ async def connect_over_sse(
message_handler: MessageHandlerFnT | None = None,
client_info: Implementation | None = None,
elicitation_callback: ElicitationFnT | None = None,
protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> AsyncIterator[Client]:
"""Yield a Client connected to the server's legacy SSE transport, entirely in process."""
app, _ = build_sse_app(server)