diff --git a/docs/advanced/events.md b/docs/advanced/events.md new file mode 100644 index 0000000000..40fe7ce84f --- /dev/null +++ b/docs/advanced/events.md @@ -0,0 +1,299 @@ +# Events + +[SEP-3415](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/3415) lets a server +tell clients that something happened: a new incident, an email, a merged PR. A client subscribes +to an **event type** with arguments, such as `incident.created` for the `billing` service. It +then receives each matching **event** by polling or at a webhook. + +The SDK ships this as the `Events` extension (`io.modelcontextprotocol/events`), with a server +side and a client side. If [Extensions](extensions.md) are new to you, skim that page first. + +!!! warning "Experimental: the wire format may change" + SEP-3415 is a draft. Events lives under `mcp.experimental.events`, and its API and wire + format may change in any release until the SEP is accepted. Push delivery (`events/stream`) + is not implemented yet. + +## Offering events + +```python title="server.py" +from pydantic import BaseModel + +from mcp.experimental.events.server import Events +from mcp.server.mcpserver import MCPServer + + +class IncidentFilter(BaseModel): + service: str + + +class Incident(BaseModel): + service: str + title: str + + +events = Events() +incidents = events.event( + "incident.created", + arguments=IncidentFilter, + payload=Incident, + description="A new incident was opened", +) + + +@incidents.match +def same_service(arguments: IncidentFilter, incident: Incident) -> bool: + return arguments.service == incident.service + + +mcp = MCPServer("pager", extensions=[events]) + + +@mcp.tool() +def open_incident(service: str, title: str) -> str: + """Open an incident.""" + incidents.emit(Incident(service=service, title=title)) + return "opened" +``` + +* `events.event(...)` declares an event type. `arguments` is the model subscribers pass, and + `payload` is the model each event carries. `events/list` publishes both as JSON Schema, and + arguments are validated before your code sees them. +* `emit` records an event. The server keeps the most recent events in a bounded buffer, so a + client that polls every few seconds misses nothing. +* `match` decides which subscribers receive an emitted event. Without it, everyone does. + Use `transform` the same way to reshape an event per subscriber, for example to redact fields. + +Use `NoArguments` for an event type that takes no arguments. + +### Reading from an upstream with a check function + +When the upstream can answer "what happened since this cursor", register a check function +instead of calling `emit`: + +```python title="server.py" +from datetime import timedelta + +from pydantic import BaseModel + +from mcp.experimental.events.server import EventBatch, EventQuery, Events, Occurrence +from mcp.server.mcpserver import MCPServer + + +class Repository(BaseModel): + name: str + + +class Push(BaseModel): + sha: str + + +events = Events() +pushes = events.event("repo.pushed", arguments=Repository, payload=Push) + + +@pushes.check +async def check(query: EventQuery[Repository]) -> EventBatch[Push]: + commits, position = await upstream_commits(query.arguments.name, after=query.cursor) + return EventBatch( + cursor=position, + events=[Occurrence(Push(sha=sha), id=sha) for sha in commits], + next_poll=timedelta(seconds=30), + ) + + +async def upstream_commits(repository: str, after: str | None) -> tuple[list[str], str]: + return [], "head" + + +mcp = MCPServer("git", extensions=[events]) +``` + +The cursor is yours: an offset, a history id, a timestamp. A `None` cursor means "start from +now", so the SDK returns your cursor with no events. Set the occurrence `id` to the upstream's +own identifier, because clients deduplicate on it. + +Return `truncated=True` when the events start later than the cursor, for example because it +fell out of the upstream's retention window. Clients use it to know they missed something. + +### Subscription hooks + +```python title="server.py" +from pydantic import BaseModel + +from mcp.experimental.events.server import Events, Subscriber +from mcp.experimental.events.types import FORBIDDEN, events_error +from mcp.server.mcpserver import MCPServer + + +class IncidentFilter(BaseModel): + service: str + + +class Incident(BaseModel): + service: str + title: str + + +events = Events() +incidents = events.event("incident.created", arguments=IncidentFilter, payload=Incident) +watched: dict[str, int] = {} + + +@incidents.on_subscribe +async def start_watching(subscriber: Subscriber[IncidentFilter]) -> None: + if subscriber.principal is None: + raise events_error(FORBIDDEN, reason="sign in to watch incidents") + service = subscriber.arguments.service + watched[service] = watched.get(service, 0) + 1 + + +@incidents.on_unsubscribe +async def stop_watching(subscriber: Subscriber[IncidentFilter]) -> None: + watched[subscriber.arguments.service] -= 1 + + +mcp = MCPServer("pager", extensions=[events]) +``` + +Hooks let you set up and tear down upstream work per subscription. `on_subscribe` is also where +you check access: the subscriber carries the caller's `principal`, and raising an `MCPError` +rejects the poll or subscription with that error. + +A poll client never says +goodbye, so the SDK leases poll subscriptions: `on_subscribe` runs on the first poll for a +principal, event type, and arguments, and `on_unsubscribe` runs once no poll renewed the lease +for `poll_lease` (three poll intervals by default). + +## Webhook delivery + +```python +from mcp.experimental.events.server import Events, Webhooks + +events = Events(webhooks=Webhooks()) +``` + +Passing `Webhooks` offers webhook delivery for every event type. The client gives a callback URL +and a secret, and the server POSTs each event there, signed with +[Standard Webhooks](https://www.standardwebhooks.com/). It retries failed deliveries, and reports +failures to the client when it refreshes the subscription. + +!!! warning "Webhooks need an authenticated principal" + A subscription is identified by who made it, the callback URL, the event type, and the + arguments. Without a principal anyone could cancel or hijack another caller's subscription, + so the server refuses webhook calls with `Forbidden` unless the caller is authenticated. + By default the principal comes from the request's OAuth access token. Pass + `Events(principal=...)` to derive it another way. + +Callback URLs come from clients, so the server treats them as hostile: + +* Only `https` URLs are accepted. +* The host is resolved before every request, and refused unless every address is public. + The server connects to the address it checked, so a DNS answer cannot change in between. +* Redirects are never followed. +* Before the first delivery, the endpoint must echo a random challenge, so a server cannot be + used to POST at a URL that never asked for events. + +For a receiver on your machine during development, allow its origin explicitly: + +```python +from mcp.experimental.events.server import Events, Webhooks + +events = Events(webhooks=Webhooks(trusted_origins=["http://127.0.0.1:9000"])) +``` + +Subscriptions live in memory. The server caps each one at `max_ttl` (one hour by default), and +clients refresh before it lapses. After a restart, clients subscribe again. + +## Consuming events + +### Polling + +```python title="client.py" +import anyio + +from mcp.client import Client +from mcp.experimental.events.client import Events + + +async def main() -> None: + async with Client("http://localhost:8000/mcp", extensions=[Events()]) as client: + events = client.extension(Events) + async for batch in events.watch("incident.created", {"service": "billing"}): + for event in batch.events: + print(event.event_id, event.data) + + +anyio.run(main) +``` + +`watch` polls in a loop. It waits as long as the server asks (`nextPollMs`), never less than +`Events(min_poll_interval=...)`, and polls again at once while the server has more. It yields +every batch, including empty ones, because an empty batch still moves the cursor. A batch with +`batch.truncated` set means events were skipped. + +To resume after a restart, save `batch.cursor` once you handled a batch, and pass it back with +`watch(..., cursor=saved)`. Delivery is at least once, so deduplicate on `event_id`. + +For a single read, use `events.poll(...)`. + +### Webhooks + +```python title="client.py" +import base64 +import secrets + +import anyio +import uvicorn +from starlette.applications import Starlette +from starlette.routing import Mount + +from mcp.client import Client +from mcp.experimental.events.client import Events +from mcp.experimental.events.receiver import Delivery, WebhookReceiver + +SECRET = "whsec_" + base64.b64encode(secrets.token_bytes(32)).decode() + + +async def handle(delivery: Delivery) -> None: + print(delivery) + + +receiver = WebhookReceiver(SECRET, handle) +app = Starlette(routes=[Mount("/hooks/events", receiver)]) + + +async def main() -> None: + config = uvicorn.Config(app, port=9000) + async with anyio.create_task_group() as task_group: + task_group.start_soon(uvicorn.Server(config).serve) + async with Client("https://pager.example.com/mcp", extensions=[Events()]) as client: + events = client.extension(Events) + async with events.subscribe( + "incident.created", + {"service": "billing"}, + url="https://hooks.example.com/hooks/events", + secret=SECRET, + ): + await anyio.sleep_forever() + + +anyio.run(main) +``` + +`subscribe` holds the subscription for the duration of the block. It refreshes it before it +expires, passing the latest cursor so a server that lost the subscription replays from there, +and unsubscribes on exit. + +`WebhookReceiver` is an ASGI app for the callback URL. It answers the verification challenge, +rejects requests that are unsigned, signed with another secret, or more than five minutes old, +and passes each delivery to your handler: + +| Delivery | Meaning | +| --- | --- | +| `EventOccurrence` | An event. `cursor` is the subscription's position after it. | +| `WebhookGap` | Events were skipped. Reconcile if you need every event. | +| `WebhookTerminated` | The server ended the subscription, for example because access was revoked. | + +The receiver drops repeated `event_id`s, since servers retry. A secret is `whsec_` followed by +the base64 of 24 to 64 random bytes, as `SECRET` shows. Store it, since the receiver needs the +same secret after a restart. diff --git a/docs/advanced/extensions.md b/docs/advanced/extensions.md index e0ba0669d6..49489acea1 100644 --- a/docs/advanced/extensions.md +++ b/docs/advanced/extensions.md @@ -133,6 +133,43 @@ The client is its own program, and it carries both halves of the client story: so the client declares them for itself. A published extension would ship them in a package that both sides import. +### Running background work + +Override `lifespan()` when the extension needs tasks that outlive a request, such as +delivery loops or lease timers: + +```python +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +import anyio + +from mcp.server.extension import Extension +from mcp.server.mcpserver import MCPServer + + +class Heartbeat(Extension): + identifier = "com.example/heartbeat" + + @asynccontextmanager + async def lifespan(self) -> AsyncIterator[None]: + async with anyio.create_task_group() as task_group: + task_group.start_soon(self._beat) + yield + task_group.cancel_scope.cancel() + + async def _beat(self) -> None: + while True: + await anyio.sleep(30) + + +mcp = MCPServer("demo", extensions=[Heartbeat()]) +``` + +`MCPServer` enters every extension's `lifespan()` before its own lifespan and exits them +after it, so your tasks run whenever the server does. Cancel them after the `yield`, or +shutdown waits for them. + ### Intercepting `tools/call` The one interceptive hook. Override `intercept_tool_call` to observe, short-circuit, diff --git a/mkdocs.yml b/mkdocs.yml index 62f0640366..afae636a70 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -70,6 +70,7 @@ nav: - Extensions: advanced/extensions.md - MCP Apps: advanced/apps.md - Skills: advanced/skills.md + - Events: advanced/events.md - Troubleshooting: troubleshooting.md - Translations: translations.md - Migration Guide: migration.md diff --git a/src/mcp/experimental/__init__.py b/src/mcp/experimental/__init__.py new file mode 100644 index 0000000000..798fd3158d --- /dev/null +++ b/src/mcp/experimental/__init__.py @@ -0,0 +1,3 @@ +"""Experimental extensions whose specifications are still drafts; their APIs may change in any release.""" + +__all__: list[str] = [] diff --git a/src/mcp/experimental/events/__init__.py b/src/mcp/experimental/events/__init__.py new file mode 100644 index 0000000000..e8469a17c2 --- /dev/null +++ b/src/mcp/experimental/events/__init__.py @@ -0,0 +1,6 @@ +"""MCP Events (`io.modelcontextprotocol/events`, SEP-3415), server and client sides. + +See https://github.com/modelcontextprotocol/modelcontextprotocol/pull/3415. +""" + +__all__: list[str] = [] diff --git a/src/mcp/experimental/events/_buffer.py b/src/mcp/experimental/events/_buffer.py new file mode 100644 index 0000000000..1cf7146e09 --- /dev/null +++ b/src/mcp/experimental/events/_buffer.py @@ -0,0 +1,82 @@ +"""The in-memory ring buffer that serves event types fed by `EventType.emit`.""" + +from __future__ import annotations + +import secrets +from collections import deque +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from typing import Generic, TypeVar + +import anyio +from pydantic import BaseModel + +from mcp.experimental.events.event_type import EventBatch, Occurrence + +PayloadT = TypeVar("PayloadT", bound=BaseModel) + + +class EmitBuffer(Generic[PayloadT]): + """Recent emitted events, addressed by process-local cursors. + + A cursor is `.`. The epoch is random per process, so a cursor + from before a restart reads as unknown, and the reader starts again from now + with `truncated` set, as SEP-3415 prescribes for emit-only event types. + """ + + def __init__(self, max_events: int, max_age: timedelta) -> None: + self._epoch = secrets.token_hex(4) + self._entries: deque[tuple[int, Occurrence[PayloadT]]] = deque(maxlen=max_events) + self._max_age = max_age + self._head = 0 + self._appended = anyio.Event() + + def append(self, occurrence: Occurrence[PayloadT]) -> None: + self._head += 1 + self._entries.append((self._head, occurrence)) + self._appended.set() + self._appended = anyio.Event() + + async def wait(self, timeout: float) -> None: + """Return once an event is appended, or after `timeout` seconds.""" + with anyio.move_on_after(timeout): + await self._appended.wait() + + def read( + self, + cursor: str | None, + *, + max_age: timedelta | None, + max_events: int | None, + select: Callable[[PayloadT], PayloadT | None], + ) -> EventBatch[PayloadT]: + now = datetime.now(UTC) + while self._entries and self._entries[0][1].timestamp < now - self._max_age: + self._entries.popleft() + position = self._position(cursor) + if position is None: + return EventBatch(cursor=f"{self._epoch}.{self._head}", truncated=cursor is not None) + oldest = self._entries[0][0] if self._entries else self._head + 1 + truncated = position < oldest - 1 + floor = now - max_age if max_age is not None else None + events: list[Occurrence[PayloadT]] = [] + has_more = False + for sequence, occurrence in self._entries: + if sequence <= position: + continue + if max_events is not None and len(events) == max_events: + has_more = True + break + position = sequence + if floor is not None and occurrence.timestamp < floor: + truncated = True + continue + if (data := select(occurrence.data)) is not None: + events.append(Occurrence(data=data, id=occurrence.id, timestamp=occurrence.timestamp)) + return EventBatch(cursor=f"{self._epoch}.{position}", events=events, truncated=truncated, has_more=has_more) + + def _position(self, cursor: str | None) -> int | None: + epoch, _, sequence = (cursor or "").partition(".") + if epoch != self._epoch or not sequence.isdigit() or int(sequence) > self._head: + return None + return int(sequence) diff --git a/src/mcp/experimental/events/_callbacks.py b/src/mcp/experimental/events/_callbacks.py new file mode 100644 index 0000000000..b898c9cd63 --- /dev/null +++ b/src/mcp/experimental/events/_callbacks.py @@ -0,0 +1,132 @@ +"""The outbound side of webhook delivery: signed POSTs to client-supplied callback URLs. + +Callback URLs come from clients, so every POST goes through the SSRF rules of SEP-3415: +`https` only, the host resolved at send time and refused unless every address is +public, the connection made to the address that was checked (with the original +`Host` and TLS SNI), and redirects never followed. +""" + +from __future__ import annotations + +import hmac +import ipaddress +import json +import secrets +import socket +import ssl +import time +from collections.abc import Collection, Sequence +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +import anyio +import httpx2 + +from mcp.experimental.events.signing import sign +from mcp.experimental.events.types import DeliveryError + + +class CallbackFailure(Exception): + """An attempt to reach a callback URL failed, for the reported reason.""" + + def __init__(self, reason: DeliveryError) -> None: + super().__init__(reason) + self.reason: DeliveryError = reason + + +class NonPublicAddress(ValueError): + """A callback URL resolved to an address that is not globally routable.""" + + +class Callbacks: + def __init__(self, client: httpx2.AsyncClient, trusted_origins: Collection[str]) -> None: + self._client = client + self._trusted = {origin.rstrip("/").lower() for origin in trusted_origins} + + def validate(self, url: str) -> None: + """Raises `ValueError` unless `url` is an absolute `https` URL or under a trusted origin.""" + parts = urlsplit(url) + if not parts.hostname: + raise ValueError("delivery.url must be an absolute URL") + if parts.scheme != "https" and not self._is_trusted(url): + raise ValueError("delivery.url must use https") + + async def post( + self, url: str, subscription_id: str, secrets: Sequence[str], message_id: str, body: dict[str, Any] + ) -> httpx2.Response: + """Sign and POST `body` once. + + Raises: + NonPublicAddress: The host resolved to a non-public address. + CallbackFailure: The endpoint could not be reached. + """ + content = json.dumps(body, separators=(",", ":")).encode() + timestamp = int(time.time()) + headers = { + "content-type": "application/json", + "webhook-id": message_id, + "webhook-timestamp": str(timestamp), + "webhook-signature": " ".join(f"v1,{sign(secret, message_id, timestamp, content)}" for secret in secrets), + "x-mcp-subscription-id": subscription_id, + } + extensions: dict[str, Any] = {} + if not self._is_trusted(url): + parts = urlsplit(url) + address = await _public_address(parts.hostname or "", parts.port or 443) + host = f"[{address}]" if ":" in address else address + url = urlunsplit(parts._replace(netloc=f"{host}:{parts.port or 443}")) + headers["host"] = parts.netloc + extensions["sni_hostname"] = parts.hostname + try: + return await self._client.post( + url, content=content, headers=headers, extensions=extensions, follow_redirects=False + ) + except httpx2.TimeoutException: + raise CallbackFailure("timeout") from None + except httpx2.TransportError as exc: + raise CallbackFailure("tls_error" if _caused_by_tls(exc) else "connection_refused") from None + + async def verify(self, url: str, subscription_id: str, secret: str) -> None: + """Prove the endpoint wants deliveries: it must echo a fresh challenge in a 2xx body. + + Raises: + NonPublicAddress: The host resolved to a non-public address. + CallbackFailure: The endpoint was unreachable or did not echo the challenge. + """ + challenge = secrets.token_urlsafe(32) + message_id = f"msg_verification_{secrets.token_hex(12)}" + response = await self.post( + url, subscription_id, [secret], message_id, {"type": "verification", "challenge": challenge} + ) + if not response.is_success: + raise CallbackFailure("http_5xx" if response.status_code >= 500 else "http_4xx") + try: + echoed = response.json().get("challenge") + except (ValueError, AttributeError): + echoed = None + if not isinstance(echoed, str) or not hmac.compare_digest(echoed, challenge): + raise CallbackFailure("challenge_failed") + + def _is_trusted(self, url: str) -> bool: + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}".lower() in self._trusted + + +async def _public_address(host: str, port: int) -> str: + try: + infos = await anyio.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except OSError: + raise CallbackFailure("connection_refused") from None + addresses = [str(info[4][0]) for info in infos] + if not all(ipaddress.ip_address(address).is_global for address in addresses): + raise NonPublicAddress(f"{host} resolves to a non-public address") + return addresses[0] + + +def _caused_by_tls(exc: BaseException) -> bool: + cause: BaseException | None = exc + while cause is not None: + if isinstance(cause, ssl.SSLError): + return True + cause = cause.__cause__ or cause.__context__ + return False diff --git a/src/mcp/experimental/events/_locks.py b/src/mcp/experimental/events/_locks.py new file mode 100644 index 0000000000..23bd0db573 --- /dev/null +++ b/src/mcp/experimental/events/_locks.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +import anyio + + +class KeyedLocks: + """One lock per key, dropped once nobody holds or awaits it.""" + + def __init__(self) -> None: + self._locks: dict[str, anyio.Lock] = {} + + @asynccontextmanager + async def __call__(self, key: str) -> AsyncIterator[None]: + lock = self._locks.setdefault(key, anyio.Lock()) + try: + async with lock: + yield + finally: + if not lock.locked() and not lock.statistics().tasks_waiting: + del self._locks[key] diff --git a/src/mcp/experimental/events/client.py b/src/mcp/experimental/events/client.py new file mode 100644 index 0000000000..f8eb0468f2 --- /dev/null +++ b/src/mcp/experimental/events/client.py @@ -0,0 +1,164 @@ +"""The client side of MCP Events (SEP-3415): `Events`, a `Client` extension. + +```python +async with Client(url, extensions=[Events()]) as client: + events = client.extension(Events) + async for batch in events.watch("incident.created", {"service": "billing"}, cursor=load_cursor()): + for event in batch.events: + handle(event) + save_cursor(batch.cursor) +``` +""" + +from __future__ import annotations + +from collections.abc import AsyncGenerator, AsyncIterator +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta +from typing import Any + +import anyio + +from mcp.client.extension import ClientExtension +from mcp.client.session import ClientSession +from mcp.experimental.events.types import ( + EXTENSION_ID, + EventDescriptor, + ListEventsParams, + ListEventsRequest, + ListEventsResult, + PollEventsParams, + PollEventsRequest, + PollEventsResult, + SubscribeEventsParams, + SubscribeEventsRequest, + SubscribeEventsResult, + UnsubscribeEventsParams, + UnsubscribeEventsRequest, + UnsubscribeEventsResult, + WebhookDelivery, + WebhookTarget, +) + +__all__ = ["Events", "EventsClient"] + + +class Events(ClientExtension): + """Register with `Client(extensions=[Events()])`, then use `client.extension(Events)`. + + Args: + min_poll_interval: The shortest wait between polls, whatever the server asks for, + so a misbehaving server cannot drive a tight loop. + """ + + identifier = EXTENSION_ID + + def __init__(self, *, min_poll_interval: timedelta = timedelta(seconds=1)) -> None: + self._min_poll_interval = min_poll_interval + + def bind(self, session: ClientSession) -> EventsClient: + """Return the events verbs bound to `session`.""" + return EventsClient(session, self._min_poll_interval) + + +class EventsClient: + """The events verbs bound to one connected session. Obtain it from `client.extension(Events)`.""" + + def __init__(self, session: ClientSession, min_poll_interval: timedelta) -> None: + self._session = session + self._min_poll_interval = min_poll_interval + + async def list(self) -> list[EventDescriptor]: + """Return every event type the server offers, following pagination.""" + descriptors: list[EventDescriptor] = [] + cursor: str | None = None + while True: + request = ListEventsRequest(params=ListEventsParams(cursor=cursor)) + result = await self._session.send_request(request, ListEventsResult) + descriptors.extend(result.events) + if (cursor := result.next_cursor) is None: + return descriptors + + async def poll( + self, + name: str, + arguments: dict[str, Any] | None = None, + *, + cursor: str | None = None, + max_age: timedelta | None = None, + max_events: int | None = None, + ) -> PollEventsResult: + """Read the events after `cursor` once. A `None` cursor returns no events, only the current position.""" + params = PollEventsParams( + name=name, + arguments=arguments, + cursor=cursor, + max_age_ms=int(max_age.total_seconds() * 1000) if max_age is not None else None, + max_events=max_events, + ) + return await self._session.send_request(PollEventsRequest(params=params), PollEventsResult) + + async def watch( + self, name: str, arguments: dict[str, Any] | None = None, *, cursor: str | None = None + ) -> AsyncGenerator[PollEventsResult, None]: + """Poll forever, yielding every batch, including empty ones. + + Persist `batch.cursor` after handling a batch, and pass it back after a restart + to resume: delivery is at-least-once, so deduplicate on `event_id`. Empty batches + still move the cursor, which keeps it inside the upstream's retention window. + """ + if cursor is None: + cursor = (await self.poll(name, arguments)).cursor + while True: + batch = await self.poll(name, arguments, cursor=cursor) + cursor = batch.cursor + yield batch + if not batch.has_more: + wait = max(timedelta(milliseconds=batch.next_poll_ms), self._min_poll_interval) + await anyio.sleep(wait.total_seconds()) + + @asynccontextmanager + async def subscribe( + self, + name: str, + arguments: dict[str, Any] | None = None, + *, + url: str, + secret: str, + cursor: str | None = None, + ttl: timedelta | None = None, + ) -> AsyncIterator[SubscribeEventsResult]: + """Hold a webhook subscription: events are POSTed to `url`, signed with `secret`. + + The subscription is refreshed in the background before it expires, and removed on exit. + Receive the deliveries with `mcp.experimental.events.receiver.WebhookReceiver`. + + Raises: + MCPError: The server refused the subscription, e.g. because `url` failed verification. + """ + params = SubscribeEventsParams( + name=name, + arguments=arguments, + delivery=WebhookDelivery(url=url, secret=secret), + cursor=cursor, + ttl_ms=int(ttl.total_seconds() * 1000) if ttl is not None else None, + ) + result = await self._session.send_request(SubscribeEventsRequest(params=params), SubscribeEventsResult) + async with anyio.create_task_group() as task_group: + task_group.start_soon(self._refresh, params, result) + try: + yield result + finally: + task_group.cancel_scope.cancel() + unsubscribe = UnsubscribeEventsParams(name=name, arguments=arguments, delivery=WebhookTarget(url=url)) + with anyio.CancelScope(shield=True): + request = UnsubscribeEventsRequest(params=unsubscribe) + await self._session.send_request(request, UnsubscribeEventsResult) + + async def _refresh(self, params: SubscribeEventsParams, result: SubscribeEventsResult) -> None: + while result.refresh_before is not None: + # Refresh at two thirds of the remaining lifetime, leaving room for a failed attempt. + await anyio.sleep((result.refresh_before - datetime.now(UTC)).total_seconds() * 2 / 3) + # The latest safe cursor is a no-op for a live subscription, and the replay point if the server lost it. + request = SubscribeEventsRequest(params=params.model_copy(update={"cursor": result.cursor})) + result = await self._session.send_request(request, SubscribeEventsResult) diff --git a/src/mcp/experimental/events/event_type.py b/src/mcp/experimental/events/event_type.py new file mode 100644 index 0000000000..3804bd1f08 --- /dev/null +++ b/src/mcp/experimental/events/event_type.py @@ -0,0 +1,185 @@ +"""Event types: what a server offers, and the value objects its authors work with.""" + +from __future__ import annotations + +import secrets +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass, field, replace +from datetime import UTC, datetime, timedelta +from functools import partial +from typing import TYPE_CHECKING, Generic, Literal, TypeAlias, TypeVar + +from pydantic import BaseModel, ConfigDict + +from mcp.experimental.events.types import EventDescriptor, EventOccurrence + +if TYPE_CHECKING: + from mcp.experimental.events._buffer import EmitBuffer + +__all__ = ["EventBatch", "EventQuery", "EventType", "NoArguments", "Occurrence", "Subscriber"] + +ArgsT = TypeVar("ArgsT", bound=BaseModel) +PayloadT = TypeVar("PayloadT", bound=BaseModel) + + +class NoArguments(BaseModel): + """The arguments of an event type that takes none; any argument is rejected.""" + + model_config = ConfigDict(extra="forbid") + + +@dataclass(frozen=True) +class Occurrence(Generic[PayloadT]): + """One event. `id` should be the upstream's stable identifier, so clients can deduplicate.""" + + data: PayloadT + id: str = field(default_factory=partial(secrets.token_hex, 16)) + timestamp: datetime = field(default_factory=partial(datetime.now, UTC)) + + +@dataclass(frozen=True) +class EventQuery(Generic[ArgsT]): + """What a check function is asked: the events after `cursor` that match `arguments`. + + A `None` cursor means "start from now": return the current position and no events. + """ + + arguments: ArgsT + cursor: str | None + principal: str | None = None + max_age: timedelta | None = None + max_events: int | None = None + + +@dataclass(frozen=True) +class EventBatch(Generic[PayloadT]): + """What a check function answers: the events found, and the cursor after them. + + Return `cursor=None` for an event type that cannot replay. Set `truncated` when the + events start later than the query's cursor, and `has_more` when you stopped at + `max_events`. `next_poll` overrides the server's poll interval for this answer. + """ + + cursor: str | None + events: Sequence[Occurrence[PayloadT]] = () + truncated: bool = False + has_more: bool = False + next_poll: timedelta | None = None + + +@dataclass(frozen=True) +class Subscriber(Generic[ArgsT]): + """A subscription as the lifecycle hooks see it.""" + + id: str + arguments: ArgsT + principal: str | None + mode: Literal["poll", "webhook"] + + +CheckFn: TypeAlias = Callable[[EventQuery[ArgsT]], Awaitable[EventBatch[PayloadT]]] +MatchFn: TypeAlias = Callable[[ArgsT, PayloadT], bool] +TransformFn: TypeAlias = Callable[[ArgsT, PayloadT], PayloadT] +HookFn: TypeAlias = Callable[[Subscriber[ArgsT]], Awaitable[None]] + + +class EventType(Generic[ArgsT, PayloadT]): + """An event type a server offers. Create it with `Events.event`. + + Feed it in one of two ways. Register a `check` function when the upstream can answer + "what happened since this cursor" (Gmail history, a Kafka offset, Stripe `/v1/events`). + Otherwise call `emit` as events arrive, and the server keeps a bounded buffer for + readers, filtered per subscriber by `match` and shaped by `transform`. + """ + + def __init__( + self, + name: str, + *, + arguments: type[ArgsT], + payload: type[PayloadT], + description: str | None, + buffer: EmitBuffer[PayloadT], + ) -> None: + self.name = name + self.arguments = arguments + self.payload = payload + self.description = description + self.buffer = buffer + self._check: CheckFn[ArgsT, PayloadT] | None = None + self._match: MatchFn[ArgsT, PayloadT] | None = None + self._transform: TransformFn[ArgsT, PayloadT] | None = None + self.subscribe_hooks: list[HookFn[ArgsT]] = [] + self.unsubscribe_hooks: list[HookFn[ArgsT]] = [] + + def check(self, fn: CheckFn[ArgsT, PayloadT]) -> CheckFn[ArgsT, PayloadT]: + """Register the function that answers every read of this event type.""" + self._check = fn + return fn + + def match(self, fn: MatchFn[ArgsT, PayloadT]) -> MatchFn[ArgsT, PayloadT]: + """Register who receives an emitted event: `fn(arguments, payload)`. Default: everyone.""" + self._match = fn + return fn + + def transform(self, fn: TransformFn[ArgsT, PayloadT]) -> TransformFn[ArgsT, PayloadT]: + """Register how an emitted event is shaped per subscriber: `fn(arguments, payload)`.""" + self._transform = fn + return fn + + def on_subscribe(self, fn: HookFn[ArgsT]) -> HookFn[ArgsT]: + """Register a hook run when a subscription starts, to set up upstream listeners.""" + self.subscribe_hooks.append(fn) + return fn + + def on_unsubscribe(self, fn: HookFn[ArgsT]) -> HookFn[ArgsT]: + """Register a hook run when a subscription ends or lapses, to tear upstream listeners down.""" + self.unsubscribe_hooks.append(fn) + return fn + + def emit(self, event: PayloadT | Occurrence[PayloadT]) -> None: + """Record an event for every current and future reader of this event type. + + Pass an `Occurrence` to set the event's id (for client-side deduplication) or timestamp. + + Raises: + RuntimeError: The event type has a check function, which owns its events. + """ + if self._check is not None: + raise RuntimeError(f"event type {self.name!r} has a check function; it cannot also emit") + self.buffer.append(event if isinstance(event, Occurrence) else Occurrence(data=event)) + + async def read(self, query: EventQuery[ArgsT]) -> EventBatch[PayloadT]: + if self._check is None: + return self.buffer.read( + query.cursor, + max_age=query.max_age, + max_events=query.max_events, + select=lambda payload: self._select(query.arguments, payload), + ) + batch = await self._check(query) + # A null cursor means "start from now", whatever the check function returned. + return replace(batch, events=()) if query.cursor is None else batch + + def descriptor(self, delivery: list[Literal["poll", "push", "webhook"]]) -> EventDescriptor: + return EventDescriptor( + name=self.name, + description=self.description, + delivery=delivery, + input_schema=self.arguments.model_json_schema(), + payload_schema=self.payload.model_json_schema(), + ) + + def wire(self, occurrence: Occurrence[PayloadT], cursor: str | None = None) -> EventOccurrence: + return EventOccurrence( + event_id=occurrence.id, + name=self.name, + timestamp=occurrence.timestamp, + data=occurrence.data.model_dump(mode="json", by_alias=True), + cursor=cursor, + ) + + def _select(self, arguments: ArgsT, payload: PayloadT) -> PayloadT | None: + if self._match is not None and not self._match(arguments, payload): + return None + return self._transform(arguments, payload) if self._transform is not None else payload diff --git a/src/mcp/experimental/events/receiver.py b/src/mcp/experimental/events/receiver.py new file mode 100644 index 0000000000..e7be664799 --- /dev/null +++ b/src/mcp/experimental/events/receiver.py @@ -0,0 +1,112 @@ +"""Receive SEP-3415 webhook deliveries: an ASGI app that verifies, decodes, and hands them over. + +```python +async def handle(delivery: EventOccurrence | WebhookGap | WebhookTerminated) -> None: + ... + +receiver = WebhookReceiver(secret, handle) +app = Starlette(routes=[Mount("/hooks/events", receiver)]) +``` +""" + +from __future__ import annotations + +import json +import time +from collections import OrderedDict +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import timedelta +from typing import Any + +from pydantic import ValidationError +from starlette.requests import Request +from starlette.responses import JSONResponse, Response +from starlette.types import Receive, Scope, Send + +from mcp.experimental.events.signing import verify +from mcp.experimental.events.types import EventOccurrence + +__all__ = ["WebhookGap", "WebhookReceiver", "WebhookTerminated"] + + +@dataclass(frozen=True) +class WebhookGap: + """Events were skipped; resume from `cursor`, and reconcile if the gap matters.""" + + subscription_id: str + cursor: str | None + + +@dataclass(frozen=True) +class WebhookTerminated: + """The server ended the subscription, e.g. after access was revoked; resubscribing may fail the same way.""" + + subscription_id: str + error: dict[str, Any] + + +Delivery = EventOccurrence | WebhookGap | WebhookTerminated + + +class WebhookReceiver: + """ASGI app for the callback URL of webhook subscriptions made with one `secret`. + + Requests that are unsigned, signed with another secret, or older than `tolerance` + get a 401 and never reach `handler`. Events are deduplicated on their subscription and + `event_id` across the last `dedupe_window` events, since servers retry deliveries. + """ + + def __init__( + self, + secret: str, + handler: Callable[[Delivery], Awaitable[None]], + *, + tolerance: timedelta = timedelta(minutes=5), + dedupe_window: int = 1000, + ) -> None: + self._secret = secret + self._handler = handler + self._tolerance = tolerance + self._dedupe_window = dedupe_window + self._seen: OrderedDict[tuple[str, str], None] = OrderedDict() + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + request = Request(scope, receive) + response = await self._handle(request) if request.method == "POST" else Response(status_code=405) + await response(scope, receive, send) + + async def _handle(self, request: Request) -> Response: + body = await request.body() + message_id = request.headers.get("webhook-id", "") + timestamp = request.headers.get("webhook-timestamp", "") + signature = request.headers.get("webhook-signature", "") + if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > self._tolerance.total_seconds(): + return Response(status_code=401) + if not verify(self._secret, message_id, int(timestamp), body, signature): + return Response(status_code=401) + subscription_id = request.headers.get("x-mcp-subscription-id", "") + delivery: Delivery + try: + payload = json.loads(body) + match payload.get("type"): + case "verification": + return JSONResponse({"challenge": str(payload["challenge"])}) + case "gap": + delivery = WebhookGap(subscription_id, payload.get("cursor")) + case "terminated": + delivery = WebhookTerminated(subscription_id, dict(payload["error"])) + case _: + delivery = EventOccurrence.model_validate(payload) + except (ValueError, TypeError, KeyError, AttributeError, ValidationError): + return Response(status_code=400) + # One upstream event can match several subscriptions, and each must receive it. + seen_key = (subscription_id, delivery.event_id) if isinstance(delivery, EventOccurrence) else None + if seen_key in self._seen: + return Response(status_code=204) + await self._handler(delivery) + if seen_key is not None: + self._seen[seen_key] = None + if len(self._seen) > self._dedupe_window: + self._seen.popitem(last=False) + return Response(status_code=204) diff --git a/src/mcp/experimental/events/server.py b/src/mcp/experimental/events/server.py new file mode 100644 index 0000000000..7b3127cacb --- /dev/null +++ b/src/mcp/experimental/events/server.py @@ -0,0 +1,245 @@ +"""The server side of MCP Events (SEP-3415): `Events`, an `MCPServer` extension. + +```python +events = Events(webhooks=Webhooks()) +incidents = events.event("incident.created", arguments=IncidentFilter, payload=Incident) + +@incidents.match +def same_service(arguments: IncidentFilter, incident: Incident) -> bool: + return arguments.service == incident.service + +mcp = MCPServer("pager", extensions=[events]) +... +incidents.emit(Incident(service="billing", title="Latency spike")) +``` +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import AsyncIterator, Callable, Sequence +from contextlib import AsyncExitStack, asynccontextmanager +from datetime import timedelta +from typing import Any, TypeVar + +import anyio +import httpx2 +from anyio.abc import TaskGroup +from mcp_types import INVALID_PARAMS +from mcp_types.version import MODERN_PROTOCOL_VERSIONS +from pydantic import BaseModel, ValidationError + +from mcp.experimental.events._buffer import EmitBuffer +from mcp.experimental.events._locks import KeyedLocks +from mcp.experimental.events.event_type import EventBatch, EventQuery, EventType, NoArguments, Occurrence, Subscriber +from mcp.experimental.events.types import ( + EXTENSION_ID, + FORBIDDEN, + NOT_FOUND, + UNSUPPORTED, + DeliveryMode, + ListEventsParams, + ListEventsResult, + PollEventsParams, + PollEventsResult, + SubscribeEventsParams, + SubscribeEventsResult, + UnsubscribeEventsParams, + UnsubscribeEventsResult, + events_error, +) +from mcp.experimental.events.webhooks import Webhooks, WebhookSubscriptions +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.server.auth.provider import principal_components +from mcp.server.context import ServerRequestContext +from mcp.server.extension import Extension, MethodBinding +from mcp.server.request_state import compact_json +from mcp.shared.exceptions import MCPError + +__all__ = [ + "EventBatch", + "EventQuery", + "EventType", + "Events", + "NoArguments", + "Occurrence", + "Subscriber", + "Webhooks", + "access_token_principal", +] + +ArgsT = TypeVar("ArgsT", bound=BaseModel) +PayloadT = TypeVar("PayloadT", bound=BaseModel) + +PrincipalResolver = Callable[[ServerRequestContext[Any, Any]], str | None] + + +def access_token_principal(ctx: ServerRequestContext[Any, Any]) -> str | None: + """The caller's OAuth identity (client, issuer, subject), or `None` on an unauthenticated server.""" + token = get_access_token() + return compact_json(principal_components(token)) if token is not None else None + + +class Events(Extension): + """Offer events to clients, delivered by poll and, with `webhooks=`, by webhook. + + Args: + webhooks: Enables webhook delivery. Webhook calls need an authenticated principal. + principal: Who is calling. Subscriptions and poll leases are scoped to it. + poll_interval: How long clients wait between polls, and how often webhook loops + run check functions. A check function's `EventBatch.next_poll` overrides it. + poll_lease: How long after its last poll a poll subscription counts as ended, + firing `on_unsubscribe`. Defaults to three poll intervals. + max_events: The most events one poll returns when the client sets no `maxEvents`. + buffer_size: How many emitted events each event type keeps for readers. + buffer_max_age: How long an emitted event stays readable. + """ + + identifier = EXTENSION_ID + + def __init__( + self, + *, + webhooks: Webhooks | None = None, + principal: PrincipalResolver = access_token_principal, + poll_interval: timedelta = timedelta(seconds=5), + poll_lease: timedelta | None = None, + max_events: int = 100, + buffer_size: int = 1000, + buffer_max_age: timedelta = timedelta(minutes=10), + ) -> None: + self._webhooks = webhooks + self._principal = principal + self._poll_interval = poll_interval + self._poll_lease = poll_lease or poll_interval * 3 + self._max_events = max_events + self._buffer_size = buffer_size + self._buffer_max_age = buffer_max_age + self._event_types: dict[str, EventType[Any, Any]] = {} + self._leases: dict[str, anyio.CancelScope] = {} + self._lease_setup = KeyedLocks() + self._task_group: TaskGroup | None = None + self._subscriptions: WebhookSubscriptions | None = None + + def event( + self, name: str, *, arguments: type[ArgsT], payload: type[PayloadT], description: str | None = None + ) -> EventType[ArgsT, PayloadT]: + """Declare an event type. `arguments` is what subscribers pass; `payload` is what each event carries. + + Raises: + ValueError: An event type with this name already exists. + """ + if name in self._event_types: + raise ValueError(f"event type {name!r} is already registered") + buffer = EmitBuffer[PayloadT](self._buffer_size, self._buffer_max_age) + event_type = EventType(name, arguments=arguments, payload=payload, description=description, buffer=buffer) + self._event_types[name] = event_type + return event_type + + def methods(self) -> Sequence[MethodBinding]: + versions = frozenset(MODERN_PROTOCOL_VERSIONS) + bindings = [ + MethodBinding("events/list", ListEventsParams, self._list, protocol_versions=versions), + MethodBinding("events/poll", PollEventsParams, self._poll, protocol_versions=versions), + ] + if self._webhooks is not None: + bindings.append(MethodBinding("events/subscribe", SubscribeEventsParams, self._subscribe, versions)) + bindings.append(MethodBinding("events/unsubscribe", UnsubscribeEventsParams, self._unsubscribe, versions)) + return bindings + + @asynccontextmanager + async def lifespan(self) -> AsyncIterator[None]: + if self._task_group is not None: + # shortcut: a nested run (an in-memory Client per connection) shares the outer one's + # background work, which stops when the outer run does; give each server its own Events. + yield + return + async with AsyncExitStack() as stack: + webhooks = self._webhooks + client = None if webhooks is None else webhooks.http_client + if webhooks is not None and client is None: + client = await stack.enter_async_context(httpx2.AsyncClient(timeout=5)) + self._task_group = await stack.enter_async_context(anyio.create_task_group()) + if webhooks is not None and client is not None: + self._subscriptions = WebhookSubscriptions(webhooks, client, self._task_group, self._poll_interval) + try: + yield + finally: + self._task_group.cancel_scope.cancel() + self._task_group = self._subscriptions = None + self._leases.clear() + + async def _list(self, ctx: ServerRequestContext[Any, Any], params: ListEventsParams) -> ListEventsResult: + delivery: list[DeliveryMode] = ["poll", "webhook"] if self._webhooks is not None else ["poll"] + return ListEventsResult(events=[event_type.descriptor(delivery) for event_type in self._event_types.values()]) + + async def _poll(self, ctx: ServerRequestContext[Any, Any], params: PollEventsParams) -> PollEventsResult: + event_type, arguments = self._resolve(params.name, params.arguments) + principal = self._principal(ctx) + if event_type.subscribe_hooks or event_type.unsubscribe_hooks: + await self._renew_lease(event_type, arguments, principal) + max_age = timedelta(milliseconds=params.max_age_ms) if params.max_age_ms is not None else None + query = EventQuery(arguments, params.cursor, principal, max_age, params.max_events or self._max_events) + batch = await event_type.read(query) + return PollEventsResult( + events=[event_type.wire(occurrence) for occurrence in batch.events], + cursor=batch.cursor, + truncated=batch.truncated, + has_more=batch.has_more, + next_poll_ms=int((batch.next_poll or self._poll_interval).total_seconds() * 1000), + ) + + async def _subscribe( + self, ctx: ServerRequestContext[Any, Any], params: SubscribeEventsParams + ) -> SubscribeEventsResult: + event_type, arguments = self._resolve(params.name, params.arguments) + if params.delivery.mode != "webhook": + raise events_error(UNSUPPORTED, feature="deliveryMode", value=params.delivery.mode) + if (principal := self._principal(ctx)) is None: + raise events_error(FORBIDDEN, reason="webhook subscriptions need an authenticated principal") + assert self._subscriptions is not None, "Events runs inside the server's lifespan" + return await self._subscriptions.subscribe(event_type, arguments, principal, params) + + async def _unsubscribe( + self, ctx: ServerRequestContext[Any, Any], params: UnsubscribeEventsParams + ) -> UnsubscribeEventsResult: + event_type, _ = self._resolve(params.name, params.arguments) + if (principal := self._principal(ctx)) is None: + raise events_error(FORBIDDEN, reason="webhook subscriptions need an authenticated principal") + assert self._subscriptions is not None, "Events runs inside the server's lifespan" + self._subscriptions.unsubscribe(event_type, principal, params) + return UnsubscribeEventsResult() + + def _resolve(self, name: str, arguments: dict[str, Any] | None) -> tuple[EventType[Any, Any], BaseModel]: + if (event_type := self._event_types.get(name)) is None: + raise events_error(NOT_FOUND, kind="event", name=name) + try: + return event_type, event_type.arguments.model_validate(arguments or {}) + except ValidationError as exc: + errors = exc.errors(include_url=False, include_context=False, include_input=False) + raise MCPError(INVALID_PARAMS, f"Invalid arguments for event {name!r}", errors) from None + + async def _renew_lease(self, event_type: EventType[Any, Any], arguments: BaseModel, principal: str | None) -> None: + key = json.dumps([principal, event_type.name, arguments.model_dump(mode="json")], sort_keys=True) + deadline = anyio.current_time() + self._poll_lease.total_seconds() + assert self._task_group is not None, "Events runs inside the server's lifespan" + subscription_id = f"sub_{hashlib.sha256(key.encode()).hexdigest()[:16]}" + subscriber = Subscriber(id=subscription_id, arguments=arguments, principal=principal, mode="poll") + async with self._lease_setup(key): + if (lease := self._leases.get(key)) is not None: + lease.deadline = deadline + return + for hook in event_type.subscribe_hooks: + await hook(subscriber) + lease = self._leases[key] = anyio.CancelScope(deadline=deadline) + self._task_group.start_soon(self._expire_lease, key, lease, event_type, subscriber) + + async def _expire_lease( + self, key: str, lease: anyio.CancelScope, event_type: EventType[Any, Any], subscriber: Subscriber[Any] + ) -> None: + with lease: + await anyio.sleep_forever() + del self._leases[key] + for hook in event_type.unsubscribe_hooks: + await hook(subscriber) diff --git a/src/mcp/experimental/events/signing.py b/src/mcp/experimental/events/signing.py new file mode 100644 index 0000000000..78fc91fbcd --- /dev/null +++ b/src/mcp/experimental/events/signing.py @@ -0,0 +1,49 @@ +"""Standard Webhooks signing, the scheme SEP-3415 webhook deliveries use. + +See https://github.com/standard-webhooks/standard-webhooks/blob/main/spec/standard-webhooks.md. +""" + +from __future__ import annotations + +import base64 +import binascii +import hashlib +import hmac + +__all__ = ["decode_secret", "sign", "verify"] + +_PREFIX = "whsec_" + + +def decode_secret(secret: str) -> bytes: + """Return the key bytes of a `whsec_` secret. + + Raises: + ValueError: The secret is not `whsec_` plus base64 of 24 to 64 bytes. + """ + if not secret.startswith(_PREFIX): + raise ValueError("secret must start with 'whsec_'") + try: + key = base64.b64decode(secret.removeprefix(_PREFIX), validate=True) + except binascii.Error: + raise ValueError("secret must be 'whsec_' followed by base64") from None + if not 24 <= len(key) <= 64: + raise ValueError("secret must encode 24 to 64 bytes") + return key + + +def sign(secret: str, message_id: str, timestamp: int, body: bytes) -> str: + """Return the base64 `v1` signature of one message made with `secret`.""" + signed = f"{message_id}.{timestamp}.".encode() + body + digest = hmac.new(decode_secret(secret), signed, hashlib.sha256).digest() + return base64.b64encode(digest).decode() + + +def verify(secret: str, message_id: str, timestamp: int, body: bytes, signature_header: str) -> bool: + """Whether any `v1,` signature in a `webhook-signature` header was made with `secret` over exactly these bytes.""" + expected = sign(secret, message_id, timestamp, body) + return any( + hmac.compare_digest(candidate.removeprefix("v1,"), expected) + for candidate in signature_header.split() + if candidate.startswith("v1,") + ) diff --git a/src/mcp/experimental/events/types.py b/src/mcp/experimental/events/types.py new file mode 100644 index 0000000000..e79ff4767f --- /dev/null +++ b/src/mcp/experimental/events/types.py @@ -0,0 +1,205 @@ +"""Wire types and error codes for MCP Events (SEP-3415), shared by the server and client sides.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from mcp_types import PaginatedRequestParams, PaginatedResult, Request, RequestParams, Result +from pydantic import BaseModel, ConfigDict, Field +from pydantic.alias_generators import to_camel + +from mcp.shared.exceptions import MCPError + +__all__ = [ + "EXTENSION_ID", + "NOT_FOUND", + "FORBIDDEN", + "RESOURCE_EXHAUSTED", + "UNSUPPORTED", + "CALLBACK_ENDPOINT_ERROR", + "DeliveryMode", + "DeliveryError", + "EventDescriptor", + "EventOccurrence", + "ListEventsParams", + "ListEventsResult", + "ListEventsRequest", + "PollEventsParams", + "PollEventsResult", + "PollEventsRequest", + "WebhookDelivery", + "WebhookTarget", + "DeliveryStatus", + "SubscribeEventsParams", + "SubscribeEventsResult", + "SubscribeEventsRequest", + "UnsubscribeEventsParams", + "UnsubscribeEventsResult", + "UnsubscribeEventsRequest", + "events_error", +] + +EXTENSION_ID = "io.modelcontextprotocol/events" + +# Provisional until SEP-3415 is accepted. +NOT_FOUND = -32023 +FORBIDDEN = -32024 +RESOURCE_EXHAUSTED = -32025 +UNSUPPORTED = -32026 +CALLBACK_ENDPOINT_ERROR = -32027 + +_ERROR_MESSAGES = { + NOT_FOUND: "Not found", + FORBIDDEN: "Forbidden", + RESOURCE_EXHAUSTED: "Resource exhausted", + UNSUPPORTED: "Unsupported", + CALLBACK_ENDPOINT_ERROR: "Callback endpoint error", +} + +DeliveryMode = Literal["poll", "push", "webhook"] + +DeliveryError = Literal["connection_refused", "timeout", "tls_error", "http_4xx", "http_5xx", "challenge_failed"] +"""The categories a server reports for a failing callback; never raw endpoint responses.""" + + +def events_error(code: int, **data: Any) -> MCPError: + """Build one of the SEP-3415 errors, with `data` as its typed payload.""" + return MCPError(code=code, message=_ERROR_MESSAGES[code], data=data or None) + + +class _Model(BaseModel): + model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True) + + +class EventDescriptor(_Model): + """One event type, as listed by `events/list`.""" + + name: str + description: str | None = None + delivery: list[DeliveryMode] + input_schema: dict[str, Any] + payload_schema: dict[str, Any] + meta: dict[str, Any] | None = Field(default=None, alias="_meta") + + +class EventOccurrence(_Model): + """One event, as returned by a poll or POSTed to a webhook.""" + + event_id: str + name: str + timestamp: datetime + data: dict[str, Any] + cursor: str | None = None + """Subscription position after this event; webhook deliveries only.""" + meta: dict[str, Any] | None = Field(default=None, alias="_meta") + + +class ListEventsParams(PaginatedRequestParams): + """Parameters for `events/list`.""" + + +class ListEventsResult(PaginatedResult): + """Result of `events/list`.""" + + events: list[EventDescriptor] + + +class PollEventsParams(RequestParams): + """Parameters for `events/poll`.""" + + name: str + arguments: dict[str, Any] | None = None + cursor: str | None = None + max_age_ms: int | None = Field(default=None, ge=0) + max_events: int | None = Field(default=None, ge=1) + + +class PollEventsResult(Result): + """Result of `events/poll`.""" + + events: list[EventOccurrence] + cursor: str | None = None + truncated: bool = False + has_more: bool = False + next_poll_ms: int + + +class WebhookDelivery(_Model): + """Where and how to deliver a webhook subscription's events.""" + + mode: str = "webhook" + url: str + secret: str + + +class WebhookTarget(_Model): + """The callback URL that identifies a webhook subscription on `events/unsubscribe`.""" + + url: str + + +class DeliveryStatus(_Model): + """How delivery to a webhook subscription is going, reported on refresh.""" + + active: bool + last_delivery_at: datetime | None = None + last_error: DeliveryError | None = None + failed_since: datetime | None = None + + +class SubscribeEventsParams(RequestParams): + """Parameters for `events/subscribe`.""" + + name: str + arguments: dict[str, Any] | None = None + delivery: WebhookDelivery + cursor: str | None = None + max_age_ms: int | None = Field(default=None, ge=0) + ttl_ms: int | None = Field(default=None, ge=0) + """Suggested lifetime; omitted means the server default, an explicit `None` asks for no expiry.""" + + +class SubscribeEventsResult(Result): + """Result of `events/subscribe`.""" + + id: str + refresh_before: datetime | None = None + cursor: str | None = None + truncated: bool = False + delivery_status: DeliveryStatus | None = None + + +class UnsubscribeEventsParams(RequestParams): + """Parameters for `events/unsubscribe`.""" + + name: str + arguments: dict[str, Any] | None = None + delivery: WebhookTarget + + +class UnsubscribeEventsResult(Result): + """Result of `events/unsubscribe`.""" + + +class ListEventsRequest(Request[ListEventsParams | None, Literal["events/list"]]): + method: Literal["events/list"] = "events/list" + params: ListEventsParams | None = None + + +class PollEventsRequest(Request[PollEventsParams, Literal["events/poll"]]): + method: Literal["events/poll"] = "events/poll" + params: PollEventsParams + name_param = "name" + + +class SubscribeEventsRequest(Request[SubscribeEventsParams, Literal["events/subscribe"]]): + method: Literal["events/subscribe"] = "events/subscribe" + params: SubscribeEventsParams + name_param = "name" + + +class UnsubscribeEventsRequest(Request[UnsubscribeEventsParams, Literal["events/unsubscribe"]]): + method: Literal["events/unsubscribe"] = "events/unsubscribe" + params: UnsubscribeEventsParams + name_param = "name" diff --git a/src/mcp/experimental/events/webhooks.py b/src/mcp/experimental/events/webhooks.py new file mode 100644 index 0000000000..d2d6c85d3e --- /dev/null +++ b/src/mcp/experimental/events/webhooks.py @@ -0,0 +1,269 @@ +"""Webhook delivery (SEP-3415): subscriptions keyed by who asked, for what, and where to deliver.""" + +from __future__ import annotations + +import hashlib +import json +import logging +import secrets +from collections.abc import Collection, Sequence +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from typing import Any + +import anyio +import httpx2 +from anyio.abc import TaskGroup +from mcp_types import INVALID_PARAMS +from pydantic import BaseModel + +from mcp.experimental.events._callbacks import CallbackFailure, Callbacks, NonPublicAddress +from mcp.experimental.events._locks import KeyedLocks +from mcp.experimental.events.event_type import EventBatch, EventQuery, EventType, Subscriber +from mcp.experimental.events.signing import decode_secret +from mcp.experimental.events.types import ( + CALLBACK_ENDPOINT_ERROR, + NOT_FOUND, + DeliveryError, + DeliveryStatus, + SubscribeEventsParams, + SubscribeEventsResult, + UnsubscribeEventsParams, + events_error, +) +from mcp.shared.exceptions import MCPError + +__all__ = ["Webhooks"] + +logger = logging.getLogger(__name__) + +_ROTATION_GRACE = timedelta(minutes=5) + + +@dataclass +class Webhooks: + """Configuration for webhook delivery. Passing it to `Events` offers webhook on every event type. + + Subscriptions live in memory, so the server grants finite TTLs only and clients + resubscribe after a restart. `trusted_origins` (e.g. `"http://127.0.0.1:9000"`) are + exempt from the `https` and public-address rules, for local receivers in development. + """ + + max_ttl: timedelta = timedelta(hours=1) + min_ttl: timedelta = timedelta(minutes=1) + trusted_origins: Collection[str] = () + retry_delays: Sequence[timedelta] = ( + timedelta(seconds=5), + timedelta(seconds=30), + timedelta(minutes=2), + timedelta(minutes=5), + ) + """Waits between attempts to deliver one event; it is abandoned after the last.""" + http_client: httpx2.AsyncClient | None = None + """Client for the POSTs; by default one with a 5 second timeout, owned by the server.""" + + +@dataclass +class _Subscription: + id: str + key: str + event_type: EventType[Any, Any] + subscriber: Subscriber[Any] + url: str + secret: str + expires_at: datetime + cursor: str | None + scope: anyio.CancelScope = field(default_factory=anyio.CancelScope) + previous_secret: tuple[str, datetime] | None = None + last_delivery_at: datetime | None = None + last_error: DeliveryError | None = None + failed_since: datetime | None = None + + def signing_secrets(self) -> list[str]: + if self.previous_secret is not None and self.previous_secret[1] > datetime.now(UTC): + return [self.secret, self.previous_secret[0]] + return [self.secret] + + def status(self) -> DeliveryStatus: + return DeliveryStatus( + active=True, + last_delivery_at=self.last_delivery_at, + last_error=self.last_error, + failed_since=self.failed_since, + ) + + +class WebhookSubscriptions: + """The live webhook subscriptions of one running server, and their delivery loops.""" + + def __init__( + self, config: Webhooks, client: httpx2.AsyncClient, task_group: TaskGroup, poll_interval: timedelta + ) -> None: + self._config = config + self._poll_interval = poll_interval + self._callbacks = Callbacks(client, config.trusted_origins) + self._task_group = task_group + self._subscriptions: dict[str, _Subscription] = {} + self._verified: set[tuple[str | None, str]] = set() + self._setup = KeyedLocks() + + async def subscribe( + self, event_type: EventType[Any, Any], arguments: BaseModel, principal: str, params: SubscribeEventsParams + ) -> SubscribeEventsResult: + url, secret = params.delivery.url, params.delivery.secret + try: + self._callbacks.validate(url) + decode_secret(secret) + except ValueError as exc: + raise MCPError(code=INVALID_PARAMS, message=str(exc)) from None + key = json.dumps([principal, url, event_type.name, params.arguments or {}], sort_keys=True) + subscription_id = f"sub_{hashlib.sha256(key.encode()).hexdigest()[:16]}" + ttl = self._grant(params) + async with self._setup(key): + if (existing := self._subscriptions.get(key)) is not None: + return self._refresh(existing, secret, ttl) + try: + return await self._create(event_type, arguments, principal, params, key, subscription_id, ttl) + except BaseException: + self._forget_unused_verification(principal, url) + raise + + async def _create( + self, + event_type: EventType[Any, Any], + arguments: BaseModel, + principal: str, + params: SubscribeEventsParams, + key: str, + subscription_id: str, + ttl: timedelta, + ) -> SubscribeEventsResult: + url, secret = params.delivery.url, params.delivery.secret + if (principal, url) not in self._verified: + try: + await self._callbacks.verify(url, subscription_id, secret) + except NonPublicAddress as exc: + raise MCPError(code=INVALID_PARAMS, message=str(exc)) from None + except CallbackFailure as exc: + raise events_error(CALLBACK_ENDPOINT_ERROR, reason=exc.reason) from None + self._verified.add((principal, url)) + max_age = timedelta(milliseconds=params.max_age_ms) if params.max_age_ms is not None else None + batch = await event_type.read(EventQuery(arguments, params.cursor, principal, max_age)) + subscriber = Subscriber(id=subscription_id, arguments=arguments, principal=principal, mode="webhook") + subscription = _Subscription( + id=subscription_id, + key=key, + event_type=event_type, + subscriber=subscriber, + url=url, + secret=secret, + expires_at=datetime.now(UTC) + ttl, + # The batch is not delivered yet, so only the supplied cursor is safe to persist. + cursor=params.cursor if batch.events else batch.cursor, + ) + subscription.scope.deadline = anyio.current_time() + ttl.total_seconds() + for hook in event_type.subscribe_hooks: + await hook(subscriber) + self._subscriptions[key] = subscription + self._task_group.start_soon(self._run, subscription, batch) + return SubscribeEventsResult( + id=subscription_id, + refresh_before=subscription.expires_at, + cursor=subscription.cursor, + truncated=batch.truncated, + ) + + def unsubscribe(self, event_type: EventType[Any, Any], principal: str, params: UnsubscribeEventsParams) -> None: + key = json.dumps([principal, params.delivery.url, event_type.name, params.arguments or {}], sort_keys=True) + subscription = self._subscriptions.pop(key, None) + if subscription is None: + raise events_error(NOT_FOUND, kind="subscription") + subscription.scope.cancel() + + def _forget_unused_verification(self, principal: str | None, url: str) -> None: + # The spec scopes verification to the subscriptions it covers, so the cache cannot outgrow them. + if not any(s.subscriber.principal == principal and s.url == url for s in self._subscriptions.values()): + self._verified.discard((principal, url)) + + def _grant(self, params: SubscribeEventsParams) -> timedelta: + # In-memory subscriptions cannot honour "no expiry", so an omitted or null ttlMs gets the cap. + if params.ttl_ms is None: + return self._config.max_ttl + return min(max(timedelta(milliseconds=params.ttl_ms), self._config.min_ttl), self._config.max_ttl) + + def _refresh(self, subscription: _Subscription, secret: str, ttl: timedelta) -> SubscribeEventsResult: + if secret != subscription.secret: + subscription.previous_secret = (subscription.secret, datetime.now(UTC) + _ROTATION_GRACE) + subscription.secret = secret + subscription.expires_at = datetime.now(UTC) + ttl + subscription.scope.deadline = anyio.current_time() + ttl.total_seconds() + return SubscribeEventsResult( + id=subscription.id, + refresh_before=subscription.expires_at, + cursor=subscription.cursor, + delivery_status=subscription.status(), + ) + + async def _run(self, subscription: _Subscription, batch: EventBatch[Any]) -> None: + event_type, subscriber = subscription.event_type, subscription.subscriber + try: + with subscription.scope: + while True: + # shortcut: events go out one at a time, so a failing event delays later ones by its + # retries, where the spec retries each independently; needs per-event acks behind the cursor. + for index, occurrence in enumerate(batch.events): + last = index == len(batch.events) - 1 + body = event_type.wire(occurrence, batch.cursor if last else subscription.cursor) + await self._deliver(subscription, body.model_dump(mode="json", by_alias=True)) + subscription.cursor = batch.cursor + delay = timedelta(0) if batch.has_more else batch.next_poll or self._poll_interval + await event_type.buffer.wait(delay.total_seconds()) + batch = await self._read(subscription) + finally: + if self._subscriptions.get(subscription.key) is subscription: + del self._subscriptions[subscription.key] + self._forget_unused_verification(subscriber.principal, subscription.url) + for hook in event_type.unsubscribe_hooks: + await hook(subscriber) + + async def _read(self, subscription: _Subscription) -> EventBatch[Any]: + query = EventQuery(subscription.subscriber.arguments, subscription.cursor, subscription.subscriber.principal) + try: + batch = await subscription.event_type.read(query) + except MCPError as exc: + await self._deliver(subscription, {"type": "terminated", "error": exc.error.model_dump(exclude_none=True)}) + subscription.scope.cancel() + return EventBatch(cursor=subscription.cursor) + except Exception: + # The check function is user code: a crash must not take the server's task group down. + logger.exception("check function for %r raised", subscription.event_type.name) + return EventBatch(cursor=subscription.cursor) + if batch.truncated: + await self._deliver(subscription, {"type": "gap", "cursor": batch.cursor}) + return batch + + async def _deliver(self, subscription: _Subscription, body: dict[str, Any]) -> None: + kind = body.get("type", "event") + message_id = f"msg_{kind}_{secrets.token_hex(12)}" if kind != "event" else f"msg_{secrets.token_hex(16)}" + for delay in [*self._config.retry_delays, None]: + try: + response = await self._callbacks.post( + subscription.url, subscription.id, subscription.signing_secrets(), message_id, body + ) + except NonPublicAddress: + logger.warning("abandoned a delivery to %s: it resolves to a non-public address", subscription.url) + subscription.last_error = "connection_refused" + return + except CallbackFailure as exc: + subscription.last_error = exc.reason + else: + if response.is_success: + subscription.last_delivery_at = datetime.now(UTC) + subscription.last_error = subscription.failed_since = None + return + subscription.last_error = "http_5xx" if response.status_code >= 500 else "http_4xx" + if response.status_code in (410, 413): + return + subscription.failed_since = subscription.failed_since or datetime.now(UTC) + if delay is not None: + await anyio.sleep(delay.total_seconds()) diff --git a/src/mcp/server/extension.py b/src/mcp/server/extension.py index 6205a7033d..880e0360b6 100644 --- a/src/mcp/server/extension.py +++ b/src/mcp/server/extension.py @@ -20,6 +20,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Sequence +from contextlib import AbstractAsyncContextManager, nullcontext from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any @@ -134,6 +135,13 @@ def methods(self) -> Sequence[MethodBinding]: """New request methods this extension serves (additive).""" return () + def lifespan(self) -> AbstractAsyncContextManager[None]: + """Context the server holds open while it runs, for background work such as delivery loops. + + Entered before the server's own lifespan and exited after it. Default: nothing to run. + """ + return nullcontext() + async def intercept_tool_call( self, params: CallToolRequestParams, diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index 7efa29108d..664a6bd0da 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/src/mcp/server/mcpserver/server.py @@ -5,7 +5,7 @@ import base64 import inspect from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Mapping, Sequence -from contextlib import AbstractAsyncContextManager, asynccontextmanager +from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager from typing import Any, Generic, Literal, TypeVar, cast, overload import anyio @@ -154,6 +154,20 @@ async def wrap(_: Server[LifespanResultT]) -> AsyncIterator[LifespanResultT]: return wrap +def _with_extension_lifespans( + lifespan: Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]], + extensions: Sequence[Extension], +) -> Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]]: + @asynccontextmanager + async def wrap(server: Server[LifespanResultT]) -> AsyncIterator[LifespanResultT]: + async with AsyncExitStack() as stack: + for extension in extensions: + await stack.enter_async_context(extension.lifespan()) + yield await stack.enter_async_context(lifespan(server)) + + return wrap + + class MCPServer(Generic[LifespanResultT]): def __init__( self, @@ -270,6 +284,7 @@ def __init__( for extension in extensions or (): self._apply_extension(extension) self._install_extension_interceptor() + self._lowlevel_server.lifespan = _with_extension_lifespans(self._lowlevel_server.lifespan, self._extensions) @property def name(self) -> str: diff --git a/tests/experimental/__init__.py b/tests/experimental/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/experimental/events/__init__.py b/tests/experimental/events/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/experimental/events/test_client.py b/tests/experimental/events/test_client.py new file mode 100644 index 0000000000..403212a250 --- /dev/null +++ b/tests/experimental/events/test_client.py @@ -0,0 +1,176 @@ +"""Tests for the client side of MCP Events (SEP-3415): `list`, `watch`, and `subscribe`'s refresh loop. + +Webhook subscribe against the SDK's own server is covered in `test_webhooks.py`. +""" + +from datetime import UTC, datetime, timedelta +from typing import Any + +import anyio +import pytest +from pydantic import BaseModel + +from mcp.client.client import Client +from mcp.experimental.events.client import Events as ClientEvents +from mcp.experimental.events.server import EventBatch, EventQuery, Events, NoArguments, Occurrence +from mcp.experimental.events.types import ( + EXTENSION_ID, + EventDescriptor, + ListEventsParams, + ListEventsResult, + SubscribeEventsParams, + SubscribeEventsResult, + UnsubscribeEventsParams, + UnsubscribeEventsResult, +) +from mcp.server.context import ServerRequestContext +from mcp.server.extension import Extension, MethodBinding +from mcp.server.mcpserver import MCPServer + +pytestmark = pytest.mark.anyio + + +class Ping(BaseModel): + n: int + + +async def test_watch_yields_batches_with_events_and_resumes_from_the_latest_cursor() -> None: + """`watch` polls in a loop and drains `hasMore` batches at once (SDK-defined loop over + spec-mandated poll semantics).""" + events = Events(poll_interval=timedelta(0), max_events=2) + pings = events.event("ping", arguments=NoArguments, payload=Ping) + mcp = MCPServer("pinger", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents(min_poll_interval=timedelta(0))]) as client: + events_client = client.extension(ClientEvents) + start = await events_client.poll("ping") + for n in range(3): + pings.emit(Ping(n=n)) + watch = events_client.watch("ping", cursor=start.cursor) + with anyio.fail_after(5): + first = await anext(watch) + second = await anext(watch) + await watch.aclose() + + assert [e.data for e in first.events] == [{"n": 0}, {"n": 1}] + assert first.has_more is True + assert [e.data for e in second.events] == [{"n": 2}] + + +async def test_watch_from_a_saved_cursor_yields_the_gap_when_the_cursor_is_unknown() -> None: + """Resuming from a cursor the server cannot read yields a `truncated` batch so the caller can + reconcile (spec-mandated gap signal).""" + events = Events(poll_interval=timedelta(0)) + events.event("ping", arguments=NoArguments, payload=Ping) + mcp = MCPServer("pinger", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents(min_poll_interval=timedelta(0))]) as client: + watch = client.extension(ClientEvents).watch("ping", cursor="from-a-previous-life") + with anyio.fail_after(5): + batch = await anext(watch) + await watch.aclose() + + assert (batch.events, batch.truncated) == ([], True) + + +async def test_watch_yields_empty_batches_and_waits_at_least_the_clients_floor_between_polls() -> None: + """Empty batches are yielded because they still advance the cursor (spec-mandated quiet-period + advancement), and the client's `min_poll_interval` overrides a server asking for a tighter loop + (spec-mandated floor; a time-based feature, so the test lets the floor elapse).""" + polls: list[str | None] = [] + events = Events(poll_interval=timedelta(0)) + pings = events.event("ping", arguments=NoArguments, payload=Ping) + + @pings.check + async def check(query: EventQuery[NoArguments]) -> EventBatch[Ping]: + polls.append(query.cursor) + if len(polls) < 3: + return EventBatch(cursor=f"c{len(polls)}") + return EventBatch(cursor=f"c{len(polls)}", events=[Occurrence(Ping(n=len(polls)))]) + + mcp = MCPServer("pinger", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents(min_poll_interval=timedelta(milliseconds=20))]) as client: + watch = client.extension(ClientEvents).watch("ping") + started = anyio.current_time() + with anyio.fail_after(5): + quiet = await anext(watch) + busy = await anext(watch) + await watch.aclose() + elapsed = anyio.current_time() - started + + assert polls == [None, "c1", "c2"] + assert (quiet.events, quiet.cursor) == ([], "c2") + assert [e.data for e in busy.events] == [{"n": 3}] + assert elapsed >= 0.02 + + +class _OtherEventsServer(Extension): + """A third-party Events server: pages `events/list`, and grants a short-lived webhook subscription + that it then stops expiring. The SDK's own server can do neither.""" + + identifier = EXTENSION_ID + + def __init__(self) -> None: + self.subscribes: list[SubscribeEventsParams] = [] + self.refreshed = anyio.Event() + self.unsubscribed = anyio.Event() + + def methods(self) -> list[MethodBinding]: + return [ + MethodBinding("events/list", ListEventsParams, self._list), + MethodBinding("events/subscribe", SubscribeEventsParams, self._subscribe), + MethodBinding("events/unsubscribe", UnsubscribeEventsParams, self._unsubscribe), + ] + + async def _list(self, ctx: ServerRequestContext[Any, Any], params: ListEventsParams) -> ListEventsResult: + name = "first" if params.cursor is None else "second" + descriptor = EventDescriptor(name=name, delivery=["poll"], input_schema={}, payload_schema={}) + return ListEventsResult(events=[descriptor], next_cursor="page-2" if params.cursor is None else None) + + async def _subscribe( + self, ctx: ServerRequestContext[Any, Any], params: SubscribeEventsParams + ) -> SubscribeEventsResult: + self.subscribes.append(params) + if len(self.subscribes) == 1: + refresh_before = datetime.now(UTC) + timedelta(milliseconds=30) + return SubscribeEventsResult(id="sub-1", refresh_before=refresh_before, cursor="c1") + self.refreshed.set() + return SubscribeEventsResult(id="sub-1", refresh_before=None) + + async def _unsubscribe( + self, ctx: ServerRequestContext[Any, Any], params: UnsubscribeEventsParams + ) -> UnsubscribeEventsResult: + self.unsubscribed.set() + return UnsubscribeEventsResult() + + +async def test_list_follows_pagination_to_the_last_page() -> None: + """`list` returns every page's event types (spec-mandated pagination).""" + mcp = MCPServer("other", extensions=[_OtherEventsServer()]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + descriptors = await client.extension(ClientEvents).list() + + assert [d.name for d in descriptors] == ["first", "second"] + + +async def test_subscribe_refreshes_before_the_subscription_expires_and_unsubscribes_on_exit() -> None: + """`subscribe` resends the subscription with the latest cursor before `refreshBefore`, stops once + the server grants no expiry, and unsubscribes when the block exits (spec-mandated refresh; a + time-based feature).""" + server = _OtherEventsServer() + mcp = MCPServer("other", extensions=[server]) + secret = "whsec_" + "a" * 43 + "=" + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + async with events_client.subscribe("first", url="https://client.test/hooks", secret=secret, cursor="c0"): + with anyio.fail_after(5): + await server.refreshed.wait() + with anyio.fail_after(5): + await server.unsubscribed.wait() + + first, refresh = server.subscribes + assert (first.cursor, refresh.cursor) == ("c0", "c1") + assert refresh.delivery == first.delivery diff --git a/tests/experimental/events/test_receiver.py b/tests/experimental/events/test_receiver.py new file mode 100644 index 0000000000..a14bbc3955 --- /dev/null +++ b/tests/experimental/events/test_receiver.py @@ -0,0 +1,149 @@ +"""Tests for `WebhookReceiver`, the ASGI app that receives SEP-3415 webhook deliveries. + +These drive it with raw HTTP: status codes and signature headers are the behaviour under test. +""" + +import base64 +import json +import time +from typing import Any + +import httpx2 +import pytest + +from mcp.experimental.events.receiver import Delivery, WebhookGap, WebhookReceiver, WebhookTerminated +from mcp.experimental.events.signing import sign +from mcp.experimental.events.types import EventOccurrence + +pytestmark = pytest.mark.anyio + +SECRET = "whsec_" + base64.b64encode(b"k" * 32).decode() +EVENT = {"eventId": "e1", "name": "ping", "timestamp": "2026-11-05T12:00:00Z", "data": {"n": 1}, "cursor": "c1"} + + +def signed( + body: dict[str, Any] | bytes, + *, + secret: str = SECRET, + timestamp: int | None = None, + subscription_id: str = "sub_1", +) -> dict[str, Any]: + content = body if isinstance(body, bytes) else json.dumps(body).encode() + timestamp = int(time.time()) if timestamp is None else timestamp + headers = { + "webhook-id": "msg_1", + "webhook-timestamp": str(timestamp), + "webhook-signature": f"v1,{sign(secret, 'msg_1', timestamp, content)}", + "x-mcp-subscription-id": subscription_id, + } + return {"content": content, "headers": headers} + + +def receiver_client(received: list[Delivery], **kwargs: Any) -> httpx2.AsyncClient: + async def handle(delivery: Delivery) -> None: + received.append(delivery) + + app = WebhookReceiver(SECRET, handle, **kwargs) + return httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), base_url="https://client.test") + + +async def test_receiver_echoes_the_verification_challenge() -> None: + """A signed verification request is answered with its challenge, proving the endpoint opted in (spec-mandated).""" + received: list[Delivery] = [] + async with receiver_client(received) as http: + response = await http.post("/", **signed({"type": "verification", "challenge": "abc"})) + + assert (response.status_code, response.json()) == (200, {"challenge": "abc"}) + assert received == [] + + +async def test_receiver_hands_over_events_gaps_and_terminations() -> None: + """Each delivery body becomes a typed value for the handler (spec-mandated body shapes).""" + received: list[Delivery] = [] + async with receiver_client(received) as http: + statuses = [ + (await http.post("/", **signed(body))).status_code + for body in ( + EVENT, + {"type": "gap", "cursor": "c9"}, + {"type": "terminated", "error": {"code": -32024, "message": "Forbidden"}}, + ) + ] + + assert statuses == [204, 204, 204] + assert received == [ + EventOccurrence.model_validate(EVENT), + WebhookGap(subscription_id="sub_1", cursor="c9"), + WebhookTerminated(subscription_id="sub_1", error={"code": -32024, "message": "Forbidden"}), + ] + + +async def test_receiver_hands_over_a_retried_event_once_per_subscription_within_the_dedupe_window() -> None: + """Servers deliver at least once, so an `eventId` repeated on one subscription is acknowledged + without reaching the handler until it falls out of the window, while the same `eventId` on + another subscription is a separate delivery (spec-mandated subscription-scoped deduplication).""" + received: list[Delivery] = [] + async with receiver_client(received, dedupe_window=2) as http: + for subscription_id, event_id in ( + ("sub_1", "e1"), + ("sub_1", "e1"), + ("sub_2", "e1"), + ("sub_1", "e2"), + ("sub_1", "e1"), + ): + response = await http.post("/", **signed({**EVENT, "eventId": event_id}, subscription_id=subscription_id)) + assert response.status_code == 204 + + assert [d.event_id for d in received if isinstance(d, EventOccurrence)] == ["e1", "e1", "e2", "e1"] + + +@pytest.mark.parametrize( + "request_kwargs", + [ + pytest.param(signed(EVENT, secret="whsec_" + base64.b64encode(b"x" * 32).decode()), id="other-secret"), + pytest.param(signed(EVENT, timestamp=int(time.time()) - 3600), id="stale"), + pytest.param( + {**signed(EVENT), "headers": {**signed(EVENT)["headers"], "webhook-timestamp": "soon"}}, id="bad-timestamp" + ), + pytest.param( + {**signed(EVENT), "headers": {**signed(EVENT)["headers"], "webhook-signature": "v2,abc"}}, + id="unknown-version", + ), + pytest.param({"content": json.dumps(EVENT).encode()}, id="unsigned"), + ], +) +async def test_receiver_rejects_unauthenticated_requests_with_401(request_kwargs: dict[str, Any]) -> None: + """Only bodies signed with the subscription's secret within the tolerance reach the handler + (spec-mandated Standard Webhooks verification).""" + received: list[Delivery] = [] + async with receiver_client(received) as http: + response = await http.post("/", **request_kwargs) + + assert response.status_code == 401 + assert received == [] + + +@pytest.mark.parametrize( + "body", + [ + pytest.param(b"not json", id="not-json"), + pytest.param({"eventId": "e1"}, id="incomplete-event"), + pytest.param({"type": "terminated"}, id="terminated-without-error"), + ], +) +async def test_receiver_rejects_malformed_bodies_with_400(body: dict[str, Any] | bytes) -> None: + """A signed body that is not a delivery is refused (SDK-defined).""" + received: list[Delivery] = [] + async with receiver_client(received) as http: + response = await http.post("/", **signed(body)) + + assert response.status_code == 400 + assert received == [] + + +async def test_receiver_only_accepts_post() -> None: + """Deliveries are POSTs; anything else is `405` (SDK-defined).""" + async with receiver_client([]) as http: + response = await http.get("/") + + assert response.status_code == 405 diff --git a/tests/experimental/events/test_server.py b/tests/experimental/events/test_server.py new file mode 100644 index 0000000000..a0deacfbf8 --- /dev/null +++ b/tests/experimental/events/test_server.py @@ -0,0 +1,438 @@ +"""Tests for the server side of MCP Events (SEP-3415): `events/list`, `events/poll`, and poll leases. + +Webhook delivery is covered in `test_webhooks.py`. +""" + +from datetime import UTC, datetime, timedelta +from typing import Any + +import anyio +import pytest +from inline_snapshot import snapshot +from mcp_types import INVALID_PARAMS +from pydantic import BaseModel, ConfigDict + +from mcp.client.client import Client +from mcp.experimental.events.client import Events as ClientEvents +from mcp.experimental.events.server import ( + EventBatch, + EventQuery, + Events, + NoArguments, + Occurrence, + Subscriber, + Webhooks, +) +from mcp.experimental.events.types import ( + EXTENSION_ID, + FORBIDDEN, + NOT_FOUND, + PollEventsParams, + PollEventsRequest, + events_error, +) +from mcp.server.mcpserver import MCPServer +from mcp.shared.exceptions import MCPError + +pytestmark = pytest.mark.anyio + + +class ServiceFilter(BaseModel): + service: str + + +class Incident(BaseModel): + service: str + title: str + + +class _RawResult(BaseModel): + model_config = ConfigDict(extra="allow") + + +async def test_events_list_describes_each_event_type_with_json_schemas() -> None: + """`events/list` derives `inputSchema`/`payloadSchema` from the Pydantic models, and lists poll alone + without webhooks (spec-mandated shape, SDK-defined schema source).""" + events = Events() + events.event("incident.created", arguments=ServiceFilter, payload=Incident, description="A new incident") + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + descriptors = await client.extension(ClientEvents).list() + capabilities = client.server_capabilities + + assert capabilities.extensions == {EXTENSION_ID: {}} + assert [d.model_dump(by_alias=True, exclude_none=True) for d in descriptors] == snapshot( + [ + { + "name": "incident.created", + "description": "A new incident", + "delivery": ["poll"], + "inputSchema": { + "properties": {"service": {"title": "Service", "type": "string"}}, + "required": ["service"], + "title": "ServiceFilter", + "type": "object", + }, + "payloadSchema": { + "properties": { + "service": {"title": "Service", "type": "string"}, + "title": {"title": "Title", "type": "string"}, + }, + "required": ["service", "title"], + "title": "Incident", + "type": "object", + }, + } + ] + ) + + +async def test_events_list_offers_webhook_when_webhooks_are_configured() -> None: + """Configuring `Webhooks` adds webhook to every event type's delivery modes (SDK-defined).""" + events = Events(webhooks=Webhooks()) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + [descriptor] = await client.extension(ClientEvents).list() + + assert descriptor.delivery == ["poll", "webhook"] + + +async def test_poll_returns_emitted_events_that_match_the_subscription_after_the_cursor() -> None: + """A null cursor returns only the current position; the next poll returns the events emitted + since, filtered by `match` and shaped by `transform` (spec-mandated cursor semantics).""" + events = Events() + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.match + def same_service(arguments: ServiceFilter, incident: Incident) -> bool: + return arguments.service == incident.service + + @incidents.transform + def shout(arguments: ServiceFilter, incident: Incident) -> Incident: + return incident.model_copy(update={"title": incident.title.upper()}) + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + incidents.emit(Incident(service="billing", title="before the subscription")) + start = await events_client.poll("incident.created", {"service": "billing"}) + incidents.emit(Occurrence(Incident(service="billing", title="latency"), id="inc-1")) + incidents.emit(Incident(service="search", title="someone else's")) + batch = await events_client.poll("incident.created", {"service": "billing"}, cursor=start.cursor) + + assert start.events == [] + assert [(event.event_id, event.data) for event in batch.events] == [ + ("inc-1", {"service": "billing", "title": "LATENCY"}) + ] + assert (batch.truncated, batch.has_more, batch.next_poll_ms) == (False, False, 5000) + + +async def test_poll_result_on_the_wire_uses_camel_case_and_a_complete_result_type() -> None: + """The poll result carries `resultType: "complete"` and camelCase fields (spec-mandated).""" + events = Events(poll_interval=timedelta(seconds=30)) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + request = PollEventsRequest(params=PollEventsParams(name="incident.created", arguments={"service": "a"})) + result = await client.session.send_request(request, _RawResult) + + dumped = result.model_dump() + dumped.pop("_meta") + assert dumped.pop("cursor") is not None + assert dumped == snapshot( + { + "resultType": "complete", + "events": [], + "truncated": False, + "hasMore": False, + "nextPollMs": 30000, + } + ) + + +async def test_poll_with_max_events_returns_a_partial_batch_and_has_more() -> None: + """When more events are pending than `maxEvents`, the poll returns that many with `hasMore`, + and the next poll continues from the intermediate cursor (spec-mandated).""" + events = Events() + pings = events.event("ping", arguments=NoArguments, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + start = await events_client.poll("ping") + for title in ("a", "b", "c"): + pings.emit(Incident(service="s", title=title)) + first = await events_client.poll("ping", cursor=start.cursor, max_events=2) + rest = await events_client.poll("ping", cursor=first.cursor, max_events=2) + + assert [e.data["title"] for e in first.events] == ["a", "b"] + assert first.has_more is True + assert [e.data["title"] for e in rest.events] == ["c"] + assert rest.has_more is False + + +async def test_poll_with_an_unknown_cursor_restarts_from_now_and_reports_truncated() -> None: + """A cursor the server cannot read (e.g. from before a restart) returns a fresh cursor and + `truncated: true`, never an error (spec-mandated).""" + events = Events() + pings = events.event("ping", arguments=NoArguments, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + pings.emit(Incident(service="s", title="a")) + batch = await events_client.poll("ping", cursor="from-another-server.1") + after = await events_client.poll("ping", cursor=batch.cursor) + + assert (batch.events, batch.truncated) == ([], True) + assert (after.events, after.truncated) == ([], False) + + +async def test_poll_reports_truncated_when_the_buffer_dropped_events_after_the_cursor() -> None: + """Events that fell out of the bounded buffer are a gap: the poll returns what is left with + `truncated: true` (spec-mandated gap signal, SDK-defined buffer).""" + events = Events(buffer_size=2) + pings = events.event("ping", arguments=NoArguments, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + start = await events_client.poll("ping") + for title in ("a", "b", "c"): + pings.emit(Incident(service="s", title=title)) + batch = await events_client.poll("ping", cursor=start.cursor) + + assert [e.data["title"] for e in batch.events] == ["b", "c"] + assert batch.truncated is True + + +async def test_poll_skips_events_older_than_max_age_and_reports_truncated() -> None: + """`maxAgeMs` bounds replay; skipped events are reported as `truncated` (spec-mandated).""" + events = Events() + pings = events.event("ping", arguments=NoArguments, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + start = await events_client.poll("ping") + old = datetime.now(UTC) - timedelta(minutes=5) + pings.emit(Occurrence(Incident(service="s", title="old"), timestamp=old)) + pings.emit(Incident(service="s", title="new")) + batch = await events_client.poll("ping", cursor=start.cursor, max_age=timedelta(minutes=1)) + + assert [e.data["title"] for e in batch.events] == ["new"] + assert batch.truncated is True + + +async def test_emitted_events_older_than_buffer_max_age_are_dropped() -> None: + """The buffer forgets events past `buffer_max_age`, so a reader behind them sees a gap (SDK-defined).""" + events = Events(buffer_max_age=timedelta(minutes=1)) + pings = events.event("ping", arguments=NoArguments, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + start = await events_client.poll("ping") + pings.emit(Occurrence(Incident(service="s", title="stale"), timestamp=datetime.now(UTC) - timedelta(hours=1))) + batch = await events_client.poll("ping", cursor=start.cursor) + + assert (batch.events, batch.truncated) == ([], True) + + +async def test_poll_answers_from_the_check_function_with_the_callers_query() -> None: + """A check function receives the validated arguments, cursor, and principal, and its batch + becomes the poll result; a null cursor still returns no events (SDK-defined, spec-mandated null cursor).""" + queries: list[EventQuery[ServiceFilter]] = [] + events = Events(principal=lambda ctx: "alice") + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.check + async def check(query: EventQuery[ServiceFilter]) -> EventBatch[Incident]: + queries.append(query) + incident = Occurrence(Incident(service=query.arguments.service, title="from upstream"), id="up-1") + return EventBatch(cursor="upstream-7", events=[incident], next_poll=timedelta(seconds=1)) + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + start = await events_client.poll("incident.created", {"service": "billing"}) + batch = await events_client.poll("incident.created", {"service": "billing"}, cursor="upstream-6") + + assert (start.events, start.cursor) == ([], "upstream-7") + assert [(e.event_id, e.data["title"]) for e in batch.events] == [("up-1", "from upstream")] + assert (batch.cursor, batch.next_poll_ms) == ("upstream-7", 1000) + assert [(q.arguments, q.cursor, q.principal, q.max_events) for q in queries] == [ + (ServiceFilter(service="billing"), None, "alice", 100), + (ServiceFilter(service="billing"), "upstream-6", "alice", 100), + ] + + +async def test_poll_for_an_unknown_event_type_raises_not_found() -> None: + """Polling a name the server does not offer is `NotFound` (spec-mandated).""" + mcp = MCPServer("pager", extensions=[Events()]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + with pytest.raises(MCPError) as exc_info: + await client.extension(ClientEvents).poll("nope") + + assert exc_info.value.code == NOT_FOUND + assert exc_info.value.data == {"kind": "event", "name": "nope"} + + +async def test_poll_with_arguments_that_fail_validation_raises_invalid_params() -> None: + """Arguments are validated against the event type's model before any read (SDK-defined).""" + events = Events() + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + with pytest.raises(MCPError) as exc_info: + await client.extension(ClientEvents).poll("incident.created", {"service": 1}) + + assert exc_info.value.code == INVALID_PARAMS + assert exc_info.value.message == snapshot("Invalid arguments for event 'incident.created'") + + +def test_registering_an_event_type_twice_raises() -> None: + """Event type names are unique per server (SDK-defined).""" + events = Events() + events.event("ping", arguments=NoArguments, payload=Incident) + + with pytest.raises(ValueError): + events.event("ping", arguments=NoArguments, payload=Incident) + + +def test_emitting_on_an_event_type_with_a_check_function_raises() -> None: + """A check function owns its event type's events, so `emit` is refused (SDK-defined).""" + events = Events() + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.check + async def check(query: EventQuery[ServiceFilter]) -> EventBatch[Incident]: # pragma: no cover + raise NotImplementedError + + with pytest.raises(RuntimeError): + incidents.emit(Incident(service="s", title="t")) + + +async def test_poll_lease_fires_on_subscribe_once_and_on_unsubscribe_when_polls_stop() -> None: + """Poll subscriptions are leased: `on_subscribe` fires on the first poll for a key, later polls + renew the lease, and `on_unsubscribe` fires once it lapses (spec SDK guidance).""" + seen: list[tuple[str, Subscriber[ServiceFilter]]] = [] + lapsed = anyio.Event() + # Time-based feature: a short lease, so the test waits for its real expiry. + events = Events(poll_lease=timedelta(milliseconds=50)) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def subscribed(subscriber: Subscriber[ServiceFilter]) -> None: + seen.append(("subscribe", subscriber)) + + @incidents.on_unsubscribe + async def unsubscribed(subscriber: Subscriber[ServiceFilter]) -> None: + seen.append(("unsubscribe", subscriber)) + lapsed.set() + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + await events_client.poll("incident.created", {"service": "billing"}) + await events_client.poll("incident.created", {"service": "billing"}) + with anyio.fail_after(5): + await lapsed.wait() + + [(first, subscriber), (second, same)] = seen + assert (first, second) == ("subscribe", "unsubscribe") + assert subscriber is same + assert (subscriber.arguments, subscriber.principal, subscriber.mode) == ( + ServiceFilter(service="billing"), + None, + "poll", + ) + + +async def test_failing_on_subscribe_fails_the_poll_and_runs_again_on_the_next_one() -> None: + """A poll whose `on_subscribe` hook raises does not take the lease, so the next poll retries + the hook (SDK-defined).""" + attempts: list[str] = [] + events = Events() + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def subscribed(subscriber: Subscriber[ServiceFilter]) -> None: + attempts.append(subscriber.id) + if len(attempts) == 1: + raise MCPError(code=INVALID_PARAMS, message="upstream refused") + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + with pytest.raises(MCPError): + await events_client.poll("incident.created", {"service": "billing"}) + await events_client.poll("incident.created", {"service": "billing"}) + assert len(attempts) == 2 + assert attempts[0] == attempts[1] + + +async def test_a_poll_during_another_polls_failing_on_subscribe_runs_the_hook_itself() -> None: + """Concurrent first polls for one key wait for the in-flight `on_subscribe`, so a hook that + rejects one of them is never skipped for the other (SDK-defined; the hook is where access is checked).""" + attempts: list[str] = [] + entered = anyio.Event() + release = anyio.Event() + events = Events() + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def check_access(subscriber: Subscriber[ServiceFilter]) -> None: + attempts.append(subscriber.id) + if len(attempts) == 1: + entered.set() + await release.wait() + raise events_error(FORBIDDEN, reason="access revoked") + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + + async def rejected_poll() -> None: + with pytest.raises(MCPError) as exc_info: + await events_client.poll("incident.created", {"service": "billing"}) + assert exc_info.value.code == FORBIDDEN + + with anyio.fail_after(5): + async with anyio.create_task_group() as task_group: + task_group.start_soon(rejected_poll) + await entered.wait() + task_group.start_soon(events_client.poll, "incident.created", {"service": "billing"}) + await anyio.wait_all_tasks_blocked() + release.set() + assert len(attempts) == 2 + + +async def test_concurrent_connections_share_one_running_events_extension() -> None: + """Every in-memory connection enters the server lifespan; a nested entry reuses the running + extension, so poll leases keep working on the second connection (pins a known shortcut).""" + subscribed: list[Any] = [] + events = Events() + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def on_subscribe(subscriber: Subscriber[ServiceFilter]) -> None: + subscribed.append(subscriber.arguments.service) + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as first, Client(mcp, extensions=[ClientEvents()]) as second: + await first.extension(ClientEvents).poll("incident.created", {"service": "a"}) + await second.extension(ClientEvents).poll("incident.created", {"service": "b"}) + assert subscribed == ["a", "b"] diff --git a/tests/experimental/events/test_webhooks.py b/tests/experimental/events/test_webhooks.py new file mode 100644 index 0000000000..3751a60fb7 --- /dev/null +++ b/tests/experimental/events/test_webhooks.py @@ -0,0 +1,756 @@ +"""Tests for webhook delivery of MCP Events (SEP-3415), server side and the client's `subscribe`. + +Deliveries go through an in-process ASGI transport into a real `WebhookReceiver`. Tests about a +failing or hostile endpoint script it with `httpx2.MockTransport` instead, since a well-behaved +receiver cannot produce those responses. +""" + +import base64 +import json +import socket +import ssl +from datetime import UTC, datetime, timedelta +from typing import Any + +import anyio +import httpx2 +import pytest +from inline_snapshot import snapshot +from mcp_types import INVALID_PARAMS +from pydantic import BaseModel + +from mcp.client.client import Client +from mcp.experimental.events.client import Events as ClientEvents +from mcp.experimental.events.receiver import Delivery, WebhookGap, WebhookReceiver, WebhookTerminated +from mcp.experimental.events.server import EventBatch, EventQuery, Events, Occurrence, Subscriber, Webhooks +from mcp.experimental.events.types import ( + CALLBACK_ENDPOINT_ERROR, + FORBIDDEN, + NOT_FOUND, + UNSUPPORTED, + EventOccurrence, + SubscribeEventsParams, + SubscribeEventsRequest, + SubscribeEventsResult, + UnsubscribeEventsParams, + UnsubscribeEventsRequest, + UnsubscribeEventsResult, + WebhookDelivery, + WebhookTarget, + events_error, +) +from mcp.server.auth.middleware.auth_context import auth_context_var +from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser +from mcp.server.auth.provider import AccessToken +from mcp.server.mcpserver import MCPServer +from mcp.shared.exceptions import MCPError + +pytestmark = pytest.mark.anyio + +SECRET = "whsec_" + base64.b64encode(b"k" * 32).decode() +ROTATED_SECRET = "whsec_" + base64.b64encode(b"r" * 32).decode() +URL = "https://client.test/hooks" + + +class ServiceFilter(BaseModel): + service: str + + +class Incident(BaseModel): + service: str + title: str + + +class Inbox: + """Collects deliveries, and lets a test wait until a number of them arrived.""" + + def __init__(self) -> None: + self.deliveries: list[Delivery] = [] + self._arrived = anyio.Event() + + async def __call__(self, delivery: Delivery) -> None: + self.deliveries.append(delivery) + self._arrived.set() + self._arrived = anyio.Event() + + async def wait_for(self, count: int) -> None: + with anyio.fail_after(5): + while len(self.deliveries) < count: + await self._arrived.wait() + + +def echo_challenge(request: httpx2.Request) -> httpx2.Response: + """Answer the verification request the way a willing endpoint does.""" + challenge = json.loads(request.content)["challenge"] + return httpx2.Response(200, json={"challenge": challenge}) + + +def subscribe_request( + ttl_ms: int | None = None, secret: str = SECRET, url: str = URL, cursor: str | None = None +) -> SubscribeEventsRequest: + params = SubscribeEventsParams( + name="incident.created", + arguments={"service": "billing"}, + delivery=WebhookDelivery(url=url, secret=secret), + cursor=cursor, + ttl_ms=ttl_ms, + ) + return SubscribeEventsRequest(params=params) + + +async def test_webhook_subscription_delivers_matching_events_until_the_client_leaves() -> None: + """Events are POSTed, signed, to the verified callback with the subscription's cursor, and leaving + `subscribe` unsubscribes, ending the subscription (spec-mandated flow). + + Steps: + 1. Subscribe: the server verifies the endpoint, then answers with the subscription id and cursor. + 2. Emit two matching events and one that does not match; the receiver gets the two. + 3. Leave the block: the subscription ends, firing `on_unsubscribe`. + """ + inbox = Inbox() + hooks: list[tuple[str, Subscriber[ServiceFilter]]] = [] + ended = anyio.Event() + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.ASGITransport(app=WebhookReceiver(SECRET, inbox))), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.match + def same_service(arguments: ServiceFilter, incident: Incident) -> bool: + return arguments.service == incident.service + + @incidents.on_subscribe + async def subscribed(subscriber: Subscriber[ServiceFilter]) -> None: + hooks.append(("subscribe", subscriber)) + + @incidents.on_unsubscribe + async def unsubscribed(subscriber: Subscriber[ServiceFilter]) -> None: + hooks.append(("unsubscribe", subscriber)) + ended.set() + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp, extensions=[ClientEvents()]) as client: + events_client = client.extension(ClientEvents) + async with events_client.subscribe("incident.created", {"service": "billing"}, url=URL, secret=SECRET) as sub: + incidents.emit(Occurrence(Incident(service="billing", title="a"), id="inc-1")) + incidents.emit(Incident(service="search", title="not ours")) + incidents.emit(Occurrence(Incident(service="billing", title="b"), id="inc-2")) + await inbox.wait_for(2) + with anyio.fail_after(5): + await ended.wait() + + first, second = inbox.deliveries + assert isinstance(first, EventOccurrence) and isinstance(second, EventOccurrence) + assert [(first.event_id, first.data), (second.event_id, second.data)] == [ + ("inc-1", {"service": "billing", "title": "a"}), + ("inc-2", {"service": "billing", "title": "b"}), + ] + assert first.cursor == sub.cursor + assert second.cursor is not None and second.cursor != sub.cursor + [(_, subscriber), (_, same)] = hooks + assert subscriber is same + assert (subscriber.id, subscriber.principal, subscriber.mode) == (sub.id, "alice", "webhook") + + +async def test_resubscribing_refreshes_the_subscription_and_signs_with_both_secrets_during_rotation() -> None: + """Subscribing again with the same key refreshes the same subscription and reports its delivery + status; after a secret change, deliveries carry both signatures for a grace period (spec-mandated).""" + inbox = Inbox() + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.ASGITransport(app=WebhookReceiver(SECRET, inbox))), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + created = await client.session.send_request(subscribe_request(), SubscribeEventsResult) + refreshed = await client.session.send_request(subscribe_request(secret=ROTATED_SECRET), SubscribeEventsResult) + incidents.emit(Incident(service="billing", title="a")) + await inbox.wait_for(1) + again = await client.session.send_request(subscribe_request(secret=ROTATED_SECRET), SubscribeEventsResult) + assert refreshed.id == created.id + assert created.delivery_status is None + assert refreshed.delivery_status is not None + assert refreshed.delivery_status.model_dump(exclude_none=True) == snapshot({"active": True}) + assert again.delivery_status is not None and again.delivery_status.last_delivery_at is not None + + +async def test_unsubscribing_an_unknown_subscription_raises_not_found() -> None: + """`events/unsubscribe` for a key with no subscription is `NotFound` (spec-mandated).""" + events = Events(webhooks=Webhooks(), principal=lambda ctx: "alice") + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + params = UnsubscribeEventsParams( + name="incident.created", arguments={"service": "billing"}, delivery=WebhookTarget(url=URL) + ) + + async with Client(mcp) as client: + with pytest.raises(MCPError) as exc_info: + await client.session.send_request(UnsubscribeEventsRequest(params=params), UnsubscribeEventsResult) + + assert exc_info.value.code == NOT_FOUND + + +async def test_a_verified_callback_is_verified_again_only_after_its_last_subscription_ends() -> None: + """Verification is per principal and callback URL, so another subscription to the same endpoint + skips the challenge, and the cache lives only as long as those subscriptions (spec-mandated + scope of verification).""" + challenges: list[str] = [] + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + challenges.append(json.loads(request.content)["challenge"]) + return echo_challenge(request) + + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint)), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + ) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + search = subscribe_request() + search.params.arguments = {"service": "search"} + + async with Client(mcp) as client: + billing_sub = await client.session.send_request(subscribe_request(), SubscribeEventsResult) + search_sub = await client.session.send_request(search, SubscribeEventsResult) + assert billing_sub.id != search_sub.id + assert len(challenges) == 1 + + for request in (subscribe_request(), search): + unsubscribe = UnsubscribeEventsParams( + name="incident.created", arguments=request.params.arguments, delivery=WebhookTarget(url=URL) + ) + await client.session.send_request(UnsubscribeEventsRequest(params=unsubscribe), UnsubscribeEventsResult) + await anyio.wait_all_tasks_blocked() + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + assert len(challenges) == 2 + + +async def test_concurrent_identical_subscribes_upsert_one_subscription() -> None: + """Two `events/subscribe` calls for one key that race each other create a single subscription: + one verification and one `on_subscribe` (spec-mandated idempotent upsert).""" + challenges: list[str] = [] + subscribed: list[str] = [] + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + challenges.append(json.loads(request.content)["challenge"]) + await anyio.wait_all_tasks_blocked() + return echo_challenge(request) + + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint)), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def on_subscribe(subscriber: Subscriber[ServiceFilter]) -> None: + subscribed.append(subscriber.id) + + mcp = MCPServer("pager", extensions=[events]) + results: list[SubscribeEventsResult] = [] + + async with Client(mcp) as client: + + async def subscribe() -> None: + results.append(await client.session.send_request(subscribe_request(), SubscribeEventsResult)) + + async with anyio.create_task_group() as task_group: + task_group.start_soon(subscribe) + task_group.start_soon(subscribe) + + assert len(challenges) == 1 + assert len(subscribed) == 1 + assert results[0].id == results[1].id + + +async def test_an_on_subscribe_hook_that_raises_rejects_the_subscription() -> None: + """`on_subscribe` is where the server checks the principal's access: raising `Forbidden` fails + `events/subscribe` and leaves no subscription behind (spec-mandated subscribe-time authorization).""" + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(echo_challenge)), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "mallory", + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def check_access(subscriber: Subscriber[ServiceFilter]) -> None: + raise events_error(FORBIDDEN, reason=f"{subscriber.principal} cannot see {subscriber.arguments.service}") + + mcp = MCPServer("pager", extensions=[events]) + unsubscribe = UnsubscribeEventsParams( + name="incident.created", arguments={"service": "billing"}, delivery=WebhookTarget(url=URL) + ) + + async with Client(mcp) as client: + with pytest.raises(MCPError) as rejected: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + with pytest.raises(MCPError) as missing: + await client.session.send_request(UnsubscribeEventsRequest(params=unsubscribe), UnsubscribeEventsResult) + assert rejected.value.error.model_dump(exclude_none=True) == snapshot( + {"code": -32024, "message": "Forbidden", "data": {"reason": "mallory cannot see billing"}} + ) + assert missing.value.code == NOT_FOUND + + +async def test_subscribe_keeps_the_supplied_cursor_until_the_first_batch_is_delivered() -> None: + """The subscribe response's cursor must be safe to persist, so while the first batch is still + undelivered it stays the client's own, even when the batch also reports a gap (spec-mandated + watermark).""" + inbox = Inbox() + answers = [ + EventBatch(cursor="c9", events=[Occurrence(Incident(service="billing", title="a"), id="e1")], truncated=True) + ] + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.ASGITransport(app=WebhookReceiver(SECRET, inbox))), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.check + async def check(query: EventQuery[ServiceFilter]) -> EventBatch[Incident]: + return answers.pop() if answers else EventBatch(cursor="c9") + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + result = await client.session.send_request(subscribe_request(cursor="c1"), SubscribeEventsResult) + await inbox.wait_for(1) + assert (result.cursor, result.truncated) == ("c1", True) + [event] = inbox.deliveries + assert isinstance(event, EventOccurrence) and event.cursor == "c9" + + +async def test_webhook_calls_without_an_authenticated_principal_raise_forbidden() -> None: + """Webhook subscriptions are keyed by principal, so unauthenticated callers are refused (spec-mandated).""" + events = Events(webhooks=Webhooks()) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + unsubscribe = UnsubscribeEventsParams( + name="incident.created", arguments={"service": "billing"}, delivery=WebhookTarget(url=URL) + ) + + async with Client(mcp) as client: + with pytest.raises(MCPError) as subscribe_error: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + with pytest.raises(MCPError) as unsubscribe_error: + request = UnsubscribeEventsRequest(params=unsubscribe) + await client.session.send_request(request, UnsubscribeEventsResult) + + assert subscribe_error.value.code == FORBIDDEN + assert unsubscribe_error.value.code == FORBIDDEN + + +async def test_webhook_subscriptions_belong_to_the_access_token_principal_by_default() -> None: + """Without a `principal` resolver, the caller's OAuth client, issuer, and subject identify it (SDK-defined).""" + subscribers: list[Subscriber[ServiceFilter]] = [] + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(echo_challenge)), + trusted_origins=["https://client.test"], + ) + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_subscribe + async def subscribed(subscriber: Subscriber[ServiceFilter]) -> None: + subscribers.append(subscriber) + + mcp = MCPServer("pager", extensions=[events]) + token = AccessToken(token="t", client_id="app", scopes=[], subject="alice", claims={"iss": "https://idp.test"}) + reset = auth_context_var.set(AuthenticatedUser(token)) + try: + async with Client(mcp) as client: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + assert [s.principal for s in subscribers] == ['["app","https://idp.test","alice"]'] + finally: + auth_context_var.reset(reset) # pragma: lax no cover - python/cpython#106749: 3.11 drops this line event + + +async def test_subscribe_with_a_delivery_mode_other_than_webhook_raises_unsupported() -> None: + """`events/subscribe` only serves webhook delivery (spec-mandated `Unsupported`).""" + events = Events(webhooks=Webhooks(), principal=lambda ctx: "alice") + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + request = subscribe_request() + request.params.delivery.mode = "carrier-pigeon" + + async with Client(mcp) as client: + with pytest.raises(MCPError) as exc_info: + await client.session.send_request(request, SubscribeEventsResult) + + assert exc_info.value.code == UNSUPPORTED + assert exc_info.value.data == {"feature": "deliveryMode", "value": "carrier-pigeon"} + + +@pytest.mark.parametrize( + ("url", "secret"), + [ + pytest.param("http://client.test/hooks", SECRET, id="plain-http"), + pytest.param("/hooks", SECRET, id="relative-url"), + pytest.param("https://localhost/hooks", SECRET, id="loopback-host"), + pytest.param(URL, "not-a-whsec-secret", id="secret-prefix"), + pytest.param(URL, "whsec_!!!", id="secret-not-base64"), + pytest.param(URL, "whsec_" + base64.b64encode(b"short").decode(), id="secret-too-short"), + ], +) +async def test_subscribe_with_an_unsafe_callback_or_malformed_secret_raises_invalid_params( + url: str, secret: str +) -> None: + """Callback URLs must be https and publicly routable, and secrets `whsec_` base64 of 24-64 bytes + (spec-mandated SSRF rules and Standard Webhooks format).""" + events = Events( + webhooks=Webhooks(http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(echo_challenge))), + principal=lambda ctx: "alice", + ) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + with pytest.raises(MCPError) as exc_info: + await client.session.send_request(subscribe_request(url=url, secret=secret), SubscribeEventsResult) + + assert exc_info.value.code == INVALID_PARAMS + + +_TLS_FAILURE = httpx2.ConnectError("certificate verify failed") +_TLS_FAILURE.__cause__ = ssl.SSLError("bad certificate") + + +@pytest.mark.parametrize( + ("outcome", "reason"), + [ + pytest.param(httpx2.Response(503), "http_5xx", id="5xx"), + pytest.param(httpx2.Response(404), "http_4xx", id="4xx"), + pytest.param(httpx2.Response(200, json={"challenge": "guessed"}), "challenge_failed", id="wrong-challenge"), + pytest.param(httpx2.Response(200, text="ok"), "challenge_failed", id="not-json"), + pytest.param(httpx2.Response(200, json=["ok"]), "challenge_failed", id="json-without-challenge"), + pytest.param(httpx2.ConnectError("connection refused"), "connection_refused", id="refused"), + pytest.param(_TLS_FAILURE, "tls_error", id="tls"), + pytest.param(httpx2.ReadTimeout("timed out"), "timeout", id="timeout"), + ], +) +async def test_subscribe_to_an_endpoint_that_fails_verification_raises_callback_endpoint_error( + outcome: httpx2.Response | Exception, reason: str +) -> None: + """A callback that does not echo the verification challenge is refused with a categorized reason, + never the endpoint's own response (spec-mandated).""" + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + if isinstance(outcome, Exception): + raise outcome + return outcome + + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint)), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + ) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + with pytest.raises(MCPError) as exc_info: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + + assert exc_info.value.code == CALLBACK_ENDPOINT_ERROR + assert exc_info.value.data == {"reason": reason} + + +@pytest.mark.parametrize( + ("family", "address"), + [ + pytest.param(socket.AF_INET, "93.184.215.14", id="ipv4"), + pytest.param(socket.AF_INET6, "2606:2800:21f:cb07:6820:80da:af6b:8b2c", id="ipv6"), + ], +) +async def test_callbacks_connect_to_the_checked_public_address_with_the_original_host( + monkeypatch: pytest.MonkeyPatch, family: socket.AddressFamily, address: str +) -> None: + """The server connects to the address it validated, keeping `Host` and TLS SNI, so a DNS + answer cannot change between the check and the connection (spec-mandated SSRF rule).""" + seen: list[httpx2.Request] = [] + + async def resolve(host: str, port: int, **kwargs: Any) -> list[Any]: + assert (host, port) == ("hooks.example.com", 443) + return [(family, socket.SOCK_STREAM, 6, "", (address, port))] + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + seen.append(request) + return echo_challenge(request) + + monkeypatch.setattr(anyio, "getaddrinfo", resolve) + events = Events( + webhooks=Webhooks(http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint))), + principal=lambda ctx: "alice", + ) + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + await client.session.send_request(subscribe_request(url="https://hooks.example.com/mcp"), SubscribeEventsResult) + [verification] = seen + assert (verification.url.host, verification.url.path) == (address, "/mcp") + assert verification.headers["host"] == "hooks.example.com" + assert verification.extensions["sni_hostname"] == "hooks.example.com" + + +async def test_subscribe_to_a_host_that_does_not_resolve_raises_callback_endpoint_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unresolvable callback host is reported as `connection_refused` (spec-mandated categories).""" + + async def resolve(host: str, port: int, **kwargs: Any) -> list[Any]: + raise socket.gaierror("no such host") + + monkeypatch.setattr(anyio, "getaddrinfo", resolve) + events = Events(webhooks=Webhooks(), principal=lambda ctx: "alice") + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + with pytest.raises(MCPError) as exc_info: + await client.session.send_request(subscribe_request(url="https://nowhere.test/"), SubscribeEventsResult) + + assert exc_info.value.data == {"reason": "connection_refused"} + + +async def test_a_callback_host_that_rebinds_to_a_private_address_gets_no_deliveries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Addresses are checked again on every delivery, so a host that passed verification and then + resolves to a private address is not contacted (spec-mandated delivery-time SSRF check).""" + answers = ["93.184.215.14", "127.0.0.1"] + rebound = anyio.Event() + posted: list[httpx2.Request] = [] + + async def resolve(host: str, port: int, **kwargs: Any) -> list[Any]: + address = answers.pop(0) + if not answers: + rebound.set() + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, port))] + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + posted.append(request) + return echo_challenge(request) + + monkeypatch.setattr(anyio, "getaddrinfo", resolve) + events = Events( + webhooks=Webhooks(http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint))), + principal=lambda ctx: "alice", + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + request = subscribe_request(url="https://hooks.example.com/mcp") + + async with Client(mcp) as client: + await client.session.send_request(request, SubscribeEventsResult) + incidents.emit(Incident(service="billing", title="a")) + with anyio.fail_after(5): + await rebound.wait() + refreshed = await client.session.send_request(request, SubscribeEventsResult) + assert len(posted) == 1 + assert refreshed.delivery_status is not None + assert refreshed.delivery_status.last_error == "connection_refused" + + +async def test_failed_deliveries_are_retried_then_abandoned_and_reported_on_refresh() -> None: + """A delivery is retried with the same `webhook-id`, abandoned after the last retry, and the + failure shows in `deliveryStatus` until a later delivery succeeds (spec-mandated).""" + attempts: list[tuple[str, str]] = [] + third_attempt = anyio.Event() + release = anyio.Event() + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + body = json.loads(request.content) + if body.get("type") == "verification": + return echo_challenge(request) + attempts.append((request.headers["webhook-id"], body["data"]["title"])) + if len(attempts) == 1: + raise httpx2.ConnectError("connection refused") + if len(attempts) == 2: + return httpx2.Response(503) + third_attempt.set() + await release.wait() + return httpx2.Response(204) + + webhooks = Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint)), + trusted_origins=["https://client.test"], + retry_delays=[timedelta(0)], + ) + events = Events(webhooks=webhooks, principal=lambda ctx: "alice") + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + incidents.emit(Incident(service="billing", title="lost")) + incidents.emit(Incident(service="billing", title="kept")) + with anyio.fail_after(5): + await third_attempt.wait() + failing = await client.session.send_request(subscribe_request(), SubscribeEventsResult) + release.set() + assert [title for _, title in attempts] == ["lost", "lost", "kept"] + assert attempts[0][0] == attempts[1][0] != attempts[2][0] + assert failing.delivery_status is not None + assert failing.delivery_status.last_error == "http_5xx" + assert failing.delivery_status.failed_since is not None + + +async def test_a_gone_response_abandons_the_delivery_without_retrying() -> None: + """`410 Gone` from the callback stops retries for that delivery (SDK-defined, after Standard Webhooks).""" + titles: list[str] = [] + second = anyio.Event() + + async def endpoint(request: httpx2.Request) -> httpx2.Response: + body = json.loads(request.content) + if body.get("type") == "verification": + return echo_challenge(request) + titles.append(body["data"]["title"]) + if len(titles) == 2: + second.set() + return httpx2.Response(410 if len(titles) == 1 else 204) + + webhooks = Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(endpoint)), + trusted_origins=["https://client.test"], + retry_delays=[timedelta(0)], + ) + events = Events(webhooks=webhooks, principal=lambda ctx: "alice") + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + incidents.emit(Incident(service="billing", title="gone")) + incidents.emit(Incident(service="billing", title="next")) + with anyio.fail_after(5): + await second.wait() + assert titles == ["gone", "next"] + + +async def test_requested_ttls_are_clamped_to_the_servers_bounds() -> None: + """The server grants the suggested `ttlMs` within `min_ttl`..`max_ttl` and answers with + `refreshBefore` (spec-mandated TTL negotiation; this tests a time-based feature).""" + webhooks = Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(echo_challenge)), + trusted_origins=["https://client.test"], + min_ttl=timedelta(hours=1), + max_ttl=timedelta(hours=2), + ) + events = Events(webhooks=webhooks, principal=lambda ctx: "alice") + events.event("incident.created", arguments=ServiceFilter, payload=Incident) + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + before = datetime.now(UTC) + short = await client.session.send_request(subscribe_request(ttl_ms=1), SubscribeEventsResult) + long = await client.session.send_request(subscribe_request(ttl_ms=10**9), SubscribeEventsResult) + assert short.refresh_before is not None and long.refresh_before is not None + assert timedelta(minutes=59) < short.refresh_before - before <= timedelta(hours=1, minutes=1) + assert timedelta(hours=1, minutes=59) < long.refresh_before - before <= timedelta(hours=2, minutes=1) + + +async def test_a_subscription_that_is_not_refreshed_ends_when_its_ttl_lapses() -> None: + """An unrefreshed subscription expires, firing `on_unsubscribe` (spec-mandated; a time-based feature, + so the test waits for a 50 ms TTL to really lapse).""" + lapsed = anyio.Event() + webhooks = Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(echo_challenge)), + trusted_origins=["https://client.test"], + min_ttl=timedelta(0), + ) + events = Events(webhooks=webhooks, principal=lambda ctx: "alice") + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.on_unsubscribe + async def unsubscribed(subscriber: Subscriber[ServiceFilter]) -> None: + lapsed.set() + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + await client.session.send_request(subscribe_request(ttl_ms=50), SubscribeEventsResult) + with anyio.fail_after(5): + await lapsed.wait() + + +async def test_check_function_subscriptions_deliver_gaps_survive_crashes_and_end_on_errors() -> None: + """A webhook loop over a check function drains `hasMore` batches, POSTs a gap for `truncated`, + logs and retries a crashing read, and terminates the subscription on an `MCPError` (spec-mandated + control messages; SDK-defined crash handling). + + Steps: + 1. Subscribe: the first read gives the starting cursor. + 2. The loop reads one event with `has_more`, then crashes, then reads a truncated batch, then fails. + 3. The receiver gets the event, a gap, and a terminated message; `on_unsubscribe` fires. + """ + inbox = Inbox() + ended = anyio.Event() + reads: list[str | None] = [] + answers: list[EventBatch[Incident] | Exception] = [ + EventBatch(cursor="c0"), + EventBatch(cursor="c1", events=[Occurrence(Incident(service="billing", title="a"), id="e1")], has_more=True), + RuntimeError("upstream crashed"), + EventBatch(cursor="c3", truncated=True), + MCPError(code=FORBIDDEN, message="Forbidden", data={"reason": "access revoked"}), + ] + events = Events( + webhooks=Webhooks( + http_client=httpx2.AsyncClient(transport=httpx2.ASGITransport(app=WebhookReceiver(SECRET, inbox))), + trusted_origins=["https://client.test"], + ), + principal=lambda ctx: "alice", + poll_interval=timedelta(0), + ) + incidents = events.event("incident.created", arguments=ServiceFilter, payload=Incident) + + @incidents.check + async def check(query: EventQuery[ServiceFilter]) -> EventBatch[Incident]: + reads.append(query.cursor) + answer = answers.pop(0) + if isinstance(answer, Exception): + raise answer + return answer + + @incidents.on_unsubscribe + async def unsubscribed(subscriber: Subscriber[ServiceFilter]) -> None: + ended.set() + + mcp = MCPServer("pager", extensions=[events]) + + async with Client(mcp) as client: + await client.session.send_request(subscribe_request(), SubscribeEventsResult) + with anyio.fail_after(5): + await ended.wait() + + assert reads == [None, "c0", "c1", "c1", "c3"] + event, gap, terminated = inbox.deliveries + assert isinstance(event, EventOccurrence) and event.event_id == "e1" + assert isinstance(gap, WebhookGap) and gap.cursor == "c3" + assert isinstance(terminated, WebhookTerminated) + assert terminated.error == {"code": FORBIDDEN, "message": "Forbidden", "data": {"reason": "access revoked"}} diff --git a/tests/server/mcpserver/test_extension.py b/tests/server/mcpserver/test_extension.py index 0d665a3d6e..9e8038d788 100644 --- a/tests/server/mcpserver/test_extension.py +++ b/tests/server/mcpserver/test_extension.py @@ -5,6 +5,8 @@ the highest-level public surface (in-memory `Client`). """ +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from dataclasses import replace from typing import Any, Literal @@ -501,3 +503,28 @@ async def test_require_client_extension_raises_minus_32021_when_client_did_not_d assert exc_info.value.code == MISSING_REQUIRED_CLIENT_CAPABILITY assert exc_info.value.error.data == snapshot({"requiredCapabilities": {"extensions": {_NEEDS_EXT: {}}}}) + + +async def test_extension_lifespan_wraps_the_server_lifespan() -> None: + """An extension's `lifespan()` is entered before the server's own lifespan and exited after it (SDK-defined).""" + order: list[str] = [] + + class _BackgroundExt(Extension): + identifier = "com.example/background" + + @asynccontextmanager + async def lifespan(self) -> AsyncIterator[None]: + order.append("extension enter") + yield + order.append("extension exit") + + @asynccontextmanager + async def server_lifespan(server: MCPServer) -> AsyncIterator[dict[str, Any]]: + order.append("server enter") + yield {} + order.append("server exit") + + mcp = MCPServer("test", lifespan=server_lifespan, extensions=[_BackgroundExt()]) + async with Client(mcp): + assert order == ["extension enter", "server enter"] + assert order == ["extension enter", "server enter", "server exit", "extension exit"]