From 2936252132a419f70ee656aef5eac7011bca902a Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 13:08:36 +0100 Subject: [PATCH 1/5] feat: add the MCP events catalog and error codes Six v1 events, each declared as data: arguments, Appwrite pattern templates, an IDs-only payload schema and the status filter the ingress will apply. Arguments are checked against a dot- and wildcard-free Appwrite ID pattern before they are spliced into an Appwrite event pattern, so a subscription can never widen past the named resource. Error codes sit behind constants because SEP-3415 renumbers them. The MCP_EVENTS flag gates the feature (off by default). --- src/mcp_server_appwrite/events/__init__.py | 1 + src/mcp_server_appwrite/events/catalog.py | 358 +++++++++++++++++++++ src/mcp_server_appwrite/events/errors.py | 103 ++++++ src/mcp_server_appwrite/flags.py | 20 +- tests/unit/test_events_catalog.py | 266 +++++++++++++++ 5 files changed, 747 insertions(+), 1 deletion(-) create mode 100644 src/mcp_server_appwrite/events/__init__.py create mode 100644 src/mcp_server_appwrite/events/catalog.py create mode 100644 src/mcp_server_appwrite/events/errors.py create mode 100644 tests/unit/test_events_catalog.py diff --git a/src/mcp_server_appwrite/events/__init__.py b/src/mcp_server_appwrite/events/__init__.py new file mode 100644 index 0000000..c4a06d7 --- /dev/null +++ b/src/mcp_server_appwrite/events/__init__.py @@ -0,0 +1 @@ +"""MCP Events: webhook subscriptions backed by Appwrite project webhooks.""" diff --git a/src/mcp_server_appwrite/events/catalog.py b/src/mcp_server_appwrite/events/catalog.py new file mode 100644 index 0000000..e3ab9b7 --- /dev/null +++ b/src/mcp_server_appwrite/events/catalog.py @@ -0,0 +1,358 @@ +"""Declarative catalog of the MCP events this server can deliver. + +Each :class:`Event` maps one MCP event name onto the Appwrite project-webhook +events that feed it. Everything about an event is data: + +* ``arguments`` — what a subscriber passes besides the always-required + ``project_id``. Every value is checked against :data:`ID_PATTERN` (or a fixed + set of choices) before it is spliced into an Appwrite event pattern, so a + ``*`` or a ``.`` can never widen a subscription to resources the caller did + not name. +* ``templates`` — the Appwrite event patterns, with ``{argument}`` placeholders. +* ``payload`` — the fields delivered to the subscriber. IDs and metadata only: + Appwrite sends whole rows and users, and the agent fetches details itself + with ``appwrite_call_tool``. +* ``status`` — the payload status filter. Appwrite webhooks filter only by event + name, so the ingress applies this one before delivering. + +The JSON Schemas served by ``events/list`` are derived from those fields, and +:meth:`Event.validate` enforces exactly what the input schema promises. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from enum import StrEnum +from types import MappingProxyType +from typing import Any + +from .errors import EventsError + +ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9_-]{0,35}$" +"""Appwrite IDs as accepted in a subscription: up to 36 characters, starting +with a letter or digit. Stricter than Appwrite itself, which also allows ``.``: +a dot is the separator in Appwrite event names, so it is rejected along with +``*`` to keep every pattern pinned to the named resource.""" + +_ID = re.compile(ID_PATTERN) + +NOT_FOUND_KIND = "event" +"""``data.kind`` of the NotFound error for an unknown event name.""" + + +class Status(StrEnum): + """Terminal Appwrite statuses an event can be filtered on.""" + + READY = "ready" + FAILED = "failed" + + +@dataclass(frozen=True) +class Argument: + """One subscription argument. + + Without ``choices`` the value is an Appwrite ID and must match + :data:`ID_PATTERN`; with ``choices`` it must be one of them. + """ + + name: str + description: str + required: bool = True + choices: tuple[str, ...] = () + + def schema(self) -> dict[str, Any]: + if self.choices: + return { + "type": "string", + "enum": [str(choice) for choice in self.choices], + "description": self.description, + } + return { + "type": "string", + "pattern": ID_PATTERN, + "description": self.description, + } + + def check(self, value: object) -> str: + """The value, if it is valid for this argument.""" + if isinstance(value, str): + if self.choices and value in self.choices: + return value + if not self.choices and _ID.fullmatch(value): + return value + if self.choices: + expected = " or ".join(self.choices) + raise EventsError.invalid_params(f"{self.name} must be {expected}") + raise EventsError.invalid_params( + f"{self.name} must be an Appwrite ID: 1-36 letters, digits, " + "'_' or '-', starting with a letter or digit" + ) + + +@dataclass(frozen=True) +class PayloadField: + """One field of a delivered event's ``data``.""" + + name: str + type: str + description: str + nullable: bool = False + format: str | None = None + choices: tuple[str, ...] = () + + def schema(self) -> dict[str, Any]: + schema: dict[str, Any] = { + "type": [self.type, "null"] if self.nullable else self.type, + "description": self.description, + } + if self.format is not None: + schema["format"] = self.format + if self.choices: + choices: list[str | None] = [str(choice) for choice in self.choices] + schema["enum"] = [*choices, None] if self.nullable else choices + return schema + + +@dataclass(frozen=True) +class StatusFilter: + """Which Appwrite payload statuses qualify an event for delivery. + + ``field`` is the status field in the Appwrite webhook body, ``accepted`` + the statuses that ever qualify, and ``argument`` the optional subscription + argument that narrows delivery to one of them. + """ + + accepted: tuple[Status, ...] + field: str = "status" + argument: str | None = None + + def matches(self, status: object, arguments: Mapping[str, Any]) -> bool: + """Whether an Appwrite payload with ``status`` should be delivered to a + subscription made with ``arguments``.""" + if status not in self.accepted: + return False + if self.argument is None: + return True + wanted = arguments.get(self.argument) + return wanted is None or wanted == status + + +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.""" + + +@dataclass(frozen=True) +class Event: + """One deliverable MCP event and the Appwrite events behind it.""" + + name: str + description: str + templates: tuple[str, ...] + payload: tuple[PayloadField, ...] + arguments: tuple[Argument, ...] = () + status: StatusFilter | None = None + + @property + def parameters(self) -> tuple[Argument, ...]: + """Every argument the event accepts, ``project_id`` first.""" + return (PROJECT, *self.arguments) + + @property + def input_schema(self) -> dict[str, Any]: + return { + "type": "object", + "properties": { + argument.name: argument.schema() for argument in self.parameters + }, + "required": [ + argument.name for argument in self.parameters if argument.required + ], + "additionalProperties": False, + } + + @property + def payload_schema(self) -> dict[str, Any]: + return { + "type": "object", + "properties": {field.name: field.schema() for field in self.payload}, + "required": [field.name for field in self.payload], + "additionalProperties": False, + } + + def validate(self, arguments: object) -> dict[str, str]: + """Check subscription arguments against :attr:`input_schema`. + + Returns the arguments in declaration order; raises an invalid-params + :class:`EventsError` on anything the schema would reject.""" + if not isinstance(arguments, Mapping): + raise EventsError.invalid_params("arguments must be an object") + known = {argument.name for argument in self.parameters} + unknown = sorted(str(key) for key in arguments if key not in known) + if unknown: + raise EventsError.invalid_params( + f"Unknown arguments for {self.name}: {', '.join(unknown)}" + ) + validated: dict[str, str] = {} + for argument in self.parameters: + if argument.name not in arguments: + if argument.required: + raise EventsError.invalid_params(f"{argument.name} is required") + continue + validated[argument.name] = argument.check(arguments[argument.name]) + return validated + + def patterns(self, arguments: object) -> list[str]: + """The Appwrite webhook event patterns for a subscription.""" + validated = self.validate(arguments) + return [template.format(**validated) for template in self.templates] + + +def _timestamp(name: str, description: str) -> PayloadField: + return PayloadField(name, "string", description, nullable=True, format="date-time") + + +_DEPLOYMENT_STATUS = Argument( + "status", + "Only deliver deployments that end in this status. Omit for both.", + required=False, + choices=(Status.READY, Status.FAILED), +) + +_DEPLOYMENT_FILTER = StatusFilter( + accepted=(Status.READY, Status.FAILED), argument=_DEPLOYMENT_STATUS.name +) + + +def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: + return ( + PayloadField(resource, "string", f"ID of the {kind}."), + PayloadField("deployment_id", "string", "ID of the deployment."), + PayloadField( + "status", + "string", + "Final deployment status.", + choices=(Status.READY, Status.FAILED), + ), + _timestamp("updated_at", "When the deployment reached its final status."), + ) + + +EVENTS: tuple[Event, ...] = ( + Event( + name="functions.execution.failed", + description=( + "A function execution finished with status failed: synchronous, " + "asynchronous, scheduled or event-triggered runs. Executions served " + "through function domains are not covered." + ), + arguments=(Argument("function_id", "ID of the function to watch."),), + # Synchronous runs fire only `.create`, after they finish; async, + # scheduled and event-triggered runs fire `.create` while waiting and + # `.update` when they finish. The status filter drops everything but + # failures. + templates=( + "functions.{function_id}.executions.*.update", + "functions.{function_id}.executions.*.create", + ), + payload=( + PayloadField("function_id", "string", "ID of the function."), + PayloadField("execution_id", "string", "ID of the execution."), + PayloadField( + "status", "string", "Execution status.", choices=(Status.FAILED,) + ), + PayloadField( + "trigger", + "string", + "What started the execution: http, schedule or event.", + nullable=True, + ), + PayloadField( + "response_status_code", + "integer", + "HTTP status code the function responded with.", + nullable=True, + ), + _timestamp("created_at", "When the execution started."), + ), + status=StatusFilter(accepted=(Status.FAILED,)), + ), + Event( + name="functions.deployment.completed", + description="A function deployment finished building, as ready or failed.", + arguments=( + Argument("function_id", "ID of the function to watch."), + _DEPLOYMENT_STATUS, + ), + templates=("functions.{function_id}.deployments.*.update",), + payload=_deployment_payload("function_id", "function"), + status=_DEPLOYMENT_FILTER, + ), + Event( + name="sites.deployment.completed", + description="A site deployment finished building, as ready or failed.", + arguments=( + Argument("site_id", "ID of the site to watch."), + _DEPLOYMENT_STATUS, + ), + templates=("sites.{site_id}.deployments.*.update",), + payload=_deployment_payload("site_id", "site"), + status=_DEPLOYMENT_FILTER, + ), + Event( + name="tablesdb.row.created", + description="A row was created in a TablesDB table.", + arguments=( + Argument("database_id", "ID of the database."), + Argument("table_id", "ID of the table to watch."), + ), + templates=("tablesdb.{database_id}.tables.{table_id}.rows.*.create",), + payload=( + PayloadField("database_id", "string", "ID of the database."), + PayloadField("table_id", "string", "ID of the table."), + PayloadField("row_id", "string", "ID of the new row."), + _timestamp("created_at", "When the row was created."), + ), + ), + Event( + name="storage.file.created", + description="A file was uploaded to a storage bucket.", + arguments=(Argument("bucket_id", "ID of the bucket to watch."),), + templates=("buckets.{bucket_id}.files.*.create",), + payload=( + PayloadField("bucket_id", "string", "ID of the bucket."), + PayloadField("file_id", "string", "ID of the new file."), + PayloadField( + "mime_type", "string", "MIME type of the file.", nullable=True + ), + PayloadField( + "size", "integer", "Original file size in bytes.", nullable=True + ), + _timestamp("created_at", "When the file was uploaded."), + ), + ), + Event( + name="users.user.created", + description="A user signed up or was created in the project.", + templates=("users.*.create",), + payload=( + PayloadField("user_id", "string", "ID of the new user."), + _timestamp("created_at", "When the user was created."), + ), + ), +) +"""The v1 catalog, in the order ``events/list`` serves it.""" + +CATALOG: Mapping[str, Event] = MappingProxyType({event.name: event for event in EVENTS}) +"""Read-only index of :data:`EVENTS` by name.""" + + +def lookup(name: object) -> Event: + """The event called ``name``; raises a NotFound :class:`EventsError` + (``data.kind = "event"``) when there is none.""" + event = CATALOG.get(name) if isinstance(name, str) else None + if event is None: + raise EventsError.not_found(f"Unknown event {name!r}", kind=NOT_FOUND_KIND) + return event diff --git a/src/mcp_server_appwrite/events/errors.py b/src/mcp_server_appwrite/events/errors.py new file mode 100644 index 0000000..720d19c --- /dev/null +++ b/src/mcp_server_appwrite/events/errors.py @@ -0,0 +1,103 @@ +"""JSON-RPC errors for the MCP Events methods. + +The events specification is still a draft and its error codes are moving: the +values below are the ones ChatGPT and the working-group design sketch use today. +SEP-3415 renumbers them, so every code lives behind a constant and switching is +a one-line edit per code: + +===================== =================== ================== +Meaning ChatGPT / sketch SEP-3415 (draft) +===================== =================== ================== +Bad arguments -32602 -32602 +NotFound -32011 -32023 +Forbidden -32012 -32024 +ResourceExhausted -32013 -32025 +Unsupported -32014 -32026 +CallbackEndpointError -32015 -32027 +===================== =================== ================== + +Handlers raise :class:`EventsError`. It is an ``MCPError``, so the SDK's +dispatcher turns it into the JSON-RPC ``error`` member (code, message, data) +without any extra plumbing. +""" + +from __future__ import annotations + +from enum import StrEnum + +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 + +__all__ = [ + "CALLBACK_ENDPOINT", + "FORBIDDEN", + "INVALID_PARAMS", + "NOT_FOUND", + "RESOURCE_EXHAUSTED", + "UNSUPPORTED", + "CallbackFailure", + "EventsError", +] + +NOT_FOUND = -32011 +"""Unknown event name or target resource; ``data.kind`` names which. SEP-3415: -32023.""" + +FORBIDDEN = -32012 +"""The caller may not watch the target resource. SEP-3415: -32024.""" + +RESOURCE_EXHAUSTED = -32013 +"""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.""" + +CALLBACK_ENDPOINT = -32015 +"""The callback URL failed verification; ``data.reason`` says how. SEP-3415: -32027.""" + + +class CallbackFailure(StrEnum): + """``data.reason`` values for :data:`CALLBACK_ENDPOINT` errors.""" + + CHALLENGE_FAILED = "challenge_failed" + TIMEOUT = "timeout" + CONNECTION_REFUSED = "connection_refused" + TLS_ERROR = "tls_error" + HTTP_4XX = "http_4xx" + HTTP_5XX = "http_5xx" + + +class EventsError(MCPError): + """A JSON-RPC error raised by an events method. + + Prefer the named constructors: they keep each code paired with the + ``data`` shape the specification expects for it. + """ + + @classmethod + def invalid_params(cls, message: str) -> EventsError: + return cls(INVALID_PARAMS, message) + + @classmethod + def not_found(cls, message: str, *, kind: str) -> EventsError: + return cls(NOT_FOUND, message, {"kind": kind}) + + @classmethod + def forbidden(cls, message: str) -> EventsError: + return cls(FORBIDDEN, message) + + @classmethod + def resource_exhausted( + cls, message: str, *, limit: str, maximum: int + ) -> EventsError: + return cls(RESOURCE_EXHAUSTED, message, {"limit": limit, "max": maximum}) + + @classmethod + def unsupported(cls, message: str) -> EventsError: + return cls(UNSUPPORTED, message) + + @classmethod + def callback_endpoint(cls, message: str, *, reason: CallbackFailure) -> EventsError: + return cls(CALLBACK_ENDPOINT, message, {"reason": reason.value}) diff --git a/src/mcp_server_appwrite/flags.py b/src/mcp_server_appwrite/flags.py index a1d7e27..68af131 100644 --- a/src/mcp_server_appwrite/flags.py +++ b/src/mcp_server_appwrite/flags.py @@ -45,7 +45,19 @@ class Flag: ), ) -FLAGS: tuple[Flag, ...] = (CONSOLE_URL,) +EVENTS = Flag( + name="events", + env="MCP_EVENTS", + help=( + "Set to 1 to advertise MCP Events and serve the events/* methods. " + "HTTP transport only." + ), +) + +FLAGS: tuple[Flag, ...] = (CONSOLE_URL, EVENTS) + +ENABLED_VALUES = frozenset({"1", "true", "yes", "on"}) +"""Values that switch an on/off flag on (case-insensitive).""" def value(flag: Flag) -> str | None: @@ -53,6 +65,12 @@ def value(flag: Flag) -> str | None: return os.getenv(flag.env, "").strip().rstrip("/") or None +def enabled(flag: Flag) -> bool: + """Whether an on/off flag is switched on. Anything outside + :data:`ENABLED_VALUES`, including unset, means off.""" + return (value(flag) or "").lower() in ENABLED_VALUES + + def register_cli_args(parser: argparse.ArgumentParser) -> None: """Add a ``--`` argument per flag. The default is ``None`` (not the environment variable) so an explicit ``-- ""`` is distinguishable diff --git a/tests/unit/test_events_catalog.py b/tests/unit/test_events_catalog.py new file mode 100644 index 0000000..7c20272 --- /dev/null +++ b/tests/unit/test_events_catalog.py @@ -0,0 +1,266 @@ +import unittest + +from jsonschema import Draft202012Validator + +from mcp_server_appwrite.events import errors +from mcp_server_appwrite.events.catalog import ( + CATALOG, + EVENTS, + Status, + StatusFilter, + lookup, +) +from mcp_server_appwrite.events.errors import EventsError + +EXPECTED_PATTERNS = { + "functions.execution.failed": ( + {"project_id": "p1", "function_id": "fn1"}, + [ + "functions.fn1.executions.*.update", + "functions.fn1.executions.*.create", + ], + ), + "functions.deployment.completed": ( + {"project_id": "p1", "function_id": "fn1", "status": "ready"}, + ["functions.fn1.deployments.*.update"], + ), + "sites.deployment.completed": ( + {"project_id": "p1", "site_id": "site-1"}, + ["sites.site-1.deployments.*.update"], + ), + "tablesdb.row.created": ( + {"project_id": "p1", "database_id": "db_1", "table_id": "tickets"}, + ["tablesdb.db_1.tables.tickets.rows.*.create"], + ), + "storage.file.created": ( + {"project_id": "p1", "bucket_id": "uploads"}, + ["buckets.uploads.files.*.create"], + ), + "users.user.created": ({"project_id": "p1"}, ["users.*.create"]), +} + +# Values that would widen an Appwrite event pattern or are not Appwrite IDs. +BAD_IDS = ["*", "a*", "a.b", "fn1.executions", "", "-lead", "_lead", "a/b", "a b"] + + +def _arguments(name: str) -> dict: + return dict(EXPECTED_PATTERNS[name][0]) + + +class CatalogShapeTests(unittest.TestCase): + def test_catalog_is_the_v1_set_in_order(self): + self.assertEqual( + [event.name for event in EVENTS], list(EXPECTED_PATTERNS.keys()) + ) + self.assertEqual(list(CATALOG), list(EXPECTED_PATTERNS.keys())) + + def test_schemas_are_valid_json_schema(self): + for event in EVENTS: + with self.subTest(event=event.name): + Draft202012Validator.check_schema(event.input_schema) + Draft202012Validator.check_schema(event.payload_schema) + + def test_every_event_requires_project_id(self): + for event in EVENTS: + with self.subTest(event=event.name): + schema = event.input_schema + self.assertIn("project_id", schema["required"]) + self.assertFalse(schema["additionalProperties"]) + + def test_payload_timestamps_are_nullable(self): + for event in EVENTS: + for name, field in event.payload_schema["properties"].items(): + if field.get("format") == "date-time": + with self.subTest(event=event.name, field=name): + self.assertEqual(field["type"], ["string", "null"]) + + def test_payloads_carry_ids_and_metadata_only(self): + # Appwrite sends whole rows and users; nothing that could hold their + # content or PII may be declared for delivery. + forbidden = {"email", "phone", "name", "prefs", "data", "body", "logs"} + for event in EVENTS: + with self.subTest(event=event.name): + self.assertFalse(forbidden & set(event.payload_schema["properties"])) + self.assertEqual( + list(lookup("users.user.created").payload_schema["properties"]), + ["user_id", "created_at"], + ) + + def test_status_filters_are_declared_as_data(self): + failed = lookup("functions.execution.failed").status + self.assertEqual(failed, StatusFilter(accepted=(Status.FAILED,))) + for name in ("functions.deployment.completed", "sites.deployment.completed"): + with self.subTest(event=name): + event = lookup(name) + assert event.status is not None + self.assertEqual(event.status.accepted, (Status.READY, Status.FAILED)) + self.assertEqual(event.status.argument, "status") + self.assertIn("status", event.input_schema["properties"]) + for name in ( + "tablesdb.row.created", + "storage.file.created", + "users.user.created", + ): + self.assertIsNone(lookup(name).status) + + +class StatusFilterTests(unittest.TestCase): + def test_execution_failed_accepts_only_failures(self): + status = lookup("functions.execution.failed").status + assert status is not None + self.assertTrue(status.matches("failed", {})) + for value in ("completed", "waiting", "processing", None): + self.assertFalse(status.matches(value, {})) + + def test_deployment_filter_honours_the_status_argument(self): + status = lookup("functions.deployment.completed").status + assert status is not None + self.assertTrue(status.matches("ready", {})) + self.assertTrue(status.matches("failed", {})) + self.assertFalse(status.matches("building", {})) + self.assertTrue(status.matches("failed", {"status": "failed"})) + self.assertFalse(status.matches("ready", {"status": "failed"})) + + +class PatternTests(unittest.TestCase): + def test_patterns_per_event(self): + for name, (arguments, expected) in EXPECTED_PATTERNS.items(): + with self.subTest(event=name): + self.assertEqual(lookup(name).patterns(arguments), expected) + + def test_patterns_never_contain_project_or_status(self): + for name, (arguments, _) in EXPECTED_PATTERNS.items(): + for pattern in lookup(name).patterns(arguments): + self.assertNotIn("p1", pattern.split(".")) + self.assertNotIn("ready", pattern.split(".")) + + +class ArgumentValidationTests(unittest.TestCase): + def assertInvalid(self, event_name: str, arguments: object) -> EventsError: + with self.assertRaises(EventsError) as caught: + lookup(event_name).patterns(arguments) + self.assertEqual(caught.exception.code, errors.INVALID_PARAMS) + return caught.exception + + def test_rejects_wildcards_and_dots_in_every_id(self): + for name in EXPECTED_PATTERNS: + event = lookup(name) + for argument in event.parameters: + if argument.choices: + continue + for bad in BAD_IDS: + with self.subTest(event=name, argument=argument.name, value=bad): + arguments = {**_arguments(name), argument.name: bad} + self.assertInvalid(name, arguments) + + def test_rejects_ids_longer_than_36(self): + arguments = {"project_id": "p1", "bucket_id": "a" * 37} + self.assertInvalid("storage.file.created", arguments) + arguments["bucket_id"] = "a" * 36 + self.assertEqual( + lookup("storage.file.created").patterns(arguments), + [f"buckets.{'a' * 36}.files.*.create"], + ) + + def test_rejects_missing_unknown_and_mistyped_arguments(self): + self.assertInvalid("users.user.created", {}) + self.assertInvalid("users.user.created", None) + self.assertInvalid("users.user.created", ["p1"]) + self.assertInvalid("users.user.created", {"project_id": 1}) + self.assertInvalid("tablesdb.row.created", {"project_id": "p1"}) + error = self.assertInvalid( + "users.user.created", {"project_id": "p1", "user_id": "u1"} + ) + self.assertIn("user_id", error.message) + + def test_status_argument_must_be_a_terminal_status(self): + base = {"project_id": "p1", "site_id": "s1"} + for bad in ("building", "READY", None, "*"): + with self.subTest(value=bad): + self.assertInvalid( + "sites.deployment.completed", {**base, "status": bad} + ) + self.assertEqual( + lookup("sites.deployment.completed").validate({**base, "status": "failed"}), + {"project_id": "p1", "site_id": "s1", "status": "failed"}, + ) + + def test_validate_agrees_with_the_published_input_schema(self): + cases = [{}, {"project_id": "p1"}, {"project_id": "*"}, {"extra": "x"}] + for name in EXPECTED_PATTERNS: + event = lookup(name) + validator = Draft202012Validator(event.input_schema) + samples = [*cases, _arguments(name)] + samples += [ + {**_arguments(name), argument.name: bad} + for argument in event.parameters + for bad in [*BAD_IDS, None, 7] + ] + for sample in samples: + with self.subTest(event=name, sample=sample): + schema_accepts = validator.is_valid(sample) + try: + event.validate(sample) + accepts = True + except EventsError: + accepts = False + self.assertEqual(accepts, schema_accepts) + + +class LookupTests(unittest.TestCase): + def test_lookup_returns_the_event(self): + self.assertIs(lookup("users.user.created"), CATALOG["users.user.created"]) + + def test_unknown_event_is_not_found_with_kind(self): + for name in ("users.user.deleted", "", None, 3): + with self.subTest(name=name): + with self.assertRaises(EventsError) as caught: + lookup(name) + self.assertEqual(caught.exception.code, errors.NOT_FOUND) + self.assertEqual(caught.exception.data, {"kind": "event"}) + + +class ErrorTests(unittest.TestCase): + def test_codes_match_the_chatgpt_values(self): + self.assertEqual( + ( + errors.INVALID_PARAMS, + errors.NOT_FOUND, + errors.FORBIDDEN, + errors.RESOURCE_EXHAUSTED, + errors.UNSUPPORTED, + errors.CALLBACK_ENDPOINT, + ), + (-32602, -32011, -32012, -32013, -32014, -32015), + ) + + def test_constructors_carry_spec_data(self): + exhausted = EventsError.resource_exhausted( + "Upgrade to Pro", limit="webhooks", maximum=2 + ) + self.assertEqual(exhausted.code, errors.RESOURCE_EXHAUSTED) + self.assertEqual(exhausted.data, {"limit": "webhooks", "max": 2}) + callback = EventsError.callback_endpoint( + "Callback timed out", reason=errors.CallbackFailure.TIMEOUT + ) + self.assertEqual(callback.code, errors.CALLBACK_ENDPOINT) + self.assertEqual(callback.data, {"reason": "timeout"}) + self.assertEqual(EventsError.forbidden("no").error.data, None) + self.assertEqual(EventsError.unsupported("no").code, errors.UNSUPPORTED) + + def test_callback_reasons(self): + self.assertEqual( + {reason.value for reason in errors.CallbackFailure}, + { + "challenge_failed", + "timeout", + "connection_refused", + "tls_error", + "http_4xx", + "http_5xx", + }, + ) + + +if __name__ == "__main__": + unittest.main() From 4e34f73e44f76cbe5afb13ef85ca876c8cc39327 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 13:08:37 +0100 Subject: [PATCH 2/5] feat: advertise MCP events and serve events/list The SDK drops unknown capability keys (python-sdk#3640), so a Server.middleware adds capabilities.events for ChatGPT and the SEP-3415 extension entry to the discover and initialize results. Only on the HTTP transport with MCP_EVENTS on. Tests assert on raw HTTP through the hosted app because the SDK client models drop the key too. --- src/mcp_server_appwrite/events/protocol.py | 122 +++++++++++ src/mcp_server_appwrite/server.py | 6 +- tests/unit/test_events_protocol.py | 243 +++++++++++++++++++++ 3 files changed, 370 insertions(+), 1 deletion(-) create mode 100644 src/mcp_server_appwrite/events/protocol.py create mode 100644 tests/unit/test_events_protocol.py diff --git a/src/mcp_server_appwrite/events/protocol.py b/src/mcp_server_appwrite/events/protocol.py new file mode 100644 index 0000000..0ad59c2 --- /dev/null +++ b/src/mcp_server_appwrite/events/protocol.py @@ -0,0 +1,122 @@ +"""MCP Events protocol layer on the low-level SDK server. + +The ``mcp`` SDK has no events support yet, so this module adds it from the +outside and keeps the workaround in one place: + +* **Capability.** ``ServerCapabilities`` drops unknown keys, so ``events`` + never reaches the wire through the SDK (python-sdk#3640). A + ``Server.middleware`` patches the already-serialized ``server/discover`` (and + legacy ``initialize``) result instead. It advertises both shapes clients read + 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`` is registered with ``Server.add_request_handler``; + the SDK adds ``resultType: "complete"`` to custom-method results on the + 2026-07-28 protocol. + +Events are served only on the hosted HTTP transport and only behind the +``events`` flag (``MCP_EVENTS``); see :func:`enabled`. +""" + +from __future__ import annotations + +from typing import Any + +from mcp.server import Server, ServerRequestContext +from mcp.server.context import CallNext, HandlerResult +from mcp.types import RequestParams +from pydantic import ConfigDict + +from .. import flags +from .catalog import EVENTS, Event +from .errors import EventsError + +CAPABILITY = "events" +"""Key under ``capabilities`` that ChatGPT reads.""" + +EXTENSION = "io.modelcontextprotocol/events" +"""SEP-3415 extension identifier under ``capabilities.extensions``.""" + +EXTENSION_SETTINGS: dict[str, Any] = {"listChanged": False} +"""The catalog is static, so there is never a list-changed notification.""" + +DELIVERY_WEBHOOK = "webhook" +"""The only delivery mode served (v1 is webhook-only, like ChatGPT).""" + +LIST_METHOD = "events/list" + +TRANSPORT = "http" +"""Events need the hosted ingress, so stdio never advertises them.""" + +CAPABILITY_METHODS = frozenset({"server/discover", "initialize"}) +"""Methods whose result carries ``capabilities``: ``server/discover`` on the +2026-07-28 protocol and ``initialize`` on earlier ones.""" + + +class ListParams(RequestParams): + """``events/list`` params. Unknown members are tolerated, as the draft + specification may add some.""" + + model_config = ConfigDict(extra="allow") + + cursor: str | None = None + + +def enabled(transport: str) -> bool: + """Whether this server should advertise and serve events.""" + return transport == TRANSPORT and flags.enabled(flags.EVENTS) + + +def describe(event: Event) -> dict[str, Any]: + """One ``events/list`` entry.""" + return { + "name": event.name, + "description": event.description, + "delivery": [DELIVERY_WEBHOOK], + "inputSchema": event.input_schema, + "payloadSchema": event.payload_schema, + } + + +def list_events(cursor: str | None = None) -> 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]} + + +def advertise(result: dict[str, Any]) -> dict[str, Any]: + """``result`` with both events capability shapes added. Existing + capabilities and extensions are kept.""" + capabilities = dict(result.get("capabilities") or {}) + extensions = dict(capabilities.get("extensions") or {}) + extensions[EXTENSION] = dict(EXTENSION_SETTINGS) + capabilities[CAPABILITY] = {} + capabilities["extensions"] = extensions + return {**result, "capabilities": capabilities} + + +class CapabilityMiddleware: + """Adds the events capability to discovery results (python-sdk#3640).""" + + # `ctx` matches the SDK's `ServerMiddleware` protocol parameter name. + async def __call__( + self, ctx: ServerRequestContext[Any, Any], call_next: CallNext + ) -> HandlerResult: + result = await call_next(ctx) + if ctx.method in CAPABILITY_METHODS and isinstance(result, dict): + return advertise(result) + return result + + +async def _handle_list( + context: ServerRequestContext[Any, Any], params: ListParams +) -> dict[str, Any]: + return list_events(params.cursor) + + +def register(server: Server[Any]) -> None: + """Advertise events and serve ``events/list`` on ``server``.""" + server.add_request_handler(LIST_METHOD, ListParams, _handle_list) + server.middleware.append(CapabilityMiddleware()) diff --git a/src/mcp_server_appwrite/server.py b/src/mcp_server_appwrite/server.py index a927abd..136d6d5 100644 --- a/src/mcp_server_appwrite/server.py +++ b/src/mcp_server_appwrite/server.py @@ -74,6 +74,7 @@ ) from .docs_search import DocsSearch from .error_classification import HostedBinaryResponseTooLarge, is_response_parse_error +from .events import protocol as events from .operator import Operator, _parse_tool_name from .service import Service from .tool_manager import ToolManager @@ -1502,7 +1503,7 @@ def tool_input_schema(name: str) -> dict[str, Any] | None: return tool.input_schema return None - return Server( + server = Server( "Appwrite MCP Server", version=SERVER_VERSION, instructions=instructions, @@ -1516,6 +1517,9 @@ def tool_input_schema(name: str) -> dict[str, Any] | None: on_subscriptions_listen=subscriptions, get_tool_input_schema=tool_input_schema, ) + if events.enabled(transport): + events.register(server) + return server def _jsonrpc_error_code(exc: Exception) -> int: diff --git a/tests/unit/test_events_protocol.py b/tests/unit/test_events_protocol.py new file mode 100644 index 0000000..0ff3039 --- /dev/null +++ b/tests/unit/test_events_protocol.py @@ -0,0 +1,243 @@ +"""MCP Events advertisement and ``events/list``, asserted on raw HTTP through the +real hosted Starlette app: the SDK's own client models drop the unknown +``events`` capability, so only the wire shows what clients actually receive.""" + +import asyncio +import json +import os +import unittest +from typing import Any +from unittest import mock + +from mcp.server import Server +from mcp.server.auth.provider import AccessToken +from starlette.testclient import TestClient + +from mcp_server_appwrite import auth, flags +from mcp_server_appwrite.events import protocol +from mcp_server_appwrite.events.catalog import EVENTS +from mcp_server_appwrite.http_app import build_app + +PROTOCOL_VERSION = "2026-07-28" +LEGACY_PROTOCOL_VERSION = "2025-11-25" +EXTENSION = "io.modelcontextprotocol/events" + + +def _request(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "id": 1, + "method": method, + "params": { + **(params or {}), + "_meta": { + "io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION, + "io.modelcontextprotocol/clientInfo": {"name": "test", "version": "1"}, + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, + } + + +def _message(response) -> dict[str, Any]: + """The JSON-RPC message from a JSON or single-event SSE response.""" + text = response.text + if response.headers.get("content-type", "").startswith("text/event-stream"): + data = [line[5:] for line in text.splitlines() if line.startswith("data:")] + text = data[-1] + return json.loads(text) + + +class HostedEventsTests(unittest.TestCase): + HEADERS = { + "Authorization": "Bearer test-token", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + + def setUp(self): + from mcp_server_appwrite import server as server_module + + original_transport = server_module._UPLOAD_TRANSPORT + self.addCleanup(setattr, server_module, "_UPLOAD_TRANSPORT", original_transport) + + async def verify_token(_verifier, token): + return AccessToken(token=token, client_id="test-client", scopes=[]) + + patcher = mock.patch.object( + auth.AppwriteTokenVerifier, "verify_token", verify_token + ) + patcher.start() + self.addCleanup(patcher.stop) + + def _flag(self, value: str | None): + environment = dict(os.environ) + environment.pop(flags.EVENTS.env, None) + if value is not None: + environment[flags.EVENTS.env] = value + return mock.patch.dict(os.environ, environment, clear=True) + + def _post(self, client, method: str, params=None, path: str = "/"): + headers = { + **self.HEADERS, + "MCP-Protocol-Version": PROTOCOL_VERSION, + "Mcp-Method": method, + } + response = client.post(path, json=_request(method, params), headers=headers) + return response.status_code, _message(response) + + def _initialize(self, client) -> dict[str, Any]: + body = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": LEGACY_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "test", "version": "1"}, + }, + } + response = client.post("/mcp", json=body, headers=self.HEADERS) + self.assertEqual(response.status_code, 200, response.text) + return _message(response)["result"] + + def test_discover_advertises_both_capability_shapes(self): + with self._flag("1"), TestClient(build_app()) as client: + for path in ("/", "/mcp"): + with self.subTest(path=path): + status, message = self._post(client, "server/discover", path=path) + self.assertEqual(status, 200, message) + capabilities = message["result"]["capabilities"] + self.assertEqual(capabilities["events"], {}) + self.assertEqual( + capabilities["extensions"][EXTENSION], {"listChanged": False} + ) + # The SDK-derived capabilities survive the patch. + self.assertIn("tools", capabilities) + self.assertIn("resources", capabilities) + + def test_legacy_initialize_advertises_events_too(self): + with self._flag("1"), TestClient(build_app()) as client: + capabilities = self._initialize(client)["capabilities"] + self.assertEqual(capabilities["events"], {}) + self.assertEqual(capabilities["extensions"][EXTENSION], {"listChanged": False}) + + def test_events_list_serves_the_catalog(self): + with self._flag("1"), TestClient(build_app()) as client: + status, message = self._post(client, "events/list") + self.assertEqual(status, 200, message) + result = message["result"] + self.assertEqual(result["resultType"], "complete") + self.assertNotIn("nextCursor", result) + self.assertEqual( + [event["name"] for event in result["events"]], + [event.name for event in EVENTS], + ) + for entry, event in zip(result["events"], EVENTS, strict=True): + with self.subTest(event=event.name): + self.assertEqual( + set(entry), + {"name", "description", "delivery", "inputSchema", "payloadSchema"}, + ) + self.assertEqual(entry["delivery"], ["webhook"]) + self.assertEqual(entry["description"], event.description) + self.assertEqual(entry["inputSchema"], event.input_schema) + self.assertEqual(entry["payloadSchema"], event.payload_schema) + + def test_events_list_rejects_a_cursor_it_never_issued(self): + with self._flag("1"), TestClient(build_app()) as client: + status, message = self._post(client, "events/list", {"cursor": "abc"}) + self.assertEqual(status, 400) + self.assertEqual(message["error"]["code"], -32602) + + def test_subscribe_and_unsubscribe_are_not_served_yet(self): + with self._flag("1"), TestClient(build_app()) as client: + for method in ("events/subscribe", "events/unsubscribe"): + with self.subTest(method=method): + _, message = self._post(client, method) + self.assertEqual(message["error"]["code"], -32601) + + def test_nothing_is_advertised_when_the_flag_is_off(self): + for value in (None, "", "0", "false"): + with ( + self.subTest(flag=value), + self._flag(value), + TestClient(build_app()) as client, + ): + status, message = self._post(client, "server/discover") + self.assertEqual(status, 200, message) + capabilities = message["result"]["capabilities"] + self.assertNotIn("events", capabilities) + self.assertNotIn(EXTENSION, capabilities.get("extensions") or {}) + self.assertNotIn("events", self._initialize(client)["capabilities"]) + + _, message = self._post(client, "events/list") + self.assertEqual(message["error"]["code"], -32601) + + +class EnabledTests(unittest.TestCase): + def test_requires_flag_and_http_transport(self): + cases = [ + ("http", "1", True), + ("http", "true", True), + ("http", "ON", True), + ("http", "0", False), + ("http", None, False), + ("stdio", "1", False), + ] + for transport, value, expected in cases: + environment = {} if value is None else {flags.EVENTS.env: value} + with ( + self.subTest(transport=transport, flag=value), + mock.patch.dict(os.environ, environment, clear=True), + ): + self.assertIs(protocol.enabled(transport), expected) + + +class CapabilityMiddlewareTests(unittest.TestCase): + def _run(self, method: str, result: Any) -> Any: + context = mock.Mock(method=method) + + async def call_next(_context): + return result + + return asyncio.run(protocol.CapabilityMiddleware()(context, call_next)) + + def test_keeps_existing_extensions(self): + result = { + "capabilities": { + "tools": {}, + "extensions": {"io.example/other": {"enabled": True}}, + } + } + patched = self._run("server/discover", result) + self.assertEqual( + patched["capabilities"]["extensions"], + { + "io.example/other": {"enabled": True}, + EXTENSION: {"listChanged": False}, + }, + ) + self.assertEqual(patched["capabilities"]["tools"], {}) + # The handler's result is not mutated in place. + self.assertNotIn("events", result["capabilities"]) + + def test_other_methods_pass_through(self): + result = {"tools": []} + self.assertIs(self._run("tools/list", result), result) + + def test_register_wires_handler_and_middleware(self): + server = Server("test") + protocol.register(server) + self.assertIsNotNone(server.get_request_handler("events/list")) + self.assertIsNone(server.get_request_handler("events/subscribe")) + self.assertTrue( + any( + isinstance(middleware, protocol.CapabilityMiddleware) + for middleware in server.middleware + ) + ) + + +if __name__ == "__main__": + unittest.main() From 101d8def695f6ed509107022be4e20e2b1894edc Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 13:08:37 +0100 Subject: [PATCH 3/5] docs: document MCP events and the events flag --- README.md | 2 ++ docs/events.md | 77 ++++++++++++++++++++++++++++++++++++++++++++++++++ docs/flags.md | 39 +++++++++++++++++++++++++ 3 files changed, 118 insertions(+) create mode 100644 docs/events.md diff --git a/README.md b/README.md index 7524057..1f446da 100644 --- a/README.md +++ b/README.md @@ -320,6 +320,8 @@ for per-client setup. - [How Cloud authentication works](docs/authentication.md) — the OAuth 2.1 flow. - [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). - [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 diff --git a/docs/events.md b/docs/events.md new file mode 100644 index 0000000..4b3eb42 --- /dev/null +++ b/docs/events.md @@ -0,0 +1,77 @@ +# Events + +> **In progress.** Events are being built behind the `--events` flag +> ([#127](https://github.com/appwrite/mcp/issues/127)). Today the server +> advertises the capability and lists the catalog; subscribing and delivery are +> not available yet. + +[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". + +Events are served only by the hosted HTTP server (`mcp.appwrite.io`), and v1 +delivers only by webhook. Each subscription will be backed by one Appwrite +webhook in the subscriber's project, so the MCP server keeps no subscription +store. + +## Enabling + +```bash +MCP_PUBLIC_URL=http://localhost:8000 \ + uv run mcp-server-appwrite --transport http --events 1 +``` + +`MCP_EVENTS=1` works too. See [flags.md](flags.md#--events--mcp-events-in-progress) +for quick checks with `curl`. + +With the flag on, the server advertises the capability in both shapes that +clients read today: + +| Where | Read by | +| --- | --- | +| `capabilities.events: {}` | ChatGPT | +| `capabilities.extensions["io.modelcontextprotocol/events"]: {"listChanged": false}` | SEP-3415 clients | + +The `mcp` SDK does not model this capability yet +([python-sdk#3640](https://github.com/modelcontextprotocol/python-sdk/issues/3640)), +so a server middleware adds it to the `server/discover` and `initialize` +results. + +## Catalog + +`events/list` returns these events. Every event takes a required `project_id`. +ID arguments accept 1–36 letters, digits, `_` or `-`, so a `*` or `.` can never +widen a subscription beyond the named resource. Payloads carry IDs and metadata +only; the agent fetches details with `appwrite_call_tool`. + +| Event | Arguments | Appwrite events | Delivered when | Payload | +| --- | --- | --- | --- | --- | +| `functions.execution.failed` | `function_id` | `functions.{function_id}.executions.*.update`, `functions.{function_id}.executions.*.create` | status is `failed` | `function_id`, `execution_id`, `status`, `trigger`, `response_status_code`, `created_at` | +| `functions.deployment.completed` | `function_id`, `status?` (`ready` \| `failed`) | `functions.{function_id}.deployments.*.update` | status is `ready` or `failed` (or the requested one) | `function_id`, `deployment_id`, `status`, `updated_at` | +| `sites.deployment.completed` | `site_id`, `status?` (`ready` \| `failed`) | `sites.{site_id}.deployments.*.update` | status is `ready` or `failed` (or the requested one) | `site_id`, `deployment_id`, `status`, `updated_at` | +| `tablesdb.row.created` | `database_id`, `table_id` | `tablesdb.{database_id}.tables.{table_id}.rows.*.create` | always | `database_id`, `table_id`, `row_id`, `created_at` | +| `storage.file.created` | `bucket_id` | `buckets.{bucket_id}.files.*.create` | always | `bucket_id`, `file_id`, `mime_type`, `size`, `created_at` | +| `users.user.created` | none | `users.*.create` | always | `user_id`, `created_at` | + +Synchronous function executions fire only `.create`, after they finish, which +is why `functions.execution.failed` listens to both. Executions served through +function domains fire no Appwrite event and are not covered. + +Timestamps are ISO 8601 strings and may be `null`. The catalog lives in +[`events/catalog.py`](../src/mcp_server_appwrite/events/catalog.py). + +## Errors + +Error codes follow the values ChatGPT uses today. SEP-3415 renumbers them, so +they are kept behind constants in +[`events/errors.py`](../src/mcp_server_appwrite/events/errors.py). + +| Meaning | Code | `data` | +| --- | --- | --- | +| Bad arguments, URL or secret | `-32602` | | +| NotFound | `-32011` | `kind` | +| Forbidden | `-32012` | | +| ResourceExhausted | `-32013` | `limit`, `max` | +| Unsupported | `-32014` | | +| CallbackEndpointError | `-32015` | `reason`: `challenge_failed`, `timeout`, `connection_refused`, `tls_error`, `http_4xx`, `http_5xx` | diff --git a/docs/flags.md b/docs/flags.md index 4ca12b4..263e077 100644 --- a/docs/flags.md +++ b/docs/flags.md @@ -34,6 +34,7 @@ Flags are for testing overrides only — permanent configuration belongs in | CLI | Env | What it does | | --- | --- | --- | | `--console-url` | `MCP_CONSOLE_URL` | Send OAuth login/consent to an alternative Appwrite Console (HTTP transport only). | +| `--events` | `MCP_EVENTS` | Advertise MCP Events and serve `events/list` (HTTP transport only). See [events.md](events.md). | ### `--console-url` — test OAuth against a pre-release console @@ -80,3 +81,41 @@ claude mcp add --transport http appwrite-test http://localhost:8000/ Then run `/mcp` in Claude Code and authenticate — the browser should open the override console's sign-in page, and after consent the client completes the 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, and the server answers `events/list` with the +event catalog. It is off by default while the feature is built +([#127](https://github.com/appwrite/mcp/issues/127)); `events/subscribe` and +delivery are not available yet. Any of `1`, `true`, `yes` or `on` enables it. + +**Enable it:** + +```bash +MCP_PUBLIC_URL=http://localhost:8000 \ + uv run mcp-server-appwrite --transport http --events 1 +``` + +**Quick checks** (any valid Appwrite OAuth access token works as `$TOKEN`): + +```bash +mcp() { + curl -s 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: $1" \ + -d '{"jsonrpc":"2.0","id":1,"method":"'"$1"'","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}' +} + +# Both shapes: capabilities.events and capabilities.extensions["io.modelcontextprotocol/events"] +mcp server/discover | jq .result.capabilities + +# The catalog +mcp events/list | jq '.result.events[].name' +``` + +With the flag off, neither capability appears and `events/list` returns +`-32601` (method not found). From e96848f189d5fae358660c04b355faa9419295ec Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 14:38:01 +0100 Subject: [PATCH 4/5] test: cover events discovery end to end against the hosted server Boot the real Starlette app under uvicorn and drive it over HTTP like ChatGPT, so the wire is what is asserted. Unit tests the e2e flow now covers are removed; only what HTTP cannot reach before subscribe stays. --- docs/events.md | 17 ++ pyproject.toml | 4 + tests/e2e/support.py | 227 +++++++++++++++++++++++ tests/e2e/test_events_protocol.py | 178 ++++++++++++++++++ tests/unit/test_events_catalog.py | 277 ++++------------------------- tests/unit/test_events_protocol.py | 244 ++----------------------- uv.lock | 4 + 7 files changed, 475 insertions(+), 476 deletions(-) create mode 100644 tests/e2e/support.py create mode 100644 tests/e2e/test_events_protocol.py diff --git a/docs/events.md b/docs/events.md index 4b3eb42..3422dd9 100644 --- a/docs/events.md +++ b/docs/events.md @@ -75,3 +75,20 @@ they are kept behind constants in | ResourceExhausted | `-32013` | `limit`, `max` | | Unsupported | `-32014` | | | CallbackEndpointError | `-32015` | `reason`: `challenge_failed`, `timeout`, `connection_refused`, `tls_error`, `http_4xx`, `http_5xx` | + +## Testing + +Events are tested end to end in `tests/e2e/`: the real hosted app runs under +uvicorn on a random localhost port, and the tests talk to it over HTTP exactly +like ChatGPT does (`MCP-Protocol-Version: 2026-07-28`, `Mcp-Method`, `_meta`). +Only the OAuth token verifier is stubbed, because Cloud OAuth is not reachable +from CI. The suite needs no credentials and runs on every PR: + +```bash +uv run --group e2e python -m unittest discover -s tests/e2e -v +``` + +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) +ships; until then the server's surroundings are played locally. diff --git a/pyproject.toml b/pyproject.toml index cc8fd7d..4fad3c5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,10 @@ dev = [ "ruff==0.15.18", "pyright==1.1.411", ] +# tests/e2e: the hosted server driven over real HTTP; no credentials needed. +e2e = [ + "jsonschema>=4.26", +] [project.scripts] mcp-server-appwrite = "mcp_server_appwrite.__main__:main" diff --git a/tests/e2e/support.py b/tests/e2e/support.py new file mode 100644 index 0000000..72e37fc --- /dev/null +++ b/tests/e2e/support.py @@ -0,0 +1,227 @@ +"""End-to-end harness: the real hosted server, driven over real HTTP. + +:class:`Server` boots the production Starlette app (``http_app.build_app``) +under uvicorn on a random localhost port, in a background thread of the test +process. :class:`Client` talks to it the way ChatGPT does: raw JSON-RPC over +Streamable HTTP with ``MCP-Protocol-Version: 2026-07-28``, ``Mcp-Method`` and +the request ``_meta``, asserting on the wire rather than on SDK models (the SDK +client drops capabilities it does not model, such as ``events``). + +The one stub is the OAuth token verifier: Cloud OAuth is not reachable from CI, +so a fixed bearer token is accepted in place of an Appwrite access token, the +same seam the unit tests of ``http_app`` use. Everything behind it (routing, +auth middleware, the MCP session manager, the low-level server and its +middleware) is the production code path. +""" + +from __future__ import annotations + +import json +import os +import socket +import threading +import time +from collections.abc import Callable, Mapping +from types import TracebackType +from typing import Any +from unittest import mock + +import httpx +import uvicorn +from mcp.server.auth.provider import AccessToken + +from mcp_server_appwrite import auth +from mcp_server_appwrite.http_app import build_app + +TOKEN = "e2e-access-token" +"""The bearer token the stubbed verifier accepts.""" + +CLIENT_ID = "e2e-client" + +PROTOCOL_VERSION = "2026-07-28" +LEGACY_PROTOCOL_VERSION = "2025-11-25" + +PUBLIC_URL = "https://mcp.e2e.test" +"""``MCP_PUBLIC_URL`` of every server: the public origin a load balancer would +serve, deliberately not the address the server listens on.""" + +STARTUP_SECONDS = 15.0 + +CONTROLLED = ( + "MCP_EVENTS", + "MCP_EVENTS_SEALING_KEYS", + "MCP_PUBLIC_URL", + "MCP_CONSOLE_URL", + "OTEL_EXPORTER_OTLP_ENDPOINT", + "SENTRY_DSN", +) +"""Variables a server never inherits from the shell running the tests.""" + + +async def _verify_token(_verifier: Any, token: str) -> AccessToken | None: + if token != TOKEN: + return None + return AccessToken(token=token, client_id=CLIENT_ID, scopes=[]) + + +class Server: + """The hosted app on ``127.0.0.1:``. + + ``environment`` is applied on top of the test process environment (minus + :data:`CONTROLLED`) for the server's whole lifetime; a ``None`` value + removes a variable. ``build`` returns the keyword arguments for + ``build_app``; it runs with that environment in place. + """ + + def __init__( + self, + environment: Mapping[str, str | None] | None = None, + build: Callable[[], Mapping[str, Any]] | None = None, + ) -> None: + self._environment = {"MCP_PUBLIC_URL": PUBLIC_URL, **(environment or {})} + self._build = build + self._patches: list[Any] = [] + self._server: uvicorn.Server | None = None + self._thread: threading.Thread | None = None + self.url = "" + + def __enter__(self) -> Server: + environment = { + key: value for key, value in os.environ.items() if key not in CONTROLLED + } + for key, value in self._environment.items(): + if value is None: + environment.pop(key, None) + else: + environment[key] = value + self._patches = [ + mock.patch.dict(os.environ, environment, clear=True), + mock.patch.object( + auth.AppwriteTokenVerifier, "verify_token", _verify_token + ), + ] + for patch in self._patches: + patch.start() + try: + self._start() + except BaseException: + self._stop_patches() + raise + return self + + def _start(self) -> None: + app = build_app(**(self._build() if self._build else {})) + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + config = uvicorn.Config(app, log_level="warning", access_log=False) + self._server = uvicorn.Server(config) + self._thread = threading.Thread( + target=self._server.run, kwargs={"sockets": [listener]}, daemon=True + ) + self._thread.start() + deadline = time.monotonic() + STARTUP_SECONDS + while not self._server.started: + if not self._thread.is_alive() or time.monotonic() > deadline: + self._shutdown() + raise RuntimeError("the hosted server did not start") + time.sleep(0.01) + self.url = f"http://127.0.0.1:{port}" + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self._shutdown() + self._stop_patches() + + def _shutdown(self) -> None: + if self._server is not None: + self._server.should_exit = True + if self._thread is not None: + self._thread.join(STARTUP_SECONDS) + self._server = None + self._thread = None + + def _stop_patches(self) -> None: + for patch in reversed(self._patches): + patch.stop() + self._patches = [] + + def client(self) -> Client: + return Client(self.url) + + +def message(response: httpx.Response) -> dict[str, Any]: + """The JSON-RPC message of a JSON or single-event SSE response.""" + text = response.text + if response.headers.get("content-type", "").startswith("text/event-stream"): + data = [line[5:] for line in text.splitlines() if line.startswith("data:")] + text = data[-1] + return json.loads(text) + + +class Client: + """A Streamable-HTTP MCP client speaking raw JSON-RPC, like ChatGPT.""" + + def __init__(self, url: str, token: str = TOKEN) -> None: + self._http = httpx.Client(base_url=url, timeout=30.0) + self._headers = { + "Authorization": f"Bearer {token}", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + self._ids = iter(range(1, 1_000_000)) + + def __enter__(self) -> Client: + return self + + def __exit__(self, *exc_info: object) -> None: + self._http.close() + + def call( + self, method: str, params: Mapping[str, Any] | None = None, *, path: str = "/" + ) -> tuple[int, dict[str, Any]]: + """One 2026-07-28 request: protocol version and method in the headers, + client identity in ``params._meta``.""" + body = { + "jsonrpc": "2.0", + "id": next(self._ids), + "method": method, + "params": { + **(params or {}), + "_meta": { + "io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION, + "io.modelcontextprotocol/clientInfo": { + "name": "e2e", + "version": "1", + }, + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, + } + headers = { + **self._headers, + "MCP-Protocol-Version": PROTOCOL_VERSION, + "Mcp-Method": method, + } + response = self._http.post(path, json=body, headers=headers) + return response.status_code, message(response) + + def initialize(self, *, path: str = "/") -> tuple[int, dict[str, Any]]: + """A 2025-11-25 ``initialize`` handshake.""" + body = { + "jsonrpc": "2.0", + "id": next(self._ids), + "method": "initialize", + "params": { + "protocolVersion": LEGACY_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "e2e", "version": "1"}, + }, + } + response = self._http.post(path, json=body, headers=self._headers) + return response.status_code, message(response) diff --git a/tests/e2e/test_events_protocol.py b/tests/e2e/test_events_protocol.py new file mode 100644 index 0000000..48e9e64 --- /dev/null +++ b/tests/e2e/test_events_protocol.py @@ -0,0 +1,178 @@ +"""MCP Events discovery and ``events/list`` against the real hosted server. + +A client connects the way ChatGPT does, discovers the events capability and +reads the catalog. The published schemas are checked with ``jsonschema`` as a +client would use them, so the wire is the contract under test. +""" + +import unittest + +from jsonschema import Draft202012Validator +from support import Server + +EXTENSION = "io.modelcontextprotocol/events" + +CATALOG = [ + "functions.execution.failed", + "functions.deployment.completed", + "sites.deployment.completed", + "tablesdb.row.created", + "storage.file.created", + "users.user.created", +] + +VALID_ARGUMENTS = { + "functions.execution.failed": {"project_id": "p1", "function_id": "fn_1"}, + "functions.deployment.completed": { + "project_id": "p1", + "function_id": "fn_1", + "status": "ready", + }, + "sites.deployment.completed": {"project_id": "p1", "site_id": "site-1"}, + "tablesdb.row.created": { + "project_id": "p1", + "database_id": "main", + "table_id": "tickets", + }, + "storage.file.created": {"project_id": "p1", "bucket_id": "uploads"}, + "users.user.created": {"project_id": "p1"}, +} + +# Content and PII that Appwrite models carry and an event must never declare. +FORBIDDEN_PAYLOAD_FIELDS = {"email", "phone", "name", "prefs", "data", "body", "logs"} + + +class EventsEnabledFlow(unittest.TestCase): + """A client discovers events on a server with the flag on.""" + + environment = {"MCP_EVENTS": "1"} + + def test_client_discovers_and_lists_events(self): + with Server(self.environment) as server, server.client() as client: + for path in ("/", "/mcp"): + with self.subTest(step="server/discover", path=path): + status, message = client.call("server/discover", path=path) + self.assertEqual(status, 200, message) + capabilities = message["result"]["capabilities"] + self.assertEqual(capabilities["events"], {}) + self.assertEqual( + capabilities["extensions"][EXTENSION], {"listChanged": False} + ) + # The SDK-derived capabilities survive the patch. + self.assertIn("tools", capabilities) + self.assertIn("resources", capabilities) + + with self.subTest(step="initialize"): + status, message = client.initialize(path="/mcp") + self.assertEqual(status, 200, message) + capabilities = message["result"]["capabilities"] + self.assertEqual(capabilities["events"], {}) + self.assertEqual( + capabilities["extensions"][EXTENSION], {"listChanged": False} + ) + self.assertIn("tools", capabilities) + + with self.subTest(step="events/list"): + status, message = client.call("events/list") + self.assertEqual(status, 200, message) + self.check_catalog(message["result"]) + + with self.subTest(step="events/list with a foreign cursor"): + status, message = client.call("events/list", {"cursor": "abc"}) + self.assertEqual(status, 400, message) + self.assertEqual(message["error"]["code"], -32602) + + for method in ("events/subscribe", "events/unsubscribe"): + with self.subTest(step=method): + _, message = client.call(method) + self.assertEqual(message["error"]["code"], -32601) + + def check_catalog(self, result): + self.assertEqual(result["resultType"], "complete") + self.assertNotIn("nextCursor", result) + self.assertEqual([event["name"] for event in result["events"]], CATALOG) + for event in result["events"]: + name = event["name"] + with self.subTest(event=name): + self.assertEqual( + set(event), + {"name", "description", "delivery", "inputSchema", "payloadSchema"}, + ) + self.assertTrue(event["description"]) + self.assertEqual(event["delivery"], ["webhook"]) + + inputs = event["inputSchema"] + payload = event["payloadSchema"] + Draft202012Validator.check_schema(inputs) + Draft202012Validator.check_schema(payload) + + self.assertIn("project_id", inputs["required"]) + self.assertFalse(inputs["additionalProperties"]) + arguments = Draft202012Validator(inputs) + self.assertTrue(arguments.is_valid(VALID_ARGUMENTS[name])) + self.assertFalse(arguments.is_valid({})) + self.assertFalse( + arguments.is_valid({**VALID_ARGUMENTS[name], "extra": "x"}) + ) + # A wildcard or a dot in an ID would widen the Appwrite + # event pattern beyond the named resource. + for argument in VALID_ARGUMENTS[name]: + if argument == "status": + continue + for bad in ("*", "a.b", "-lead", "a" * 37): + self.assertFalse( + arguments.is_valid( + {**VALID_ARGUMENTS[name], argument: bad} + ), + (argument, bad), + ) + + fields = payload["properties"] + self.assertFalse(FORBIDDEN_PAYLOAD_FIELDS & set(fields)) + for field, schema in fields.items(): + if schema.get("format") == "date-time": + self.assertEqual(schema["type"], ["string", "null"], field) + + deployments = { + event["name"]: event["inputSchema"] + for event in result["events"] + if event["name"].endswith("deployment.completed") + } + for name, schema in deployments.items(): + with self.subTest(event=name, argument="status"): + validator = Draft202012Validator(schema) + base = VALID_ARGUMENTS[name] + self.assertTrue(validator.is_valid({**base, "status": "failed"})) + self.assertFalse(validator.is_valid({**base, "status": "building"})) + users = next(e for e in result["events"] if e["name"] == "users.user.created") + self.assertEqual( + list(users["payloadSchema"]["properties"]), ["user_id", "created_at"] + ) + + +class EventsDisabledFlow(unittest.TestCase): + """Without the flag, nothing about events is visible.""" + + def test_nothing_is_advertised_or_served(self): + for value in (None, "0", "false"): + with ( + self.subTest(flag=value), + Server({"MCP_EVENTS": value}) as server, + server.client() as client, + ): + status, message = client.call("server/discover") + self.assertEqual(status, 200, message) + capabilities = message["result"]["capabilities"] + self.assertNotIn("events", capabilities) + self.assertNotIn(EXTENSION, capabilities.get("extensions") or {}) + + status, message = client.initialize() + self.assertEqual(status, 200, message) + self.assertNotIn("events", message["result"]["capabilities"]) + + _, message = client.call("events/list") + self.assertEqual(message["error"]["code"], -32601) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_events_catalog.py b/tests/unit/test_events_catalog.py index 7c20272..124e74a 100644 --- a/tests/unit/test_events_catalog.py +++ b/tests/unit/test_events_catalog.py @@ -1,265 +1,52 @@ -import unittest +"""Server-side argument validation of the events catalog. + +Nothing reaches ``Event.patterns`` or ``lookup`` over HTTP until +``events/subscribe`` lands (PR 5 of #127). The published schemas are covered +end to end in ``tests/e2e/test_events_protocol.py``; replace this module with +the subscribe e2e flow once it exists. +""" -from jsonschema import Draft202012Validator +import unittest from mcp_server_appwrite.events import errors -from mcp_server_appwrite.events.catalog import ( - CATALOG, - EVENTS, - Status, - StatusFilter, - lookup, -) +from mcp_server_appwrite.events.catalog import EVENTS, lookup from mcp_server_appwrite.events.errors import EventsError -EXPECTED_PATTERNS = { - "functions.execution.failed": ( - {"project_id": "p1", "function_id": "fn1"}, - [ - "functions.fn1.executions.*.update", - "functions.fn1.executions.*.create", - ], - ), - "functions.deployment.completed": ( - {"project_id": "p1", "function_id": "fn1", "status": "ready"}, - ["functions.fn1.deployments.*.update"], - ), - "sites.deployment.completed": ( - {"project_id": "p1", "site_id": "site-1"}, - ["sites.site-1.deployments.*.update"], - ), - "tablesdb.row.created": ( - {"project_id": "p1", "database_id": "db_1", "table_id": "tickets"}, - ["tablesdb.db_1.tables.tickets.rows.*.create"], - ), - "storage.file.created": ( - {"project_id": "p1", "bucket_id": "uploads"}, - ["buckets.uploads.files.*.create"], - ), - "users.user.created": ({"project_id": "p1"}, ["users.*.create"]), +VALID = { + "functions.execution.failed": {"function_id": "fn1"}, + "functions.deployment.completed": {"function_id": "fn1"}, + "sites.deployment.completed": {"site_id": "site-1"}, + "tablesdb.row.created": {"database_id": "db_1", "table_id": "tickets"}, + "storage.file.created": {"bucket_id": "uploads"}, + "users.user.created": {}, } # Values that would widen an Appwrite event pattern or are not Appwrite IDs. -BAD_IDS = ["*", "a*", "a.b", "fn1.executions", "", "-lead", "_lead", "a/b", "a b"] - - -def _arguments(name: str) -> dict: - return dict(EXPECTED_PATTERNS[name][0]) - - -class CatalogShapeTests(unittest.TestCase): - def test_catalog_is_the_v1_set_in_order(self): - self.assertEqual( - [event.name for event in EVENTS], list(EXPECTED_PATTERNS.keys()) - ) - self.assertEqual(list(CATALOG), list(EXPECTED_PATTERNS.keys())) - - def test_schemas_are_valid_json_schema(self): - for event in EVENTS: - with self.subTest(event=event.name): - Draft202012Validator.check_schema(event.input_schema) - Draft202012Validator.check_schema(event.payload_schema) +BAD_IDS = ["*", "a*", "a.b", "fn1.executions", "", "-lead", "_lead", "a b", "a" * 37] - def test_every_event_requires_project_id(self): - for event in EVENTS: - with self.subTest(event=event.name): - schema = event.input_schema - self.assertIn("project_id", schema["required"]) - self.assertFalse(schema["additionalProperties"]) - - def test_payload_timestamps_are_nullable(self): - for event in EVENTS: - for name, field in event.payload_schema["properties"].items(): - if field.get("format") == "date-time": - with self.subTest(event=event.name, field=name): - self.assertEqual(field["type"], ["string", "null"]) - def test_payloads_carry_ids_and_metadata_only(self): - # Appwrite sends whole rows and users; nothing that could hold their - # content or PII may be declared for delivery. - forbidden = {"email", "phone", "name", "prefs", "data", "body", "logs"} +class ArgumentValidationTests(unittest.TestCase): + def test_ids_can_never_widen_a_pattern(self): for event in EVENTS: - with self.subTest(event=event.name): - self.assertFalse(forbidden & set(event.payload_schema["properties"])) - self.assertEqual( - list(lookup("users.user.created").payload_schema["properties"]), - ["user_id", "created_at"], - ) - - def test_status_filters_are_declared_as_data(self): - failed = lookup("functions.execution.failed").status - self.assertEqual(failed, StatusFilter(accepted=(Status.FAILED,))) - for name in ("functions.deployment.completed", "sites.deployment.completed"): - with self.subTest(event=name): - event = lookup(name) - assert event.status is not None - self.assertEqual(event.status.accepted, (Status.READY, Status.FAILED)) - self.assertEqual(event.status.argument, "status") - self.assertIn("status", event.input_schema["properties"]) - for name in ( - "tablesdb.row.created", - "storage.file.created", - "users.user.created", - ): - self.assertIsNone(lookup(name).status) - - -class StatusFilterTests(unittest.TestCase): - def test_execution_failed_accepts_only_failures(self): - status = lookup("functions.execution.failed").status - assert status is not None - self.assertTrue(status.matches("failed", {})) - for value in ("completed", "waiting", "processing", None): - self.assertFalse(status.matches(value, {})) - - def test_deployment_filter_honours_the_status_argument(self): - status = lookup("functions.deployment.completed").status - assert status is not None - self.assertTrue(status.matches("ready", {})) - self.assertTrue(status.matches("failed", {})) - self.assertFalse(status.matches("building", {})) - self.assertTrue(status.matches("failed", {"status": "failed"})) - self.assertFalse(status.matches("ready", {"status": "failed"})) - - -class PatternTests(unittest.TestCase): - def test_patterns_per_event(self): - for name, (arguments, expected) in EXPECTED_PATTERNS.items(): - with self.subTest(event=name): - self.assertEqual(lookup(name).patterns(arguments), expected) - - def test_patterns_never_contain_project_or_status(self): - for name, (arguments, _) in EXPECTED_PATTERNS.items(): - for pattern in lookup(name).patterns(arguments): + arguments = {"project_id": "p1", **VALID[event.name]} + patterns = event.patterns(arguments) + self.assertTrue(patterns) + for pattern in patterns: self.assertNotIn("p1", pattern.split(".")) - self.assertNotIn("ready", pattern.split(".")) - - -class ArgumentValidationTests(unittest.TestCase): - def assertInvalid(self, event_name: str, arguments: object) -> EventsError: - with self.assertRaises(EventsError) as caught: - lookup(event_name).patterns(arguments) - self.assertEqual(caught.exception.code, errors.INVALID_PARAMS) - return caught.exception - - def test_rejects_wildcards_and_dots_in_every_id(self): - for name in EXPECTED_PATTERNS: - event = lookup(name) for argument in event.parameters: if argument.choices: continue for bad in BAD_IDS: - with self.subTest(event=name, argument=argument.name, value=bad): - arguments = {**_arguments(name), argument.name: bad} - self.assertInvalid(name, arguments) - - def test_rejects_ids_longer_than_36(self): - arguments = {"project_id": "p1", "bucket_id": "a" * 37} - self.assertInvalid("storage.file.created", arguments) - arguments["bucket_id"] = "a" * 36 - self.assertEqual( - lookup("storage.file.created").patterns(arguments), - [f"buckets.{'a' * 36}.files.*.create"], - ) - - def test_rejects_missing_unknown_and_mistyped_arguments(self): - self.assertInvalid("users.user.created", {}) - self.assertInvalid("users.user.created", None) - self.assertInvalid("users.user.created", ["p1"]) - self.assertInvalid("users.user.created", {"project_id": 1}) - self.assertInvalid("tablesdb.row.created", {"project_id": "p1"}) - error = self.assertInvalid( - "users.user.created", {"project_id": "p1", "user_id": "u1"} - ) - self.assertIn("user_id", error.message) + with self.subTest(event=event.name, argument=argument.name): + with self.assertRaises(EventsError) as caught: + event.patterns({**arguments, argument.name: bad}) + self.assertEqual(caught.exception.code, errors.INVALID_PARAMS) - def test_status_argument_must_be_a_terminal_status(self): - base = {"project_id": "p1", "site_id": "s1"} - for bad in ("building", "READY", None, "*"): - with self.subTest(value=bad): - self.assertInvalid( - "sites.deployment.completed", {**base, "status": bad} - ) - self.assertEqual( - lookup("sites.deployment.completed").validate({**base, "status": "failed"}), - {"project_id": "p1", "site_id": "s1", "status": "failed"}, - ) - - def test_validate_agrees_with_the_published_input_schema(self): - cases = [{}, {"project_id": "p1"}, {"project_id": "*"}, {"extra": "x"}] - for name in EXPECTED_PATTERNS: - event = lookup(name) - validator = Draft202012Validator(event.input_schema) - samples = [*cases, _arguments(name)] - samples += [ - {**_arguments(name), argument.name: bad} - for argument in event.parameters - for bad in [*BAD_IDS, None, 7] - ] - for sample in samples: - with self.subTest(event=name, sample=sample): - schema_accepts = validator.is_valid(sample) - try: - event.validate(sample) - accepts = True - except EventsError: - accepts = False - self.assertEqual(accepts, schema_accepts) - - -class LookupTests(unittest.TestCase): - def test_lookup_returns_the_event(self): - self.assertIs(lookup("users.user.created"), CATALOG["users.user.created"]) - - def test_unknown_event_is_not_found_with_kind(self): - for name in ("users.user.deleted", "", None, 3): - with self.subTest(name=name): - with self.assertRaises(EventsError) as caught: - lookup(name) - self.assertEqual(caught.exception.code, errors.NOT_FOUND) - self.assertEqual(caught.exception.data, {"kind": "event"}) - - -class ErrorTests(unittest.TestCase): - def test_codes_match_the_chatgpt_values(self): - self.assertEqual( - ( - errors.INVALID_PARAMS, - errors.NOT_FOUND, - errors.FORBIDDEN, - errors.RESOURCE_EXHAUSTED, - errors.UNSUPPORTED, - errors.CALLBACK_ENDPOINT, - ), - (-32602, -32011, -32012, -32013, -32014, -32015), - ) - - def test_constructors_carry_spec_data(self): - exhausted = EventsError.resource_exhausted( - "Upgrade to Pro", limit="webhooks", maximum=2 - ) - self.assertEqual(exhausted.code, errors.RESOURCE_EXHAUSTED) - self.assertEqual(exhausted.data, {"limit": "webhooks", "max": 2}) - callback = EventsError.callback_endpoint( - "Callback timed out", reason=errors.CallbackFailure.TIMEOUT - ) - self.assertEqual(callback.code, errors.CALLBACK_ENDPOINT) - self.assertEqual(callback.data, {"reason": "timeout"}) - self.assertEqual(EventsError.forbidden("no").error.data, None) - self.assertEqual(EventsError.unsupported("no").code, errors.UNSUPPORTED) - - def test_callback_reasons(self): - self.assertEqual( - {reason.value for reason in errors.CallbackFailure}, - { - "challenge_failed", - "timeout", - "connection_refused", - "tls_error", - "http_4xx", - "http_5xx", - }, - ) + def test_unknown_event_is_not_found(self): + with self.assertRaises(EventsError) as caught: + lookup("users.user.deleted") + self.assertEqual(caught.exception.code, errors.NOT_FOUND) + self.assertEqual(caught.exception.data, {"kind": "event"}) if __name__ == "__main__": diff --git a/tests/unit/test_events_protocol.py b/tests/unit/test_events_protocol.py index 0ff3039..ab95348 100644 --- a/tests/unit/test_events_protocol.py +++ b/tests/unit/test_events_protocol.py @@ -1,242 +1,24 @@ -"""MCP Events advertisement and ``events/list``, asserted on raw HTTP through the -real hosted Starlette app: the SDK's own client models drop the unknown -``events`` capability, so only the wire shows what clients actually receive.""" +"""Events stay off on stdio. + +Everything else about the protocol layer is covered end to end against the +hosted server in ``tests/e2e/test_events_protocol.py``. The stdio transport +cannot be booted there: it validates an Appwrite API key against a live project +at startup, which only the credentialed integration suite has. +""" -import asyncio -import json import os import unittest -from typing import Any from unittest import mock -from mcp.server import Server -from mcp.server.auth.provider import AccessToken -from starlette.testclient import TestClient - -from mcp_server_appwrite import auth, flags +from mcp_server_appwrite import flags from mcp_server_appwrite.events import protocol -from mcp_server_appwrite.events.catalog import EVENTS -from mcp_server_appwrite.http_app import build_app - -PROTOCOL_VERSION = "2026-07-28" -LEGACY_PROTOCOL_VERSION = "2025-11-25" -EXTENSION = "io.modelcontextprotocol/events" - - -def _request(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]: - return { - "jsonrpc": "2.0", - "id": 1, - "method": method, - "params": { - **(params or {}), - "_meta": { - "io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION, - "io.modelcontextprotocol/clientInfo": {"name": "test", "version": "1"}, - "io.modelcontextprotocol/clientCapabilities": {}, - }, - }, - } - - -def _message(response) -> dict[str, Any]: - """The JSON-RPC message from a JSON or single-event SSE response.""" - text = response.text - if response.headers.get("content-type", "").startswith("text/event-stream"): - data = [line[5:] for line in text.splitlines() if line.startswith("data:")] - text = data[-1] - return json.loads(text) - - -class HostedEventsTests(unittest.TestCase): - HEADERS = { - "Authorization": "Bearer test-token", - "Accept": "application/json, text/event-stream", - "Content-Type": "application/json", - } - - def setUp(self): - from mcp_server_appwrite import server as server_module - - original_transport = server_module._UPLOAD_TRANSPORT - self.addCleanup(setattr, server_module, "_UPLOAD_TRANSPORT", original_transport) - - async def verify_token(_verifier, token): - return AccessToken(token=token, client_id="test-client", scopes=[]) - - patcher = mock.patch.object( - auth.AppwriteTokenVerifier, "verify_token", verify_token - ) - patcher.start() - self.addCleanup(patcher.stop) - - def _flag(self, value: str | None): - environment = dict(os.environ) - environment.pop(flags.EVENTS.env, None) - if value is not None: - environment[flags.EVENTS.env] = value - return mock.patch.dict(os.environ, environment, clear=True) - - def _post(self, client, method: str, params=None, path: str = "/"): - headers = { - **self.HEADERS, - "MCP-Protocol-Version": PROTOCOL_VERSION, - "Mcp-Method": method, - } - response = client.post(path, json=_request(method, params), headers=headers) - return response.status_code, _message(response) - - def _initialize(self, client) -> dict[str, Any]: - body = { - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": LEGACY_PROTOCOL_VERSION, - "capabilities": {}, - "clientInfo": {"name": "test", "version": "1"}, - }, - } - response = client.post("/mcp", json=body, headers=self.HEADERS) - self.assertEqual(response.status_code, 200, response.text) - return _message(response)["result"] - - def test_discover_advertises_both_capability_shapes(self): - with self._flag("1"), TestClient(build_app()) as client: - for path in ("/", "/mcp"): - with self.subTest(path=path): - status, message = self._post(client, "server/discover", path=path) - self.assertEqual(status, 200, message) - capabilities = message["result"]["capabilities"] - self.assertEqual(capabilities["events"], {}) - self.assertEqual( - capabilities["extensions"][EXTENSION], {"listChanged": False} - ) - # The SDK-derived capabilities survive the patch. - self.assertIn("tools", capabilities) - self.assertIn("resources", capabilities) - - def test_legacy_initialize_advertises_events_too(self): - with self._flag("1"), TestClient(build_app()) as client: - capabilities = self._initialize(client)["capabilities"] - self.assertEqual(capabilities["events"], {}) - self.assertEqual(capabilities["extensions"][EXTENSION], {"listChanged": False}) - - def test_events_list_serves_the_catalog(self): - with self._flag("1"), TestClient(build_app()) as client: - status, message = self._post(client, "events/list") - self.assertEqual(status, 200, message) - result = message["result"] - self.assertEqual(result["resultType"], "complete") - self.assertNotIn("nextCursor", result) - self.assertEqual( - [event["name"] for event in result["events"]], - [event.name for event in EVENTS], - ) - for entry, event in zip(result["events"], EVENTS, strict=True): - with self.subTest(event=event.name): - self.assertEqual( - set(entry), - {"name", "description", "delivery", "inputSchema", "payloadSchema"}, - ) - self.assertEqual(entry["delivery"], ["webhook"]) - self.assertEqual(entry["description"], event.description) - self.assertEqual(entry["inputSchema"], event.input_schema) - self.assertEqual(entry["payloadSchema"], event.payload_schema) - - def test_events_list_rejects_a_cursor_it_never_issued(self): - with self._flag("1"), TestClient(build_app()) as client: - status, message = self._post(client, "events/list", {"cursor": "abc"}) - self.assertEqual(status, 400) - self.assertEqual(message["error"]["code"], -32602) - - def test_subscribe_and_unsubscribe_are_not_served_yet(self): - with self._flag("1"), TestClient(build_app()) as client: - for method in ("events/subscribe", "events/unsubscribe"): - with self.subTest(method=method): - _, message = self._post(client, method) - self.assertEqual(message["error"]["code"], -32601) - - def test_nothing_is_advertised_when_the_flag_is_off(self): - for value in (None, "", "0", "false"): - with ( - self.subTest(flag=value), - self._flag(value), - TestClient(build_app()) as client, - ): - status, message = self._post(client, "server/discover") - self.assertEqual(status, 200, message) - capabilities = message["result"]["capabilities"] - self.assertNotIn("events", capabilities) - self.assertNotIn(EXTENSION, capabilities.get("extensions") or {}) - self.assertNotIn("events", self._initialize(client)["capabilities"]) - - _, message = self._post(client, "events/list") - self.assertEqual(message["error"]["code"], -32601) - - -class EnabledTests(unittest.TestCase): - def test_requires_flag_and_http_transport(self): - cases = [ - ("http", "1", True), - ("http", "true", True), - ("http", "ON", True), - ("http", "0", False), - ("http", None, False), - ("stdio", "1", False), - ] - for transport, value, expected in cases: - environment = {} if value is None else {flags.EVENTS.env: value} - with ( - self.subTest(transport=transport, flag=value), - mock.patch.dict(os.environ, environment, clear=True), - ): - self.assertIs(protocol.enabled(transport), expected) - - -class CapabilityMiddlewareTests(unittest.TestCase): - def _run(self, method: str, result: Any) -> Any: - context = mock.Mock(method=method) - - async def call_next(_context): - return result - - return asyncio.run(protocol.CapabilityMiddleware()(context, call_next)) - - def test_keeps_existing_extensions(self): - result = { - "capabilities": { - "tools": {}, - "extensions": {"io.example/other": {"enabled": True}}, - } - } - patched = self._run("server/discover", result) - self.assertEqual( - patched["capabilities"]["extensions"], - { - "io.example/other": {"enabled": True}, - EXTENSION: {"listChanged": False}, - }, - ) - self.assertEqual(patched["capabilities"]["tools"], {}) - # The handler's result is not mutated in place. - self.assertNotIn("events", result["capabilities"]) - def test_other_methods_pass_through(self): - result = {"tools": []} - self.assertIs(self._run("tools/list", result), result) - def test_register_wires_handler_and_middleware(self): - server = Server("test") - protocol.register(server) - self.assertIsNotNone(server.get_request_handler("events/list")) - self.assertIsNone(server.get_request_handler("events/subscribe")) - self.assertTrue( - any( - isinstance(middleware, protocol.CapabilityMiddleware) - for middleware in server.middleware - ) - ) +class StdioTests(unittest.TestCase): + def test_stdio_never_serves_events_even_with_the_flag_on(self): + with mock.patch.dict(os.environ, {flags.EVENTS.env: "1"}): + self.assertFalse(protocol.enabled("stdio")) + self.assertTrue(protocol.enabled("http")) if __name__ == "__main__": diff --git a/uv.lock b/uv.lock index 63050ff..d82a05d 100644 --- a/uv.lock +++ b/uv.lock @@ -700,6 +700,9 @@ dev = [ { name = "pyright" }, { name = "ruff" }, ] +e2e = [ + { name = "jsonschema" }, +] [package.metadata] requires-dist = [ @@ -731,6 +734,7 @@ dev = [ { name = "pyright", specifier = "==1.1.411" }, { name = "ruff", specifier = "==0.15.18" }, ] +e2e = [{ name = "jsonschema", specifier = ">=4.26" }] [[package]] name = "mcp-types" From a2cb6e749e071a78b23239d829b2f80cf251d0e5 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 14:38:01 +0100 Subject: [PATCH 5/5] ci: run the e2e suite on every pull request It needs no credentials, so unlike integration it also runs for forks. --- .github/workflows/ci.yml | 23 +++++++++++++++++++++++ AGENTS.md | 24 +++++++++++++++++++----- docs/development.md | 5 +++++ 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e519809..60e0f2d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,6 +59,29 @@ jobs: - name: Run unit tests run: uv run python -m unittest discover -s tests/unit -v + e2e: + name: E2E + runs-on: ubuntu-latest + steps: + - name: Check out code + uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 + + - name: Set up Python + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 + with: + python-version: "3.12" + + - name: Set up uv + uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0 + with: + version: "0.11.22" + + - name: Install dependencies + run: uv sync --frozen --group e2e + + - name: Run e2e tests + run: uv run --group e2e python -m unittest discover -s tests/e2e -v + docker: name: Docker build runs-on: ubuntu-latest diff --git a/AGENTS.md b/AGENTS.md index 14fcc06..72c5182 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -63,6 +63,7 @@ preview + resource URI. # Install uv, then sync deps uv sync # runtime deps uv sync --group dev # + black, ruff (lint/format) +uv sync --group e2e # + e2e-test deps uv sync --extra integration # + integration-test deps # Run hosted HTTP transport @@ -82,7 +83,7 @@ docker compose up --build # compose.yaml; endpoint at http://localhost:8000/ Run these locally before opening a PR. They mirror the `CI` workflow (`.github/workflows/ci.yml`), which runs on every pull request and on pushes to -`main`. **All four jobs must pass.** +`main`. **All five jobs must pass.** 1. **Lint** (`lint` job) ```bash @@ -112,13 +113,24 @@ Run these locally before opening a PR. They mirror the `CI` workflow ``` Fast, no external services or credentials required. -5. **Docker build** (`docker` job) +5. **E2E tests** (`e2e` job) + ```bash + uv sync --group e2e + uv run --group e2e python -m unittest discover -s tests/e2e -v + ``` + 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 + 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. + +6. **Docker build** (`docker` job) ```bash docker build -t appwrite-mcp:ci . ``` The hosted HTTP image must build cleanly. -6. **Integration tests** (`integration` job) — *CI runs these only for pushes and +7. **Integration tests** (`integration` job) — *CI runs these only for pushes and for PRs from branches on the same repo (not forks).* They create and delete **real** Appwrite resources, so they need live credentials and are skipped when absent: @@ -171,5 +183,7 @@ uv run python scripts/render_server_json.py 0.8.8 tool definitions. - Match existing style: black formatting, ruff-clean imports, type hints, module docstrings explaining intent (see `auth.py`, `http_app.py`, `docs_search.py`). -- Add unit tests under `tests/unit/` for any non-trivial logic; add integration - coverage under `tests/integration/` when touching real API behavior. +- Cover hosted-server behavior with end-to-end flows under `tests/e2e/` (shared + harness in `tests/e2e/support.py`). Add unit tests under `tests/unit/` only for + what no e2e flow can reach. Add integration coverage under + `tests/integration/` when touching real Appwrite API behavior. diff --git a/docs/development.md b/docs/development.md index 66f34bc..1de762e 100644 --- a/docs/development.md +++ b/docs/development.md @@ -58,8 +58,13 @@ APPWRITE_API_KEY= \ | Suite | Command | Needs credentials | | --- | --- | --- | | Unit | `uv run python -m unittest discover -s tests/unit -v` | No | +| E2E | `uv run --group e2e python -m unittest discover -s tests/e2e -v` | No | | Integration | `uv run --extra integration python -m unittest discover -s tests/integration -v` | Yes | +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. + 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.