|
| 1 | +"""`Connection` — per-client connection state and the standalone outbound channel. |
| 2 | +
|
| 3 | +Always present on `Context` (never ``None``), even in stateless deployments. |
| 4 | +Holds peer info populated at ``initialize`` time, the per-connection lifespan |
| 5 | +output, and an `Outbound` for the standalone stream (the SSE GET stream in |
| 6 | +streamable HTTP, or the single duplex stream in stdio). |
| 7 | +
|
| 8 | +`notify` is best-effort: it never raises. If there's no standalone channel |
| 9 | +(stateless HTTP) or the stream has been dropped, the notification is |
| 10 | +debug-logged and silently discarded — server-initiated notifications are |
| 11 | +inherently advisory. `send_raw_request` *does* raise `NoBackChannelError` when |
| 12 | +there's no channel; `ping` is the only spec-sanctioned standalone request. |
| 13 | +""" |
| 14 | + |
| 15 | +import logging |
| 16 | +from collections.abc import Mapping |
| 17 | +from typing import Any |
| 18 | + |
| 19 | +import anyio |
| 20 | + |
| 21 | +from mcp.server._typed_request import TypedServerRequestMixin |
| 22 | +from mcp.shared.dispatcher import CallOptions, Outbound |
| 23 | +from mcp.shared.exceptions import NoBackChannelError |
| 24 | +from mcp.shared.peer import Meta, dump_params |
| 25 | +from mcp.types import ClientCapabilities, Implementation, LoggingLevel |
| 26 | + |
| 27 | +__all__ = ["Connection"] |
| 28 | + |
| 29 | +logger = logging.getLogger(__name__) |
| 30 | + |
| 31 | + |
| 32 | +def _notification_params(payload: dict[str, Any] | None, meta: Meta | None) -> dict[str, Any] | None: |
| 33 | + if not meta: |
| 34 | + return payload |
| 35 | + out = dict(payload or {}) |
| 36 | + out["_meta"] = meta |
| 37 | + return out |
| 38 | + |
| 39 | + |
| 40 | +class Connection(TypedServerRequestMixin): |
| 41 | + """Per-client connection state and standalone-stream `Outbound`. |
| 42 | +
|
| 43 | + Constructed by `ServerRunner` once per connection. The peer-info fields are |
| 44 | + ``None`` until ``initialize`` completes; ``initialized`` is set then. |
| 45 | + """ |
| 46 | + |
| 47 | + def __init__(self, outbound: Outbound, *, has_standalone_channel: bool) -> None: |
| 48 | + self._outbound = outbound |
| 49 | + self.has_standalone_channel = has_standalone_channel |
| 50 | + |
| 51 | + self.client_info: Implementation | None = None |
| 52 | + self.client_capabilities: ClientCapabilities | None = None |
| 53 | + self.protocol_version: str | None = None |
| 54 | + self.initialized: anyio.Event = anyio.Event() |
| 55 | + # TODO: make this generic (Connection[StateT]) once connection_lifespan |
| 56 | + # wiring lands in ServerRunner — see FOLLOWUPS.md. |
| 57 | + self.state: Any = None |
| 58 | + |
| 59 | + async def send_raw_request( |
| 60 | + self, |
| 61 | + method: str, |
| 62 | + params: Mapping[str, Any] | None, |
| 63 | + opts: CallOptions | None = None, |
| 64 | + ) -> dict[str, Any]: |
| 65 | + """Send a raw request on the standalone stream. |
| 66 | +
|
| 67 | + Low-level `Outbound` channel. Prefer the typed ``send_request`` (from |
| 68 | + `TypedServerRequestMixin`) or the convenience methods below; use this |
| 69 | + directly only for off-spec messages. |
| 70 | +
|
| 71 | + Raises: |
| 72 | + MCPError: The peer responded with an error. |
| 73 | + NoBackChannelError: ``has_standalone_channel`` is ``False``. |
| 74 | + """ |
| 75 | + if not self.has_standalone_channel: |
| 76 | + raise NoBackChannelError(method) |
| 77 | + return await self._outbound.send_raw_request(method, params, opts) |
| 78 | + |
| 79 | + async def notify(self, method: str, params: Mapping[str, Any] | None) -> None: |
| 80 | + """Send a best-effort notification on the standalone stream. |
| 81 | +
|
| 82 | + Never raises. If there's no standalone channel or the stream is broken, |
| 83 | + the notification is dropped and debug-logged. |
| 84 | + """ |
| 85 | + if not self.has_standalone_channel: |
| 86 | + logger.debug("dropped %s: no standalone channel", method) |
| 87 | + return |
| 88 | + try: |
| 89 | + await self._outbound.notify(method, params) |
| 90 | + except (anyio.BrokenResourceError, anyio.ClosedResourceError): |
| 91 | + logger.debug("dropped %s: standalone stream closed", method) |
| 92 | + |
| 93 | + async def ping(self, *, meta: Meta | None = None, opts: CallOptions | None = None) -> None: |
| 94 | + """Send a ``ping`` request on the standalone stream. |
| 95 | +
|
| 96 | + Raises: |
| 97 | + MCPError: The peer responded with an error. |
| 98 | + NoBackChannelError: ``has_standalone_channel`` is ``False``. |
| 99 | + """ |
| 100 | + await self.send_raw_request("ping", dump_params(None, meta), opts) |
| 101 | + |
| 102 | + async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *, meta: Meta | None = None) -> None: |
| 103 | + """Send a ``notifications/message`` log entry on the standalone stream. Best-effort.""" |
| 104 | + params: dict[str, Any] = {"level": level, "data": data} |
| 105 | + if logger is not None: |
| 106 | + params["logger"] = logger |
| 107 | + await self.notify("notifications/message", _notification_params(params, meta)) |
| 108 | + |
| 109 | + async def send_tool_list_changed(self, *, meta: Meta | None = None) -> None: |
| 110 | + await self.notify("notifications/tools/list_changed", _notification_params(None, meta)) |
| 111 | + |
| 112 | + async def send_prompt_list_changed(self, *, meta: Meta | None = None) -> None: |
| 113 | + await self.notify("notifications/prompts/list_changed", _notification_params(None, meta)) |
| 114 | + |
| 115 | + async def send_resource_list_changed(self, *, meta: Meta | None = None) -> None: |
| 116 | + await self.notify("notifications/resources/list_changed", _notification_params(None, meta)) |
| 117 | + |
| 118 | + async def send_resource_updated(self, uri: str, *, meta: Meta | None = None) -> None: |
| 119 | + await self.notify("notifications/resources/updated", _notification_params({"uri": uri}, meta)) |
| 120 | + |
| 121 | + def check_capability(self, capability: ClientCapabilities) -> bool: |
| 122 | + """Return whether the connected client declared the given capability. |
| 123 | +
|
| 124 | + Returns ``False`` if ``initialize`` hasn't completed yet. |
| 125 | + """ |
| 126 | + # TODO: redesign — mirrors v1 ServerSession.check_client_capability |
| 127 | + # verbatim for parity. See FOLLOWUPS.md. |
| 128 | + if self.client_capabilities is None: |
| 129 | + return False |
| 130 | + have = self.client_capabilities |
| 131 | + if capability.roots is not None: |
| 132 | + if have.roots is None: |
| 133 | + return False |
| 134 | + if capability.roots.list_changed and not have.roots.list_changed: |
| 135 | + return False |
| 136 | + if capability.sampling is not None and have.sampling is None: |
| 137 | + return False |
| 138 | + if capability.elicitation is not None and have.elicitation is None: |
| 139 | + return False |
| 140 | + if capability.experimental is not None: |
| 141 | + if have.experimental is None: |
| 142 | + return False |
| 143 | + for k in capability.experimental: |
| 144 | + if k not in have.experimental: |
| 145 | + return False |
| 146 | + return True |
0 commit comments