AI Assistant Command MCP

AI Assistant — Command Dispatch → MCP Bridge (Layer 1)

The bridge from application operations (commands) to the agent's tool surface (MCP, multi-provider tool calls, OpenAPI, CLI). The unique value of this layer: one annotation pass, many surfaces, no drift.

The decision in one sentence

Treat the command dispatch registry as the single source of truth. Emit FastMCP (Python) or @modelcontextprotocol/sdk (TS) tool servers from it. Annotate every command with side-effect class, scopes, idempotency, and approval policy. JSON Schema is the universal pivot.

Audit: what to look for

# Existing command dispatch (any variant)
grep -rE "@command\(|class.*CommandSpec|register_command|@app\.tool" .
grep -rE "from acture|from wrapex|import py2mcp" .
ls commands/ command_registry/ src/commands/ 2>/dev/null

# Existing MCP server
grep -rE "from fastmcp|FastMCP\(|@modelcontextprotocol/sdk" .
ls mcp/ mcp_server/ 2>/dev/null

# Frontend command palette (cmdk/kbar) — also signal of command dispatch
grep -rE "cmdk|kbar|useCommand|registerCommand" .

# OpenAPI / FastAPI routes — could be the source if no command registry exists
grep -rE "@app\.(get|post|put|delete)" .

Audit verdicts:

• Command dispatch present + MCP absent. Build the MCP emitter (Section: Emitter A below). Easiest path.
• Command dispatch present + MCP present + duplicated definitions. REFACTOR: drive MCP from the registry; delete the duplicates.
• No command dispatch + ad-hoc @mcp.tool decorators. Acceptable for ≤ 10 tools. Annotate them with the metadata taxonomy below so migration to a command registry later is mechanical.
• No command dispatch + > 10 tools or > 1 surface (CLI + MCP, MCP + REST). Add a command registry. Strangler-fig pattern: wrap existing handlers without rewriting them.
• py2mcp present. This is Thor's Python→MCP generator. Prefer it where it covers the annotation taxonomy; supplement with FastMCP for anything it doesn't yet handle. Do NOT bypass py2mcp by hand-rolling FastMCP code in parallel — extend py2mcp instead.
• OpenAPI present (FastAPI app) but no command registry. Use FastMCP.from_openapi() for a prototype, but plan to migrate to a command registry — auto-generated MCP servers from OpenAPI underperform hand-curated ones on complex APIs (FastMCP's own docs caution this).

The minimum-viable command (canonical form)

# commands/data.py
from pydantic import BaseModel, Field
from typing import Literal
from command_registry.core import command


class ApplyFilterParams(BaseModel):
    """Filter the active dataset by a column condition."""
    column: str = Field(description="Column name to filter on")
    operator: Literal["=", "!=", ">", "<", ">=", "<="]
    value: str | float


@command(
    id="app.data.applyFilter",
    summary="Apply Filter",
    schema=ApplyFilterParams,
    side_effect="additive",          # query | additive | destructive
    idempotent=True,
    open_world=False,
    requires_confirmation=False,
    scopes=("data:read", "data:write"),
    tags=("data",),
    keywords=("filter", "where", "query", "predicate"),
    examples=({"column": "country", "operator": "=", "value": "France"},),
)
def apply_filter(params: ApplyFilterParams) -> dict:
    """Filter the active dataset by a column condition.

    Use when the user wants to narrow visible rows by a single column predicate.
    For multi-column or compound predicates, use applyFilters instead.
    """
    # ... implementation ...
    return {"ok": True, "rowsRemaining": 12_345}

The CommandSpec carries everything every emitter needs.

The full annotation taxonomy

This taxonomy is intentionally compatible with the Anthropic Skills frontmatter (see ai-assistant-prompts-skills). A command with surfaces=("mcp", "palette") and a frontmatter-bearing COMMAND.md doc file is one git-tracked artifact serving every surface.

Emitter A: MCP server (FastMCP)

# emitters/mcp_emitter.py
from fastmcp import FastMCP
from mcp.types import ToolAnnotations
from command_registry.core import registry


def build_mcp_server(name: str = "AppMCP") -> FastMCP:
    mcp = FastMCP(name)
    for spec in registry().values():
        if "mcp" not in spec.surfaces:
            continue
        annotations = ToolAnnotations(
            title=spec.summary,
            readOnlyHint=(spec.side_effect == "query"),
            destructiveHint=(spec.side_effect == "destructive"),
            idempotentHint=spec.idempotent,
            openWorldHint=spec.open_world,
        )
        @mcp.tool(
            name=spec.id,
            description=spec.description,
            annotations=annotations,
            meta={
                "x-side-effect": spec.side_effect,
                "x-requires-confirmation": spec.requires_confirmation,
                "x-can-elicit": spec.can_elicit,
                "x-scopes": list(spec.scopes),
                "x-tags": list(spec.tags),
                "x-keywords": list(spec.keywords),
                "x-examples": list(spec.examples),
            },
        )
        def _tool(params: spec.schema, _spec=spec):
            return _spec.handler(params)
    return mcp


if __name__ == "__main__":
    import commands.data  # triggers registration
    build_mcp_server().run(transport="streamable-http", host="0.0.0.0", port=8000)

Transport choice (2026): streamable-http for production. SSE was deprecated in MCP spec revision 2026-03-26. stdio only for local subprocesses.

Emitter B: Zod schemas for the frontend (Pydantic → JSON Schema → Zod)

# emitters/json_schema_emitter.py
import json
from command_registry.core import registry

def emit_json_schemas(output_path: str) -> None:
    bundle = {
        spec.id: {
            "title": spec.summary,
            "description": spec.description,
            "inputSchema": spec.schema.model_json_schema(),
            "x-side-effect": spec.side_effect,
            "x-idempotent": spec.idempotent,
            "x-requires-confirmation": spec.requires_confirmation,
            "x-scopes": list(spec.scopes),
            "x-tags": list(spec.tags),
            "x-keywords": list(spec.keywords),
        }
        for spec in registry().values()
        if "palette" in spec.surfaces or "mcp" in spec.surfaces
    }
    with open(output_path, "w") as f:
        json.dump(bundle, f, indent=2)

Frontend build step (in package.json):

"build:schemas": "json-schema-to-zod -i ../schemas/commands.json -o src/commands.gen.ts"

Emitter C: CLI (cw / argparse)

Follow Thor's python-package-architecture skill conventions. The dispatch table is {spec.id: spec.handler} — the same registry every other emitter reads, so the CLI is a projection of the SSOT, never a second definition of it.

# emitters/cli_emitter.py
import argparse, json
from command_registry.core import registry

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="app")
    subparsers = parser.add_subparsers(dest="command", required=True)
    for spec in registry().values():
        if "cli" not in spec.surfaces:
            continue
        sub = subparsers.add_parser(spec.id, help=spec.summary, description=spec.description)
        for name, field in spec.schema.model_fields.items():
            sub.add_argument(f"--{name.replace('_', '-')}", dest=name, help=field.description)
        sub.set_defaults(_handler=spec.handler, _schema=spec.schema)
    return parser

