First end-to-end 2026-07-28 stateless tools/call (experimental entry + ClientSession pin) (#2917)

This commit is contained in:
Max 2026-06-20 14:55:59 +01:00 • committed by GitHub
commit 84bf9bde05
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 1515 additions and 93 deletions

View file

@ -0,0 +1,231 @@
"""Single-exchange HTTP serving for protocol version 2026-07-28.
Private module — entry is via `StreamableHTTPSessionManager.handle_request`.
The legacy streamable-HTTP transport is untouched and remains the supported
path for earlier protocol revisions.
A 2026-07-28 request is a self-contained POST: no `initialize` handshake, no
`Mcp-Session-Id`, one JSON-RPC request in, one JSON-RPC response out. This
module handles such a request directly in the ASGI task - no memory streams,
no per-request task group, no `JSONRPCDispatcher`.
"""
from __future__ import annotations
import logging
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
import anyio
import anyio.abc
from pydantic import ValidationError
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import Receive, Scope, Send
from mcp.server.runner import (
_EXIT_STACK_CLOSE_TIMEOUT, # type: ignore[reportPrivateUsage]
ServerRunner,
otel_middleware,
)
from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest
from mcp.shared.exceptions import MCPError, NoBackChannelError
from mcp.shared.message import MessageMetadata, ServerMessageMetadata
from mcp.shared.transport_context import TransportContext
from mcp.types import (
INTERNAL_ERROR,
INVALID_PARAMS,
PARSE_ERROR,
ErrorData,
JSONRPCError,
JSONRPCRequest,
JSONRPCResponse,
RequestId,
)
if TYPE_CHECKING:
from mcp.server.lowlevel.server import Server
logger = logging.getLogger(__name__)
@dataclass
class _SingleExchangeDispatchContext:
"""`DispatchContext` for one inbound HTTP request.
Structurally satisfies `mcp.shared.dispatcher.DispatchContext`. The
back-channel is closed by construction: a 2026-07-28 server cannot send
requests to the client.
"""
transport: TransportContext
request_id: RequestId
message_metadata: MessageMetadata
cancel_requested: anyio.Event = field(default_factory=anyio.Event)
can_send_request: bool = False
async def send_raw_request(
self,
method: str,
params: Mapping[str, Any] | None,
opts: CallOptions | None = None,
) -> dict[str, Any]:
raise NoBackChannelError(method)
async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
return None
async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
# TODO: no progressToken plumbing yet.
return None
class SingleExchangeDispatcher:
"""Dispatcher for exactly one inbound JSON-RPC request over a single HTTP POST.
The exception->wire boundary lives here (mirrors `JSONRPCDispatcher`'s
role). Implements the `Dispatcher` Protocol so `ServerRunner` /
`Connection` / `ServerSession` accept it; `run()` is never driven.
"""
def __init__(self, request: Request) -> None:
self._request = request
self._tctx = TransportContext(
kind="streamable-http",
can_send_request=False,
headers=request.headers,
)
async def send_raw_request(
self,
method: str,
params: Mapping[str, Any] | None,
opts: CallOptions | None = None,
*,
_related_request_id: RequestId | None = None,
) -> dict[str, Any]:
raise NoBackChannelError(method)
async def notify(
self,
method: str,
params: Mapping[str, Any] | None,
*,
_related_request_id: RequestId | None = None,
) -> None:
# TODO: buffer and stream as SSE once the response-mode design lands.
return None
async def run(
self,
on_request: OnRequest,
on_notify: OnNotify,
*,
task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
) -> None:
raise RuntimeError("SingleExchangeDispatcher.run() is never driven; use handle()")
async def handle(self, req: JSONRPCRequest, on_request: OnRequest) -> JSONRPCResponse | JSONRPCError:
"""Dispatch one request and map any exception to a `JSONRPCError`."""
dctx = _SingleExchangeDispatchContext(
transport=self._tctx,
request_id=req.id,
message_metadata=ServerMessageMetadata(request_context=self._request),
)
try:
result = await on_request(dctx, req.method, req.params)
return JSONRPCResponse(jsonrpc="2.0", id=req.id, result=result)
except MCPError as e:
return JSONRPCError(jsonrpc="2.0", id=req.id, error=e.error)
except ValidationError:
return JSONRPCError(
jsonrpc="2.0",
id=req.id,
error=ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data=""),
)
# TODO: consolidate the three exception->ErrorData copies once the
# code=0 compat pin in JSONRPCDispatcher is lifted.
except Exception:
logger.exception("handler for %r raised", req.method)
return JSONRPCError(
jsonrpc="2.0",
id=req.id,
error=ErrorData(code=INTERNAL_ERROR, message="Internal server error"),
)
async def handle_modern_request(
app: Server[Any],
security_settings: TransportSecuritySettings | None,
protocol_version: str,
scope: Scope,
receive: Receive,
send: Send,
) -> None:
"""ASGI handler for a single stateless-era POST.
Called from `StreamableHTTPSessionManager.handle_request` when the
`MCP-Protocol-Version` header is in `MODERN_PROTOCOL_VERSIONS`; the header
value is passed as `protocol_version`. Never sets `Mcp-Session-Id`.
"""
request = Request(scope, receive)
security = TransportSecurityMiddleware(security_settings)
err = await security.validate_request(request, is_post=(request.method == "POST"))
if err is not None:
await err(scope, receive, send)
return
# TODO: validate Accept header once the JSON-vs-SSE response-mode design is settled.
if request.method != "POST":
# TODO: GET/DELETE rejection (405 + -32601) lands with the validation ladder.
await Response(status_code=405, headers={"Allow": "POST"})(scope, receive, send)
return
body = await request.body()
try:
req = JSONRPCRequest.model_validate_json(body)
except ValidationError:
msg = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error"))
await Response(
msg.model_dump_json(by_alias=True),
status_code=400,
media_type="application/json",
)(scope, receive, send)
return
dispatcher = SingleExchangeDispatcher(request)
# TODO: per-request lifespan re-entry matches stateless_http=True today; revisit in #2893.
async with app.lifespan(app) as lifespan_state:
runner = ServerRunner(
server=app,
dispatcher=dispatcher,
lifespan_state=lifespan_state,
has_standalone_channel=False,
stateless=True,
dispatch_middleware=[otel_middleware],
)
runner.connection.protocol_version = protocol_version
try:
msg = await dispatcher.handle(req, runner._compose_on_request()) # type: ignore[reportPrivateUsage]
finally:
with anyio.move_on_after(_EXIT_STACK_CLOSE_TIMEOUT, shield=True) as cancel_scope:
try:
await runner.connection.exit_stack.aclose()
except Exception:
logger.exception("connection exit_stack cleanup raised")
if cancel_scope.cancelled_caught:
logger.warning(
"connection exit_stack cleanup exceeded %s seconds; abandoning remaining callbacks",
_EXIT_STACK_CLOSE_TIMEOUT,
)
# TODO: error.code -> HTTP status mapping is a follow-up; 200 for all JSONRPCError bodies for now.
await Response(
msg.model_dump_json(by_alias=True, exclude_none=True),
status_code=200,
media_type="application/json",
)(scope, receive, send)

