From 1bef7fcd0e2a17240ffb5c9881b0de7dd49823f9 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Mon, 12 Oct 2026 21:33:09 +0100 Subject: [PATCH] feat: push delivery for MCP Events behind the events-push flag Clients that hold a connection open no longer need a public webhook endpoint: events/stream delivers a subscription's events on the request itself. Each stream owns one leased Appwrite webhook, and events reach the replica holding the stream over Redis-protocol pub/sub (Dragonfly in the cluster). Closes #137 --- .github/workflows/ci.yml | 14 + AGENTS.md | 6 +- compose.yaml | 18 + docs/development.md | 8 + docs/events.md | 172 ++++- docs/flags.md | 44 ++ pyproject.toml | 4 +- src/mcp_server_appwrite/events/bus.py | 208 ++++++ src/mcp_server_appwrite/events/delivery.py | 21 +- src/mcp_server_appwrite/events/envelope.py | 82 ++- src/mcp_server_appwrite/events/errors.py | 7 +- src/mcp_server_appwrite/events/ingress.py | 95 ++- src/mcp_server_appwrite/events/projection.py | 5 + src/mcp_server_appwrite/events/protocol.py | 121 +++- src/mcp_server_appwrite/events/streams.py | 480 +++++++++++++ .../events/subscriptions.py | 215 ++++-- src/mcp_server_appwrite/events/webhooks.py | 13 +- src/mcp_server_appwrite/flags.py | 11 +- src/mcp_server_appwrite/http_app.py | 59 +- src/mcp_server_appwrite/server.py | 11 +- src/mcp_server_appwrite/telemetry.py | 24 +- tests/e2e/support.py | 230 +++++- tests/e2e/test_events_stream.py | 657 ++++++++++++++++++ uv.lock | 19 +- 24 files changed, 2323 insertions(+), 201 deletions(-) create mode 100644 src/mcp_server_appwrite/events/bus.py create mode 100644 src/mcp_server_appwrite/events/streams.py create mode 100644 tests/e2e/test_events_stream.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 60e0f2d..462ec32 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -76,6 +76,20 @@ jobs: with: version: "0.11.22" + - name: Install Dragonfly + # The push delivery flows start their own bus, on the Dragonfly + # version the clusters run. + env: + VERSION: v2.0.0 + SHA256: 31060f32858f99a751cae8b54b7b5b556a9b7eb855c30b79bfc3ab7acb26928a + run: | + curl -fsSL -o dragonfly.tar.gz "https://github.com/dragonflydb/dragonfly/releases/download/$VERSION/dragonfly-x86_64.tar.gz" + echo "$SHA256 dragonfly.tar.gz" | sha256sum -c - + tar -xzf dragonfly.tar.gz dragonfly-x86_64 + sudo install dragonfly-x86_64 /usr/local/bin/dragonfly + rm dragonfly.tar.gz dragonfly-x86_64 + dragonfly --version + - name: Install dependencies run: uv sync --frozen --group e2e diff --git a/AGENTS.md b/AGENTS.md index 77b0ba2..7ac9235 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,7 +34,7 @@ Source lives in `src/mcp_server_appwrite/`: | `docs_search.py` | In-process semantic docs search (`appwrite_search_docs`) over a prebuilt index. | | `docs_source.py` | Fetches the published docs (`/docs/llms.txt` + per-page `.md` exports on appwrite.io) and chunks them for the index build. | | `docs_index.py` | Writes the index artifact deterministically and atomically: page and chunk hashes, reuse of cached chunk vectors, build lock, and the added/changed/removed page report. | -| `events/` | MCP Events (behind `flags.EVENTS`, HTTP only): `catalog.py` (event catalog), `protocol.py` (capability, `events/list` and the subscribe methods), `subscriptions.py` + `webhooks.py` (`events/subscribe` / `events/unsubscribe`: one managed Appwrite webhook per subscription), `envelope.py` (subscription state sealed with the `mcp-event-subscriptions` JWE profile), `egress.py` + `delivery.py` (SSRF-safe signed delivery), `projection.py` + `ingress.py` (Appwrite webhook ingress at `/appwrite/webhooks/{id}`). See [docs/events.md](docs/events.md). | +| `events/` | MCP Events (behind `flags.EVENTS`, HTTP only): `catalog.py` (event catalog), `protocol.py` (capability, `events/list` and the subscribe methods), `subscriptions.py` + `webhooks.py` (`events/subscribe` / `events/unsubscribe`: one managed Appwrite webhook per subscription), `envelope.py` (subscription state sealed with the `mcp-event-subscriptions` JWE profile), `egress.py` + `delivery.py` (SSRF-safe signed delivery), `projection.py` + `ingress.py` (Appwrite webhook ingress at `/appwrite/webhooks/{id}`), `streams.py` + `bus.py` (push delivery: `events/stream`, behind `flags.EVENTS_PUSH`, with events routed between replicas over Redis pub/sub). See [docs/events.md](docs/events.md). | | `telemetry.py` | OpenTelemetry metrics layer (OTLP/HTTP). No-op unless an OTLP endpoint is configured and the transport is `http`. | | `data/` | Committed docs index artifact (`docs_index.npz`: vectors, chunk map, chunk hashes, embedded `meta.json`), shipped in the wheel/image. | @@ -121,7 +121,9 @@ Run these locally before opening a PR. They mirror the `CI` workflow ``` Boots the real hosted HTTP app on a random localhost port and drives it over real HTTP the way an MCP client does. No credentials: only the OAuth token - verifier is stubbed. Runs on every PR, forks included. Prefer adding + verifier is stubbed. Runs on every PR, forks included. The push delivery + flows start a real `dragonfly` (CI) or `redis-server`, which must be on + `PATH` (`brew install redis` or `apt-get install redis-server`). Prefer adding assertions to an existing flow over adding unit tests; add a unit test only for what no e2e flow can reach, and say why in its module docstring. diff --git a/compose.yaml b/compose.yaml index 1c06373..ee06e34 100644 --- a/compose.yaml +++ b/compose.yaml @@ -18,6 +18,13 @@ services: OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} SENTRY_DSN: ${SENTRY_DSN:-} SENTRY_ENVIRONMENT: ${SENTRY_ENVIRONMENT:-development} + MCP_EVENTS: ${MCP_EVENTS:-} + MCP_EVENTS_PUSH: ${MCP_EVENTS_PUSH:-} + MCP_EVENTS_SEALING_KEYS: ${MCP_EVENTS_SEALING_KEYS:-} + MCP_EVENTS_REDIS_URL: redis://dragonfly:6379/0 + depends_on: + dragonfly: + condition: service_healthy healthcheck: test: [ @@ -30,3 +37,14 @@ services: timeout: 3s retries: 3 start_period: 10s + + # Carries push events between replicas (MCP_EVENTS_PUSH), as in the cluster; + # any Redis-protocol server works. See docs/events.md. + dragonfly: + image: docker.dragonflydb.io/dragonflydb/dragonfly:v2.0.0 + command: ["dragonfly", "--logtostderr", "--dbfilename=", "--proactor_threads=2"] + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 5 diff --git a/docs/development.md b/docs/development.md index 56f13c2..c47925d 100644 --- a/docs/development.md +++ b/docs/development.md @@ -45,6 +45,10 @@ docker compose up --build > then move it first, then remove the old one 25 hours after that reached every > replica (see "Key rotation" in [events.md](events.md)). +> MCP Events push delivery (`--events-push`) routes events between replicas +> through a Redis-protocol server (Dragonfly in the cluster) at +> `MCP_EVENTS_REDIS_URL`. Compose runs Dragonfly and points the server at it; see [flags.md](flags.md#--events-push--push-delivery-in-progress). + **`uv` directly — HTTP:** ```bash @@ -73,6 +77,10 @@ E2E tests boot the real hosted HTTP app on a random localhost port and drive it over real HTTP like an MCP client. Only the OAuth token verifier is stubbed, so they need no credentials and run on every PR. +The push-delivery flows also start a real bus on a random port: `dragonfly` when +it is on `PATH`, as in CI, otherwise `redis-server` (`brew install redis` or +`apt-get install redis-server`; Dragonfly has no macOS build). + Integration tests create and delete **real** Appwrite resources. They authenticate via `APPWRITE_PROJECT_ID`, `APPWRITE_API_KEY`, `APPWRITE_ENDPOINT` (shell or `.env`) and are skipped when no credentials are present. diff --git a/docs/events.md b/docs/events.md index 9f43003..a7e7546 100644 --- a/docs/events.md +++ b/docs/events.md @@ -7,13 +7,20 @@ > lets a webhook's `authPassword` hold the sealed subscription. [MCP Events](https://developers.openai.com/plugins/build/mcp-events) let an MCP -client such as ChatGPT subscribe to things that happen in an Appwrite project -and receive a signed webhook when they occur, for example "when a function -execution fails in project X, read the logs and open a fix". +client subscribe to things that happen in an Appwrite project and hear about +them when they occur, for example "when a function execution fails in project +X, read the logs and open a fix". -Events are served only by the hosted HTTP server (`mcp.appwrite.io`), and v1 -delivers only by webhook. Each subscription is backed by one Appwrite webhook in -the subscriber's project, so the MCP server keeps no subscription store. +Events are served only by the hosted HTTP server (`mcp.appwrite.io`), in two +delivery modes: + +| Mode | Method | For | +| --- | --- | --- | +| `webhook` | `events/subscribe` | Clients with a public HTTPS endpoint, such as ChatGPT: a signed POST per event. | +| `push` | `events/stream` | Clients on a user's machine, such as Codex CLI or [`pi-mcp-events`](https://github.com/mikekelly/pi-mcp-events): events arrive on a request the client holds open. Behind its own flag; see [Push delivery](#push-delivery). | + +Each subscription is backed by one Appwrite webhook in the subscriber's +project, so the MCP server keeps no subscription store. ## Testing flag @@ -68,7 +75,7 @@ they are kept behind constants in | NotFound | `-32011` | `kind` | | Forbidden | `-32012` | | | ResourceExhausted | `-32013` | `limit`, `max` | -| Unsupported | `-32014` | | +| Unsupported | `-32014` | `feature`, `value` (`{"feature": "deliveryMode", "value": "push"}` when `events/subscribe` is asked for push) | | CallbackEndpointError | `-32015` | `reason`: `challenge_failed`, `timeout`, `connection_refused`, `tls_error`, `http_4xx`, `http_5xx` | Failures on our side, including Appwrite answering `5xx`, are `-32603`. @@ -93,7 +100,8 @@ subscriptions. `events/subscribe`, in order: 1. Looks the event up (`-32011`, `data.kind = "event"`) and checks its - arguments (`-32602`). `delivery.mode` must be `webhook` (`-32014`). The + arguments (`-32602`). `delivery.mode` must be `webhook` (`-32014`; push + goes through [`events/stream`](#push-delivery)). The callback must be `https`, without credentials, and resolve only to public addresses (`-32602`). The secret must be `whsec_` + base64 of 24 to 64 bytes (`-32602`). @@ -163,7 +171,7 @@ so `maxAgeMs` is accepted and has no effect. | Field | Value | | --- | --- | | `$id` | The subscription id | -| `name` | `MCP events \| \| until \| owner ` | +| `name` | `MCP events \| \| until \| owner ` (`MCP events push \| …` for a [push stream](#push-delivery)) | | `url` | `{MCP_PUBLIC_URL}/appwrite/webhooks/{id}` | | `events` | The event's Appwrite patterns | | `tls` | `true` | @@ -243,6 +251,114 @@ caller never hold a slot. The server already advertises these. A user can narrow them at consent, and the error message names the scope to grant. +## Push delivery + +``` +events/stream {name, arguments, cursor, maxAgeMs?} + -> notifications/events/active {cursor: null, truncated: false} + notifications/events/event {eventId, name, timestamp, data, cursor: null} + notifications/events/heartbeat {cursor: null} every 25 s + notifications/events/error {error} the stream stays open + notifications/events/terminated {error} the stream ends +``` + +A push subscription is one long-lived `events/stream` request. On Streamable +HTTP it is a POST whose SSE response carries only that subscription's +notifications, each with the request id in +`params._meta["io.modelcontextprotocol/subscriptionId"]`. Heartbeats are SSE +`data:` frames, not comments. The client cancels by aborting the request; no +result is sent then. When the server ends the stream (termination, shutdown, token +expiry) it sends the empty result `{"_meta": {}}` as the final frame. + +Push sits behind the `--events-push` flag +([flags.md](flags.md#--events-push--push-delivery-in-progress)) and needs a +Redis-protocol server (`MCP_EVENTS_REDIS_URL`): Dragonfly in the cluster, or +Redis; the bus uses only `PUBLISH`, `SUBSCRIBE` and `PING`. `events/list` advertises +`"delivery": ["webhook", "push"]` only when both are there; otherwise it +advertises `["webhook"]` and `events/stream` is `-32601`. + +### Lifecycle + +1. **Validate.** Exactly like `events/subscribe`, minus the callback, secret + and handshake: the event (`-32011`), its arguments, `cursor` (a string or + `null`) and `maxAgeMs` (`-32602`), a user behind the token and the webhook + write scope (`-32012`). An invalid subscription is an immediate JSON-RPC + error and no stream is opened (on the 2026-07-28 protocol the answer is + plain JSON, unless Appwrite takes longer than the SDK's 15-second SSE + deferral to refuse it; on 2025-era protocols it is a single SSE frame). +2. **Listen.** The stream gets a random subscription id (`sub_` + 32 hex + characters): push is connection-scoped, so two streams with the same + arguments are two subscriptions. The replica starts listening on the bus + under that id before anything else, so no event is missed. +3. **Create the webhook.** One Appwrite webhook per stream, written like a + [managed webhook](#the-managed-webhook) with the caller's token after the + same authorization read, webhook listing and cleanup as a subscribe + (`-32011` for a missing resource, `-32012` without access, `-32013` at the + plan limit). Its name starts `MCP events push`, its envelope's record says + `delivery: {"mode": "push"}`, with no callback or secret, and its lease is 5 + minutes. A client that aborts while the webhook is being created waits for + the create, so the cleanup that follows always finds it. +4. **Serve.** `active` first, then every event from the bus, a heartbeat every + 25 seconds, and a lease renewal every 2 minutes: the authorization read + again, then the signing key and the webhook rewritten with a new envelope, + as on a refresh, all with the token of the `events/stream` request. +5. **Close.** On client abort, termination, shutdown or token expiry, the + replica stops listening and deletes the webhook (best effort, within 10 + seconds). + +A stream renews with the token it was opened with, so it ends a minute before +that token expires, the way shutdown ends it: with its final result and no +`terminated`. The client reconnects with a fresh token; `terminated` with +`-32012` would tell it to give the subscription up. + +| Renewal answer | Result | +| --- | --- | +| Forbidden (`401`/`403`: access or a scope was revoked) | `terminated` with `-32012` | +| NotFound: the resource is gone (`data.kind = "resource"`) or the webhook was deleted in the Console (`data.kind = "subscription"`) | `terminated` with `-32011` | +| Anything else (`5xx`, no answer within 30 seconds) | `error`, retried after 15 seconds; `terminated` with `-32603` only once the lease has run out | + +Losing the bus connection sends an `error`; once it is back, `active` with +`truncated: true` marks the gap. Each listener pings Redis every 10 seconds and +counts its connection lost after 5 seconds without an answer, so a Redis that +vanished without closing the connection is noticed too. Events have no replay, so `cursor` is always +`null` and `maxAgeMs` is accepted with no effect: events during a reconnect are +lost, which the spec allows for emit-only event types. + +If the replica dies instead, its webhooks stay behind with their short lease. +Once that runs out the ingress drops their deliveries as `expired`, and the +subscriber's next subscribe or stream in the project deletes them a minute +later (a webhook subscription gets 10 minutes; a live stream renews long before +its lease runs out, so a lapsed push webhook belongs to a stream that is gone). +On the Free plan that gives the slot back within about six minutes. A rolling deploy drops every stream, as expected: on +SIGTERM the server ends its streams cleanly (each deletes its webhook and sends +its final result) and the clients reconnect to another replica. + +### Routing between replicas + +A stream lives on one replica, but Appwrite's webhook POST lands on any replica +behind the gateway. The ingress therefore hands push events to a bus: Redis +pub/sub on channel `mcp:events:`. The ingress publishes the +projected event (the same `EventOccurrence` a webhook delivery carries, never +the Appwrite body), and the replica holding the stream subscribes to its channel +while the stream is open. Pub/sub is fire-and-forget, which matches events +without replay. An event published while no replica listens is acknowledged and +dropped as `no_listener`; when Redis is unreachable it is dropped as +`bus_error`. Publishing is one Redis call with a 2-second timeout, so the +ingress never waits on a stream. The bus is one module +(`events/bus.py`) behind a small `Bus` interface. + +### Limits + +| Limit | Value | Beyond it | +| --- | --- | --- | +| Open streams per user, per replica | 25 | `-32013`, `data: {"limit": "streams", "max": 25}` | +| Open streams per replica | 1000 (each holds one Redis connection) | `-32013`, `data: {"limit": "server_streams", "max": 1000}` | +| Appwrite calls of the events methods at once, per replica | 40, a thread pool of their own so stream reconnects after a deploy cannot starve tool calls | they wait | +| Webhooks per project | the plan's (Free: 2); each open stream uses one | `-32013`, `data: {"limit": "webhooks", "max": …}` | + +Heartbeats every 25 seconds keep idle streams alive through proxies with a +30-second or longer idle timeout. + ## Ingress Each subscription is one Appwrite webhook in the subscriber's project. It points @@ -277,9 +393,12 @@ For each delivery the ingress: payload schema. Only the declared fields are forwarded, never the raw body: Appwrite sends whole rows, users with email and phone, and executions with request headers and logs. -5. Starts the delivery in a background task and answers `200` straight away, - well inside Appwrite's 15 second timeout. The delivery is signed with - Standard Webhooks and retried in-process (see `events/delivery.py`). +5. For a webhook subscription, starts the delivery in a background task and + answers `200` straight away, well inside Appwrite's 15 second timeout. The + delivery is signed with Standard Webhooks and retried in-process (see + `events/delivery.py`). For a push subscription, publishes the event to the + replica holding its stream (see [Routing between + replicas](#routing-between-replicas)) and answers `200`. ### Envelope @@ -294,16 +413,17 @@ adapter around the package. - **Record.** `version` 1, `id` (the subscription id), `principal` (a digest of the OAuth `iss`, `sub` and `client_id`, never a token), `event.name` and - `event.arguments`, `delivery` (`mode` `webhook`, the callback `url` and its one - `whsec_` `secret`), `lease` (`expiresAt` and `refreshBefore` in UTC RFC 3339 - with `Z`) and empty `extensions`. + `event.arguments`, `delivery` (`mode` `webhook` with the callback `url` and its + one `whsec_` `secret`, or `mode` `push` alone), `lease` (`expiresAt` and + `refreshBefore` in UTC RFC 3339 with `Z`) and empty `extensions`. - **Context binding.** Every envelope is sealed for a context of `server` (`MCP_PUBLIC_URL`), `tenant` (the project id) and `resource` (the subscription id, which is also the webhook id). The ingress takes the expected context only from its configuration, the routed webhook id and `X-Appwrite-Webhook-Project-Id`, never from the envelope, so an envelope - copied to another webhook, project or deployment does not open. The record - must also hash back to the subscription id, or it does not open either. + copied to another webhook, project or deployment does not open. A webhook + record must also hash back to the subscription id, or it does not open + either; a push record's random id is bound by the context alone. - **Keys.** Each key in `MCP_EVENTS_SEALING_KEYS` is 32 random bytes (keys that are short, long or obviously not random stop the server at startup) and is the JWE key directly. The webhook signing secret is derived from it with @@ -382,6 +502,8 @@ request that fails authentication gets `401`. | Body over 2 MiB (signature still checked) | `200` dropped `too_large` | no | | Body over 16 MiB, or slot wait and body slower than 10 s (signature never checked) | `200` dropped `unread` | no | | Delivery backlog full (process or subscription) | `200` dropped `backlog_full` | no | +| Push subscription whose stream is not open on any replica | `200` dropped `no_listener` | no | +| Push event, but the bus (Redis) is unreachable | `200` dropped `bus_error` | no | | No Basic auth, or the password is not an envelope | `401` `credentials` | no | | Signature missing or wrong | `401` `signature` | no | | Envelope tampered with, not the profile, or for another server, webhook or project | `401` `envelope` | no | @@ -428,11 +550,13 @@ lost: delivery is at most once, like Appwrite webhooks themselves. | Metric | Attributes | | --- | --- | | `mcp.events.ingress` | `outcome` (`accepted`, `dropped`, `rejected`), `reason`, `event` | -| `mcp.events.deliveries` | `event`, `outcome` (`delivered`, `rejected`, `abandoned`, `too_large`, `error`), `reason` | -| `mcp.events.subscriptions` | `operation` (`subscribe`, `unsubscribe`, `cleanup`), `outcome` (`created`, `refreshed`, `removed`, `absent`, `error`), `event`, `reason` (the JSON-RPC code of an error) | +| `mcp.events.deliveries` | `mode` (`webhook`, `push`), `event`, `outcome` (`delivered`, `rejected`, `abandoned`, `too_large`, `error`), `reason` | +| `mcp.events.subscriptions` | `operation` (`subscribe`, `unsubscribe`, `cleanup`, `stream`), `outcome` (`created`, `refreshed`, `removed`, `absent`, `opened`, `closed`, `terminated`, `expired`, `error`), `event`, `reason` (the JSON-RPC code of an error) | +| `mcp.events.streams` (up-down counter) | `event`: the streams open right now | -`events/subscribe` and `events/unsubscribe` are also counted in -`mcp.messages.received` and `mcp.jsonrpc.errors`, like `tools/call`. Errors on +`events/subscribe`, `events/unsubscribe` and `events/stream` are also counted +in `mcp.messages.received` and `mcp.jsonrpc.errors`, like `tools/call` (for a +stream, the duration is how long it was open). Errors on our side (`-32603`) go to Sentry with the same tags as `tools/call` plus `event.name` and `appwrite.project_id`. Errors the caller caused are counted but not reported. @@ -460,6 +584,12 @@ verification handshake and checks every request with the official `standardwebhooks` library. Counters are read from the server's real OTLP export. +Push streams run against a real bus the suite starts on a random port: +`dragonfly` when it is on `PATH`, as in CI (pinned to the version the clusters +run), otherwise `redis-server` (`brew install redis` or +`apt-get install redis-server`). Two server instances share it, so a stream +opened on one receives what Appwrite delivers to the other. + A live end-to-end run against Cloud staging (real OAuth tokens, real Appwrite webhooks) is planned once [appwrite/appwrite#14293](https://github.com/appwrite/appwrite/issues/14293) diff --git a/docs/flags.md b/docs/flags.md index d776d47..2f2ea06 100644 --- a/docs/flags.md +++ b/docs/flags.md @@ -35,6 +35,7 @@ Flags are for testing overrides only — permanent configuration belongs in | --- | --- | --- | | `--console-url` | `MCP_CONSOLE_URL` | Send OAuth login/consent to an alternative Appwrite Console (HTTP transport only). | | `--events` | `MCP_EVENTS` | Advertise MCP Events, serve `events/list` and mount the Appwrite webhook ingress (HTTP transport only; needs `MCP_EVENTS_SEALING_KEYS`). See [events.md](events.md). | +| `--events-push` | `MCP_EVENTS_PUSH` | Also offer push delivery (`events/stream`) for MCP Events (needs `--events` and `MCP_EVENTS_REDIS_URL`). See [events.md](events.md#push-delivery). | ### `--console-url` — test OAuth against a pre-release console @@ -134,3 +135,46 @@ curl -s -X POST http://localhost:8000/appwrite/webhooks/sub_0123456789abcdef0123 With the flag off, neither capability appears, `events/list`, `events/subscribe` and `events/unsubscribe` return `-32601` (method not found) and the ingress route returns `404`. + +### `--events-push` — push delivery (in progress) + +Adds [push delivery](events.md#push-delivery) to MCP Events: `events/list` +advertises `"delivery": ["webhook", "push"]` and the server serves +`events/stream`, a long-lived request that streams the subscription's events +back ([#137](https://github.com/appwrite/mcp/issues/137)). It needs `--events` +and a Redis-protocol server (Dragonfly in the cluster, or Redis) at +`MCP_EVENTS_REDIS_URL`, which carries events from the replica Appwrite delivers +to the one holding the stream. With the flag on but no URL, the server logs it and serves webhook delivery only. Off by default, so +webhook delivery can stay live while push rolls out. + +**Enable it:** + +```bash +redis-server --port 6379 & +MCP_PUBLIC_URL=http://localhost:8000 \ +MCP_EVENTS_SEALING_KEYS="k1:$(openssl rand -base64 32)" \ +MCP_EVENTS_REDIS_URL=redis://localhost:6379/0 \ + uv run mcp-server-appwrite --transport http --events 1 --events-push 1 +``` + +Or `docker compose up --build`, whose Dragonfly is wired in; set `MCP_EVENTS=1`, +`MCP_EVENTS_PUSH=1` and `MCP_EVENTS_SEALING_KEYS` in `.env`. + +**Quick checks**, with the `mcp` helper above: + +```bash +# "push" is advertised +mcp events/list | jq '.result.events[0].delivery' + +# A stream: active first, then events and a heartbeat every 25 s (Ctrl-C to close) +curl -sN http://localhost:8000/ \ + -H "Authorization: Bearer $TOKEN" \ + -H 'Content-Type: application/json' \ + -H 'Accept: application/json, text/event-stream' \ + -H 'MCP-Protocol-Version: 2026-07-28' \ + -H 'Mcp-Method: events/stream' \ + -d '{"jsonrpc":"2.0","id":1,"method":"events/stream","params":{"name":"users.user.created","arguments":{"project_id":""},"cursor":null,"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}' +``` + +With the flag off, `events/list` advertises webhook only and `events/stream` +returns `-32601` (method not found). diff --git a/pyproject.toml b/pyproject.toml index df3bd65..65023d4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,8 +15,9 @@ dependencies = [ "httpx>=0.27.0", "cryptography>=44.0.0", "httpcore>=1.0", - "mcp-event-subscriptions>=0.1.0a1,<0.2", + "mcp-event-subscriptions>=0.1.0a2,<0.2", "pyjwt[crypto]>=2.9.0", + "redis>=6.0,<9", "numpy>=2.0", "openai>=1.40", "opentelemetry-api>=1.28", @@ -45,6 +46,7 @@ e2e = [ "standardwebhooks>=1.1.0", ] + [project.scripts] mcp-server-appwrite = "mcp_server_appwrite.__main__:main" diff --git a/src/mcp_server_appwrite/events/bus.py b/src/mcp_server_appwrite/events/bus.py new file mode 100644 index 0000000..11db7a3 --- /dev/null +++ b/src/mcp_server_appwrite/events/bus.py @@ -0,0 +1,208 @@ +"""Carry push events from the ingress to the replica that holds their stream. + +A push subscription is one open ``events/stream`` request on one replica, but +Appwrite's webhook for it lands on whichever replica the gateway picks. So the +ingress publishes each projected event under the subscription id, and the +replica holding the stream listens on that id for as long as the stream is open. + +The bus is fire-and-forget, like the events themselves: Appwrite webhooks have +no replay, so an event published while nobody listens (the stream just closed, +or its client is reconnecting) is dropped. :meth:`Bus.publish` says whether +anyone received it, so the ingress can count the drop. + +:class:`RedisBus` is the implementation: Redis pub/sub on channel +``mcp:events:``, configured by ``MCP_EVENTS_REDIS_URL``. Every +open stream holds its own pub/sub connection; the stream caps in +:mod:`.streams` bound how many a replica opens. +""" + +from __future__ import annotations + +import json +import os +import sys +import time +from collections.abc import AsyncIterator +from typing import Protocol + +from redis.asyncio import Redis +from redis.asyncio.client import PubSub +from redis.exceptions import RedisError + +from .delivery import Event + +URL_ENV = "MCP_EVENTS_REDIS_URL" + +CHANNEL_PREFIX = "mcp:events:" + +TIMEOUT_SECONDS = 2.0 +"""Connect and command timeout. A publish sits on the ingress's request path, +so a stalled Redis must not hold Appwrite's delivery for long.""" + +POLL_SECONDS = 1.0 +"""How long a listener waits for a message before checking its connection.""" + +PING_SECONDS = 10.0 +"""How often a listener pings Redis over its own connection.""" + +PONG_SECONDS = 5.0 +"""How long a listener waits for the answer to its ping. Without one the +connection is dead (a peer that vanished without a reset can otherwise look +quiet for as long as TCP takes to give up), and :class:`BusError` is raised.""" + +PING_MESSAGE = "mcp-liveness" + + +class BusError(Exception): + """The bus is unreachable. A listener raises it when its connection is lost; + events published meanwhile are gone.""" + + +class Listener(Protocol): + """The events of one subscription, from when it was opened until + :meth:`aclose`.""" + + def __aiter__(self) -> AsyncIterator[Event]: ... + + async def aclose(self) -> None: ... + + +class Bus(Protocol): + """Routes events by subscription id between replicas.""" + + async def publish(self, id: str, event: Event) -> bool: + """Send ``event`` to the listener of subscription ``id``; whether one + received it. Raises :class:`BusError` when the bus is unreachable.""" + ... + + async def listen(self, id: str) -> Listener: + """Start receiving the events of subscription ``id``: nothing published + after this returns is missed. Raises :class:`BusError`.""" + ... + + async def aclose(self) -> None: ... + + +def channel(id: str) -> str: + return CHANNEL_PREFIX + id + + +def encode(event: Event) -> bytes: + return json.dumps(event.occurrence(), separators=(",", ":")).encode("utf-8") + + +def decode(message: bytes) -> Event | None: + """The event in a bus message, or ``None`` for anything else.""" + try: + payload = json.loads(message) + except ValueError: + return None + if not isinstance(payload, dict): + return None + id, name = payload.get("eventId"), payload.get("name") + timestamp, data = payload.get("timestamp"), payload.get("data") + if not ( + isinstance(id, str) + and isinstance(name, str) + and isinstance(timestamp, str) + and isinstance(data, dict) + ): + return None + return Event(id=id, name=name, timestamp=timestamp, data=data) + + +class RedisBus: + """:class:`Bus` over Redis pub/sub.""" + + def __init__( + self, + url: str, + *, + timeout: float = TIMEOUT_SECONDS, + ping: float = PING_SECONDS, + pong: float = PONG_SECONDS, + ) -> None: + self._ping = ping + self._pong = pong + self._client = Redis.from_url( + url, + socket_timeout=timeout, + socket_connect_timeout=timeout, + socket_keepalive=True, + ) + + @classmethod + def from_env(cls) -> RedisBus | None: + """The bus at ``MCP_EVENTS_REDIS_URL``, or ``None`` when it is unset.""" + url = os.getenv(URL_ENV, "").strip() + return cls(url) if url else None + + async def publish(self, id: str, event: Event) -> bool: + try: + receivers = await self._client.publish(channel(id), encode(event)) + except (RedisError, OSError) as error: + raise BusError(f"publish failed: {error!r}") from error + return receivers > 0 + + async def listen(self, id: str) -> RedisListener: + pubsub = self._client.pubsub(ignore_subscribe_messages=True) + try: + await pubsub.subscribe(channel(id)) + except BaseException as error: + await RedisListener(pubsub, self._ping, self._pong).aclose() + if isinstance(error, (RedisError, OSError)): + raise BusError(f"subscribe failed: {error!r}") from error + raise + return RedisListener(pubsub, self._ping, self._pong) + + async def aclose(self) -> None: + await self._client.aclose() + + +class RedisListener: + """:class:`Listener` on one Redis pub/sub connection.""" + + def __init__(self, pubsub: PubSub, ping: float, pong: float) -> None: + self._pubsub = pubsub + self._ping = ping + self._pong = pong + + async def __aiter__(self) -> AsyncIterator[Event]: + pinged: float | None = None + """When the unanswered ping went out.""" + alive = time.monotonic() + """When Redis last answered.""" + while True: + now = time.monotonic() + if pinged is not None and now - pinged > self._pong: + raise BusError("listen failed: Redis stopped answering pings") + try: + if pinged is None and now - alive > self._ping: + await self._pubsub.ping(PING_MESSAGE) + pinged = now + message = await self._pubsub.get_message( + ignore_subscribe_messages=True, + timeout=min(POLL_SECONDS, self._ping), + ) + except (RedisError, OSError) as error: + raise BusError(f"listen failed: {error!r}") from error + if message is None: + continue + alive, pinged = time.monotonic(), None + if message.get("type") != "message": + continue + event = decode(message["data"]) + if event is None: + _log(f"{message.get('channel')!r}: ignored a malformed message") + continue + yield event + + async def aclose(self) -> None: + try: + await self._pubsub.aclose() + except (RedisError, OSError): + pass + + +def _log(message: str) -> None: + print(f"[appwrite-mcp][events] {message}", file=sys.stderr, flush=True) diff --git a/src/mcp_server_appwrite/events/delivery.py b/src/mcp_server_appwrite/events/delivery.py index 9f5742b..ae5483b 100644 --- a/src/mcp_server_appwrite/events/delivery.py +++ b/src/mcp_server_appwrite/events/delivery.py @@ -152,16 +152,19 @@ class Event: timestamp: str data: Mapping[str, Any] + def occurrence(self) -> dict[str, Any]: + """The ``EventOccurrence`` sent to the subscriber. Appwrite webhooks + cannot replay, so there is never a cursor.""" + return { + "eventId": self.id, + "name": self.name, + "timestamp": self.timestamp, + "data": dict(self.data), + "cursor": None, + } + def encode(self) -> bytes: - return encode( - { - "eventId": self.id, - "name": self.name, - "timestamp": self.timestamp, - "data": dict(self.data), - "cursor": None, - } - ) + return encode(self.occurrence()) @dataclass(frozen=True) diff --git a/src/mcp_server_appwrite/events/envelope.py b/src/mcp_server_appwrite/events/envelope.py index 4930971..aa99a17 100644 --- a/src/mcp_server_appwrite/events/envelope.py +++ b/src/mcp_server_appwrite/events/envelope.py @@ -6,7 +6,9 @@ returns, and sends back as HTTP Basic auth on every delivery: * The webhook ``$id`` is the deterministic :func:`subscription_id`, so a refresh - is an upsert and an unsubscribe is a delete by id. + is an upsert and an unsubscribe is a delete by id. A push subscription (one + ``events/stream``) gets a random :func:`stream_id` instead: it lives as long + as its stream, so two streams with the same arguments are two subscriptions. * ``authPassword`` holds the envelope from :meth:`Keyring.seal`: a Subscription Record v1 sealed with the ``mcp-subscription+jwe`` profile of `mcp-event-subscriptions `_ @@ -43,6 +45,7 @@ import json import os import re +import secrets from collections.abc import Mapping from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta @@ -67,7 +70,6 @@ """``header..iv.ciphertext.tag``: JWE Compact with an empty encrypted key.""" RECORD_VERSION = 1 -DELIVERY_MODE = "webhook" SUBSCRIPTION_PREFIX = "sub_" # Appwrite custom ids: at most 36 chars of [A-Za-z0-9._-], no leading special @@ -96,6 +98,15 @@ Argument = str | int | bool | None +class Mode(StrEnum): + """How a subscription's events reach the subscriber.""" + + WEBHOOK = profile.WEBHOOK + """Signed POSTs to the subscriber's callback URL.""" + PUSH = profile.PUSH + """Notifications on the ``events/stream`` the subscriber holds open.""" + + class EnvelopeFailure(StrEnum): """Why an envelope did not open.""" @@ -165,6 +176,12 @@ def subscription_id( return SUBSCRIPTION_PREFIX + digest[:SUBSCRIPTION_HASH_CHARS] +def stream_id() -> str: + """A fresh id for a push subscription, in the :func:`subscription_id` + format so its webhook is recognized as managed.""" + return SUBSCRIPTION_PREFIX + secrets.token_hex(SUBSCRIPTION_HASH_CHARS // 2) + + def valid_subscription_id(value: str) -> bool: """Whether ``value`` is a subscription id this module could have produced.""" return SUBSCRIPTION_ID_PATTERN.fullmatch(value) is not None @@ -229,11 +246,11 @@ class Subscription: """Event name, e.g. ``tablesdb.row.created``.""" arguments: Mapping[str, Argument] """Event arguments, sorted by key and read-only.""" - callback: str - """Delivery URL.""" - secret: str = field(repr=False) - """The client's ``whsec_`` secret. Kept out of ``repr`` so logging a - subscription never leaks it.""" + callback: str | None + """Delivery URL; ``None`` for a push subscription.""" + secret: str | None = field(repr=False) + """The client's ``whsec_`` secret, ``None`` for a push subscription. Kept + out of ``repr`` so logging a subscription never leaks it.""" expires: int """Lease expiry as Unix epoch milliseconds.""" refresh: int @@ -244,12 +261,24 @@ class Subscription: def __post_init__(self) -> None: canonical = MappingProxyType(dict(sorted(self.arguments.items()))) object.__setattr__(self, "arguments", canonical) + if (self.callback is None) != (self.secret is None): + raise ValueError("a webhook subscription needs a callback and a secret") + if self.callback is None: + # A push subscription's id is random, so only its shape is checked; + # the envelope's context binds it to the webhook. + if not valid_subscription_id(self.id): + raise ValueError("push subscription id is invalid") + return expected = subscription_id( self.principal, self.callback, self.name, self.arguments ) if self.id != expected: raise ValueError("subscription id does not match its contents") + @property + def mode(self) -> Mode: + return Mode.PUSH if self.callback is None else Mode.WEBHOOK + @classmethod def create( cls, @@ -262,7 +291,7 @@ def create( refresh: int, principal: str, ) -> Subscription: - """Build a subscription, deriving its id.""" + """Build a webhook subscription, deriving its id.""" return cls( id=subscription_id(principal, callback, name, arguments), project=project, @@ -275,17 +304,41 @@ def create( principal=principal, ) + @classmethod + def stream( + cls, + project: str, + name: str, + arguments: Mapping[str, Argument], + expires: int, + refresh: int, + principal: str, + ) -> Subscription: + """Build a push subscription with a fresh :func:`stream_id`.""" + return cls( + id=stream_id(), + project=project, + name=name, + arguments=arguments, + callback=None, + secret=None, + expires=expires, + refresh=refresh, + principal=principal, + ) + @classmethod def from_record(cls, record: Mapping[str, Any], project: str) -> Subscription: """The subscription of a validated record. Raises ``ValueError`` when its id does not match its contents.""" + delivery = record["delivery"] return cls( id=record["id"], project=project, name=record["event"]["name"], arguments=record["event"]["arguments"], - callback=record["delivery"]["url"], - secret=record["delivery"]["secret"], + callback=delivery.get("url"), + secret=delivery.get("secret"), expires=milliseconds(record["lease"]["expiresAt"]), refresh=milliseconds(record["lease"]["refreshBefore"]), principal=record["principal"], @@ -294,16 +347,15 @@ def from_record(cls, record: Mapping[str, Any], project: str) -> Subscription: @property def record(self) -> dict[str, Any]: """The Subscription Record v1 that is sealed.""" + delivery: dict[str, Any] = {"mode": self.mode.value} + if self.mode is Mode.WEBHOOK: + delivery |= {"url": self.callback, "secret": self.secret} return { "version": RECORD_VERSION, "id": self.id, "principal": self.principal, "event": {"name": self.name, "arguments": dict(self.arguments)}, - "delivery": { - "mode": DELIVERY_MODE, - "url": self.callback, - "secret": self.secret, - }, + "delivery": delivery, "lease": { "expiresAt": timestamp(self.expires), "refreshBefore": timestamp(self.refresh), diff --git a/src/mcp_server_appwrite/events/errors.py b/src/mcp_server_appwrite/events/errors.py index e4aeca9..ce83e77 100644 --- a/src/mcp_server_appwrite/events/errors.py +++ b/src/mcp_server_appwrite/events/errors.py @@ -54,7 +54,8 @@ """A quota was hit; ``data.limit`` and ``data.max`` describe it. SEP-3415: -32025.""" UNSUPPORTED = -32014 -"""A delivery mode or option this server does not offer. SEP-3415: -32026.""" +"""A delivery mode or option this server does not offer; ``data.feature`` and +``data.value`` name it. SEP-3415: -32026.""" CALLBACK_ENDPOINT = -32015 """The callback URL failed verification; ``data.reason`` says how. SEP-3415: -32027.""" @@ -97,8 +98,8 @@ def resource_exhausted( return cls(RESOURCE_EXHAUSTED, message, {"limit": limit, "max": maximum}) @classmethod - def unsupported(cls, message: str) -> EventsError: - return cls(UNSUPPORTED, message) + def unsupported(cls, message: str, *, feature: str, value: object) -> EventsError: + return cls(UNSUPPORTED, message, {"feature": feature, "value": value}) @classmethod def callback_endpoint(cls, message: str, *, reason: CallbackFailure) -> EventsError: diff --git a/src/mcp_server_appwrite/events/ingress.py b/src/mcp_server_appwrite/events/ingress.py index e186024..6c6f06f 100644 --- a/src/mcp_server_appwrite/events/ingress.py +++ b/src/mcp_server_appwrite/events/ingress.py @@ -25,10 +25,14 @@ hashed as it streams in, by at most :data:`READS` requests at once. 4. Map the Appwrite event onto the subscribed catalog event and project the body down to the event's payload fields (:mod:`.projection`). -5. Take a backlog slot (:data:`BACKLOG` per process, :data:`SUBSCRIPTION_BACKLOG` - per subscription), start the delivery in the background and answer ``200`` - at once, well inside Appwrite's 15 second timeout. With no slot free the - event is acknowledged and dropped (``backlog_full``). +5. For a webhook subscription, take a backlog slot (:data:`BACKLOG` per + process, :data:`SUBSCRIPTION_BACKLOG` per subscription), start the delivery + in the background and answer ``200`` at once, well inside Appwrite's 15 + second timeout. With no slot free the event is acknowledged and dropped + (``backlog_full``). For a push subscription, publish the event on the + :mod:`.bus` for the replica holding its ``events/stream``: one bounded Redis + call, never waiting on the stream. With no stream open anywhere it is + dropped (``no_listener``). Status codes follow one rule: **anything that came from the subscriber's own webhook is answered with 2xx.** Appwrite counts every ``>= 400`` answer and, @@ -82,6 +86,7 @@ from .. import error_monitoring, telemetry from ..auth import public_base_url +from .bus import Bus, BusError, RedisBus from .catalog import CATALOG from .delivery import ( Callback, @@ -95,6 +100,7 @@ Context, EnvelopeError, Keyring, + Mode, SubscriptionExpired, appwrite_hmac, canonical_json, @@ -220,8 +226,8 @@ class Ingress: """The Appwrite webhook endpoint and the deliveries it starts. Create one per process. :meth:`run` must wrap the app's lifetime: it owns - the task group deliveries run in and closes the shared :class:`Egress` on - the way out. + the task group deliveries run in and closes the shared :class:`Egress` and + the :class:`~.bus.Bus` on the way out. """ def __init__( @@ -235,6 +241,7 @@ def __init__( subscription_backlog: int = SUBSCRIPTION_BACKLOG, reads: int = READS, read_timeout: float = READ_TIMEOUT_SECONDS, + bus: Bus | None = None, ) -> None: if backlog < 1 or subscription_backlog < 1 or reads < 1: raise ValueError("backlogs and reads must be at least 1") @@ -244,6 +251,7 @@ def __init__( self._egress = egress self._dispatcher = dispatcher or Dispatcher(egress) self._base = base.rstrip("/") + self._bus = bus self._group: TaskGroup | None = None self._backlog = backlog self._subscription_backlog = subscription_backlog @@ -255,9 +263,15 @@ def __init__( @classmethod def from_env(cls) -> Ingress: """Sealing keys from ``MCP_EVENTS_SEALING_KEYS``, the public URL from - ``MCP_PUBLIC_URL``. Raises :class:`~.envelope.KeyringError` when the - keys are missing, so a misconfigured server fails at startup.""" - return cls(Keyring.from_env(), Egress(), base=public_base_url()) + ``MCP_PUBLIC_URL`` and the bus from ``MCP_EVENTS_REDIS_URL``. Raises + :class:`~.envelope.KeyringError` when the keys are missing, so a + misconfigured server fails at startup.""" + return cls( + Keyring.from_env(), + Egress(), + base=public_base_url(), + bus=RedisBus.from_env(), + ) @property def keyring(self) -> Keyring: @@ -271,6 +285,11 @@ def egress(self) -> Egress: def dispatcher(self) -> Dispatcher: return self._dispatcher + @property + def bus(self) -> Bus | None: + """Where push events go; ``None`` when no bus is configured.""" + return self._bus + @property def server(self) -> str: """This deployment as envelopes are bound to it: ``MCP_PUBLIC_URL``.""" @@ -288,14 +307,18 @@ def url(self, id: str) -> str: async def run(self) -> AsyncIterator[None]: """Serve deliveries for the duration of the block; cancel what is still in flight and close the egress client afterwards.""" - async with self._egress: - async with anyio.create_task_group() as group: - self._group = group - try: - yield - finally: - self._group = None - group.cancel_scope.cancel() + try: + async with self._egress: + async with anyio.create_task_group() as group: + self._group = group + try: + yield + finally: + self._group = None + group.cancel_scope.cancel() + finally: + if self._bus is not None: + await self._bus.aclose() async def read(self, request: Request, id: str, keys: Sequence[str]) -> Body | None: """Stream the body, hashing all of it under each of ``keys`` and @@ -406,12 +429,6 @@ async def handle(self, request: Request) -> Response: except Dropped as dropped: return self._drop(id, dropped.reason, event.name) - try: - callback = Callback( - subscription.callback, subscription.id, subscription.secret - ) - except InvalidSecretError: - return self._drop(id, Drop.MALFORMED, event.name) delivery = ( request.headers.get(DELIVERY_HEADER) or hashlib.sha256(body.content).hexdigest() @@ -422,6 +439,14 @@ async def handle(self, request: Request) -> Response: timestamp=projection.occurred or _now(), data=projection.data, ) + if subscription.callback is None or subscription.secret is None: + return await self._publish(subscription.id, occurrence) + try: + callback = Callback( + subscription.callback, subscription.id, subscription.secret + ) + except InvalidSecretError: + return self._drop(id, Drop.MALFORMED, event.name) if self._group is None: raise RuntimeError("Ingress.run() is not active") if not self._admit(subscription.id): @@ -430,6 +455,22 @@ async def handle(self, request: Request) -> Response: telemetry.record_ingress(Outcome.ACCEPTED, None, event.name) return JSONResponse({"status": Outcome.ACCEPTED, "eventId": occurrence.id}) + async def _publish(self, id: str, event: Event) -> Response: + """Hand a push event to the replica holding its stream.""" + if self._bus is None: + return self._drop(id, Drop.NO_LISTENER, event.name) + try: + received = await self._bus.publish(id, event) + except BusError as error: + # Counted and logged, not reported: an outage would report every + # delivery. The streams report it when they cannot listen. + _log(f"{id}: {error}") + return self._drop(id, Drop.BUS_ERROR, event.name) + if not received: + return self._drop(id, Drop.NO_LISTENER, event.name) + telemetry.record_ingress(Outcome.ACCEPTED, None, event.name) + return JSONResponse({"status": Outcome.ACCEPTED, "eventId": event.id}) + def _admit(self, subscription: str) -> bool: """Take a backlog slot for one delivery of ``subscription``, unless the process or that subscription is already at its limit. Runs on the @@ -460,7 +501,7 @@ async def _attempt(self, callback: Callback, event: Event) -> None: try: result = await self._dispatcher.deliver(callback, event) except PayloadTooLargeError: - telemetry.record_event_delivery(event.name, "too_large", None) + telemetry.record_event_delivery(event.name, "too_large", None, Mode.WEBHOOK) _log(f"{callback.subscription}: {event.id} too large; not sent") return except Exception as error: @@ -469,11 +510,13 @@ async def _attempt(self, callback: Callback, event: Event) -> None: error_monitoring.capture_exception( error, tags={"mcp.events.subscription": callback.subscription} ) - telemetry.record_event_delivery(event.name, "error", None) + telemetry.record_event_delivery(event.name, "error", None, Mode.WEBHOOK) _log(f"{callback.subscription}: {event.id} failed: {error!r}") return reason = result.reason.value if result.reason is not None else None - telemetry.record_event_delivery(event.name, result.outcome, reason) + telemetry.record_event_delivery( + event.name, result.outcome, reason, Mode.WEBHOOK + ) _log( f"{callback.subscription}: {event.id} {result.outcome} after " f"{result.attempts} attempt(s) in {time.monotonic() - started:.1f}s" diff --git a/src/mcp_server_appwrite/events/projection.py b/src/mcp_server_appwrite/events/projection.py index 23479b1..454d130 100644 --- a/src/mcp_server_appwrite/events/projection.py +++ b/src/mcp_server_appwrite/events/projection.py @@ -71,6 +71,11 @@ class Drop(StrEnum): BACKLOG_FULL = "backlog_full" """The process or the subscription already has the most deliveries in flight it may hold.""" + NO_LISTENER = "no_listener" + """A push subscription whose stream is not open on any replica (it just + closed, or its client is reconnecting). Events have no replay.""" + BUS_ERROR = "bus_error" + """A push event that could not be handed to the bus.""" class Dropped(Exception): diff --git a/src/mcp_server_appwrite/events/protocol.py b/src/mcp_server_appwrite/events/protocol.py index f9e9ac0..6db0818 100644 --- a/src/mcp_server_appwrite/events/protocol.py +++ b/src/mcp_server_appwrite/events/protocol.py @@ -10,16 +10,20 @@ today: ``capabilities.events = {}`` for ChatGPT and ``capabilities.extensions["io.modelcontextprotocol/events"]`` for SEP-3415. Delete the middleware once the SDK models the capability. -* **Methods.** ``events/list``, ``events/subscribe`` and ``events/unsubscribe`` - are registered with ``Server.add_request_handler``; the SDK adds - ``resultType: "complete"`` to custom-method results on the 2026-07-28 - protocol. Subscribe and unsubscribe are counted like ``tools/call`` - (``mcp.messages.received`` / ``mcp.jsonrpc.errors``), and failures on our - side go to Sentry with the same tags; errors the caller caused (bad input, - no access, a quota, an unreachable callback) are counted but not reported. +* **Methods.** ``events/list``, ``events/subscribe``, ``events/unsubscribe`` + and, with push on, ``events/stream`` (:mod:`.streams`) are registered with + ``Server.add_request_handler``; the SDK adds ``resultType: "complete"`` to + custom-method results on the 2026-07-28 protocol. All but the listing are + counted like ``tools/call`` (``mcp.messages.received`` / + ``mcp.jsonrpc.errors``), and failures on our side go to Sentry with the same + tags; errors the caller caused (bad input, no access, a quota, an + unreachable callback) are counted but not reported. Events are served only on the hosted HTTP transport and only behind the -``events`` flag (``MCP_EVENTS``); see :func:`enabled`. +``events`` flag (``MCP_EVENTS``); see :func:`enabled`. Push delivery also needs +the ``events-push`` flag (``MCP_EVENTS_PUSH``) and a bus +(``MCP_EVENTS_REDIS_URL``); see :func:`push_enabled`. ``events/list`` +advertises ``push`` only then. """ from __future__ import annotations @@ -36,9 +40,12 @@ from .. import error_monitoring, flags, telemetry from .catalog import CATALOG, EVENTS, Event +from .envelope import Mode from .errors import EventsError +from .ingress import Ingress +from .streams import METHOD as STREAM_METHOD +from .streams import StreamParams, Streams from .subscriptions import ( - WEBHOOK, Operation, SubscribeParams, Subscriptions, @@ -80,23 +87,34 @@ def enabled(transport: str) -> bool: return transport == TRANSPORT and flags.enabled(flags.EVENTS) -def describe(event: Event) -> dict[str, Any]: +def push_enabled(transport: str, ingress: Ingress) -> bool: + """Whether this server should also offer push delivery (``events/stream``): + the events and events-push flags are on and ``ingress`` has a bus.""" + return ( + enabled(transport) + and flags.enabled(flags.EVENTS_PUSH) + and ingress.bus is not None + ) + + +def describe(event: Event, push: bool) -> dict[str, Any]: """One ``events/list`` entry.""" + delivery = [Mode.WEBHOOK.value, *([Mode.PUSH.value] if push else [])] return { "name": event.name, "description": event.description, - "delivery": [WEBHOOK], + "delivery": delivery, "inputSchema": event.input_schema, "payloadSchema": event.payload_schema, } -def list_events(cursor: str | None = None) -> dict[str, Any]: +def list_events(cursor: str | None = None, *, push: bool = False) -> dict[str, Any]: """The ``events/list`` result. The catalog fits in one page, so a cursor can only be one this server never issued.""" if cursor is not None: raise EventsError.invalid_params("Invalid cursor") - return {"events": [describe(event) for event in EVENTS]} + return {"events": [describe(event, push) for event in EVENTS]} def advertise(result: dict[str, Any]) -> dict[str, Any]: @@ -123,25 +141,25 @@ async def __call__( return result -async def _handle_list( - context: ServerRequestContext[Any, Any], params: ListParams -) -> dict[str, Any]: - return list_events(params.cursor) - - Identify = Callable[[ServerRequestContext[Any, Any]], Mapping[str, Any]] """Sentry tags naming the client behind a request (``mcp.client.name``, …).""" -_Params = TypeVar("_Params", SubscribeParams, UnsubscribeParams) +_Params = TypeVar("_Params", SubscribeParams, UnsubscribeParams, StreamParams) + +Handler = Callable[[ServerRequestContext[Any, Any], _Params], Awaitable[dict[str, Any]]] def _observed( method: str, operation: Operation, - handler: Callable[[_Params], Awaitable[dict[str, Any]]], + handler: Handler[_Params], identify: Identify, -) -> Callable[[ServerRequestContext[Any, Any], _Params], Awaitable[dict[str, Any]]]: - """``handler`` with the telemetry and error reporting ``tools/call`` has.""" + *, + successes: bool = True, +) -> Handler[_Params]: + """``handler`` with the telemetry and error reporting ``tools/call`` has. + Without ``successes`` only errors are counted here: ``events/stream`` + counts its own success once the stream is open.""" async def handle( context: ServerRequestContext[Any, Any], params: _Params @@ -150,7 +168,7 @@ async def handle( name = params.name event = name if isinstance(name, str) and name in CATALOG else None try: - result = await handler(params) + result = await handler(context, params) except Exception as error: # A bug is reported, and its text never sent to the client. failure = ( @@ -172,7 +190,8 @@ async def handle( if failure is error: raise raise failure from error - telemetry.record_message(method, "success", time.monotonic() - started) + if successes: + telemetry.record_message(method, "success", time.monotonic() - started) return result return handle @@ -182,7 +201,7 @@ def _report( error: Exception, method: str, event: str | None, - params: SubscribeParams | UnsubscribeParams, + params: SubscribeParams | UnsubscribeParams | StreamParams, context: ServerRequestContext[Any, Any], identify: Identify, ) -> None: @@ -208,26 +227,52 @@ def _report( def register( - server: Server[Any], subscriptions: Subscriptions, identify: Identify + server: Server[Any], + subscriptions: Subscriptions, + identify: Identify, + streams: Streams | None = None, ) -> None: """Advertise events and serve ``events/list``, ``events/subscribe`` and - ``events/unsubscribe`` on ``server``.""" - server.add_request_handler(LIST_METHOD, ListParams, _handle_list) + ``events/unsubscribe`` on ``server``, and ``events/stream`` when + ``streams`` is given.""" + push = streams is not None + + async def handle_list( + context: ServerRequestContext[Any, Any], params: ListParams + ) -> dict[str, Any]: + return list_events(params.cursor, push=push) + + async def subscribe( + context: ServerRequestContext[Any, Any], params: SubscribeParams + ) -> dict[str, Any]: + return await subscriptions.subscribe(params) + + async def unsubscribe( + context: ServerRequestContext[Any, Any], params: UnsubscribeParams + ) -> dict[str, Any]: + return await subscriptions.unsubscribe(params) + + server.add_request_handler(LIST_METHOD, ListParams, handle_list) server.add_request_handler( SUBSCRIBE_METHOD, SubscribeParams, - _observed( - SUBSCRIBE_METHOD, Operation.SUBSCRIBE, subscriptions.subscribe, identify - ), + _observed(SUBSCRIBE_METHOD, Operation.SUBSCRIBE, subscribe, identify), ) server.add_request_handler( UNSUBSCRIBE_METHOD, UnsubscribeParams, - _observed( - UNSUBSCRIBE_METHOD, - Operation.UNSUBSCRIBE, - subscriptions.unsubscribe, - identify, - ), + _observed(UNSUBSCRIBE_METHOD, Operation.UNSUBSCRIBE, unsubscribe, identify), ) + if streams is not None: + server.add_request_handler( + STREAM_METHOD, + StreamParams, + _observed( + STREAM_METHOD, + Operation.STREAM, + streams.stream, + identify, + successes=False, + ), + ) server.middleware.append(CapabilityMiddleware()) diff --git a/src/mcp_server_appwrite/events/streams.py b/src/mcp_server_appwrite/events/streams.py new file mode 100644 index 0000000..3b6b517 --- /dev/null +++ b/src/mcp_server_appwrite/events/streams.py @@ -0,0 +1,480 @@ +"""``events/stream``: push delivery of MCP events. + +A push subscription is one long-lived ``events/stream`` request. On the hosted +Streamable HTTP transport it is a POST whose SSE response carries only that +subscription's ``notifications/events/*``, each with the request id in +``params._meta["io.modelcontextprotocol/subscriptionId"]``. The client cancels by +aborting the request. + +**Protocol layer.** The low-level SDK serves this as an ordinary +``add_request_handler`` method, with no ASGI branch of its own: on the +stateless SSE transport ``http_app`` configures, a handler may run for as long +as it likes, notifications it sends for its own request go out as SSE ``data:`` +frames on that POST, a client abort cancels the handler, and nothing caps +concurrent requests. An invalid subscription is refused before anything is +sent, so its JSON-RPC error is the only message and no stream is opened. On +the 2026-07-28 protocol that answer is plain JSON as long as it comes within +the SDK's 15 s SSE deferral; 2025-era transports always answer in SSE. + +**Lifecycle.** Each stream owns exactly one Appwrite webhook in the subscriber's +project, whose sealed envelope says the delivery mode is push (see +:mod:`.envelope`). The stream: + +1. Validates like ``events/subscribe`` (catalog event, arguments, a user behind + the token, webhook write scope), minus the callback, secret and handshake, + and takes a slot under :data:`PRINCIPAL_STREAMS` and :data:`STREAMS` + (``-32013`` beyond them). +2. Starts listening on the :mod:`.bus` under a random subscription id, then + creates the webhook with a short lease (:data:`LEASE_MS`) through + :meth:`.subscriptions.Subscriptions.open`, which also authorizes the caller + and cleans up their expired subscriptions. Listening first means no event + the webhook delivers is missed. +3. Sends ``notifications/events/active {cursor: null, truncated: false}``, then + forwards each event from the bus as ``notifications/events/event``, sends + ``notifications/events/heartbeat {cursor: null}`` every + :data:`HEARTBEAT_SECONDS`, and renews the lease every + :data:`RENEW_SECONDS` with the request's own bearer token. +4. On client abort, a terminal error or server shutdown, stops listening and + deletes the webhook (best effort, shielded from the cancellation). If the + process dies instead, the lease runs out: the ingress drops the webhook's + deliveries as ``expired`` and the caller's next subscribe or stream deletes + it. + +A stream renews with the token it was opened with, so it ends shortly before +that token expires (:data:`EXPIRY_SECONDS`), the way shutdown ends it: with +its final result and no ``terminated``, so the client reconnects with a fresh +token instead of giving the subscription up. A renewal refused with Forbidden +or NotFound (access or a scope was revoked, the resource or the webhook was +deleted) sends ``notifications/events/terminated`` and ends the stream. Other +renewal failures, a call slower than :data:`APPWRITE_SECONDS` among them, send +``notifications/events/error`` and retry after :data:`RETRY_SECONDS`, and only +once the lease has run out is the stream terminated. Losing the bus connection sends an error too; once it +is back, ``notifications/events/active`` with ``truncated: true`` marks the +gap. Events have no replay, so ``cursor`` is always ``null`` and ``maxAgeMs`` +is accepted without effect. + +**Shutdown.** :meth:`Streams.close`, called when the server is asked to stop, +ends every stream: each deletes its webhook and sends its final result, and the +client reconnects to another replica. +""" + +from __future__ import annotations + +import asyncio +import sys +import time +from collections.abc import Callable, Generator +from contextlib import contextmanager +from dataclasses import dataclass, replace +from typing import Any, cast + +import anyio +from anyio.abc import TaskGroup +from mcp.server import ServerRequestContext +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.shared.subscriptions import SUBSCRIPTION_ID_META_KEY +from mcp.types import RequestParams +from mcp_types import Notification, ServerNotification +from pydantic import ConfigDict + +from .. import telemetry +from .bus import Bus, BusError, Listener +from .envelope import Mode, Subscription +from .errors import FORBIDDEN, INTERNAL_ERROR, NOT_FOUND, EventsError +from .subscriptions import ( + Grant, + Operation, + Outcome, + Subscriptions, + Target, + check_max_age, + current_principal, + require_writable, + resolve_target, +) + +METHOD = "events/stream" + +ACTIVE = "notifications/events/active" +EVENT = "notifications/events/event" +HEARTBEAT = "notifications/events/heartbeat" +ERROR = "notifications/events/error" +TERMINATED = "notifications/events/terminated" + +LEASE_MS = 5 * 60 * 1000 +"""Lease written into a stream's webhook. Short, because a stream renews it +while open: a replica that dies leaves a webhook that stops delivering within +this long.""" + +RENEW_SECONDS = 120.0 +"""How often an open stream renews its lease. Leaves room for one failed +renewal and a retry before the lease runs out.""" + +RETRY_SECONDS = 15.0 +"""Wait before retrying a failed renewal or reconnecting to the bus.""" + +HEARTBEAT_SECONDS = 25.0 +"""Heartbeat interval: within the spec's 30 s, and inside proxy idle timeouts.""" + +PRINCIPAL_STREAMS = 25 +"""Open streams one user may hold on one replica.""" + +STREAMS = 1000 +"""Open streams one replica holds. Each holds a bus connection.""" + +CLOSE_SECONDS = 10.0 +"""How long deleting a closed stream's webhook, or announcing its end to a +client that stopped reading, may take.""" + +APPWRITE_SECONDS = 30.0 +"""How long renewing a stream's lease may take. The Appwrite SDK sets no +request timeout, and a renewal that hangs would let the lease run out while +heartbeats make the stream look healthy.""" + +EXPIRY_SECONDS = 60.0 +"""How long before its token expires a stream ends, so the renewal that would +be refused never happens.""" + + +@dataclass(frozen=True) +class Settings: + """Clocks (in seconds) and caps of the streams; tests shrink them.""" + + lease: float = LEASE_MS / 1000 + renew: float = RENEW_SECONDS + retry: float = RETRY_SECONDS + heartbeat: float = HEARTBEAT_SECONDS + principal_streams: int = PRINCIPAL_STREAMS + streams: int = STREAMS + appwrite: float = APPWRITE_SECONDS + expiry: float = EXPIRY_SECONDS + + +class StreamParams(RequestParams): + """``events/stream`` params. Unknown members are tolerated, as the draft + specification may add some.""" + + model_config = ConfigDict(extra="allow") + + name: Any = None + arguments: Any = None + cursor: Any = None + max_age_ms: Any = None + + +class Streams: + """Serves ``events/stream`` for the subscriptions of one ingress, with + events arriving on ``bus``.""" + + def __init__( + self, + subscriptions: Subscriptions, + bus: Bus, + *, + clock: Callable[[], float] = time.time, + settings: Settings = Settings(), + ) -> None: + self._subscriptions = subscriptions + self._bus = bus + self._clock = clock + self._settings = settings + self._open: dict[str, int] = {} + """Open streams per principal; an entry exists only while above zero.""" + self._count = 0 + self._closed = False + self._closing: anyio.Event | None = None + self._loop: asyncio.AbstractEventLoop | None = None + + @property + def bus(self) -> Bus: + return self._bus + + @property + def subscriptions(self) -> Subscriptions: + return self._subscriptions + + @property + def settings(self) -> Settings: + return self._settings + + async def closed(self) -> None: + """Return once :meth:`close` has been called.""" + if self._closed: + return + await (self._closing or anyio.Event()).wait() + + def close(self) -> None: + """End every open stream and refuse new ones. Safe to call from a + signal handler or another thread: ``_closed`` is set first, and every + stream checks it only after the event it waits on exists.""" + self._closed = True + if self._loop is not None and self._closing is not None: + self._loop.call_soon_threadsafe(self._closing.set) + + async def stream( + self, context: ServerRequestContext[Any, Any], params: StreamParams + ) -> dict[str, Any]: + started = time.monotonic() + if context.request_id is None: + raise EventsError.invalid_params("events/stream needs a request id") + target = resolve_target(params.name, params.arguments) + if params.cursor is not None and not isinstance(params.cursor, str): + raise EventsError.invalid_params("cursor must be a string or null") + check_max_age(params.max_age_ms) + principal = current_principal() + require_writable() + token = get_access_token() + expires = token.expires_at if token is not None else None + if self._closing is None: + self._closing = anyio.Event() + self._loop = asyncio.get_running_loop() + if self._closed: + raise EventsError.internal("The server is shutting down; reconnect") + with self._slot(principal): + grant = self.grant() + subscription = Subscription.stream( + project=target.project, + name=target.event.name, + arguments=target.arguments, + expires=grant.expires, + refresh=grant.refresh, + principal=principal, + ) + try: + listener = await self._bus.listen(subscription.id) + except BusError as error: + raise EventsError.internal("Push delivery is unavailable") from error + stream = Stream(self, context, target, subscription, listener, expires) + try: + await self._subscriptions.open(target, subscription) + except BaseException as error: + with anyio.CancelScope(shield=True): + await stream.listener.aclose() + # A refusal created nothing; an abort mid-create waited for + # the create, so the webhook exists by now. + if not isinstance(error, EventsError): + await stream.delete() + raise + await stream.run(started) + return {"_meta": {}} + + @contextmanager + def _slot(self, principal: str) -> Generator[None]: + held = self._open.get(principal, 0) + if held >= self._settings.principal_streams: + raise EventsError.resource_exhausted( + f"You already hold {held} event streams on this server; close one " + "before opening another", + limit="streams", + maximum=self._settings.principal_streams, + ) + if self._count >= self._settings.streams: + raise EventsError.resource_exhausted( + "This server holds as many event streams as it can; retry later", + limit="server_streams", + maximum=self._settings.streams, + ) + self._open[principal] = held + 1 + self._count += 1 + try: + yield + finally: + self._count -= 1 + remaining = self._open[principal] - 1 + if remaining: + self._open[principal] = remaining + else: + del self._open[principal] + + def grant(self) -> Grant: + """A full lease from now; its ``refresh`` is when the stream renews + it.""" + now = self.now() + expires = now + int(self._settings.lease * 1000) + return Grant(expires, min(now + int(self._settings.renew * 1000), expires)) + + def now(self) -> int: + """Unix epoch milliseconds.""" + return int(self._clock() * 1000) + + +class Stream: + """One open ``events/stream``.""" + + def __init__( + self, + streams: Streams, + context: ServerRequestContext[Any, Any], + target: Target, + subscription: Subscription, + listener: Listener, + expires: int | None, + ) -> None: + self._streams = streams + self._context = context + self._target = target + self.subscription = subscription + self.listener = listener + self._expires = expires + """When the stream's token expires, in Unix epoch seconds.""" + self._group: TaskGroup | None = None + self._outcome = Outcome.CLOSED + + @property + def _event(self) -> str: + return self._target.event.name + + async def run(self, started: float) -> None: + """Serve the stream until the client leaves, it is terminated or the + server closes it; then delete its webhook. ``started`` is when the + request arrived (``time.monotonic``).""" + telemetry.record_stream(self._event, 1) + telemetry.record_subscription(Operation.STREAM, Outcome.OPENED, self._event) + _log(f"{self.subscription.id}: stream opened for {self._event}") + try: + await self._notify(ACTIVE, {"cursor": None, "truncated": False}) + # Counted once open, not when it ends: a client abort, the usual + # end of a stream, is a cancellation. + telemetry.record_message(METHOD, "success", time.monotonic() - started) + async with anyio.create_task_group() as group: + self._group = group + group.start_soon(self._forward) + group.start_soon(self._beat) + group.start_soon(self._keep) + group.start_soon(self._wait_closing) + group.start_soon(self._wait_expiry) + finally: + with anyio.CancelScope(shield=True): + await self.listener.aclose() + await self.delete() + telemetry.record_stream(self._event, -1) + telemetry.record_subscription( + Operation.STREAM, self._outcome, self._event + ) + _log(f"{self.subscription.id}: stream {self._outcome}") + + def _end(self) -> None: + if self._group is not None: + self._group.cancel_scope.cancel() + + async def _notify(self, method: str, params: dict[str, Any]) -> None: + request = self._context.request_id + notification = Notification[dict[str, Any], str]( + method=method, + params={**params, "_meta": {SUBSCRIPTION_ID_META_KEY: request}}, + ) + # The SDK's notification union has no events notifications yet + # (python-sdk#3640); it only serializes the model it is given. + await self._context.session.send_notification( + cast(ServerNotification, notification), related_request_id=request + ) + + async def _forward(self) -> None: + while True: + try: + async for event in self.listener: + if event.name != self._event: + continue + await self._notify(EVENT, event.occurrence()) + telemetry.record_event_delivery( + event.name, "delivered", None, Mode.PUSH + ) + except BusError as error: + _log(f"{self.subscription.id}: {error}") + await self._error( + INTERNAL_ERROR, "Lost the connection that carries events" + ) + await self.listener.aclose() + self.listener = await self._reconnect() + await self._notify(ACTIVE, {"cursor": None, "truncated": True}) + + async def _reconnect(self) -> Listener: + while True: + await anyio.sleep(self._streams.settings.retry) + try: + return await self._streams.bus.listen(self.subscription.id) + except BusError as error: + _log(f"{self.subscription.id}: {error}") + + async def _beat(self) -> None: + while True: + await anyio.sleep(self._streams.settings.heartbeat) + await self._notify(HEARTBEAT, {"cursor": None}) + + async def _keep(self) -> None: + """Renew the lease until the stream ends.""" + settings = self._streams.settings + delay = settings.renew + while True: + await anyio.sleep(delay) + grant = self._streams.grant() + renewed = replace( + self.subscription, expires=grant.expires, refresh=grant.refresh + ) + try: + with anyio.fail_after(settings.appwrite): + await self._streams.subscriptions.renew(self._target, renewed) + except (EventsError, TimeoutError) as caught: + error = ( + caught + if isinstance(caught, EventsError) + else EventsError.internal("Appwrite did not answer in time") + ) + if error.code in (FORBIDDEN, NOT_FOUND): + await self._terminate(error) + return + _log(f"{self.subscription.id}: renewal failed ({error.code})") + if self._streams.now() >= self.subscription.expires: + await self._terminate( + EventsError.internal("The subscription could not be renewed") + ) + return + await self._error(error.code, error.message) + delay = settings.retry + continue + self.subscription = renewed + delay = settings.renew + + async def _wait_closing(self) -> None: + await self._streams.closed() + self._end() + + async def _wait_expiry(self) -> None: + """End the stream before its token expires; see the module docstring.""" + if self._expires is None: + return + now = self._streams.now() / 1000 + ending = self._expires - self._streams.settings.expiry + await anyio.sleep(max(ending if ending > now else self._expires, now) - now) + self._outcome = Outcome.EXPIRED + self._end() + + async def _error(self, code: int, message: str) -> None: + await self._notify(ERROR, {"error": {"code": code, "message": message}}) + + async def _terminate(self, error: EventsError) -> None: + payload: dict[str, Any] = {"code": error.code, "message": error.message} + if error.data is not None: + payload["data"] = error.data + # A client that stopped reading must not keep the stream open. + with anyio.move_on_after(CLOSE_SECONDS): + await self._notify(TERMINATED, {"error": payload}) + self._outcome = Outcome.TERMINATED + _log(f"{self.subscription.id}: terminated ({error.code})") + self._end() + + async def delete(self) -> None: + """Delete the stream's webhook, best effort, within + :data:`CLOSE_SECONDS`.""" + try: + with anyio.fail_after(CLOSE_SECONDS): + await self._streams.subscriptions.close( + self._target, self.subscription.id + ) + except Exception as error: + # The lease runs out and cleanup removes the webhook later. + _log(f"{self.subscription.id}: webhook not deleted ({error!r})") + + +def _log(message: str) -> None: + print(f"[appwrite-mcp][events] {message}", file=sys.stderr, flush=True) diff --git a/src/mcp_server_appwrite/events/subscriptions.py b/src/mcp_server_appwrite/events/subscriptions.py index 4255375..ccf8929 100644 --- a/src/mcp_server_appwrite/events/subscriptions.py +++ b/src/mcp_server_appwrite/events/subscriptions.py @@ -50,6 +50,11 @@ success, so unsubscribe is idempotent; a missing project is ``-32012``, as on subscribe. The id is bound to the principal, so a caller can only ever reach their own subscriptions. + +Push subscriptions (:mod:`.streams`) use the same checks and webhooks without a +callback: :meth:`Subscriptions.open` creates the webhook of a stream, +:meth:`Subscriptions.renew` extends its lease and :meth:`Subscriptions.close` +deletes it. """ from __future__ import annotations @@ -61,6 +66,7 @@ from enum import StrEnum from typing import Any +import anyio import httpx from anyio import to_thread from appwrite_console.client import Client @@ -76,6 +82,7 @@ from .envelope import ( EnvelopeTooLarge, InvalidSubscription, + Mode, Principal, Subscription, subscription_id, @@ -83,10 +90,7 @@ ) from .errors import CallbackFailure, EventsError from .ingress import Ingress -from .webhooks import Label, Webhooks, managed, owner - -WEBHOOK = "webhook" -"""The only delivery mode served (v1 is webhook-only, like ChatGPT).""" +from .webhooks import NOT_FOUND, Label, Webhooks, managed, owner TTL_DEFAULT_MS = 60 * 60 * 1000 """Granted when the client suggests no TTL, or asks for none (``ttlMs: null``): @@ -111,6 +115,17 @@ cleans it up. A client still refreshing that subscription would have done so well before; past this it has gone, and its deliveries are already dropped.""" +PUSH_CLEANUP_GRACE_MS = 60 * 1000 +"""The same for the webhook of a push stream. An open stream renews its lease +long before it runs out and ends the stream once it has, so a push webhook past +its lease belongs to a stream that is gone (its replica died). The short grace +gives the Free plan's slots back soon after.""" + +THREADS = 40 +"""Appwrite calls of the events methods running at once. They have their own +pool so a burst of stream reconnects after a deploy cannot starve tool calls, +and the other way round.""" + PROJECT = "project_id" SCOPE_PREFIX = "project:" @@ -123,6 +138,7 @@ class Operation(StrEnum): SUBSCRIBE = "subscribe" UNSUBSCRIBE = "unsubscribe" CLEANUP = "cleanup" + STREAM = "stream" class Outcome(StrEnum): @@ -130,6 +146,10 @@ class Outcome(StrEnum): REFRESHED = "refreshed" REMOVED = "removed" ABSENT = "absent" + OPENED = "opened" + CLOSED = "closed" + TERMINATED = "terminated" + EXPIRED = "expired" class AppwriteError(StrEnum): @@ -205,11 +225,10 @@ class State: @dataclass(frozen=True) class Target: - """What a subscribe or unsubscribe request names, validated.""" + """The event and arguments a request names, validated.""" event: Event arguments: dict[str, str] - delivery: Mapping[str, Any] @property def project(self) -> str: @@ -230,19 +249,36 @@ def __init__( self._ingress = ingress self._client = client self._clock = clock + self._limiter: anyio.CapacityLimiter | None = None def _now(self) -> int: return int(self._clock() * 1000) + async def _run( + self, function: Callable[..., Any], *args: Any, abandon: bool = True + ) -> Any: + """``function(*args)`` on the events thread pool. With ``abandon``, a + cancelled caller stops waiting for it; otherwise the cancellation + takes effect once it returns.""" + if self._limiter is None: + self._limiter = anyio.CapacityLimiter(THREADS) + return await to_thread.run_sync( + function, *args, abandon_on_cancel=abandon, limiter=self._limiter + ) + async def subscribe(self, params: SubscribeParams) -> dict[str, Any]: - target = _target(params.name, params.arguments, params.delivery) - mode = target.delivery.get("mode", WEBHOOK) - if mode != WEBHOOK: + target = resolve_target(params.name, params.arguments) + delivery = _delivery(params.delivery) + mode = delivery.get("mode", Mode.WEBHOOK) + if mode != Mode.WEBHOOK: raise EventsError.unsupported( - f"Delivery mode {mode!r} is not supported; use 'webhook'" + f"events/subscribe delivers only by {Mode.WEBHOOK!r}; " + "for push, open an events/stream", + feature="deliveryMode", + value=mode, ) - callback = _callback(target.delivery) - secret = target.delivery.get("secret") + callback = _callback(delivery) + secret = delivery.get("secret") if not isinstance(secret, str): raise EventsError.invalid_params("delivery.secret is required") try: @@ -250,9 +286,9 @@ async def subscribe(self, params: SubscribeParams) -> dict[str, Any]: except InvalidSecretError as error: raise EventsError.invalid_params(f"delivery.secret: {error}") from error suggested = _ttl(params) - _max_age(params.max_age_ms) - principal = _principal() - _writable() + check_max_age(params.max_age_ms) + principal = current_principal() + require_writable() grant = Grant.of(suggested, self._now()) subscription = Subscription.create( @@ -265,27 +301,12 @@ async def subscribe(self, params: SubscribeParams) -> dict[str, Any]: refresh=grant.refresh, principal=principal, ) - try: - envelope = self._ingress.keyring.seal(subscription, self._ingress.server) - except EnvelopeTooLarge as error: - raise EventsError.invalid_params( - "delivery.url is too long to fit in the subscription" - ) from error - except InvalidSubscription as error: - raise EventsError.invalid_params( - "delivery is not a valid subscription: the url must be https " - "without credentials or a fragment, and the secret canonical " - "padded base64" - ) from error + envelope = self._seal(subscription) await self._check(callback) - state = await to_thread.run_sync( - self._prepare, target, subscription, abandon_on_cancel=True - ) + state = await self._run(self._prepare, target, subscription) await self._verify(subscription) - outcome = await to_thread.run_sync( - self._write, state, target, subscription, envelope, abandon_on_cancel=True - ) + outcome = await self._run(self._write, state, target, subscription, envelope) telemetry.record_subscription(Operation.SUBSCRIBE, outcome, target.event.name) _log(f"{subscription.id}: {outcome} {target.event.name} until {grant.expires}") return { @@ -296,19 +317,56 @@ async def subscribe(self, params: SubscribeParams) -> dict[str, Any]: } async def unsubscribe(self, params: UnsubscribeParams) -> dict[str, Any]: - target = _target(params.name, params.arguments, params.delivery) - url = target.delivery.get("url") + target = resolve_target(params.name, params.arguments) + url = _delivery(params.delivery).get("url") if not isinstance(url, str) or not url: raise EventsError.invalid_params("delivery.url is required") - principal = _principal() + principal = current_principal() id = subscription_id(principal, url, target.event.name, target.arguments) - outcome = await to_thread.run_sync( - self._remove, target, id, abandon_on_cancel=True - ) + outcome = await self._run(self._remove, target, id) telemetry.record_subscription(Operation.UNSUBSCRIBE, outcome, target.event.name) _log(f"{id}: unsubscribe {outcome}") return {} + async def open(self, target: Target, subscription: Subscription) -> None: + """Create the webhook of push subscription ``subscription``, after the + checks a subscribe makes (authorization, webhook scopes, cleanup of + the caller's expired subscriptions). + + Not abandoned when cancelled: a client that aborts while the webhook + is being created waits for the create, so the stream's cleanup always + comes after it and never leaves the webhook behind.""" + envelope = self._seal(subscription) + state = await self._run(self._prepare, target, subscription, abandon=False) + await self._run( + self._write, state, target, subscription, envelope, abandon=False + ) + + async def renew(self, target: Target, subscription: Subscription) -> None: + """Rewrite a push subscription's webhook with its new lease. Repeats + the authorization read, so access revoked mid-stream surfaces here.""" + envelope = self._seal(subscription) + await self._run(self._renew, target, subscription, envelope) + + async def close(self, target: Target, id: str) -> bool: + """Delete the webhook of push subscription ``id``; whether it was + there.""" + return await self._run(self._delete, target, id) + + def _seal(self, subscription: Subscription) -> str: + try: + return self._ingress.keyring.seal(subscription, self._ingress.server) + except EnvelopeTooLarge as error: + raise EventsError.invalid_params( + "delivery.url is too long to fit in the subscription" + ) from error + except InvalidSubscription as error: + raise EventsError.invalid_params( + "delivery is not a valid subscription: the url must be https " + "without credentials or a fragment, and the secret canonical " + "padded base64" + ) from error + async def _check(self, url: str) -> None: """Refuse a callback whose host is not public before anything is sent to Appwrite or to the callback. The connect-time check still runs on @@ -328,6 +386,8 @@ async def _check(self, url: str) -> None: ) from error async def _verify(self, subscription: Subscription) -> None: + if subscription.callback is None or subscription.secret is None: + raise ValueError("only a webhook subscription has a callback to verify") callback = Callback(subscription.callback, subscription.id, subscription.secret) try: await self._ingress.dispatcher.verify(callback) @@ -385,15 +445,7 @@ def _write( envelope: str, ) -> Outcome: webhooks, count = state.webhooks, state.count - label = Label( - target.event.name, subscription.expires, owner(subscription.principal) - ) - fields: dict[str, Any] = { - "url": self._ingress.url(subscription.id), - "label": label, - "events": target.event.patterns(target.arguments), - "password": envelope, - } + fields = self._fields(target, subscription, envelope) secret = self._ingress.keyring.signing_key(subscription.id) if not state.exists: try: @@ -423,6 +475,48 @@ def _write( raise _webhook_error(error, target.project, count) from error return Outcome.REFRESHED + def _fields( + self, target: Target, subscription: Subscription, envelope: str + ) -> dict[str, Any]: + """What a subscription's webhook is written with, besides its id and + signing key.""" + label = Label( + target.event.name, + subscription.expires, + owner(subscription.principal), + push=subscription.mode is Mode.PUSH, + ) + return { + "url": self._ingress.url(subscription.id), + "label": label, + "events": target.event.patterns(target.arguments), + "password": envelope, + } + + def _renew(self, target: Target, subscription: Subscription, envelope: str) -> None: + webhooks = Webhooks(self._authorize(target)) + secret = self._ingress.keyring.signing_key(subscription.id) + try: + # The signing key goes first, as on a refresh. + webhooks.sign(subscription.id, secret) + webhooks.update( + subscription.id, **self._fields(target, subscription, envelope) + ) + except AppwriteException as error: + if error.type == NOT_FOUND: + raise EventsError.not_found( + f"The subscription's webhook {subscription.id} was deleted", + kind="subscription", + ) from error + raise _webhook_error(error, target.project, 0) from error + + def _delete(self, target: Target, id: str) -> bool: + webhooks = Webhooks(self._project_client(target.project)) + try: + return webhooks.delete(id) + except AppwriteException as error: + raise _webhook_error(error, target.project, 0) from error + def _clean( self, webhooks: Webhooks, existing: list[Any], subscription: Subscription ) -> int: @@ -432,20 +526,21 @@ def _clean( Appwrite has no conditional delete, so a refresh of a candidate can race this. Two guards keep that from deleting a live subscription: - only webhooks expired for longer than :data:`CLEANUP_GRACE_MS` are - candidates, and each is read again right before the delete and kept + only webhooks expired for longer than :data:`CLEANUP_GRACE_MS` + (:data:`PUSH_CLEANUP_GRACE_MS` for a push stream's) are candidates, and each is read again right before the delete and kept if its name or ``$updatedAt`` changed since the listing. What remains is a refresh landing between that read and the delete, of a subscription its client left expired for over ten minutes; the client then gets its webhook back on the following refresh.""" tag = owner(subscription.principal) - cutoff = self._now() - CLEANUP_GRACE_MS + now = self._now() removed = 0 for webhook in existing: label = managed(webhook) if label is None or webhook.id == subscription.id: continue - if label.owner != tag or not label.expired(cutoff): + grace = PUSH_CLEANUP_GRACE_MS if label.push else CLEANUP_GRACE_MS + if label.owner != tag or not label.expired(now - grace): continue try: current = webhooks.get(webhook.id) @@ -480,12 +575,16 @@ def _remove(self, target: Target, id: str) -> Outcome: raise _webhook_error(error, target.project, 0) from error -def _target(name: object, arguments: object, delivery: object) -> Target: +def resolve_target(name: object, arguments: object) -> Target: + """The catalog event ``name`` and its validated ``arguments``.""" event = lookup(name) - validated = event.validate(arguments) + return Target(event, event.validate(arguments)) + + +def _delivery(delivery: object) -> Mapping[str, Any]: if not isinstance(delivery, Mapping): raise EventsError.invalid_params("delivery must be an object") - return Target(event, validated, delivery) + return delivery def _callback(delivery: Mapping[str, Any]) -> str: @@ -512,7 +611,7 @@ def _ttl(params: SubscribeParams) -> int | None: return value -def _max_age(value: object) -> None: +def check_max_age(value: object) -> None: """``maxAgeMs`` bounds replay. Appwrite webhooks have none, so it is only checked for shape.""" if value is None: @@ -523,7 +622,7 @@ def _max_age(value: object) -> None: ) -def _writable() -> None: +def require_writable() -> None: """Refuse a token whose granted scopes cannot write webhooks before the callback gets a verification POST. @@ -546,7 +645,7 @@ def _scopes_message() -> str: ) -def _principal() -> str: +def current_principal() -> str: """Digest of the OAuth principal behind the current request.""" token = get_access_token() claims = (token.claims or {}) if token is not None else {} diff --git a/src/mcp_server_appwrite/events/webhooks.py b/src/mcp_server_appwrite/events/webhooks.py index 406c32e..cebba38 100644 --- a/src/mcp_server_appwrite/events/webhooks.py +++ b/src/mcp_server_appwrite/events/webhooks.py @@ -9,7 +9,8 @@ :func:`.envelope.subscription_id`), so a refresh is an update and an unsubscribe a delete by id. * ``name`` is a :class:`Label`, for example - ``MCP events | tablesdb.row.created | until 2026-10-09T11:00:00Z | owner OPxPhaa1``. + ``MCP events | tablesdb.row.created | until 2026-10-09T11:00:00Z | owner OPxPhaa1`` + (``MCP events push | …`` for the webhook of a push stream). Appwrite stores ``authPassword`` write-only, so the sealed envelope (and the expiry inside it) can never be read back. The label repeats the two facts a later subscribe needs to clean up without opening it: when the subscription @@ -47,6 +48,7 @@ Appwrite send the envelope (the password) as Basic auth.""" PREFIX = "MCP events" +PUSH_PREFIX = "MCP events push" SEPARATOR = " | " """ASCII on purpose: Appwrite's worker sends the name back in the ``X-Appwrite-Webhook-Name`` header of every delivery.""" @@ -86,11 +88,14 @@ class Label: """Expiry as Unix epoch milliseconds. Written rounded up to the second, so the label never claims an earlier expiry than the envelope.""" owner: str + push: bool = False + """Whether it backs a push stream, which renews a short lease while open.""" def __str__(self) -> str: seconds = math.ceil(self.expires / 1000) stamp = datetime.fromtimestamp(seconds, UTC).strftime(STAMP) - return SEPARATOR.join((PREFIX, self.event, UNTIL + stamp, OWNER + self.owner)) + prefix = PUSH_PREFIX if self.push else PREFIX + return SEPARATOR.join((prefix, self.event, UNTIL + stamp, OWNER + self.owner)) @classmethod def parse(cls, name: str) -> Label | None: @@ -99,7 +104,7 @@ def parse(cls, name: str) -> Label | None: if len(parts) != 4: return None prefix, event, until, tag = parts - if prefix != PREFIX or not until.startswith(UNTIL): + if prefix not in (PREFIX, PUSH_PREFIX) or not until.startswith(UNTIL): return None if not tag.startswith(OWNER): return None @@ -111,7 +116,7 @@ def parse(cls, name: str) -> Label | None: except ValueError: return None expires = int(stamp.replace(tzinfo=UTC).timestamp()) * 1000 - return cls(event=event, expires=expires, owner=tag) + return cls(event=event, expires=expires, owner=tag, push=prefix == PUSH_PREFIX) def expired(self, now: int) -> bool: return now >= self.expires diff --git a/src/mcp_server_appwrite/flags.py b/src/mcp_server_appwrite/flags.py index 68af131..dfd388f 100644 --- a/src/mcp_server_appwrite/flags.py +++ b/src/mcp_server_appwrite/flags.py @@ -54,7 +54,16 @@ class Flag: ), ) -FLAGS: tuple[Flag, ...] = (CONSOLE_URL, EVENTS) +EVENTS_PUSH = Flag( + name="events-push", + env="MCP_EVENTS_PUSH", + help=( + "Set to 1 to also offer push delivery (events/stream) for MCP Events. " + "Needs the events flag and MCP_EVENTS_REDIS_URL." + ), +) + +FLAGS: tuple[Flag, ...] = (CONSOLE_URL, EVENTS, EVENTS_PUSH) ENABLED_VALUES = frozenset({"1", "true", "yes", "on"}) """Values that switch an on/off flag on (case-insensitive).""" diff --git a/src/mcp_server_appwrite/http_app.py b/src/mcp_server_appwrite/http_app.py index 2b91d48..ff1b769 100644 --- a/src/mcp_server_appwrite/http_app.py +++ b/src/mcp_server_appwrite/http_app.py @@ -18,6 +18,10 @@ with the events flag on. Unauthenticated by bearer token: the Appwrite signature authenticates it (see :mod:`.events.ingress`). +Push delivery (``events/stream``, :mod:`.events.streams`) runs on the MCP +endpoint itself. Streams never finish on their own, so :class:`Uvicorn` closes +them when the server is asked to stop instead of waiting for them. + Auth uses the SDK primitives (``BearerAuthBackend`` + ``AuthContextMiddleware``) so the validated token is reachable from tool handlers via ``get_access_token()``. """ @@ -32,9 +36,11 @@ import re import sys from importlib.resources import files +from types import FrameType from urllib.parse import urlsplit import httpx +import uvicorn from mcp.server.auth.middleware.auth_context import AuthContextMiddleware from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, BearerAuthBackend from mcp.server.streamable_http_manager import StreamableHTTPSessionManager @@ -53,7 +59,7 @@ from starlette.routing import Route from starlette.types import ASGIApp, Message, Receive, Scope, Send -from . import error_monitoring, telemetry +from . import error_monitoring, flags, telemetry from .auth import ( AppwriteTokenVerifier, authorization_server_metadata, @@ -68,10 +74,13 @@ from .constants import CORS_HEADERS, SERVER_VERSION from .events import protocol as events_protocol from .events.ingress import Ingress +from .events.streams import Settings, Streams +from .events.subscriptions import Subscriptions from .server import ( build_catalog_tools_manager, build_mcp_server, build_operator, + resolve_client, ) @@ -461,18 +470,31 @@ async def favicon_ico_endpoint(request: Request) -> RedirectResponse: return RedirectResponse("/favicon.svg", status_code=307) -def build_app(ingress: Ingress | None = None) -> Starlette: +def build_app( + ingress: Ingress | None = None, settings: Settings = Settings() +) -> Starlette: """The hosted app. ``ingress`` replaces the events ingress built from the - environment; it is mounted only when events are enabled.""" + environment; it is mounted only when events are enabled. ``settings`` + tunes the push streams. ``app.state.streams`` holds the push streams, + or ``None`` when push is off.""" error_monitoring.init_error_monitoring("http", SERVER_VERSION) telemetry.init_telemetry("http", SERVER_VERSION) tools_manager = build_catalog_tools_manager() operator = build_operator(tools_manager, store_results=False) + streams: Streams | None = None if events_protocol.enabled("http"): ingress = ingress or Ingress.from_env() + if ingress.bus is not None and events_protocol.push_enabled("http", ingress): + streams = Streams( + Subscriptions(ingress, resolve_client), ingress.bus, settings=settings + ) + elif flags.enabled(flags.EVENTS_PUSH): + _log("events-push is on but MCP_EVENTS_REDIS_URL is not set: webhook only") else: ingress = None - server = build_mcp_server(operator, transport="http", ingress=ingress) + server = build_mcp_server( + operator, transport="http", ingress=ingress, streams=streams + ) # Streamable HTTP with SSE responses (the MCP SDK/ecosystem default). Stateless, # so each request opens and closes its own short-lived stream — no session to pin. @@ -557,12 +579,30 @@ async def lifespan(app: Starlette): Middleware(AuthContextMiddleware), ] - return Starlette(routes=routes, middleware=middleware, lifespan=lifespan) + app = Starlette(routes=routes, middleware=middleware, lifespan=lifespan) + app.state.streams = streams + return app + + +class Uvicorn(uvicorn.Server): + """uvicorn for the hosted app. Asked to stop, it first closes the app's + push streams: uvicorn waits for open requests before it shuts down, and a + stream only ends when it is closed. Each one deletes its webhook and sends + its final result, and its client reconnects to another replica.""" + + def __init__(self, config: uvicorn.Config, app: Starlette) -> None: + super().__init__(config) + self._streams: Streams | None = app.state.streams + + def handle_exit(self, sig: int, frame: FrameType | None) -> None: + if self._streams is not None: + self._streams.close() + super().handle_exit(sig, frame) def run_http(*, host: str = "0.0.0.0", port: int = 8000) -> None: - import uvicorn from uvicorn.config import LOGGING_CONFIG + from uvicorn.main import STARTUP_FAILURE app = build_app() log_config = copy.deepcopy(LOGGING_CONFIG) @@ -572,4 +612,9 @@ def run_http(*, host: str = "0.0.0.0", port: int = 8000) -> None: log_config["handlers"]["access"]["filters"] = ["hide_healthz"] _log(f"Serving on http://{host}:{port}/ (also available at /mcp)") - uvicorn.run(app, host=host, port=port, log_config=log_config) + config = uvicorn.Config(app, host=host, port=port, log_config=log_config) + server = Uvicorn(config, app) + server.run() + if not server.started: + # As uvicorn.run does, so a failed startup is not a clean exit. + sys.exit(STARTUP_FAILURE) diff --git a/src/mcp_server_appwrite/server.py b/src/mcp_server_appwrite/server.py index 6881f59..4a2f151 100644 --- a/src/mcp_server_appwrite/server.py +++ b/src/mcp_server_appwrite/server.py @@ -76,6 +76,7 @@ from .error_classification import HostedBinaryResponseTooLarge, is_response_parse_error from .events import protocol as events from .events.ingress import Ingress +from .events.streams import Streams from .events.subscriptions import Subscriptions from .operator import Operator, _parse_tool_name from .service import Service @@ -1268,11 +1269,16 @@ def build_instructions(transport: str = "http", *, docs_enabled: bool = True) -> def build_mcp_server( - operator: Operator, *, transport: str = "http", ingress: Ingress | None = None + operator: Operator, + *, + transport: str = "http", + ingress: Ingress | None = None, + streams: Streams | None = None, ) -> Server: """The low-level MCP server. With events enabled, ``events/subscribe`` writes webhooks that point at ``ingress`` (built from the environment when - not given) and verifies callbacks through its egress.""" + not given) and verifies callbacks through its egress. ``streams``, when + given, serves push delivery (``events/stream``).""" _configure_uploads(transport) instructions = build_instructions( transport, docs_enabled=bool(getattr(operator, "docs_enabled", False)) @@ -1529,6 +1535,7 @@ def tool_input_schema(name: str) -> dict[str, Any] | None: server, Subscriptions(ingress or Ingress.from_env(), resolve_client), lambda ctx: _mcp_request_context(ctx).tags, + streams, ) return server diff --git a/src/mcp_server_appwrite/telemetry.py b/src/mcp_server_appwrite/telemetry.py index 277d248..ef92152 100644 --- a/src/mcp_server_appwrite/telemetry.py +++ b/src/mcp_server_appwrite/telemetry.py @@ -292,7 +292,15 @@ def _build_instruments(meter: Any, transport: str, version: str) -> None: _instruments["events_deliveries"] = meter.create_counter( "mcp.events.deliveries", unit="{event}", - description="MCP event deliveries to subscriber callbacks by outcome.", + description=( + "MCP event deliveries by mode (webhook callbacks, push streams) " + "and outcome." + ), + ) + _instruments["events_streams"] = meter.create_up_down_counter( + "mcp.events.streams", + unit="{stream}", + description="Open events/stream push subscriptions.", ) _instruments["events_subscriptions"] = meter.create_counter( "mcp.events.subscriptions", @@ -737,10 +745,18 @@ def record_subscription( ) -def record_event_delivery(event: str, outcome: str, reason: str | None) -> None: - """How one MCP event delivery to a subscriber callback ended.""" +def record_event_delivery( + event: str, outcome: str, reason: str | None, mode: str +) -> None: + """How one MCP event delivery ended: to a subscriber callback (``mode`` + ``webhook``) or on an ``events/stream`` (``push``).""" _safe_add( "events_deliveries", 1, - {"event": event, "outcome": outcome, "reason": reason}, + {"event": event, "outcome": outcome, "reason": reason, "mode": mode}, ) + + +def record_stream(event: str, delta: int) -> None: + """An ``events/stream`` opened (``delta`` 1) or closed (-1).""" + _safe_add("events_streams", delta, {"event": event}) diff --git a/tests/e2e/support.py b/tests/e2e/support.py index fe2a34d..1393f77 100644 --- a/tests/e2e/support.py +++ b/tests/e2e/support.py @@ -28,10 +28,15 @@ ``standardwebhooks`` library and echoes verification challenges. * :class:`Collector` is an OTLP/HTTP metrics endpoint. Every server exports to it, so tests read the counters the server really emits. +* :class:`Redis` is a real Dragonfly (production's bus), or ``redis-server`` + where Dragonfly has no build, on a random port: the bus that push streams + share across server instances. :meth:`Client.stream` opens an + ``events/stream`` and reads its SSE frames as they arrive. """ from __future__ import annotations +import atexit import base64 import datetime import hashlib @@ -39,10 +44,14 @@ import itertools import json import os +import queue import re import secrets +import shutil +import signal import socket import ssl +import subprocess import tempfile import threading import time @@ -66,11 +75,12 @@ from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( ExportMetricsServiceRequest, ) +from sse_starlette.sse import AppStatus from standardwebhooks import Webhook as StandardWebhook from standardwebhooks.webhooks import WebhookVerificationError from mcp_server_appwrite import auth -from mcp_server_appwrite.http_app import build_app +from mcp_server_appwrite.http_app import Uvicorn, build_app TOKEN = "e2e-access-token" """The bearer token the stubbed verifier accepts, for :data:`SUBJECT`.""" @@ -97,6 +107,10 @@ """The OAuth ``scope`` claim of a token, for tests that set one; others carry none.""" +EXPIRES: dict[str, int] = {} +"""The ``exp`` claim (Unix seconds) of a token, for tests that set one; others +never expire.""" + PROTOCOL_VERSION = "2026-07-28" LEGACY_PROTOCOL_VERSION = "2025-11-25" @@ -109,6 +123,8 @@ CONTROLLED = ( "APPWRITE_ENDPOINT", "MCP_EVENTS", + "MCP_EVENTS_PUSH", + "MCP_EVENTS_REDIS_URL", "MCP_EVENTS_SEALING_KEYS", "MCP_PUBLIC_URL", "MCP_CONSOLE_URL", @@ -135,6 +151,7 @@ async def _verify_token(_verifier: Any, token: str) -> AccessToken | None: token=token, client_id=CLIENT_ID, scopes=list(GRANTED.get(token, [])), + expires_at=EXPIRES.get(token), subject=subject, claims=claims, ) @@ -161,7 +178,7 @@ def __init__( } self._build = build self._patches: list[Any] = [] - self._server: uvicorn.Server | None = None + self._server: Uvicorn | None = None self._thread: threading.Thread | None = None self.url = "" @@ -196,7 +213,7 @@ def _start(self) -> None: listener.bind(("127.0.0.1", 0)) port = listener.getsockname()[1] config = uvicorn.Config(app, log_level="warning", access_log=False, ws="none") - self._server = uvicorn.Server(config) + self._server = Uvicorn(config, app) self._thread = threading.Thread( target=self._server.run, kwargs={"sockets": [listener]}, daemon=True ) @@ -219,20 +236,25 @@ def __exit__( self._stop_patches() def _shutdown(self) -> None: + """Stop the way a pod is stopped: uvicorn's SIGTERM handling.""" if self._server is not None: - self._server.should_exit = True + self._server.handle_exit(signal.SIGTERM, None) if self._thread is not None: self._thread.join(STARTUP_SECONDS) self._server = None self._thread = None + # sse_starlette, which serves the SDK's 2025-era SSE responses, patches + # uvicorn's exit handler to drain every SSE stream in the process. A + # pod runs one server; here the next server would inherit the drain. + AppStatus.should_exit = False def _stop_patches(self) -> None: for patch in reversed(self._patches): patch.stop() self._patches = [] - def client(self) -> Client: - return Client(self.url) + def client(self, token: str = TOKEN) -> Client: + return Client(self.url, token) def message(response: httpx.Response) -> dict[str, Any]: @@ -291,6 +313,32 @@ def call( response = self._http.post(path, json=body, headers=headers) return response.status_code, message(response) + def stream( + self, params: Mapping[str, Any], *, legacy: bool = False, path: str = "/" + ) -> Stream: + """Open an ``events/stream``: as a 2026-07-28 request, or with + ``legacy`` as a 2025-11-25 one, the way handshake-era clients send it + after ``initialize``.""" + id = next(self._ids) + body: dict[str, Any] = { + "jsonrpc": "2.0", + "id": id, + "method": "events/stream", + "params": dict(params), + } + headers = {**self._headers} + if legacy: + headers["MCP-Protocol-Version"] = LEGACY_PROTOCOL_VERSION + else: + body["params"]["_meta"] = { + "io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION, + "io.modelcontextprotocol/clientInfo": {"name": "e2e", "version": "1"}, + "io.modelcontextprotocol/clientCapabilities": {}, + } + headers["MCP-Protocol-Version"] = PROTOCOL_VERSION + headers["Mcp-Method"] = "events/stream" + return Stream(self._http.base_url, path, id, body, headers) + def initialize(self, *, path: str = "/") -> tuple[int, dict[str, Any]]: """A 2025-11-25 ``initialize`` handshake.""" body = { @@ -307,6 +355,171 @@ def initialize(self, *, path: str = "/") -> tuple[int, dict[str, Any]]: return response.status_code, message(response) +_END = object() + + +class Stream: + """One open ``events/stream``, read on a thread of its own. + + :meth:`next` returns its JSON-RPC messages in order: the notifications, + then the final result or error. A response that is not an SSE stream (an + immediate error) is one message. :meth:`close` aborts the request, which is + how a client cancels a stream.""" + + def __init__( + self, + base: httpx.URL, + path: str, + id: int, + body: Mapping[str, Any], + headers: Mapping[str, str], + ) -> None: + self.id = id + self.status = 0 + self.content_type = "" + self._http = httpx.Client(base_url=base, timeout=httpx.Timeout(30.0, read=None)) + self._messages: queue.Queue[Any] = queue.Queue() + self._opened = threading.Event() + self._thread = threading.Thread( + target=self._read, args=(path, body, headers), daemon=True + ) + self._thread.start() + + def _read( + self, path: str, body: Mapping[str, Any], headers: Mapping[str, str] + ) -> None: + try: + with self._http.stream( + "POST", path, json=body, headers=headers + ) as response: + self.status = response.status_code + self.content_type = response.headers.get("content-type", "") + self._opened.set() + if not self.content_type.startswith("text/event-stream"): + self._messages.put(json.loads(response.read())) + return + for line in response.iter_lines(): + if line.startswith("data:"): + self._messages.put(json.loads(line[5:])) + except (httpx.HTTPError, RuntimeError, ValueError): + pass + finally: + self._opened.set() + self._messages.put(_END) + + def next(self, timeout: float = 10.0) -> dict[str, Any]: + """The next message; fails when the stream ended or ``timeout`` passed.""" + try: + message = self._messages.get(timeout=timeout) + except queue.Empty: + raise AssertionError(f"no message within {timeout}s") from None + if message is _END: + self._messages.put(_END) + raise AssertionError("the stream ended") + return message + + def until(self, method: str, timeout: float = 10.0) -> dict[str, Any]: + """Skip heartbeats until a ``method`` notification arrives.""" + deadline = time.monotonic() + timeout + while True: + message = self.next(max(deadline - time.monotonic(), 0.01)) + if message.get("method") == method: + return message + if message.get("method") != "notifications/events/heartbeat": + raise AssertionError(f"expected {method}, got {message}") + + def ended(self, timeout: float = 10.0) -> bool: + """Whether the response finished within ``timeout``.""" + self._thread.join(timeout) + return not self._thread.is_alive() + + def close(self) -> None: + self._http.close() + self._thread.join(5.0) + + def __enter__(self) -> Stream: + self._opened.wait(30.0) + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() + + +class Redis: + """A Redis-protocol server on ``127.0.0.1:``, without + persistence: ``dragonfly`` when it is on ``PATH``, as in CI, otherwise + ``redis-server``. + + :meth:`stop` and :meth:`start` take it down and bring it back on the same + port, for outages.""" + + def __init__(self) -> None: + dragonfly = shutil.which("dragonfly") + redis = shutil.which("redis-server") + if dragonfly is not None: + self._arguments = [dragonfly, "--dbfilename=", "--proactor_threads=2"] + elif redis is not None: + self._arguments = [redis, "--save", "", "--appendonly", "no"] + else: + raise RuntimeError( + "push streams need dragonfly or redis-server on PATH: " + "`brew install redis` or `apt-get install redis-server`" + ) + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + self.port = listener.getsockname()[1] + listener.close() + self.url = f"redis://127.0.0.1:{self.port}/0" + self._process: subprocess.Popen[bytes] | None = None + self.start() + + def start(self) -> None: + self._process = subprocess.Popen( + [*self._arguments, "--port", str(self.port), "--bind", "127.0.0.1"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + deadline = time.monotonic() + STARTUP_SECONDS + while time.monotonic() < deadline: + try: + with socket.create_connection(("127.0.0.1", self.port), 0.2) as probe: + probe.sendall(b"PING\r\n") + if probe.recv(16).startswith(b"+PONG"): + return + except OSError: + time.sleep(0.02) + raise RuntimeError(f"{self._arguments[0]} did not start") + + def stop(self) -> None: + if self._process is not None: + self._process.kill() + self._process.wait() + self._process = None + + def pause(self) -> None: + """Freeze the server with its connections open: a peer that is gone + without a reset, as after a node loss.""" + if self._process is not None: + self._process.send_signal(signal.SIGSTOP) + + def resume(self) -> None: + if self._process is not None: + self._process.send_signal(signal.SIGCONT) + + +_redis: Redis | None = None + + +def redis() -> Redis: + """The process-wide Redis every push server shares.""" + global _redis + with _collector_lock: + if _redis is None: + _redis = Redis() + atexit.register(_redis.stop) + return _redis + + class _QuietServer(ThreadingHTTPServer): daemon_threads = True @@ -955,6 +1168,11 @@ def store(self, project: str, document: Mapping[str, Any]) -> None: **document, } + def remove(self, project: str, id: str) -> None: + """Delete a webhook, as a person in the Console would.""" + with self._lock: + del self._projects[project].webhooks[id] + def documents(self, project: str) -> dict[str, dict[str, Any]]: """The stored webhook documents, write-only fields included.""" with self._lock: diff --git a/tests/e2e/test_events_stream.py b/tests/e2e/test_events_stream.py new file mode 100644 index 0000000..a1b46aa --- /dev/null +++ b/tests/e2e/test_events_stream.py @@ -0,0 +1,657 @@ +"""Push delivery (``events/stream``) through real hosted servers. + +Two server instances share one real Redis, the way replicas behind the gateway +do. A client opens a stream on one, Appwrite's webhooks worker +(:meth:`support.Appwrite.fire`) delivers to the other from the webhook the +stream stored in :class:`support.Cloud`, and the event arrives on the stream: +the cross-replica hop this feature exists for. The flows then cover what a +stream owns over its life: one webhook with a short lease, renewed while the +stream is open and deleted when it closes, heartbeats, immediate errors for +invalid subscriptions, termination when access goes away, recovery from a bus +outage, a clean close on shutdown and a lease that runs out after a crash. + +The servers run with short clocks (:data:`SETTINGS`) so renewals and +heartbeats happen within seconds. +""" + +import base64 +import json +import re +import threading +import time +import unittest +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import replace +from pathlib import Path +from typing import Any + +from support import ( + EXPIRES, + OTHER_TOKEN, + PUBLIC_URL, + SUBJECTLESS_TOKEN, + SUBJECTS, + TOKEN, + Appwrite, + Client, + Cloud, + Project, + Server, + Stream, + collector, + redis, +) + +from mcp_server_appwrite.auth import public_base_url +from mcp_server_appwrite.events.bus import RedisBus +from mcp_server_appwrite.events.egress import Egress +from mcp_server_appwrite.events.envelope import Context, Keyring, Mode +from mcp_server_appwrite.events.ingress import Ingress +from mcp_server_appwrite.events.streams import Settings + +FIXTURES = Path(__file__).parent / "fixtures" / "appwrite" +KEYS = "k1:" + base64.b64encode(bytes(range(64, 96))).decode() + +SETTINGS = Settings(heartbeat=0.5, renew=1.0, retry=0.3, lease=30.0, appwrite=0.5) +PRINCIPAL_STREAMS = 2 +"""Streams one user may hold per server in these flows.""" + +MAIN = "6630f1a2b3c4d5e6f7a8" +FREE = "7740f1a2b3c4d5e6f7a8" +"""A Free-plan project whose two webhooks are taken.""" + +REVOKED = "e2e-stream-revoked" +"""A user whose access to the project is revoked mid-stream.""" +EXPIRING = "e2e-stream-expiring" +"""A user whose token expires while the stream is open.""" + +ROWS = {"project_id": MAIN, "database_id": "main", "table_id": "support_tickets"} +ROW_EVENT = "tablesdb.main.tables.support_tickets.rows.row_8f2a.create" +ROW_DATA = { + "database_id": "main", + "table_id": "support_tickets", + "row_id": "row_8f2a", + "created_at": "2026-10-09T10:00:00.250Z", +} + +LABEL = re.compile( + r"MCP events push \| tablesdb\.row\.created \| until " + r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ \| owner [A-Za-z0-9_-]{8}" +) +SUBSCRIPTION_ID = "io.modelcontextprotocol/subscriptionId" + +ACTIVE = "notifications/events/active" +EVENT = "notifications/events/event" +HEARTBEAT = "notifications/events/heartbeat" +ERROR = "notifications/events/error" +TERMINATED = "notifications/events/terminated" + +INVALID_PARAMS = -32602 +METHOD_NOT_FOUND = -32601 +INTERNAL_ERROR = -32603 +NOT_FOUND = -32011 +FORBIDDEN = -32012 +RESOURCE_EXHAUSTED = -32013 +UNSUPPORTED = -32014 + +cloud: Cloud + + +def setUpModule() -> None: + global cloud + cloud = Cloud() + SUBJECTS[REVOKED] = "cc17c8d9e0f1a2b3c4d5" + SUBJECTS[EXPIRING] = "dd28d9e0f1a2b3c4d5e6" + cloud.add(Project(MAIN, tables={("main", "support_tickets")})) + cloud.add(Project(FREE, tables={("main", "tickets")}, webhook_limit=2)) + for id in ("console_hook_1", "console_hook_2"): + cloud.store(FREE, {"$id": id, "name": "Mine", "url": "https://example.com"}) + cloud.grant(TOKEN, {MAIN, FREE}) + cloud.grant(OTHER_TOKEN, {MAIN}) + cloud.grant(EXPIRING, {MAIN}) + + +def tearDownModule() -> None: + cloud.close() + SUBJECTS.pop(REVOKED, None) + SUBJECTS.pop(EXPIRING, None) + EXPIRES.clear() + + +def fixture(name: str) -> dict[str, Any]: + return json.loads((FIXTURES / f"{name}.json").read_text()) + + +def environment(**extra: str | None) -> dict[str, str | None]: + return { + "MCP_EVENTS": "1", + "MCP_EVENTS_PUSH": "1", + "MCP_EVENTS_SEALING_KEYS": KEYS, + "MCP_EVENTS_REDIS_URL": redis().url, + "APPWRITE_ENDPOINT": cloud.endpoint, + **extra, + } + + +def server(settings: Settings = SETTINGS) -> Server: + """A push server. Its bus pings Redis every 0.3 s, so a dead connection + shows within a second.""" + + def build() -> dict[str, Any]: + bus = RedisBus(redis().url, ping=0.3, pong=0.5) + ingress = Ingress(Keyring.from_env(), Egress(), base=public_base_url(), bus=bus) + return {"settings": settings, "ingress": ingress} + + return Server(environment(), build) + + +def params(arguments: dict[str, str] = ROWS, **extra: Any) -> dict[str, Any]: + """``events/stream`` params as a push client sends them.""" + return { + "name": "tablesdb.row.created", + "arguments": dict(arguments), + "cursor": None, + **extra, + } + + +def wait(condition, timeout: float = 10.0) -> bool: + deadline = time.monotonic() + timeout + while not condition(): + if time.monotonic() > deadline: + return False + time.sleep(0.05) + return True + + +class StreamFlow(unittest.TestCase): + """Streams on one replica receive what Appwrite delivers to another.""" + + a: Server + b: Server + appwrite: Appwrite + + @classmethod + def setUpClass(cls) -> None: + settings = replace(SETTINGS, principal_streams=PRINCIPAL_STREAMS) + cls.a = server(settings).__enter__() + cls.b = server(settings).__enter__() + cls.appwrite = Appwrite(cls.b) + + @classmethod + def tearDownClass(cls) -> None: + cls.appwrite.close() + cls.b.__exit__(None, None, None) + cls.a.__exit__(None, None, None) + + def ours(self, project: str = MAIN) -> dict[str, dict[str, Any]]: + return { + id: document + for id, document in cloud.documents(project).items() + if LABEL.fullmatch(document["name"]) + } + + @contextmanager + def opened( + self, client: Client, request: dict[str, Any] | None = None, **options: Any + ) -> Iterator[tuple[Stream, str]]: + """A stream once ``active`` arrived, and the id of its webhook.""" + before = set(self.ours()) + with client.stream(request or params(), **options) as stream: + yield stream, self.activated(stream, before) + + def activated(self, stream: Stream, before: set[str]) -> str: + active = stream.next() + self.assertEqual(stream.status, 200) + self.assertTrue(stream.content_type.startswith("text/event-stream")) + self.assertEqual(active["method"], ACTIVE, active) + self.assertEqual( + active["params"], + {"cursor": None, "truncated": False, "_meta": {SUBSCRIPTION_ID: stream.id}}, + ) + created = set(self.ours()) - before + self.assertEqual(len(created), 1, created) + return created.pop() + + def refused(self, stream: Stream, code: int) -> dict[str, Any]: + """An invalid stream: an immediate JSON error, no SSE, no webhook.""" + message = stream.next() + self.assertTrue(stream.content_type.startswith("application/json"), message) + self.assertEqual(message["id"], stream.id) + self.assertEqual(message["error"]["code"], code, message) + return message["error"] + + def gone(self, id: str, project: str = MAIN) -> None: + self.assertTrue( + wait(lambda: id not in cloud.documents(project)), f"{id} still stored" + ) + + def test_stream_receives_events_delivered_to_another_replica(self): + for legacy in (False, True): + with self.subTest(legacy=legacy): + self.receive(legacy) + + def receive(self, legacy: bool) -> None: + metrics = collector() + streams = metrics.count("mcp.events.streams", event="tablesdb.row.created") + delivered = metrics.count( + "mcp.events.deliveries", mode="push", outcome="delivered" + ) + with self.a.client() as client, self.opened(client, legacy=legacy) as opened: + stream, id = opened + document = cloud.documents(MAIN)[id] + self.assertEqual(document["url"], f"{PUBLIC_URL}/appwrite/webhooks/{id}") + self.assertEqual( + document["events"], + ["tablesdb.main.tables.support_tickets.rows.*.create"], + ) + self.assertEqual(document["authUsername"], "mcp-events") + subscription = Keyring.parse(KEYS).open( + document["httpPass"], Context(PUBLIC_URL, MAIN, id) + ) + self.assertIs(subscription.mode, Mode.PUSH) + self.assertIsNone(subscription.callback) + self.assertAlmostEqual( + subscription.expires / 1000, time.time() + SETTINGS.lease, delta=5 + ) + self.assertEqual( + metrics.wait( + "mcp.events.streams", streams + 1, event=subscription.name + ), + streams + 1, + ) + + [answer] = self.appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + self.assertEqual(answer.status_code, 200, answer.text) + self.assertEqual(answer.json()["status"], "accepted", answer.text) + event = stream.until(EVENT) + self.assertEqual( + event["params"], + { + "eventId": answer.json()["eventId"], + "name": "tablesdb.row.created", + "timestamp": "2026-10-09T10:00:00.250Z", + "data": ROW_DATA, + "cursor": None, + "_meta": {SUBSCRIPTION_ID: stream.id}, + }, + ) + self.assertEqual( + metrics.wait( + "mcp.events.deliveries", + delivered + 1, + mode="push", + outcome="delivered", + ), + delivered + 1, + ) + + heartbeat = stream.next(SETTINGS.heartbeat * 4) + self.assertEqual( + heartbeat, + { + "jsonrpc": "2.0", + "method": HEARTBEAT, + "params": {"cursor": None, "_meta": {SUBSCRIPTION_ID: stream.id}}, + }, + ) + + # The lease is renewed while the stream is open. + renewals = cloud.calls("PUT", f"/v1/webhooks/{id}") + self.assertTrue( + wait(lambda: cloud.calls("PUT", f"/v1/webhooks/{id}") > renewals) + ) + renewed = Keyring.parse(KEYS).open( + cloud.documents(MAIN)[id]["httpPass"], Context(PUBLIC_URL, MAIN, id) + ) + self.assertGreater(renewed.expires, subscription.expires) + webhook = next(w for w in cloud.webhooks(MAIN) if w.id == id) + + # Aborting the request ends the subscription and deletes its webhook. + self.gone(id) + self.assertEqual( + metrics.wait("mcp.events.streams", streams, event="tablesdb.row.created"), + streams, + ) + # A delivery Appwrite was already sending finds no stream: 200, dropped. + dropped = metrics.count("mcp.events.ingress", reason="no_listener") + late = self.appwrite.deliver(webhook, ROW_EVENT, fixture("row_created")) + self.assertEqual(late.status_code, 200, late.text) + self.assertEqual(late.json(), {"status": "dropped", "reason": "no_listener"}) + self.assertEqual( + metrics.wait("mcp.events.ingress", dropped + 1, reason="no_listener"), + dropped + 1, + ) + + def test_each_stream_is_its_own_subscription(self): + with self.a.client() as client, self.b.client() as other: + with ( + self.opened(client) as (first, one), + self.opened(other) as ( + second, + two, + ), + ): + ids = {one, two} + self.assertEqual(len(ids), 2, ids) + answers = self.appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + self.assertEqual(len(answers), 2) + for stream in (first, second): + self.assertEqual(stream.until(EVENT)["params"]["data"], ROW_DATA) + for id in ids: + self.gone(id) + + def test_invalid_streams_are_refused_before_any_stream_opens(self): + before = cloud.documents(MAIN) + cases = [ + ("unknown event", TOKEN, {**params(), "name": "rows.deleted"}, NOT_FOUND), + ( + "bad arguments", + TOKEN, + params({**ROWS, "table_id": "tickets.*"}), + INVALID_PARAMS, + ), + ("bad cursor", TOKEN, params(cursor=7), INVALID_PARAMS), + ("bad maxAgeMs", TOKEN, params(maxAgeMs=-1), INVALID_PARAMS), + ("missing table", TOKEN, params({**ROWS, "table_id": "orders"}), NOT_FOUND), + ("no user", SUBJECTLESS_TOKEN, params(), FORBIDDEN), + ( + "no access", + OTHER_TOKEN, + params({**ROWS, "project_id": FREE, "table_id": "tickets"}), + FORBIDDEN, + ), + ] + for name, token, request, code in cases: + with self.subTest(case=name), Client(self.a.url, token) as client: + with client.stream(request) as stream: + self.refused(stream, code) + self.assertEqual(cloud.documents(MAIN), before) + + with self.subTest(case="plan limit"), self.a.client() as client: + with client.stream( + params({**ROWS, "project_id": FREE, "table_id": "tickets"}) + ) as stream: + error = self.refused(stream, RESOURCE_EXHAUSTED) + self.assertEqual(error["data"], {"limit": "webhooks", "max": 2}) + + with self.subTest(case="stream cap"), Client(self.a.url, OTHER_TOKEN) as client: + held = [client.stream(params()) for _ in range(PRINCIPAL_STREAMS)] + try: + for stream in held: + stream.__enter__() + self.assertEqual(stream.next()["method"], ACTIVE) + with client.stream(params()) as refused: + error = self.refused(refused, RESOURCE_EXHAUSTED) + self.assertEqual( + error["data"], {"limit": "streams", "max": PRINCIPAL_STREAMS} + ) + finally: + for stream in held: + stream.close() + # The other replica keeps its own count. + with Client(self.b.url, OTHER_TOKEN) as elsewhere: + with elsewhere.stream(params()) as stream: + self.assertEqual(stream.next()["method"], ACTIVE) + + with self.subTest(case="push through events/subscribe"): + with self.a.client() as client: + _, message = client.call( + "events/subscribe", + {**params(), "delivery": {"mode": "push"}}, + ) + self.assertEqual(message["error"]["code"], UNSUPPORTED, message) + self.assertEqual( + message["error"]["data"], {"feature": "deliveryMode", "value": "push"} + ) + + def terminated(self, stream: Stream, code: int) -> dict[str, Any]: + """``terminated``, then the final result, then the end of the stream.""" + message = stream.until(TERMINATED) + self.assertEqual(message["params"]["_meta"], {SUBSCRIPTION_ID: stream.id}) + error = message["params"]["error"] + self.assertEqual(error["code"], code, message) + final = stream.next() + self.assertEqual(final["id"], stream.id, final) + self.assertIn("result", final) + self.assertTrue(stream.ended()) + return error + + def test_revoked_access_terminates_the_stream(self): + cloud.grant(REVOKED, {MAIN}) + with Client(self.a.url, REVOKED) as client, self.opened(client) as (stream, id): + cloud.grant(REVOKED, set()) + self.terminated(stream, FORBIDDEN) + # Its webhook cannot be deleted without access; its lease runs out. + self.assertIn(id, cloud.documents(MAIN)) + cloud.remove(MAIN, id) + + def test_deleted_webhook_terminates_the_stream(self): + with self.a.client() as client, self.opened(client) as (stream, id): + cloud.remove(MAIN, id) + error = self.terminated(stream, NOT_FOUND) + self.assertEqual(error["data"], {"kind": "subscription"}) + + def test_failed_renewal_is_reported_and_retried(self): + with self.a.client() as client, self.opened(client) as (stream, id): + cloud.fail("PUT", f"/v1/webhooks/{id}") + error = stream.until(ERROR, timeout=SETTINGS.renew * 4) + self.assertEqual(error["params"]["error"]["code"], INTERNAL_ERROR, error) + renewals = cloud.calls("PUT", f"/v1/webhooks/{id}") + self.assertTrue( + wait(lambda: cloud.calls("PUT", f"/v1/webhooks/{id}") > renewals) + ) + self.assertEqual(stream.until(HEARTBEAT)["method"], HEARTBEAT) + [answer] = self.appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + self.assertEqual(answer.json()["status"], "accepted", answer.text) + self.assertEqual(stream.until(EVENT)["params"]["data"], ROW_DATA) + self.gone(id) + + def test_bus_outage_is_reported_then_marked_as_a_gap(self): + bus = redis() + with self.a.client() as client, self.opened(client) as (stream, id): + bus.stop() + try: + error = stream.until(ERROR) + self.assertEqual(error["params"]["error"]["code"], INTERNAL_ERROR) + # Appwrite's delivery meanwhile is acknowledged and dropped. + [answer] = self.appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + self.assertEqual( + answer.json(), {"status": "dropped", "reason": "bus_error"} + ) + finally: + bus.start() + active = stream.until(ACTIVE) + self.assertEqual( + active["params"], + { + "cursor": None, + "truncated": True, + "_meta": {SUBSCRIPTION_ID: stream.id}, + }, + ) + [answer] = self.appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + self.assertEqual(answer.json()["status"], "accepted", answer.text) + self.assertEqual(stream.until(EVENT)["params"]["data"], ROW_DATA) + self.gone(id) + + def test_stream_ends_cleanly_before_its_token_expires(self): + EXPIRES[EXPIRING] = int(time.time()) + 3 + try: + with ( + Client(self.a.url, EXPIRING) as client, + self.opened(client) as ( + stream, + id, + ), + ): + # No terminated: the client is to reconnect with a new token, + # not drop the subscription. + final = stream.next(6.0) + while final.get("method") == HEARTBEAT: + final = stream.next(6.0) + self.assertEqual(final["id"], stream.id, final) + self.assertIn("result", final) + self.assertTrue(stream.ended()) + self.gone(id) + finally: + EXPIRES.pop(EXPIRING, None) + + def test_hung_renewal_is_reported_and_retried(self): + table = "/v1/tablesdb/main/tables/support_tickets" + with self.a.client() as client, self.opened(client) as (stream, id): + cloud.before("GET", table, lambda: time.sleep(SETTINGS.appwrite * 3)) + error = stream.until(ERROR, timeout=SETTINGS.renew * 4) + self.assertEqual(error["params"]["error"]["code"], INTERNAL_ERROR, error) + renewals = cloud.calls("PUT", f"/v1/webhooks/{id}") + self.assertTrue( + wait(lambda: cloud.calls("PUT", f"/v1/webhooks/{id}") > renewals) + ) + self.assertEqual(stream.until(HEARTBEAT)["method"], HEARTBEAT) + self.gone(id) + + def test_frozen_bus_is_noticed(self): + bus = redis() + with self.a.client() as client, self.opened(client) as (stream, id): + bus.pause() + try: + error = stream.until(ERROR, timeout=5.0) + self.assertEqual(error["params"]["error"]["code"], INTERNAL_ERROR) + finally: + bus.resume() + active = stream.until(ACTIVE, timeout=10.0) + self.assertTrue(active["params"]["truncated"]) + [answer] = self.appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + self.assertEqual(answer.json()["status"], "accepted", answer.text) + self.assertEqual(stream.until(EVENT)["params"]["data"], ROW_DATA) + self.gone(id) + + def test_abort_while_the_webhook_is_created_leaves_none(self): + before = set(self.ours()) + creates = cloud.calls("POST", "/v1/webhooks") + cloud.before("POST", "/v1/webhooks", lambda: time.sleep(1.0)) + with self.a.client() as client: + stream = client.stream(params()) + time.sleep(0.5) + stream.close() + self.assertTrue(wait(lambda: cloud.calls("POST", "/v1/webhooks") > creates)) + self.assertTrue(wait(lambda: set(self.ours()) == before), self.ours()) + + +class LifecycleFlow(unittest.TestCase): + """How a stream's webhook goes away when its server stops or dies.""" + + def ours(self) -> set[str]: + return { + id + for id, document in cloud.documents(MAIN).items() + if LABEL.fullmatch(document["name"]) + } + + def test_shutdown_closes_streams_and_deletes_their_webhooks(self): + before = self.ours() + instance = server().__enter__() + try: + client = instance.client() + stream = client.stream(params()).__enter__() + self.assertEqual(stream.next()["method"], ACTIVE) + [id] = self.ours() - before + except BaseException: + instance.__exit__(None, None, None) + raise + stopping = threading.Thread(target=instance.__exit__, args=(None, None, None)) + stopping.start() + try: + final = stream.next() + while "method" in final: + final = stream.next() + self.assertEqual(final["id"], stream.id, final) + self.assertIn("result", final) + self.assertTrue(stream.ended()) + finally: + stopping.join(30.0) + stream.close() + client.__exit__(None, None, None) + self.assertFalse(stopping.is_alive(), "the server did not stop") + self.assertNotIn(id, cloud.documents(MAIN)) + + def test_a_crashed_stream_leaves_a_lease_that_runs_out(self): + crashing = Settings(heartbeat=5.0, renew=60.0, retry=5.0, lease=1.0) + before = self.ours() + with server(crashing) as instance: + appwrite = Appwrite(instance) + try: + # The replica dies before it can delete the webhook. + cloud.fail("DELETE", r"/v1/webhooks/sub_[0-9a-f]{32}") + with instance.client() as client, client.stream(params()) as stream: + self.assertEqual(stream.next()["method"], ACTIVE) + [id] = self.ours() - before + self.assertTrue( + wait(lambda: cloud.calls("DELETE", f"/v1/webhooks/{id}") == 1) + ) + self.assertIn(id, cloud.documents(MAIN)) + + # Past its lease, Appwrite's deliveries are dropped as expired. + time.sleep(crashing.lease + 1.1) + webhook = next(w for w in cloud.webhooks(MAIN) if w.id == id) + answer = appwrite.deliver(webhook, ROW_EVENT, fixture("row_created")) + self.assertEqual( + answer.json(), {"status": "dropped", "reason": "expired"} + ) + + # A minute past its lease (not the ten minutes a webhook + # subscription gets), the owner's next stream removes it. + document = cloud.documents(MAIN)[id] + stamp = time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 65) + ) + cloud.store( + MAIN, + { + **document, + "$id": id, + "name": re.sub( + r"until \S+", f"until {stamp}", document["name"] + ), + }, + ) + with instance.client() as client, client.stream(params()) as stream: + self.assertEqual(stream.next()["method"], ACTIVE) + self.assertNotIn(id, cloud.documents(MAIN)) + finally: + appwrite.close() + + +class PushOffFlow(unittest.TestCase): + """Without push, nothing about it is advertised or served.""" + + def test_push_needs_its_flag_and_a_bus(self): + for name, extra in ( + ("flag off", {"MCP_EVENTS_PUSH": None}), + ("no bus", {"MCP_EVENTS_REDIS_URL": None}), + ): + with self.subTest(case=name), Server(environment(**extra)) as instance: + with instance.client() as client: + _, listing = client.call("events/list") + for event in listing["result"]["events"]: + self.assertEqual(event["delivery"], ["webhook"], event) + with client.stream(params()) as stream: + message = stream.next() + self.assertEqual(message["error"]["code"], METHOD_NOT_FOUND) + + +if __name__ == "__main__": + unittest.main() diff --git a/uv.lock b/uv.lock index 4bca6b7..06a5aa9 100644 --- a/uv.lock +++ b/uv.lock @@ -680,15 +680,15 @@ cli = [ [[package]] name = "mcp-event-subscriptions" -version = "0.1.0a1" +version = "0.1.0a2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "jsonschema" }, { name = "jwcrypto" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/db/2e/f6af884063973af62b006525f9accfd889b423af7548c4fc169ed27505a6/mcp_event_subscriptions-0.1.0a1.tar.gz", hash = "sha256:2c34ea0af0605d27ca23a842092275b57b482ab2d47288e10d3bfcd88fbef19d", size = 75263, upload-time = "2026-10-11T20:24:35.521Z" } +sdist = { url = "https://files.pythonhosted.org/packages/06/d1/99ee4c8d4548ba8a6ff36da64ed95cc5517f8b0536064df03546e3457359/mcp_event_subscriptions-0.1.0a2.tar.gz", hash = "sha256:5d5a5698bd333752a2afdbad720b6d65a239dd094b9578e850a0cfb42e7b74f8", size = 76019, upload-time = "2026-10-12T20:25:25.76Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b5/13/b3c37eb259cca28a19bcd374a54f3ce8ce903a6f8a61df23c324318e36f1/mcp_event_subscriptions-0.1.0a1-py3-none-any.whl", hash = "sha256:d0bf02ca46e462b4b9d2deaf790ba5828254edbc36033cf509a64e47da31330b", size = 7437, upload-time = "2026-10-11T20:24:34.194Z" }, + { url = "https://files.pythonhosted.org/packages/ce/c5/703027fb759e9b1bac49a20c303830ef98949ec01db5f75840e2b4da4537/mcp_event_subscriptions-0.1.0a2-py3-none-any.whl", hash = "sha256:abe6d58ea312d3c93579584d795dd06a6b541376e592d7a869bd035acaabc7fb", size = 7674, upload-time = "2026-10-12T20:25:24.49Z" }, ] [[package]] @@ -710,6 +710,7 @@ dependencies = [ { name = "opentelemetry-sdk" }, { name = "pyjwt", extra = ["crypto"] }, { name = "python-dotenv" }, + { name = "redis" }, { name = "sentry-sdk", extra = ["starlette"] }, { name = "starlette" }, { name = "uvicorn", extra = ["standard"] }, @@ -745,7 +746,7 @@ requires-dist = [ { name = "httpcore", specifier = ">=1.0" }, { name = "httpx", specifier = ">=0.27.0" }, { name = "mcp", extras = ["cli"], specifier = ">=2.3,<3" }, - { name = "mcp-event-subscriptions", specifier = ">=0.1.0a1,<0.2" }, + { name = "mcp-event-subscriptions", specifier = ">=0.1.0a2,<0.2" }, { name = "numpy", specifier = ">=2.0" }, { name = "openai", specifier = ">=1.40" }, { name = "opentelemetry-api", specifier = ">=1.28" }, @@ -755,6 +756,7 @@ requires-dist = [ { name = "pycryptodome", marker = "extra == 'integration'", specifier = ">=3.20.0" }, { name = "pyjwt", extras = ["crypto"], specifier = ">=2.9.0" }, { name = "python-dotenv", specifier = ">=1.0.1" }, + { name = "redis", specifier = ">=6.0,<9" }, { name = "sentry-sdk", extras = ["starlette"], specifier = ">=2.0" }, { name = "starlette", specifier = ">=0.40.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.30.0" }, @@ -1291,6 +1293,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f1/12/de94a39c2ef588c7e6455cfbe7343d3b2dc9d6b6b2f40c4c6565744c873d/pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b", size = 149341, upload-time = "2025-09-25T21:32:56.828Z" }, ] +[[package]] +name = "redis" +version = "8.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a8/99/604f0b666d4c616d891cf77ebb9db6bb21601344c051aebf1b72b9ff915f/redis-8.1.0.tar.gz", hash = "sha256:6e1a19beef9225c83efd689c7e6b7da2d5215b1f42cd13b7fc3714d0a09c7b25", size = 5254356, upload-time = "2026-07-30T08:51:00.269Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/66/9d/c5731f6e3608663d4d3656fd8d3aecee8b509c3082818f5a13eae925baea/redis-8.1.0-py3-none-any.whl", hash = "sha256:a4fe1aac3d3b3cc791d4b3d5931c5a956045dc951ee74d1c913ee3ac4d2ee9fb", size = 560618, upload-time = "2026-07-30T08:50:58.497Z" }, +] + [[package]] name = "referencing" version = "0.37.0"