"""`Connection` - per-client connection state and the standalone outbound channel. Always present on `Context` (never `None`), even in stateless deployments. Holds peer info, per-connection scratch `state` and an `exit_stack` for teardown, and an `Outbound` for the standalone stream (the SSE GET stream in streamable HTTP, or the single duplex stream in stdio). Construct via the factories: `Connection.from_envelope` for the 2026-era single-exchange path (born ready, no back-channel) and `Connection.for_loop` for the handshake-driven loop path. Both populate `protocol_version` so the kernel reads it as a fact. `notify` is best-effort: it never raises. If there's no standalone channel or the stream has been dropped, the notification is debug-logged and silently discarded - server-initiated notifications are inherently advisory. `send_raw_request` raises `NoBackChannelError` when there's no channel; `ping` is the only spec-sanctioned standalone request. """ from __future__ import annotations import logging from collections.abc import Mapping from contextlib import AsyncExitStack from typing import Any, Final, TypeVar, get_args, overload import anyio from mcp_types import ( LOG_LEVEL_META_KEY, ClientCapabilities, CreateMessageRequest, CreateMessageResult, ElicitRequest, ElicitResult, EmptyResult, Implementation, InitializeRequestParams, ListRootsRequest, ListRootsResult, LoggingLevel, PingRequest, Request, ) from mcp_types import methods as _methods from mcp_types.version import LATEST_HANDSHAKE_VERSION, MODERN_PROTOCOL_VERSIONS from pydantic import BaseModel, ValidationError from typing_extensions import deprecated from mcp.shared.dispatcher import CallOptions, Outbound from mcp.shared.exceptions import MCPDeprecationWarning, NoBackChannelError from mcp.shared.peer import Meta, dump_params from mcp.shared.subscriptions import LISTEN_STREAM_METHODS __all__ = ["Connection"] logger = logging.getLogger(__name__) # `Connection.log`'s `logger` parameter (public API, the spec's logger-name # field) shadows the module logger inside that method; this alias keeps the # module logger reachable there. _logger = logger _LOG_LEVELS: Final[tuple[LoggingLevel, ...]] = get_args(LoggingLevel) """Severity-ascending, from the `LoggingLevel` literal's declaration order (the RFC 5424 scale) - the literal is the single source of the ordering.""" _ALL_LOG_LEVELS: Final[frozenset[LoggingLevel]] = frozenset(_LOG_LEVELS) def allowed_log_levels(protocol_version: str, meta: Mapping[str, Any] | None) -> frozenset[LoggingLevel]: """The `notifications/message` levels deliverable for one inbound request. 2026-07-28+ makes log delivery a per-request opt-in (server/utilities/ logging): the client sets the reserved `io.modelcontextprotocol/logLevel` `_meta` key, absent means no levels - the server MUST NOT send - and present means that level and above. An unrecognized value reads as absent; spec methods already reject a malformed value at surface validation before any handler runs, so that arm only serves custom methods, where dropping is the safe direction. Connection-scoped emitters pass `meta=None`: `logging/setLevel` is gone at 2026 and log delivery is request-scoped only, so they deliver nothing. Handshake versions keep their `logging/setLevel`-era semantics: every level may be sent, filtering is the application's `logging/setLevel` handler's job as before. """ if protocol_version not in MODERN_PROTOCOL_VERSIONS: return _ALL_LOG_LEVELS requested = (meta or {}).get(LOG_LEVEL_META_KEY) if requested not in _LOG_LEVELS: return frozenset() return frozenset(_LOG_LEVELS[_LOG_LEVELS.index(requested) :]) ResultT = TypeVar("ResultT", bound=BaseModel) # Result types for the spec's server-to-client request set, used by # `Connection.send_request` to infer the result type. If the spec's request # set grows substantially, consider declaring the result mapping on the # request types themselves (a `__mcp_result__` ClassVar read via a structural # protocol) so this table and the overload ladder don't need maintaining. _RESULT_FOR: dict[type[Request[Any, Any]], type[BaseModel]] = { CreateMessageRequest: CreateMessageResult, ElicitRequest: ElicitResult, ListRootsRequest: ListRootsResult, PingRequest: EmptyResult, } _ModelT = TypeVar("_ModelT", bound=BaseModel) def _typed(model: type[_ModelT], raw: Any) -> _ModelT | None: """Validate a raw envelope value into a typed model. A missing, null or mis-shaped value falls through to `ValidationError` and is treated as not supplied so the request still routes. Spec methods are separately re-validated by the kernel's per-version params surface, which types the reserved `_meta` keys strictly. """ try: return model.model_validate(raw, by_name=False) except ValidationError: return None def _notification_params(payload: dict[str, Any] | None, meta: Meta | None) -> dict[str, Any] | None: if not meta: return payload out = dict(payload or {}) out["_meta"] = meta return out class _NoChannelOutbound: """Connection-scoped `Outbound` for the no-back-channel case. The structural answer to "this connection cannot push to its peer": `send_raw_request` raises `NoBackChannelError`; `notify` drops with a debug log. `Connection.from_envelope` installs this so the modern single-exchange path never needs a mode flag - the channel itself says no. """ 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, opts: CallOptions | None = None) -> None: logger.debug("dropped %s: no standalone channel", method) _NO_CHANNEL = _NoChannelOutbound() class NotifyOnlyOutbound(_NoChannelOutbound): """Connection-scoped `Outbound` that forwards notifications and refuses requests. Installed by `serve_dual_era_loop` for modern (2026-07-28+) connections over duplex stream transports: the pipe is real, so server notifications ride it, but the modern protocol forbids server-initiated JSON-RPC requests, so `send_raw_request` (inherited) refuses by construction. Change notifications (`notifications/*/list_changed`, `notifications/resources/updated`) are dropped with a debug log: at this era they reach a client only through a `subscriptions/listen` stream it opened, so a bare copy on the shared channel would be an unrequested notification. Publish them on the server's `SubscriptionBus` instead. """ def __init__(self, outbound: Outbound) -> None: self._outbound = outbound async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: # At the 2026-07-28 era these are `subscriptions/listen` stream goods # only: the spec forbids sending a change notification a subscription # did not request, and listen streams deliver them (stamped, filtered) # via the request-scoped outbound, never this connection-scoped channel. if method in LISTEN_STREAM_METHODS: logger.debug("dropped %s: delivered via subscriptions/listen at this era", method) return await self._outbound.notify(method, params, opts) class Connection: """Per-client connection state and standalone-stream `Outbound`. Construct via `from_envelope` (modern single-exchange: born ready, no back-channel) or `for_loop` (handshake-driven: ready once the client's `notifications/initialized` arrives). Either way `protocol_version` is populated at construction. """ outbound: Outbound """The connection-scoped channel for server-initiated messages.""" session_id: str | None client_capabilities: ClientCapabilities | None """The capabilities the peer declared: the handshake's on the loop path, the request envelope's on the modern path. `None` when none were declared. Kept in lockstep with `client_params` by its setter, and settable on its own for the modern envelope, where capabilities are required but client info is optional (spec PR #3002) - capability checks must not depend on the peer having identified itself.""" protocol_version: str """The protocol version this connection speaks. Populated at construction by the factory and overwritten by `_handle_initialize` once the handshake commits on the loop path.""" initialized: anyio.Event """Set when `notifications/initialized` arrives (matches TS `oninitialized`); the point from which the spec permits server-initiated requests beyond ping/logging. Pre-set on connections built via `from_envelope`.""" state: dict[str, Any] """Per-connection scratch state; persists across requests on this connection.""" exit_stack: AsyncExitStack """Per-connection teardown, unwound LIFO (shielded) when the connection closes. Push cleanup from handlers or middleware; exceptions are logged and swallowed.""" def __init__( self, outbound: Outbound, *, protocol_version: str, session_id: str | None = None, client_params: InitializeRequestParams | None = None, ) -> None: self.outbound = outbound self.protocol_version = protocol_version self.session_id = session_id self.client_capabilities = None self.client_params = client_params self.initialized = anyio.Event() self.state = {} self.exit_stack = AsyncExitStack() @property def client_params(self) -> InitializeRequestParams | None: """The full `initialize` request params, or the equivalent built from the 2026-era envelope. `None` when no client info was supplied.""" return self._client_params @client_params.setter def client_params(self, value: InitializeRequestParams | None) -> None: # Assignment is the sync point: recording full client params (the # handshake commit, or a modern envelope carrying client info) also # records the capabilities fact, so the two can never drift. Clearing # to `None` leaves `client_capabilities` alone - the modern envelope # declares capabilities without client info. self._client_params = value if value is not None: self.client_capabilities = value.capabilities @classmethod def from_envelope( cls, protocol_version: str, client_info: Any, client_capabilities: Any, *, outbound: Outbound = _NO_CHANNEL, ) -> Connection: """A born-ready connection populated from a request's `_meta` envelope. `protocol_version` must be an already-validated version string - the inbound classification ladder owns rejecting non-string or unsupported values. `client_info` and `client_capabilities` are the raw envelope values: this constructor owns turning them into connection identity, identically on every modern entry, so a mis-shaped value degrades to not-supplied rather than failing the request. `initialized` is set, well-formed capabilities are recorded as `client_capabilities` (client info is optional per spec PR #3002, so capability checks never depend on it), and the full `client_params` is additionally synthesized when client info was supplied too. `outbound` defaults to the no-channel sentinel for the single-exchange HTTP path; duplex modern transports (e.g. stdio) pass a notify-only wrapper around the dispatcher so server notifications ride the pipe while server-initiated requests stay refused. """ info = _typed(Implementation, client_info) capabilities = _typed(ClientCapabilities, client_capabilities) client_params = None if info is not None and capabilities is not None: client_params = InitializeRequestParams( protocol_version=protocol_version, capabilities=capabilities, client_info=info, ) connection = cls(outbound, protocol_version=protocol_version, client_params=client_params) connection.client_capabilities = capabilities connection.initialized.set() return connection @classmethod def for_loop( cls, outbound: Outbound, *, session_id: str | None = None, protocol_version_hint: str | None = None, ) -> Connection: """A connection for the handshake-driven loop path. Not born-ready: `initialized` is set later by the kernel when `notifications/initialized` arrives. `protocol_version` is seeded from the transport hint (or `LATEST_HANDSHAKE_VERSION`) so it's never `None`; the handshake overwrites it once negotiated. """ return cls( outbound, protocol_version=protocol_version_hint if protocol_version_hint is not None else LATEST_HANDSHAKE_VERSION, session_id=session_id, ) @property def has_standalone_channel(self) -> bool: """Whether this connection has a real back-channel for server-initiated messages. Derived from `outbound` - the no-channel sentinel is the only case that doesn't. Channel presence, not request permission: a modern (2026-07-28+) duplex connection has a channel that carries notifications while `send_raw_request` still refuses, because the protocol forbids server-initiated requests.""" return self.outbound is not _NO_CHANNEL @property def initialize_accepted(self) -> bool: """True once the inbound request gate is open: `initialize` recorded the peer info, or the handshake completed outright (born-ready, or a bare `notifications/initialized`). Derived, never stored.""" return self.client_params is not None or self.initialized.is_set() async def send_raw_request( self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None, ) -> dict[str, Any]: """Send a raw request on the standalone stream. Low-level `Outbound` channel. Prefer the typed `send_request` or the convenience methods below; use this directly only for off-spec messages. `opts` carries per-call `timeout` / `on_progress` / resumption hints; see `CallOptions`. Raises: MCPError: The peer responded with an error. NoBackChannelError: no back-channel for server-initiated requests - `has_standalone_channel` is `False`, or a modern (2026-07-28+) connection, where the protocol forbids them. """ return await self.outbound.send_raw_request(method, params, opts) @overload async def send_request( self, req: CreateMessageRequest, *, opts: CallOptions | None = None ) -> CreateMessageResult: ... @overload async def send_request(self, req: ElicitRequest, *, opts: CallOptions | None = None) -> ElicitResult: ... @overload async def send_request(self, req: ListRootsRequest, *, opts: CallOptions | None = None) -> ListRootsResult: ... @overload async def send_request(self, req: PingRequest, *, opts: CallOptions | None = None) -> EmptyResult: ... @overload async def send_request( self, req: Request[Any, Any], *, result_type: type[ResultT], opts: CallOptions | None = None ) -> ResultT: ... async def send_request( self, req: Request[Any, Any], *, result_type: type[BaseModel] | None = None, opts: CallOptions | None = None, ) -> BaseModel: """Send a typed server-to-client request and return its typed result. For spec request types the result type is inferred. For custom requests pass `result_type=` explicitly. Raises: MCPError: The peer responded with an error. NoBackChannelError: No back-channel for server-initiated requests. pydantic.ValidationError: The peer's result does not match the expected result type. KeyError: `result_type` omitted for a non-spec request type. """ raw = await self.send_raw_request(req.method, dump_params(req.params), opts) if req.method in _methods.MONOLITH_REQUESTS: try: _methods.validate_client_result(req.method, self.protocol_version, raw) except KeyError: pass cls = result_type if result_type is not None else _RESULT_FOR[type(req)] return cls.model_validate(raw, by_name=False) async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: """Send a best-effort notification on the standalone stream. Never raises. If there's no standalone channel or the stream is broken, the notification is dropped and debug-logged. """ try: await self.outbound.notify(method, params, opts) except (anyio.BrokenResourceError, anyio.ClosedResourceError): logger.debug("dropped %s: standalone stream closed", method) async def ping(self, *, meta: Meta | None = None, opts: CallOptions | None = None) -> None: """Send a `ping` request on the standalone stream. Raises: MCPError: The peer responded with an error. NoBackChannelError: no back-channel for server-initiated requests - `has_standalone_channel` is `False`, or a modern (2026-07-28+) connection, where the protocol forbids them. """ await self.send_raw_request("ping", dump_params(None, meta), opts) @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning) async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *, meta: Meta | None = None) -> None: """Send a `notifications/message` log entry on the standalone stream. Best-effort. On 2026-07-28+ connections this never sends: log delivery is a per-request opt-in that rides the requesting stream (`ctx.log`, `ctx.session.send_log_message`), and the standalone stream is forbidden from carrying `notifications/message`, so the entry is debug-logged and dropped. """ if level not in allowed_log_levels(self.protocol_version, None): _logger.debug("dropped notifications/message: no connection-wide log delivery at %s", self.protocol_version) return params: dict[str, Any] = {"level": level, "data": data} if logger is not None: params["logger"] = logger await self.notify("notifications/message", _notification_params(params, meta)) async def send_tool_list_changed(self, *, meta: Meta | None = None) -> None: await self.notify("notifications/tools/list_changed", _notification_params(None, meta)) async def send_prompt_list_changed(self, *, meta: Meta | None = None) -> None: await self.notify("notifications/prompts/list_changed", _notification_params(None, meta)) async def send_resource_list_changed(self, *, meta: Meta | None = None) -> None: await self.notify("notifications/resources/list_changed", _notification_params(None, meta)) async def send_resource_updated(self, uri: str, *, meta: Meta | None = None) -> None: await self.notify("notifications/resources/updated", _notification_params({"uri": uri}, meta)) def check_capability(self, capability: ClientCapabilities) -> bool: """Return whether the connected client declared the given capability. Returns `False` when no capabilities have been recorded. """ # TODO(L53): redesign - mirrors v1 ServerSession.check_client_capability # verbatim for parity. if self.client_capabilities is None: return False have = self.client_capabilities if capability.roots is not None: if have.roots is None: return False if capability.roots.list_changed and not have.roots.list_changed: return False if capability.sampling is not None: if have.sampling is None: return False if capability.sampling.context is not None and have.sampling.context is None: return False if capability.sampling.tools is not None and have.sampling.tools is None: return False if capability.elicitation is not None and have.elicitation is None: return False if capability.experimental is not None: if have.experimental is None: return False for k, v in capability.experimental.items(): if k not in have.experimental or have.experimental[k] != v: return False if capability.extensions is not None: # SEP-2133: an extension is supported when the client declares its # identifier. Settings are negotiated per-extension (the client may # advertise more than the server asks for), so presence - not value # equality - is the meaningful check. if have.extensions is None: return False for identifier in capability.extensions: if identifier not in have.extensions: return False return True