def main(argv=None) -> None:
    args = vars(build_parser().parse_args(argv))
    handler, schema = args.pop("_handler"), args.pop("_schema")
    args.pop("command", None)
    payload = schema(**{k: v for k, v in args.items() if v is not None})
    print(json.dumps(handler(payload), default=str))

Do not use argh. It is LGPL-3.0-or-later and has been removed across this fleet.

The house CLI adapter is cw (MIT, no dependencies) — the third string-referenced adapter beside py2mcp (MCP) and qh (HTTP) over the same plain Python functions. It reads the dispatch table directly, so the whole emitter above collapses to:

# emitters/cli_emitter.py
import cw
from command_registry.core import registry

COMMANDS = {
    spec.id: spec.handler for spec in registry().values() if "cli" in spec.surfaces
}

def main() -> None:
    raise SystemExit(cw.dispatch(COMMANDS, prog="app"))

cw.dispatch reads each handler's signature and docstring; cw.mk_parser(COMMANDS) returns the same thing as a plain argparse.ArgumentParser when you want to inspect or extend it. Per-parameter add_argument particulars that a signature cannot express go in a config dict shaped like COMMANDS — see coact/__main__.py for a worked example.

cw.dispatch returns an exit code rather than exiting, so the raise SystemExit(...) above is load-bearing: drop it and every usage error exits 0.

Emitter D: OpenAPI (FastAPI)

Mount each command as a route. FastMCP.from_fastapi() can then re-emit an MCP server from the OpenAPI doc — useful if you want both surfaces from one source. Thor's qh does the same job straight from plain Python functions; prefer it where it covers the case, on the same "one registry, many adapters" principle as py2mcp.

Emitter E: Multi-provider tool catalogues

def to_openai_tool(spec):
    return {"type": "function", "function": {
        "name": spec.id,
        "description": spec.description,
        "parameters": spec.schema.model_json_schema(),
    }}

def to_anthropic_tool(spec):
    return {
        "name": spec.id,
        "description": spec.description,
        "input_schema": spec.schema.model_json_schema(),
    }

def to_gemini_tool(spec):
    return {
        "name": spec.id,
        "description": spec.description,
        "parameters": spec.schema.model_json_schema(),
    }

Or hand the JSON Schema to LiteLLM (Python) / AI SDK (TS) and let them do the envelope translation.

Per-user permission scoping (TWO-LAYER enforcement)

Always enforce both:

1. Filter tools/list by the authenticated user's JWT scopes:

# emitters/scope_middleware.py
from fastmcp.server.middleware import ListToolsMiddleware

