feat: serve events/subscribe and events/unsubscribe

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`), `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}`). 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. |
 

README.md

@@ -321,7 +321,7 @@ for per-client setup.
 - [Documentation search](docs/documentation-search.md) — the in-process
   `appwrite_search_docs` tool and how to rebuild its index.
 - [Events](docs/events.md) — MCP Events: webhook subscriptions to changes in
-  your Appwrite project (in progress, behind a flag).
+  your Appwrite project (in development, not available yet).
 - [Self-hosted Appwrite](docs/self-hosted.md) — run the server locally with a
   project API key.
 - [Local development](docs/development.md) — running, testing, and debugging the

docs/development.md

@@ -41,8 +41,9 @@ docker compose up --build
 > [`mcp-subscription+jwe`](https://github.com/ChiragAgg5k/mcp-event-subscriptions/blob/main/profiles/encrypted-envelope.md)
 > envelope, with the keys in `MCP_EVENTS_SEALING_KEYS` (`<id>:<base64 of 32
 > random bytes>`, comma separated, first one active). Generate a key with
-> `openssl rand -base64 32`. To rotate, put a new key first and remove the old
-> one once the longest subscription TTL (24h) has passed.
+> `openssl rand -base64 32`. To rotate, add the new key last and roll that out,
+> then move it first, then remove the old one 25 hours after that reached every
+> replica (see "Key rotation" in [events.md](events.md)).
 
 **`uv` directly — HTTP:**
 

docs/flags.md

@@ -85,11 +85,14 @@ token exchange against the real authorization server.
 ### `--events` — MCP Events (in progress)
 
 Turns on [MCP Events](events.md): `server/discover` (and legacy `initialize`)
-advertise the events capability, the server answers `events/list` with the
-event catalog, and the Appwrite webhook ingress is mounted at
-`/appwrite/webhooks/{id}`. It is off by default while the feature is built
-([#127](https://github.com/appwrite/mcp/issues/127)); `events/subscribe` is not
-available yet. Any of `1`, `true`, `yes` or `on` enables it. With the flag on,
+advertise the events capability, the server answers `events/list`,
+`events/subscribe` and `events/unsubscribe`, and the Appwrite webhook ingress
+is mounted at `/appwrite/webhooks/{id}`. It is a testing flag, off by default
+and in production, while the feature is built
+([#127](https://github.com/appwrite/mcp/issues/127)): subscriptions cannot work
+on Appwrite Cloud until
+[appwrite/appwrite#14293](https://github.com/appwrite/appwrite/issues/14293)
+ships. Any of `1`, `true`, `yes` or `on` enables it. With the flag on,
 `MCP_EVENTS_SEALING_KEYS` is required and the server refuses to start without
 it.
 
@@ -128,5 +131,6 @@ curl -s -X POST http://localhost:8000/appwrite/webhooks/sub_0123456789abcdef0123
 # {"status":"rejected","reason":"credentials"}
 ```
 
