"""Tool annotations (MCP spec `ToolAnnotations`) for every Helix tool. Clients use these hints to decide how much ceremony a call deserves: ``readOnlyHint=True`` tools can run freely, mutating tools may prompt, and ``destructiveHint=True`` tools should always be confirmed. Helix has a single ground truth for read-vs-write — the ``mutates=`` flag on each tool body's ``tool_runtime(...)`` call (it drives rate-limit class + the Leaver write gate) — so the hints are DERIVED from it rather than hand-maintained on 400+ decorators: * :func:`mutating_tools_from_source` AST-parses ``tools.py`` once at startup and maps ``tool name -> mutates``. Parsing our own shipped source is deliberate: the flag lives inside function bodies, which no registry introspection can reach, and a hand-written mirror table of 400+ rows WILL drift. A tool whose body has no ``tool_runtime`` call (or a dynamic ``mutates=`` value) fails loud at import — that's a bug in the tool, not a case to paper over. * ``DESTRUCTIVE_TOOLS`` curates the truly-irreversible / hand-over / control-bypass subset. It must stay in lockstep with the CLI's ``_DESTRUCTIVE_MCP_TOOLS`` confirmation list (same judgement, second enforcement point) — ``tests/test_mcp/test_tool_annotations.py`` asserts set-equality by parsing the sibling ``cli/`` checkout. * :func:`apply_annotations` stamps a ``ToolAnnotations`` onto every registered tool. Mutating tools get ``idempotentHint=True`` — the platform invariant (CLAUDE.md rule 5) is that every state-changing operation is idempotent — EXCEPT the curated ``NON_IDEMPOTENT_TOOLS`` set, see below. ``openWorldHint=False`` everywhere: Helix is a closed, enumerated domain. """ from __future__ import annotations import ast import logging from functools import lru_cache from pathlib import Path from typing import Final from mcp.types import ToolAnnotations logger = logging.getLogger(__name__) _TOOLS_SOURCE: Final[Path] = Path(__file__).with_name("tools.py") #: Mirror of the CLI's ``_DESTRUCTIVE_MCP_TOOLS`` (cli/helix_cli/main.py). #: Keep the two in sync — the annotations test enforces equality. DESTRUCTIVE_TOOLS: Final[frozenset[str]] = frozenset({ "revoke_api_key", "update_api_key_assignment", "claim_asset", "return_asset", "borrow_asset", "return_borrowed_asset", "mark_asset_held_by_leaver", "recover_asset_from_leaver", "cancel_asset_transfer", "reject_asset_transfer", "admin_cancel_asset_transfer", "delete_my_email_alias", "delete_email_group", "reset_my_email_password", "transfer_email_group_owner", "cancel_my_seat_next_month", "admin_change_seat_plan", "revoke_my_cloud_grant", "transfer_cloud_scope_owner", "withdraw_my_cloud_scope_request", "cancel_my_dns_request", "cancel_my_sso_grant", "cancel_my_domain_registration", "cancel_my_vpn_application", "cancel_my_announcement_request", "cancel_my_ai_tool_application", "delete_sso_app", "create_sso_app", "import_sso_apps_csv", "mark_person_leaver_manual", "revoke_offboarding_blocker", "leaver_blocked_force_bypass_grace", "cancel_onboarding", "skip_onboarding_blocker", "server_destroy", "transfer_my_dns_records", "transfer_my_servers", "revoke_domain_registration", "bypass_4_eyes_dns_request", "recycle_asset_to_office", "release_asset_from_office", "remove_api_key_member", "cancel_my_api_key_release", "cancel_my_api_key_transfer", "execute_provisioning_task", "apply_provisioning_bundle_to_person", "cleanup_cloud_orphan_assignments", "reveal_api_key", "reveal_my_contact", }) #: Mutating tools whose repeat WITH THE SAME ARGUMENTS is not a no-op. #: #: The platform invariant (CLAUDE.md rule 5) holds for state-changing #: operations: replaying them converges on the same state. Model #: generation is the one family where that is false in a way a client #: can act on — ``force=True`` deliberately bypasses the cache, debits #: the caller's daily quota and issues a fresh BILLED model call, so an #: agent that reads ``idempotentHint=True`` and retries on a timeout or #: a 502 burns quota and money for a second narrative it did not want. #: The hint is advertised honestly instead of weakening the invariant #: for the other ~200 mutating tools. #: #: Deliberately a two-name allow-list rather than a heuristic: a tool #: only belongs here when a retry has an external, non-convergent side #: effect (spend, an outbound message duplicate, a counter debit). It #: is NOT "this write is risky" — that judgement is DESTRUCTIVE_TOOLS. NON_IDEMPOTENT_TOOLS: Final[frozenset[str]] = frozenset({ "intelligence_generate_platform_analysis", "cloud_scope_generate_ai_analysis", }) def _tool_name_from_decorator(node: ast.AST) -> str | None: """Return the ``name=`` kwarg of an ``@mcp_server.tool(...)`` decorator.""" if not isinstance(node, ast.Call): return None func = node.func if not ( isinstance(func, ast.Attribute) and func.attr == "tool" and isinstance(func.value, ast.Name) and func.value.id == "mcp_server" ): return None for kw in node.keywords: if kw.arg == "name" and isinstance(kw.value, ast.Constant): return str(kw.value.value) return None def _mutates_from_body(fn: ast.AsyncFunctionDef | ast.FunctionDef) -> bool | None: """Find the ``tool_runtime(...)`` call in ``fn`` and read ``mutates=``. Returns ``None`` when no ``tool_runtime`` call exists — the caller treats that as a hard error so a structurally-novel tool can't silently ship without hints. """ for node in ast.walk(fn): if not isinstance(node, ast.Call): continue callee = node.func name = ( callee.id if isinstance(callee, ast.Name) else callee.attr if isinstance(callee, ast.Attribute) else None ) if name != "tool_runtime": continue for kw in node.keywords: if kw.arg == "mutates": if isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, bool): return kw.value.value # Conditionally-mutating (e.g. list_approval_flow_stages' # ``mutates=bool(refresh)``): annotate conservatively as # mutating — a client that over-confirms is annoying, one # that under-confirms a write is wrong. return True return False # tool_runtime present, mutates omitted → read return None def _called_helper_names(fn: ast.AsyncFunctionDef | ast.FunctionDef) -> list[str]: """Names of module-level helpers this function calls (``_foo(...)``).""" names: list[str] = [] for node in ast.walk(fn): if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): names.append(node.func.id) return names @lru_cache(maxsize=1) def mutating_tools_from_source() -> dict[str, bool]: """Parse ``tools.py`` → ``{tool_name: mutates}`` for every ``@mcp_server.tool``-decorated function. Cached for the process. Tools that delegate to a shared module-level helper (e.g. the ``server_power_on`` family → ``_mcp_run_server_lifecycle``) are resolved by following the delegation up to 3 hops. """ tree = ast.parse(_TOOLS_SOURCE.read_text(encoding="utf-8")) module_fns: dict[str, ast.AsyncFunctionDef | ast.FunctionDef] = { n.name: n for n in tree.body if isinstance(n, (ast.AsyncFunctionDef, ast.FunctionDef)) } def resolve(fn: ast.AsyncFunctionDef | ast.FunctionDef, depth: int) -> bool | None: direct = _mutates_from_body(fn) if direct is not None: return direct if depth <= 0: return None for helper in _called_helper_names(fn): target = module_fns.get(helper) if target is None: continue resolved = resolve(target, depth - 1) if resolved is not None: return resolved return None out: dict[str, bool] = {} for node in ast.walk(tree): if not isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)): continue tool_name = next( (n for d in node.decorator_list if (n := _tool_name_from_decorator(d))), None, ) if tool_name is None: continue mutates = resolve(node, 3) if mutates is None: raise ValueError( f"tool {tool_name!r} has no reachable tool_runtime(...) call — " "cannot derive its read/write annotation; add the call or " "extend app/mcp/annotations.py for the new pattern" ) out[tool_name] = mutates return out def annotations_for(name: str, *, mutates: bool) -> ToolAnnotations: if not mutates: return ToolAnnotations(readOnlyHint=True, openWorldHint=False) return ToolAnnotations( readOnlyHint=False, destructiveHint=name in DESTRUCTIVE_TOOLS, # Platform invariant: every state-changing operation is # idempotent (CLAUDE.md non-negotiable rule 5) — except the # curated cost-bearing set, see NON_IDEMPOTENT_TOOLS. idempotentHint=name not in NON_IDEMPOTENT_TOOLS, openWorldHint=False, ) def apply_annotations(mcp_server) -> int: """Stamp ToolAnnotations onto every registered tool; returns count. Called once from ``build_mcp_app()`` after ``app.mcp.tools`` import. Fails loud on a tool that exists in the registry but not in the parsed source (impossible unless registration grows a second path — in which case the hints MUST learn about it). """ flags = mutating_tools_from_source() tools = mcp_server._tool_manager._tools missing = sorted(set(tools) - set(flags)) if missing: raise RuntimeError( f"tools registered outside tools.py have no annotation source: {missing}" ) for name, tool in tools.items(): tool.annotations = annotations_for(name, mutates=flags[name]) logger.info("mcp: applied ToolAnnotations to %d tools", len(tools)) return len(tools) __all__ = [ "DESTRUCTIVE_TOOLS", "NON_IDEMPOTENT_TOOLS", "annotations_for", "apply_annotations", "mutating_tools_from_source", ]