class ScopeFilteringMiddleware(ListToolsMiddleware):
    async def on_list_tools(self, context, tools):
        user_scopes = set(context.auth_info.scopes)
        return [t for t in tools if set(t.meta.get("x-scopes", ())).issubset(user_scopes)]

2. Re-check scopes inside each tool handler so direct calls bypassing tools/list are still rejected. This is the Atlassian MCP Server pattern — inherit permissions from the underlying product rather than granting any of its own.

Multi-server composition (namespacing to avoid collisions)

Three patterns:

• Compose in the server. FastMCP.mount() attaches a server prefix (weather_get_forecast).
• Compose in the client. mcp-use's MCPClient.from_config_file() + MCPAgent(use_server_manager=True) routes per step.
• Gateway. Stacklok MCP Optimizer / MCP Manager / Apigene / Higress between many clients and many servers — multi-tenant production pattern.

For a single-codebase SSOT, mounting is the default. Each domain (commands/data/, commands/files/, commands/admin/) gets its own sub-server.

Tool Search Tool (large catalogues)

Beyond ~50 tools, context overflow degrades selection accuracy. Anthropic's Tool Search Tool (GA Feb 2026):

• Mark long-tail commands with defer_loading=True.
• Keep frequently-used commands defer_loading=False.
• Anthropic's benchmarks: ~85% reduction in tool-definition tokens; Opus 4 from 49% → 74% MCP-eval accuracy.
• Real cap: ~60% retrieval accuracy on 4000+ tools (third-party Arcade.dev test). Don't over-defer.

For non-Anthropic providers, Pydantic AI issue #3590 tracks universal deferred loading; mcp-use's Server Manager is the client-side equivalent today.

Approval flows (human-in-the-loop)

Three escalation levels mapped to side_effect:

requires_confirmation=True overrides the default. Cloudflare Agents and Mastra both have first-class approval patterns to reference.

OAuth (when MCP server is remote)

For Streamable HTTP transports: OAuth 2.1 + PKCE + Protected Resource Metadata (RFC 9728). Scopes from the JWT drive the tools/list filter. Use a gateway (Stacklok MCP Optimizer, Apigene) for multi-tenant production.

Anti-patterns

• Authoring MCP tools and command-palette entries separately. They are the same idea. Pick the registry as SSOT.
• Skipping description quality. Tool descriptions are the rate-limiting factor for agent reliability. Write them like API docs for an intern.
• Using passthrough() Zod schemas on the TS side. They let extra fields slip through validation. Use mcp-types (community fork without passthrough).
• Authoring per-provider tool definitions by hand. JSON Schema → LiteLLM/AI SDK normalization is one line.
• MCP at the chat-UI layer. MCP is a tool/resource protocol, not a UI streaming protocol. Use AI SDK UI Message Stream Protocol (see ai-assistant-chat-ui) on the wire; MCP behind FastAPI for tools.

Tool description style (the lint)

A tool description should answer:

1. What does it do? (one sentence)
2. When should the agent reach for it? (situations / triggers)
3. When should it NOT? (disambiguation from similar tools)
4. What does the output look like? (so the agent can plan downstream calls)

Lint rule (CI): every command's docstring is ≥ 3 sentences, contains the word "use when" or "Use this when", and references at least one related command if any exist with similar tags.

Versions known good as of 2026-05-18

• FastMCP 3.x (2.x folded into official Python SDK; 3.x adds AggregateProvider, Transform)
• MCP Python SDK matching spec rev 2026-03-26
• @modelcontextprotocol/sdk (TS) with Standard Schema support
• MCP spec rev 2026-03-26 — SSE deprecated, Streamable HTTP default
• Tool Search Tool — GA Feb 2026, beta header advanced-tool-use-2025-11-20 no longer required for GA
• OAuth 2.1 + PKCE + RFC 9728 for remote MCP

Freshness check

• MCP spec revision — check for new annotation vocabulary (the working SEPs on reads_private_data, sees_untrusted_content, can_exfiltrate could land).
• SEP-1382 (description-vs-schema convention) — if it graduates to normative, update the description style lint above.
• Pydantic AI #3590 (universal deferred loading) — when shipped, remove the Anthropic-only caveat on Tool Search Tool.
• py2mcp capabilities — if it grows to cover the full annotation taxonomy, deprecate the FastMCP emitter in this skill in favor of py2mcp.

Related skills

• ai-assistant-architect — overall architecture
• ai-assistant-prompts-skills — frontmatter taxonomy is compatible; the same git repo holds commands + skills + prompts
• ai-assistant-agent-runtime — how the agent calls these tools

Source reports (in Thor's project knowledge)

• Command_Dispatch_to_MCP_Bridge.md (primary)
• command_dispatch_journal_article.md (philosophical foundation — rule of three, strangler fig migration)