-With the flag off, neither capability appears, `events/list` returns `-32601`
-(method not found) and the ingress route returns `404`.
+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`.

src/mcp_server_appwrite/events/catalog.py

@@ -14,6 +14,9 @@
   with ``appwrite_call_tool``.
 * ``status`` — the payload status filter. Appwrite webhooks filter only by event
   name, so the ingress applies this one before delivering.
+* ``access`` — the Appwrite read that proves a caller may watch the event.
+  ``events/subscribe`` makes it with the caller's own token on every subscribe
+  and refresh, so a refresh doubles as a periodic access check.
 
 The JSON Schemas served by ``events/list`` are derived from those fields, and
 :meth:`Event.validate` enforces exactly what the input schema promises.
@@ -28,6 +31,8 @@
 from types import MappingProxyType
 from typing import Any
 
+from appwrite_console.query import Query
+
 from .errors import EventsError
 
 ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9_-]{0,35}$"
@@ -139,6 +144,27 @@ def matches(self, status: object, arguments: Mapping[str, Any]) -> bool:
         return wanted is None or wanted == status
 
 
+@dataclass(frozen=True)
+class Access:
+    """The least-privilege Appwrite read that proves access to an event's
+    resource: ``GET {path}`` in the subscribed project, which needs ``scope``.
+
+    ``path`` and ``resource`` (how errors name what was read) have
+    ``{argument}`` placeholders like the pattern templates; ``params`` go in
+    the query string."""
+
+    path: str
+    scope: str
+    resource: str
+    params: Mapping[str, Any] = MappingProxyType({})
+
+    def resolve(self, arguments: Mapping[str, str]) -> str:
+        return self.path.format(**arguments)
+
+    def describe(self, arguments: Mapping[str, str]) -> str:
+        return self.resource.format(**arguments)
+
+
 PROJECT = Argument("project_id", "Appwrite project ID to watch.")
 """Required by every event; it is never part of the Appwrite pattern because
 project webhooks are already scoped to one project."""
@@ -152,6 +178,7 @@ class Event:
     description: str
     templates: tuple[str, ...]
     payload: tuple[PayloadField, ...]
+    access: Access
     arguments: tuple[Argument, ...] = ()
     status: StatusFilter | None = None
 
@@ -221,6 +248,10 @@ def _timestamp(name: str, description: str) -> PayloadField:
     choices=(Status.READY, Status.FAILED),
 )
 
+_FUNCTION = Access(
+    "/functions/{function_id}", "functions.read", "function {function_id}"
+)
+
 _DEPLOYMENT_FILTER = StatusFilter(
     accepted=(Status.READY, Status.FAILED), argument=_DEPLOYMENT_STATUS.name
 )
@@ -277,6 +308,7 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]:
             ),
             _timestamp("created_at", "When the execution started."),
         ),
+        access=_FUNCTION,
         status=StatusFilter(accepted=(Status.FAILED,)),
     ),
     Event(
@@ -288,6 +320,7 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]:
         ),
         templates=("functions.{function_id}.deployments.*.update",),
         payload=_deployment_payload("function_id", "function"),
+        access=_FUNCTION,
         status=_DEPLOYMENT_FILTER,
     ),
     Event(
@@ -299,6 +332,7 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]:
         ),
         templates=("sites.{site_id}.deployments.*.update",),
         payload=_deployment_payload("site_id", "site"),
+        access=Access("/sites/{site_id}", "sites.read", "site {site_id}"),
         status=_DEPLOYMENT_FILTER,
     ),
     Event(
@@ -315,6 +349,11 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]:
             PayloadField("row_id", "string", "ID of the new row."),
             _timestamp("created_at", "When the row was created."),
         ),
+        access=Access(
+            "/tablesdb/{database_id}/tables/{table_id}",
+            "tables.read",
+            "table {table_id} in database {database_id}",
+        ),
     ),
     Event(
         name="storage.file.created",
@@ -332,6 +371,9 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]:
             ),
             _timestamp("created_at", "When the file was uploaded."),
         ),
+        access=Access(
+            "/storage/buckets/{bucket_id}", "buckets.read", "bucket {bucket_id}"
+        ),
     ),
     Event(
         name="users.user.created",
@@ -341,6 +383,14 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]:
             PayloadField("user_id", "string", "ID of the new user."),
             _timestamp("created_at", "When the user was created."),
         ),
+        # Listing one user is the narrowest read that proves the caller may
+        # see users; the agent needs users.read to act on the event anyway.
+        access=Access(
+            "/users",
+            "users.read",
+            "users",
+            MappingProxyType({"queries": [Query.limit(1)], "total": False}),
+        ),
     ),
 )
 """The v1 catalog, in the order ``events/list`` serves it."""

src/mcp_server_appwrite/events/envelope.py

@@ -28,8 +28,10 @@
 
 Sealing keys come from ``MCP_EVENTS_SEALING_KEYS`` (``<id>:<base64 32 bytes>``,
 comma separated, first one active). Every key in the ring opens envelopes, so a
-rotation adds the new key in front and drops the old one once every subscription
-has refreshed (the TTL cap bounds how long that takes).
+rotation first adds the new key behind the active one on every replica, then
+moves it in front, and drops the old one once every subscription has refreshed
+(the TTL cap bounds how long that takes). Sealing with a key some replica does
+not hold yet would make that replica drop the deliveries as ``retired_key``.
 """
 
 from __future__ import annotations

src/mcp_server_appwrite/events/errors.py

@@ -27,13 +27,15 @@
 
 from mcp import MCPError
 
-# Malformed arguments, callback URL or secret: the JSON-RPC code, unchanged by
-# SEP-3415. Re-exported so callers take every events code from this module.
-from mcp.types import INVALID_PARAMS
+# Malformed arguments, callback URL or secret, and failures on our side (such
+# as Appwrite answering 5xx): the JSON-RPC codes, unchanged by SEP-3415.
+# Re-exported so callers take every events code from this module.
+from mcp.types import INTERNAL_ERROR, INVALID_PARAMS
 
 __all__ = [
     "CALLBACK_ENDPOINT",
     "FORBIDDEN",
+    "INTERNAL_ERROR",
     "INVALID_PARAMS",
     "NOT_FOUND",
     "RESOURCE_EXHAUSTED",
@@ -101,3 +103,14 @@ def unsupported(cls, message: str) -> EventsError:
     @classmethod
     def callback_endpoint(cls, message: str, *, reason: CallbackFailure) -> EventsError:
         return cls(CALLBACK_ENDPOINT, message, {"reason": reason.value})
+
+    @classmethod
+    def internal(cls, message: str) -> EventsError:
+        return cls(INTERNAL_ERROR, message)
+
+    @property
+    def expected(self) -> bool:
+        """Whether the caller caused this error (bad input, no access, a quota,
+        an unreachable callback) rather than a failure on our side. Only
+        unexpected errors are reported to Sentry."""
+        return self.code != INTERNAL_ERROR

src/mcp_server_appwrite/events/ingress.py

@@ -259,6 +259,23 @@ def from_env(cls) -> Ingress:
         keys are missing, so a misconfigured server fails at startup."""
         return cls(Keyring.from_env(), Egress(), base=public_base_url())
 
+    @property
+    def keyring(self) -> Keyring:
+        return self._keyring
+
+    @property
+    def egress(self) -> Egress:
+        return self._egress
+
+    @property
+    def dispatcher(self) -> Dispatcher:
+        return self._dispatcher
+
+    @property
+    def server(self) -> str:
+        """This deployment as envelopes are bound to it: ``MCP_PUBLIC_URL``."""
+        return self._base
+
     @property
     def route(self) -> Route:
         return Route(PATH, endpoint=self.handle, methods=["POST"])