From e6495d25116c13b7a3abd699cd0759172440da08 Mon Sep 17 00:00:00 2001 From: tangkikodo Date: Tue, 13 Oct 2026 14:45:07 +0800 Subject: [PATCH 1/2] feat(mcp): annotate tools with readOnlyHint/destructiveHint (AI mutation safety P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MCP annotations are the hint layer of AI mutation safety: agents and clients read them from tools/list for UI routing and planning caution (the spec marks them untrusted hints — the contract layer is a separate concern). Rules applied on all three server surfaces (entity single/multi-app, UseCase compose): - discovery/introspection tools → readOnlyHint - execution tools (graphql_query / compose_query) → readOnlyHint only when NO reachable schema exposes a Mutation operation — computed from schema content, not from the allow_mutation flag, because the flag gates tool advertising while the handler still executes whatever the schema contains - graphql_mutation → destructiveHint Verified end to end that the annotations reach tools/list. Co-Authored-By: Claude Code --- src/nexusx/mcp/annotations.py | 34 ++++ src/nexusx/mcp/tools/multi_app_tools.py | 25 ++- src/nexusx/mcp/tools/simple_tools.py | 13 +- src/nexusx/use_case/compose_mcp_server.py | 19 +- tests/mcp/test_tool_annotations.py | 220 ++++++++++++++++++++++ 5 files changed, 297 insertions(+), 14 deletions(-) create mode 100644 src/nexusx/mcp/annotations.py create mode 100644 tests/mcp/test_tool_annotations.py diff --git a/src/nexusx/mcp/annotations.py b/src/nexusx/mcp/annotations.py new file mode 100644 index 00000000..f615acd4 --- /dev/null +++ b/src/nexusx/mcp/annotations.py @@ -0,0 +1,34 @@ +"""MCP tool annotations — the hint layer of AI mutation safety. + +Per the MCP spec, annotations are **untrusted hints**: clients use them for +UI routing and model-side caution (e.g. an agent planning around a +``destructiveHint: true`` tool), never as a security boundary. The contract +layer (server-side gates) is a separate concern. + +Rules applied across nexusx MCP servers: + +- Discovery/introspection tools are read-only — they cannot change data. +- A tool that executes arbitrary GraphQL (``graphql_query`` / + ``compose_query``) is only read-only when **no reachable schema exposes a + Mutation operation**. The ``allow_mutation`` flag gates tool *registration* + (whether mutation tooling is advertised), not the handler — an execution + tool pointed at a schema with mutations can still run them, so the + annotation is computed from schema content, not from the flag. +- Mutation-executing tools carry ``destructiveHint``. +""" + +from __future__ import annotations + +from mcp.types import ToolAnnotations + +__all__ = ["destructive", "read_only"] + + +def read_only() -> ToolAnnotations: + """Annotations for tools that cannot change data.""" + return ToolAnnotations(read_only_hint=True) + + +def destructive() -> ToolAnnotations: + """Annotations for tools that may change or destroy data.""" + return ToolAnnotations(read_only_hint=False, destructive_hint=True) diff --git a/src/nexusx/mcp/tools/multi_app_tools.py b/src/nexusx/mcp/tools/multi_app_tools.py index ac3c89c9..8422bd2e 100644 --- a/src/nexusx/mcp/tools/multi_app_tools.py +++ b/src/nexusx/mcp/tools/multi_app_tools.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any +from nexusx.mcp.annotations import destructive, read_only from nexusx.mcp.types.errors import ( MCPErrors, create_error_response, @@ -32,8 +33,18 @@ def register_multi_app_tools( in list_apps. Default is False (read-only mode). """ + # AI mutation safety (hint layer): graphql_query executes arbitrary + # GraphQL, so its read-only annotation must reflect whether any reachable + # schema actually exposes a Mutation operation — not the allow_mutation + # flag, which only gates which tools are advertised. + has_mutations = any( + app.tracer.list_group_operations("Mutation") + for app in manager.apps.values() + ) + exec_annotations = destructive() if has_mutations else read_only() + # Layer 0: Application discovery - @mcp.tool() + @mcp.tool(annotations=read_only()) def list_apps() -> dict[str, Any]: """List all available applications. @@ -99,7 +110,7 @@ def list_apps() -> dict[str, Any]: return create_error_response(str(e), MCPErrors.INTERNAL_ERROR) # Layer 1: List operations - @mcp.tool() + @mcp.tool(annotations=read_only()) def list_queries(app_name: str) -> dict[str, Any]: """List all available GraphQL queries for a specific application. @@ -139,7 +150,7 @@ def list_queries(app_name: str) -> dict[str, Any]: if allow_mutation: - @mcp.tool() + @mcp.tool(annotations=read_only()) def list_mutations(app_name: str) -> dict[str, Any]: """List all available GraphQL mutations for a specific application. @@ -178,7 +189,7 @@ def list_mutations(app_name: str) -> dict[str, Any]: return create_error_response(str(e), MCPErrors.INTERNAL_ERROR) # Layer 2: Get operation schema - @mcp.tool() + @mcp.tool(annotations=read_only()) def get_query_schema( entity: str, method: str, app_name: str, response_type: str = "sdl" ) -> dict[str, Any]: @@ -258,7 +269,7 @@ def get_query_schema( if allow_mutation: - @mcp.tool() + @mcp.tool(annotations=read_only()) def get_mutation_schema( entity: str, method: str, app_name: str, response_type: str = "sdl" ) -> dict[str, Any]: @@ -336,7 +347,7 @@ def get_mutation_schema( return create_error_response(str(e), MCPErrors.INTERNAL_ERROR) # Layer 3: Execute operations - @mcp.tool() + @mcp.tool(annotations=exec_annotations) async def graphql_query(query: str, app_name: str) -> dict[str, Any]: """Execute a GraphQL query on a specific application. @@ -406,7 +417,7 @@ async def graphql_query(query: str, app_name: str) -> dict[str, Any]: if allow_mutation: - @mcp.tool() + @mcp.tool(annotations=destructive()) async def graphql_mutation(mutation: str, app_name: str) -> dict[str, Any]: """Execute a GraphQL mutation on a specific application. diff --git a/src/nexusx/mcp/tools/simple_tools.py b/src/nexusx/mcp/tools/simple_tools.py index 9a97150a..38474e9c 100644 --- a/src/nexusx/mcp/tools/simple_tools.py +++ b/src/nexusx/mcp/tools/simple_tools.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any +from nexusx.mcp.annotations import destructive, read_only from nexusx.mcp.types.errors import ( MCPErrors, create_error_response, @@ -36,7 +37,13 @@ def register_simple_tools( Mutation type in schema. Default is False (read-only mode). """ - @mcp.tool() + # AI mutation safety (hint layer): graphql_query executes arbitrary + # GraphQL against the handler; read-only only when the schema has no + # Mutation operation (allow_mutation gates advertising, not execution). + has_mutations = bool(manager.tracer.list_group_operations("Mutation")) + exec_annotations = destructive() if has_mutations else read_only() + + @mcp.tool(annotations=read_only()) def get_schema() -> dict[str, Any]: """Get the complete GraphQL schema in SDL format. @@ -79,7 +86,7 @@ def get_schema() -> dict[str, Any]: except Exception as e: return create_error_response(str(e), MCPErrors.INTERNAL_ERROR) - @mcp.tool() + @mcp.tool(annotations=exec_annotations) async def graphql_query(query: str) -> dict[str, Any]: """Execute a GraphQL query. @@ -131,7 +138,7 @@ async def graphql_query(query: str) -> dict[str, Any]: if allow_mutation: - @mcp.tool() + @mcp.tool(annotations=destructive()) async def graphql_mutation(mutation: str) -> dict[str, Any]: """Execute a GraphQL mutation. diff --git a/src/nexusx/use_case/compose_mcp_server.py b/src/nexusx/use_case/compose_mcp_server.py index 8dc19d4e..a6c49dce 100644 --- a/src/nexusx/use_case/compose_mcp_server.py +++ b/src/nexusx/use_case/compose_mcp_server.py @@ -27,6 +27,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any +from nexusx.mcp.annotations import destructive, read_only from nexusx.mcp.types.errors import ( MCPErrors, create_error_response, @@ -109,7 +110,7 @@ def _get_app(registry: dict[str, _ComposeAppEntry], name: str) -> _ComposeAppEnt def _register_list_apps(mcp: FastMCP, registry: dict[str, _ComposeAppEntry]) -> None: - @mcp.tool() + @mcp.tool(annotations=read_only()) def list_apps() -> dict[str, Any]: """List all available UseCase applications registered on this server. @@ -141,7 +142,7 @@ def list_apps() -> dict[str, Any]: def _register_describe_compose_schema( mcp: FastMCP, registry: dict[str, _ComposeAppEntry] ) -> None: - @mcp.tool() + @mcp.tool(annotations=read_only()) def describe_compose_schema(app_name: str) -> dict[str, Any]: """List services and methods for an app. @@ -199,7 +200,7 @@ def describe_compose_schema(app_name: str) -> dict[str, Any]: def _register_describe_compose_method( mcp: FastMCP, registry: dict[str, _ComposeAppEntry] ) -> None: - @mcp.tool() + @mcp.tool(annotations=read_only()) def describe_compose_method( app_name: str, service_name: str, @@ -285,7 +286,17 @@ def _arg_to_dict(a: Any) -> dict[str, Any]: def _register_compose_query( mcp: FastMCP, registry: dict[str, _ComposeAppEntry] ) -> None: - @mcp.tool() + # AI mutation safety (hint layer): compose_query executes methods by name, + # query or mutation alike — read-only only when no service exposes a + # mutation method (a `{Service}Mutation` entry in the compose registry). + has_mutations = any( + entry.schema.registry.get(f"{service_cls.__name__}Mutation") is not None + for entry in registry.values() + for service_cls in entry.config.services + ) + exec_annotations = destructive() if has_mutations else read_only() + + @mcp.tool(annotations=exec_annotations) async def compose_query( app_name: str, query: str, diff --git a/tests/mcp/test_tool_annotations.py b/tests/mcp/test_tool_annotations.py new file mode 100644 index 00000000..b83f4c55 --- /dev/null +++ b/tests/mcp/test_tool_annotations.py @@ -0,0 +1,220 @@ +"""MCP tool annotations — the hint layer of AI mutation safety. + +Locks the annotation rules across all three server surfaces: +- discovery tools are read-only; +- an execution tool (graphql_query / compose_query) is read-only ONLY when + no reachable schema exposes a Mutation operation — computed from schema + content, NOT from the allow_mutation flag (the flag gates advertising, + while the handler still executes whatever the schema contains); +- mutation tools carry destructiveHint. +""" + +from __future__ import annotations + +import pytest +from pydantic import BaseModel +from sqlmodel import Field, SQLModel + +from nexusx import query +from nexusx.decorator import mutation +from nexusx.use_case.business import UseCaseService +from nexusx.use_case.compose_mcp_server import create_use_case_graphql_mcp_server +from nexusx.use_case.types import UseCaseAppConfig + + +def _get_tools_dict(mcp): + """Get tools as dict {name: tool} from FastMCP (compatible with fastmcp 3.x).""" + components = mcp._local_provider._components + return { + key.split(":")[1].split("@")[0]: value + for key, value in components.items() + if key.startswith("tool:") + } + + +def _hints(mcp, name: str) -> tuple[bool | None, bool | None]: + tool = _get_tools_dict(mcp)[name] + ann = tool.annotations + return (ann.read_only_hint, ann.destructive_hint) + + +# ── entity-side fixtures ──────────────────────────────────────────────── + + +class _AnnBase(SQLModel): + pass + + +class _AnnBothEntity(_AnnBase, table=True): + """Has a @mutation method → its schema exposes a Mutation operation.""" + + __tablename__ = "ann_both_entity" + __test__ = False + + id: int | None = Field(default=None, primary_key=True) + + @query + async def get_all(cls) -> list[_AnnBothEntity]: + return [] + + @mutation + async def create(cls) -> _AnnBothEntity: + return _AnnBothEntity(id=1) + + +class _AnnQueryOnlyBase(SQLModel): + pass + + +class _AnnQueryOnlyEntity(_AnnQueryOnlyBase, table=True): + """@query only → schema has no Mutation operation.""" + + __tablename__ = "ann_query_only_entity" + __test__ = False + + id: int | None = Field(default=None, primary_key=True) + + @query + async def get_all(cls) -> list[_AnnQueryOnlyEntity]: + return [] + + +# ── compose-side fixtures ─────────────────────────────────────────────── + + +class _AnnItem(BaseModel): + id: int + + +class _AnnQueryService(UseCaseService): + @query + async def list_items(cls) -> list[_AnnItem]: + return [_AnnItem(id=1)] + + +class _AnnMixedService(UseCaseService): + @query + async def list_items(cls) -> list[_AnnItem]: + return [] + + @mutation + async def create_item(cls) -> _AnnItem: + return _AnnItem(id=99) + + +# ── entity single-app server ──────────────────────────────────────────── + + +class TestSimpleServerAnnotations: + @pytest.fixture + def mutation_app_mcp(self): + from fastmcp import FastMCP + + from nexusx.mcp.managers.single_app_manager import SingleAppManager + from nexusx.mcp.tools.simple_tools import register_simple_tools + + mcp = FastMCP("ann-simple") + manager = SingleAppManager(base=_AnnBase) + register_simple_tools(mcp, manager, allow_mutation=False) + return mcp + + @pytest.fixture + def query_only_mcp(self): + from fastmcp import FastMCP + + from nexusx.mcp.managers.single_app_manager import SingleAppManager + from nexusx.mcp.tools.simple_tools import register_simple_tools + + mcp = FastMCP("ann-simple-qo") + manager = SingleAppManager(base=_AnnQueryOnlyBase) + register_simple_tools(mcp, manager, allow_mutation=False) + return mcp + + def test_discovery_tools_are_read_only(self, mutation_app_mcp): + assert _hints(mutation_app_mcp, "get_schema") == (True, None) + + def test_execution_tool_follows_schema_content_not_flag(self, mutation_app_mcp): + """App HAS a mutation method but allow_mutation=False: graphql_query + still executes arbitrary GraphQL against a schema with a Mutation + operation → must NOT claim read-only (content over flag).""" + ro, de = _hints(mutation_app_mcp, "graphql_query") + assert ro is False + assert de is True + + def test_query_only_app_execution_tool_is_read_only(self, query_only_mcp): + assert _hints(query_only_mcp, "graphql_query") == (True, None) + + +# ── entity multi-app server ───────────────────────────────────────────── + + +class TestMultiAppServerAnnotations: + def _build(self, base, allow_mutation: bool): + from nexusx.mcp import Application + from nexusx.mcp.server import create_multi_app_mcp_server + + app = Application(name="ann-app", base=base) + return create_multi_app_mcp_server( + apps=[app], name="ann-multi", allow_mutation=allow_mutation + ) + + def test_discovery_tools_are_read_only(self): + mcp = self._build(_AnnBase, allow_mutation=False) + for name in ("list_apps", "list_queries", "get_query_schema"): + assert _hints(mcp, name) == (True, None), name + + def test_execution_tool_follows_schema_content_not_flag(self): + mcp = self._build(_AnnBase, allow_mutation=False) + ro, de = _hints(mcp, "graphql_query") + assert ro is False + assert de is True + + def test_mutation_tools_when_advertised(self): + mcp = self._build(_AnnBase, allow_mutation=True) + assert _hints(mcp, "graphql_mutation") == (False, True) + assert _hints(mcp, "graphql_query") == (False, True) + for name in ("list_mutations", "get_mutation_schema"): + assert _hints(mcp, name) == (True, None), name + + def test_query_only_apps_execution_tool_is_read_only(self): + mcp = self._build(_AnnQueryOnlyBase, allow_mutation=False) + assert _hints(mcp, "graphql_query") == (True, None) + + +# ── compose server ────────────────────────────────────────────────────── + + +class TestComposeServerAnnotations: + def test_discovery_tools_are_read_only(self): + mcp = create_use_case_graphql_mcp_server( + apps=[ + UseCaseAppConfig( + name="annq", services=[_AnnQueryService], description="q" + ) + ], + name="ann-compose", + ) + for name in ("list_apps", "describe_compose_schema", "describe_compose_method"): + assert _hints(mcp, name) == (True, None), name + + def test_query_only_services_execution_tool_is_read_only(self): + mcp = create_use_case_graphql_mcp_server( + apps=[ + UseCaseAppConfig( + name="annq", services=[_AnnQueryService], description="q" + ) + ], + name="ann-compose-qo", + ) + assert _hints(mcp, "compose_query") == (True, None) + + def test_mutation_service_makes_execution_tool_destructive(self): + mcp = create_use_case_graphql_mcp_server( + apps=[ + UseCaseAppConfig( + name="annm", services=[_AnnMixedService], description="m" + ) + ], + name="ann-compose-m", + ) + assert _hints(mcp, "compose_query") == (False, True) From c149840ed498036bb571ac985372b3a4d669912d Mon Sep 17 00:00:00 2001 From: tangkikodo Date: Wed, 14 Oct 2026 11:24:54 +0800 Subject: [PATCH 2/2] feat(mcp): gate compose mutations behind human confirmation (AI mutation safety P2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mutation confirmation is the contract layer of AI mutation safety: the annotations layer (P1) advertises risk as untrusted hints, while this gate enforces a human decision before a mutation executes. create_use_case_graphql_mcp_server gains a confirm_mutations flag (default False). When on, compose_query extracts top-level mutation fields (Service.method form) and asks the human: - handshake-era connections: synchronous ctx.elicit() round-trip; - 2026-07-28 connections (server-initiated elicitation unavailable): the SEP-2322 guard pattern — the ask is an InputRequiredResult (a legitimate result, not an error); the client re-issues the identical call with the answer, read back via ctx.input_responses. Declined mutations return an errors payload without executing. Queries are never gated; parse-broken mutations pass through so the executor surfaces the real parse error. Clients that did not declare elicitation capability pass ungated — fail-open, demo scope; a spec'd contract layer fails closed (constant request_state token, no nonce yet). Co-Authored-By: Claude Code --- src/nexusx/use_case/compose_mcp_server.py | 166 +++++++++++++++++++++- tests/mcp/test_mutation_confirmation.py | 127 +++++++++++++++++ 2 files changed, 291 insertions(+), 2 deletions(-) create mode 100644 tests/mcp/test_mutation_confirmation.py diff --git a/src/nexusx/use_case/compose_mcp_server.py b/src/nexusx/use_case/compose_mcp_server.py index a6c49dce..0b6d8bea 100644 --- a/src/nexusx/use_case/compose_mcp_server.py +++ b/src/nexusx/use_case/compose_mcp_server.py @@ -27,6 +27,17 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any +from fastmcp import Context +from fastmcp.exceptions import ToolError +from fastmcp.server.elicitation import AcceptedElicitation +from graphql import FieldNode, OperationDefinitionNode, OperationType, parse +from mcp.types import ( + ElicitRequest, + ElicitRequestFormParams, + ElicitResult, + InputRequiredResult, +) + from nexusx.mcp.annotations import destructive, read_only from nexusx.mcp.types.errors import ( MCPErrors, @@ -58,6 +69,8 @@ class _ComposeAppEntry: def create_use_case_graphql_mcp_server( apps: list[UseCaseAppConfig], name: str = "nexusx UseCase GraphQL API", + *, + confirm_mutations: bool = False, ) -> FastMCP: """Create a 4-layer progressive-disclosure MCP server. @@ -66,6 +79,14 @@ def create_use_case_graphql_mcp_server( built eagerly at this call, so schema errors surface here (not at query time). name: MCP server name (shown in protocol handshake). + confirm_mutations: Contract layer of AI mutation safety (experimental). + When True, ``compose_query`` gates mutations behind a human + confirmation via the SEP-2322 guard pattern: the first round + returns an ``InputRequiredResult`` (an ask, not an error); the + client re-calls with the human's answer, and only ``accept`` + executes. Clients that did not declare elicitation capability + (embedded agents with empty capabilities) pass through ungated — + fail-open, demo scope; a spec'd contract layer fails closed. Returns: ``FastMCP`` instance. @@ -89,7 +110,7 @@ def create_use_case_graphql_mcp_server( _register_list_apps(mcp, registry) _register_describe_compose_schema(mcp, registry) _register_describe_compose_method(mcp, registry) - _register_compose_query(mcp, registry) + _register_compose_query(mcp, registry, confirm_mutations=confirm_mutations) return mcp @@ -282,9 +303,79 @@ def _arg_to_dict(a: Any) -> dict[str, Any]: # Layer 3 — compose_query # --------------------------------------------------------------------------- +# Guard-pattern constants (SEP-2322). ``request_state`` is an opaque +# correlation token the client echoes back; a constant is enough for the +# demo — a spec'd contract layer would mint a nonce (anti-replay). +_CONFIRM_KEY = "confirm" +_CONFIRM_STATE = "mutation-confirm:v1" + + +def mutation_fields(query: str) -> list[str]: + """Top-level mutation fields in ``Service.method`` form ([] if none). + + Parse errors yield ``[]`` — the executor surfaces the real parse error + with full context; the confirmation gate must not shadow it with its own. + """ + try: + doc = parse(query) + except Exception: + return [] + found: list[str] = [] + for defn in doc.definitions: + if ( + not isinstance(defn, OperationDefinitionNode) + or defn.operation != OperationType.MUTATION + ): + continue + for service_sel in defn.selection_set.selections: + if not isinstance(service_sel, FieldNode): + continue + if service_sel.selection_set is None: + found.append(service_sel.name.value) + continue + for method_sel in service_sel.selection_set.selections: + if isinstance(method_sel, FieldNode): + found.append(f"{service_sel.name.value}.{method_sel.name.value}") + return found + + +def _client_can_confirm(ctx: Context | None) -> bool: + """True when the connected client declared elicitation capability.""" + if ctx is None: + return False + try: + caps = ctx.session.client_capabilities + except Exception: + return False + return getattr(caps, "elicitation", None) is not None + + +def _confirm_verdict(ctx: Context | None) -> bool | None: + """Read the human's answer from a prior ask leg. + + True — confirmed (accept + ``value`` true); False — declined, cancelled, + or answered false; None — no answer yet (first round, or a retry without + responses — ask again). + """ + if ctx is None: + return None + responses = ctx.input_responses + if not responses: + return None + result = responses.get(_CONFIRM_KEY) + if not isinstance(result, ElicitResult): + return None + if result.action != "accept": + return False + content = result.content or {} + return content.get("value") is True + def _register_compose_query( - mcp: FastMCP, registry: dict[str, _ComposeAppEntry] + mcp: FastMCP, + registry: dict[str, _ComposeAppEntry], + *, + confirm_mutations: bool, ) -> None: # AI mutation safety (hint layer): compose_query executes methods by name, # query or mutation alike — read-only only when no service exposes a @@ -300,6 +391,7 @@ def _register_compose_query( async def compose_query( app_name: str, query: str, + ctx: Context, variables: dict[str, Any] | None = None, ) -> dict[str, Any]: """Execute a GraphQL query against an app's UseCase compose schema. @@ -310,6 +402,11 @@ async def compose_query( rejected — use ``describe_compose_schema`` and ``describe_compose_method`` for schema discovery. + When the server runs with mutation confirmation, a mutation call + first returns an input-required ask naming the mutation fields + (``Service.method``). Surface it to the human; re-issue the identical + call with the human's answer to execute (accept) or abort (decline). + Pass string arguments via ``variables`` — never inline them as GraphQL string literals. Inline strings containing quotes (``"``), backslashes or newlines produce unparseable queries; ``variables`` sidesteps all @@ -335,6 +432,71 @@ async def compose_query( }], } + # AI mutation safety (contract layer): a mutation from a + # confirmation-capable client must be confirmed by the human behind + # it. Two ask mechanisms, by connection era: + # - handshake-era connections (pre 2026-07-28): server-initiated + # ``ctx.elicit()`` — a synchronous round-trip; + # - modern connections: elicit raises ToolError → the SEP-2322 guard + # pattern (the ask is the legitimate result of this leg; the + # continuation round carries the answer in ctx.input_responses). + if confirm_mutations: + targets = mutation_fields(query) + if targets and _client_can_confirm(ctx): + message = ( + "Confirm mutation: " + ", ".join(targets) + + ". Decline to abort without executing." + ) + try: + answer = await ctx.elicit(message, response_type=bool) + confirmed = ( + isinstance(answer, AcceptedElicitation) + and answer.data is True + ) + if not confirmed: + return { + "data": None, + "errors": [{ + "message": ( + "Mutation declined by user: " + + ", ".join(targets) + ), + }], + } + except ToolError: + verdict = _confirm_verdict(ctx) + if verdict is None: + return InputRequiredResult( + input_requests={ + _CONFIRM_KEY: ElicitRequest( + params=ElicitRequestFormParams( + message=message, + requested_schema={ + "type": "object", + "properties": { + "value": { + "type": "boolean", + "title": "Confirm", + } + }, + "required": ["value"], + }, + ) + ) + }, + request_state=_CONFIRM_STATE, + ) + if verdict is False: + return { + "data": None, + "errors": [{ + "message": ( + "Mutation declined by user: " + + ", ".join(targets) + ), + }], + } + context: dict[str, Any] = {} if entry.config.context_extractor is not None: try: diff --git a/tests/mcp/test_mutation_confirmation.py b/tests/mcp/test_mutation_confirmation.py new file mode 100644 index 00000000..9ceb4b55 --- /dev/null +++ b/tests/mcp/test_mutation_confirmation.py @@ -0,0 +1,127 @@ +"""Mutation confirmation — the contract layer of AI mutation safety. + +Drives the SEP-2322 guard pattern through the fastmcp in-memory client: +- leg 1: ``compose_query`` with a mutation returns an ``InputRequiredResult`` + ask (the elicitation handler fires exactly once — the ask is the result of + the leg, not an error); +- leg 2: the client re-calls with the human's answer — + accept + true executes the mutation, decline aborts it; +- a client without an elicitation handler (no capability declared) passes + through ungated — fail-open, demo scope; +- queries are never gated, and parse-broken mutations are not gated either + (the executor owns surfacing parse errors). +""" + +from __future__ import annotations + +import json +from typing import Any + +from fastmcp import Client +from fastmcp.client.elicitation import ElicitResult +from pydantic import BaseModel + +from nexusx.decorator import mutation, query +from nexusx.use_case.business import UseCaseService +from nexusx.use_case.compose_mcp_server import create_use_case_graphql_mcp_server +from nexusx.use_case.types import UseCaseAppConfig + + +class _ConfItem(BaseModel): + done: bool + + +class _ConfService(UseCaseService): + """Mutation records invocations so tests can prove execution vs abort.""" + + calls: list[str] = [] + + @query + async def peek(cls) -> list[str]: + return list(_ConfService.calls) + + @mutation + async def do_thing(cls) -> _ConfItem: + _ConfService.calls.append("do_thing") + return _ConfItem(done=True) + + +def _server(*, confirm: bool): + _ConfService.calls = [] + return create_use_case_graphql_mcp_server( + apps=[UseCaseAppConfig(name="conf", services=[_ConfService])], + name="conf-test", + confirm_mutations=confirm, + ) + + +_MUTATION = "mutation { _ConfService { do_thing { done } } }" +_QUERY = "query { _ConfService { peek } }" + + +async def _call_tool(client: Client, query: str) -> dict[str, Any]: + res = await client.call_tool( + "compose_query", {"app_name": "conf", "query": query} + ) + return json.loads(res.content[0].text) + + +class TestGuardRounds: + async def test_accept_executes_mutation(self): + mcp = _server(confirm=True) + asked: list[str] = [] + + async def answer(message, response_type, params, context): + asked.append(message) + return True # auto-wrapped: accept + {"value": true} + + async with Client(mcp, elicitation_handler=answer) as client: + data = await _call_tool(client, _MUTATION) + + assert asked == [f"Confirm mutation: _ConfService.do_thing. " + "Decline to abort without executing."] + assert data["data"]["_ConfService"]["do_thing"]["done"] is True + assert _ConfService.calls == ["do_thing"] + + async def test_decline_aborts_mutation(self): + mcp = _server(confirm=True) + + async def answer(message, response_type, params, context): + return ElicitResult(action="decline") + + async with Client(mcp, elicitation_handler=answer) as client: + data = await _call_tool(client, _MUTATION) + + assert data["data"] is None + assert "declined" in data["errors"][0]["message"] + assert _ConfService.calls == [] # never executed + + async def test_query_never_gated(self): + mcp = _server(confirm=True) + + async def fail_if_asked(message, response_type, params, context): + raise AssertionError("queries must not elicit") + + async with Client(mcp, elicitation_handler=fail_if_asked) as client: + data = await _call_tool(client, _QUERY) + + assert data["data"]["_ConfService"]["peek"] == [] + + async def test_client_without_capability_passes_ungated(self): + mcp = _server(confirm=True) + async with Client(mcp) as client: # no handler → no elicitation capability + data = await _call_tool(client, _MUTATION) + + assert data["data"]["_ConfService"]["do_thing"]["done"] is True + assert _ConfService.calls == ["do_thing"] + + async def test_disabled_flag_never_gates(self): + mcp = _server(confirm=False) + + async def fail_if_asked(message, response_type, params, context): + raise AssertionError("confirm_mutations=False must not elicit") + + async with Client(mcp, elicitation_handler=fail_if_asked) as client: + data = await _call_tool(client, _MUTATION) + + assert data["data"]["_ConfService"]["do_thing"]["done"] is True