View file

@ -31,9 +31,8 @@ from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, Server
from mcp.server.models import InitializationOptions
from mcp.server.session import ServerSession
from mcp.shared._otel import extract_trace_context, otel_span
from mcp.shared.dispatcher import DispatchContext, DispatchMiddleware, OnRequest
from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnRequest
from mcp.shared.exceptions import MCPError
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
from mcp.shared.message import MessageMetadata, ServerMessageMetadata
from mcp.shared.transport_context import TransportContext
from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
@ -175,7 +174,7 @@ class ServerRunner(Generic[LifespanT]):
"""Per-connection orchestrator. One instance per client connection."""
server: Server[LifespanT]
dispatcher: JSONRPCDispatcher[Any]
dispatcher: Dispatcher[Any]
lifespan_state: LifespanT
has_standalone_channel: bool
init_options: InitializationOptions | None = None

View file

@ -9,16 +9,15 @@ The receive-loop, initialize handling, and per-request task isolation that
used to live here are now owned by `JSONRPCDispatcher` and `ServerRunner`.
"""
from typing import Any, TypeVar, overload
from typing import Any, TypeVar, cast, overload
from pydantic import AnyUrl, BaseModel
from mcp import types
from mcp.server.connection import Connection
from mcp.server.validation import validate_sampling_tools, validate_tool_use_result_messages
from mcp.shared.dispatcher import CallOptions, ProgressFnT
from mcp.shared.dispatcher import CallOptions, Dispatcher, ProgressFnT
from mcp.shared.exceptions import NoBackChannelError, StatelessModeNotSupported
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
from mcp.shared.message import ServerMessageMetadata
from mcp.types import methods as _methods
@ -37,7 +36,7 @@ class ServerSession:
def __init__(
self,
dispatcher: JSONRPCDispatcher[Any],
dispatcher: Dispatcher[Any],
connection: Connection,
*,
stateless: bool = False,
@ -92,8 +91,19 @@ class ServerSession:
# Fail fast instead of parking forever on a response that cannot
# arrive; matches `Connection.send_raw_request`.
raise NoBackChannelError(data["method"])
result = await self._dispatcher.send_raw_request(
data["method"], data.get("params"), opts or None, _related_request_id=related
# TODO: _related_request_id is not on the Dispatcher Protocol (and must not
# be — it's transport-specific). The fix is to give `ctx.session` a per-request
# Outbound (the DispatchContext, which threads its own request_id) alongside
# the connection-level one, with `related_request_id` as the selector; that
# belongs with the ServerSession/Context rework, not here.
result = cast(
"dict[str, Any]",
await self._dispatcher.send_raw_request(
data["method"],
data.get("params"),
opts or None,
_related_request_id=related, # type: ignore[call-arg]
),
)
# Literal fallback covers pre-handshake and stateless; matches runner.py.
version = self.protocol_version or "2025-11-25"
@ -110,7 +120,7 @@ class ServerSession:
) -> None:
"""Send a typed server-to-client notification."""
data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
await self._dispatcher.notify(data["method"], data.get("params"), _related_request_id=related_request_id)
await self._dispatcher.notify(data["method"], data.get("params"), _related_request_id=related_request_id) # type: ignore[call-arg]
def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
"""Check if the client supports a specific capability."""

View file

@ -14,6 +14,7 @@ from starlette.requests import Request
from starlette.responses import Response
from starlette.types import Receive, Scope, Send
from mcp.server._streamable_http_modern import handle_modern_request
from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
from mcp.server.streamable_http import (
MCP_SESSION_ID_HEADER,
@ -22,6 +23,7 @@ from mcp.server.streamable_http import (
)
from mcp.server.transport_security import TransportSecuritySettings
from mcp.shared._compat import resync_tracer
from mcp.shared.version import MODERN_PROTOCOL_VERSIONS
from mcp.types import INVALID_REQUEST, ErrorData, JSONRPCError
if TYPE_CHECKING:
@ -150,6 +152,13 @@ class StreamableHTTPSessionManager:
if self._task_group is None:
raise RuntimeError("Task group is not initialized. Make sure to use run().")
# TODO: header-only routing for now; body-primary classification
# (per SEP-2575) is a follow-up. 2025 paths below remain unchanged.
pv = next((v.decode("latin-1") for k, v in scope["headers"] if k == b"mcp-protocol-version"), None)
if pv in MODERN_PROTOCOL_VERSIONS:
await handle_modern_request(self.app, self.security_settings, pv, scope, receive, send)
return
# Dispatch to the appropriate handler
if self.stateless:
await self._handle_stateless_request(scope, receive, send)