同一个 agent 能力，平台通过四种"面（surface）"暴露给不同用户：

关键点：CLI 和 MCP 本身都不含业务逻辑，它们最终都是把请求转成对 Day 12 那个 HTTP server 的调用。二者甚至共用同一个 HTTP 客户端 PlatformClient（就在 gov_agents_mcp/client.py 里，CLI 直接 import 它）。今天就是把这三个"门"的真实源码逐段读一遍。

packages/gov-agents-cli 提供命令 agentctl（基于 Typer）。先看它的顶部——写死的 agent 名单 + Typer app 装配：

# 跟 platform AGENT_REGISTRY 对齐(避免运行时 import platform 包 · 写死即可)
AGENT_NAMES = ["biz-link-gov", "rca", "risk-review", "alert-triage", "doc-check"]  # main.py:31

app = typer.Typer(
    name="agentctl",
    help="调 gov-agents-platform v1 endpoint · CI/CD 脚本入口。",
    no_args_is_help=True,          # 不带参数就打印 help，不报错
)
app.add_typer(skill_app)           # 挂子命令组：agentctl skill ...（管 L4 skills）
app.add_typer(new_app)             # 挂子命令组：agentctl new agent（脚手架，Day 19）
register_validate_agent_id(app)    # agentctl validate-agent-id

再看两个"守门函数"——校验 agent 名、把所有异常翻成退出码：

def _validate_agent(name: str) -> str:            # main.py:43
    if name not in AGENT_NAMES:
        stderr().print(f"[red]未知 agent {name!r} · 合法值: {AGENT_NAMES}[/red]")
        raise typer.Exit(code=2)                   # ← 输入错 = 退出码 2
    return name

def _run(coro):                                    # main.py:58 · 所有命令都用它跑
    """跑 async coro · 统一 exception 处理 · 翻成退出码。"""
    try:
        return asyncio.run(coro)
    except McpProxyError as e:                      # ← 复用 MCP 那套错误类型！
        if e.code in ("INVALID_INPUT",):
            stderr().print(f"[red][{e.code}] {e.message}[/red]")
            raise typer.Exit(code=1) from e
        if e.code == "UPSTREAM_ERROR" and "无法连接" in e.message:
            stderr().print(f"[red]{e.message}[/red]")
            raise typer.Exit(code=3) from e         # ← 网络错 = 退出码 3
        if e.code == "NOT_FOUND":
            stderr().print(f"[yellow]{e.message}[/yellow]")
            raise typer.Exit(code=1) from e
        stderr().print(f"[red][{e.code}] {e.message}[/red]")
        raise typer.Exit(code=1) from e             # ← 其它 HTTP 错 = 退出码 1
    except KeyboardInterrupt:
        stderr().print("\n[yellow]Ctrl+C · 中断[/yellow]")
        raise typer.Exit(code=130) from None        # ← Ctrl+C = 退出码 130（Unix 惯例）

拿最常用的 agentctl invoke 走一遍，看参数怎么变成一次 HTTP 调用（main.py:99-122）：

@app.command()
def invoke(
    agent: Annotated[str, typer.Argument(help="5 个之一:biz-link-gov / rca / ...")],
    data: Annotated[str | None, typer.Option(help="JSON body string")] = None,
    data_file: Annotated[Path | None, typer.Option(help="JSON body 文件路径")] = None,
    table: Annotated[bool, typer.Option("--table", help="人类格式 · 默认 JSON")] = False,
    no_color: Annotated[bool, typer.Option("--no-color")] = False,
) -> None:
    """sync invoke · POST /v1/agent/{name}/invoke。"""
    agent = _validate_agent(agent)          # ① 先卡 agent 名（不合法退出码 2）
    body = _parse_data(data, data_file)     # ② 解析 JSON body

    async def go():
        client = _make_client()             # ③ 造 PlatformClient
        try:
            return await client.invoke(agent, body)   # ④ 真打 HTTP
        finally:
            await _close(client)            # ⑤ 关连接池（无论成败）
    env = _run(go())                        # ⑥ _run 统一跑 + 翻退出码
    if table:
        render_envelope(env, no_color=no_color)       # 人类友好表格
    else:
        stdout_console(no_color=no_color).print_json(dump_json(env))  # 默认吐 JSON

其中 _parse_data（main.py:79）负责把 --data 字符串或 --data-file 文件读成 dict，任何一步失败都退出码 2：

def _parse_data(data, data_file) -> dict:          # main.py:79
    if data_file:
        try:
            return json.loads(data_file.read_text())
        except (OSError, json.JSONDecodeError) as e:
            stderr().print(f"[red]--data-file 解析失败: {e}[/red]")
            raise typer.Exit(code=2) from e         # 文件读不了/JSON 坏 = 输入错
    if data:
        try:
            return json.loads(data)
        except json.JSONDecodeError as e:
            raise typer.Exit(code=2) from e
    stderr().print("[red]必须传 --data 或 --data-file 之一[/red]")
    raise typer.Exit(code=2)                         # 两个都没给也是输入错

CLI 和 MCP 都靠同一个 PlatformClient（gov-agents-mcp/src/gov_agents_mcp/client.py:49）打平台。先看它怎么派生请求身份 X-User-Id——这个函数很能体现"一份代码两处复用"：

def _resolve_user_id(env=None, *, prefix: str = "mcp") -> str:     # client.py:28
    """派生 X-User-Id:{PREFIX}_USER_ID → MCP_USER_ID → {prefix}:$USER → {prefix}:anonymous。"""
    e = env if env is not None else dict(os.environ)
    explicit_env = f"{prefix.upper()}_USER_ID"      # MCP→MCP_USER_ID / CLI→AGENTCTL_USER_ID
    explicit = (e.get(explicit_env) or "").strip()
    if explicit:
        return explicit
    # 兼容:CLI 没设 AGENTCTL_USER_ID 时也认 MCP_USER_ID
    if prefix != "mcp":
        mcp_explicit = (e.get("MCP_USER_ID") or "").strip()
        if mcp_explicit:
            return mcp_explicit
    user = (e.get("USER") or "").strip()
    if user:
        return f"{prefix}:{user}"                   # 例 mcp:alice / agentctl:alice
    return f"{prefix}:anonymous"

再看 client 构造 + 请求头，理解它怎么把 env 变成一次带鉴权的 HTTP：

class PlatformClient:                               # client.py:49
    def __init__(self, *, base_url=None, timeout=120.0, env=None, user_id_prefix="mcp"):
        e = env if env is not None else dict(os.environ)
        self.base_url = (base_url or e.get("PLATFORM_BASE_URL") or "http://localhost:8080").rstrip("/")
        self.user_id = _resolve_user_id(e, prefix=user_id_prefix)
        self.anthropic_api_key = (e.get("ANTHROPIC_API_KEY") or "").strip() or None
        self._client = httpx.AsyncClient(timeout=timeout)   # 复用连接池

    def _headers(self) -> dict[str, str]:           # client.py:66
        h = {"X-User-Id": self.user_id, "Content-Type": "application/json"}
        if self.anthropic_api_key:
            h["X-Anthropic-Api-Key"] = self.anthropic_api_key   # BYOK:自带 key 透传
        return h

MCP（Model Context Protocol）是一个标准协议，让 IDE 里的 AI 助手（Cursor / Claude Code / Cline / Windsurf）能"调用外部工具"。你在 IDE 里说一句人话，AI 就能触发一个 MCP tool 帮你干活。IDE 通过 stdio（标准输入输出）启动 MCP server 子进程，用 JSON-RPC 通信。

本框架把 agent 暴露成 MCP tool，于是你在 IDE 里说"帮我分析下 service-A 的超时根因"，AI 就会调 invoke_sre_rca 这个 MCP tool，背后其实是打了平台的 HTTP 接口。

MCP tool 要告诉 IDE"我接受什么参数"（inputSchema）。本框架不手写 JSON Schema，而是定义 Pydantic 模型 → 自动生成。先看数据结构层（gov-agents-mcp/src/gov_agents_mcp/schemas.py）：

class SreRcaInput(BaseModel):                       # schemas.py:27
    """sre-rca · 微服务根因分析。"""
    trace_id: str = Field(..., description="SkyWalking trace ID")
    service_name: str | None = Field(None, description="可选服务名 · 提示重点看哪个服务")
    user_question: str = Field(..., description="用户问题描述 · 例 'service-A 慢,DB 连接池打满'")
    thread_id: str | None = Field(None, description="可选 thread_id · 中断恢复用")

# tool_name → (输入模型, platform agent 名) 的总映射表   # schemas.py:68
AGENT_INPUTS: dict[str, tuple[type[BaseModel], str]] = {
    "invoke_biz_link_gov": (BizLinkGovInput, "biz-link-gov"),
    "invoke_sre_rca":      (SreRcaInput,      "rca"),          # ← MCP tool 名 → 路由名
    "invoke_risk_review":  (RiskReviewInput,  "risk-review"),
    "invoke_alert_triage": (AlertTriageInput, "alert-triage"),
    "invoke_doc_check":    (DocCheckInput,    "doc-check"),
}

再看控制层怎么用这张表：list_tools 生成 schema、call_tool 校验后转发（gov-agents-mcp/src/gov_agents_mcp/tools.py）：

def list_tools() -> list[Tool]:                     # tools.py:14
    tools = []
    for tool_name, (model_cls, _agent_name) in AGENT_INPUTS.items():
        schema = model_cls.model_json_schema()      # ← Pydantic 自动出 JSON Schema！
        tools.append(Tool(name=tool_name,
                          description=TOOL_DESCRIPTIONS[tool_name],
                          inputSchema=schema))
    return tools

async def call_tool(name, arguments, *, client) -> list[TextContent]:   # tools.py:29
    if name not in AGENT_INPUTS:
        raise McpProxyError("INVALID_INPUT", f"unknown tool {name!r}")
    model_cls, agent_name = AGENT_INPUTS[name]
    try:
        validated = model_cls.model_validate(arguments)   # ← 二道校验（MCP SDK 已校一次）
    except Exception as e:
        raise McpProxyError("INVALID_INPUT", f"参数校验失败: {e}") from e
    body = validated.model_dump(exclude_none=True)        # 去掉没填的可选字段
    envelope = await client.invoke(agent_name, body)      # ← 转成 HTTP！用 AGENT_INPUTS 里的路由名
    return [TextContent(type="text", text=json.dumps(envelope, ensure_ascii=False, indent=2))]

平台可能返回各种 HTTP 错误码。PlatformClient.invoke 把它们归成 4 类友好错误（client.py:75-115）——CLI 和 MCP 都靠这套分类：

async def invoke(self, agent_name, body) -> dict:          # client.py:75
    url = f"{self.base_url}/v1/agent/{agent_name}/invoke"
    try:
        r = await self._client.post(url, json=body, headers=self._headers())
    except httpx.HTTPError as e:
        raise McpProxyError("UPSTREAM_ERROR", f"无法连接 platform ...")  # 连不上=网络错
    if r.status_code == 200:
        return r.json()                                     # 正常：返 envelope
    # ── 错误归类 ──
    try:    err_body = r.json()
    except Exception:  err_body = {"detail": r.text[:500]}
    if r.status_code == 422:                                # 输入不合法
        raise McpProxyError("INVALID_INPUT", str(err_body.get("detail", err_body)))
    if r.status_code in (401, 403):                         # 鉴权失败
        raise McpProxyError("UNAUTHORIZED", f"platform 拒绝(HTTP {r.status_code})...")
    if r.status_code == 500:                                # agent 内部异常
        detail = err_body.get("detail") or {}
        case_id = detail.get("case_id") if isinstance(detail, dict) else None
        error_msg = "agent 内部异常"
        if isinstance(detail, dict):
            result = detail.get("result") or {}
            error_msg = result.get("error") or result.get("error_type") or error_msg
        raise McpProxyError("AGENT_ERROR", error_msg, case_id=case_id)   # ← 带上 case_id！
    raise McpProxyError("UPSTREAM_ERROR", f"HTTP {r.status_code}: {err_body}")

MCP server 主入口 build_server（gov-agents-mcp/src/gov_agents_mcp/server.py:36）把上面几块用装饰器挂到一个 Server 上：

def build_server(client=None) -> tuple[Server, PlatformClient]:   # server.py:36
    pc = client if client is not None else PlatformClient()        # 单例复用连接池
    server = Server("gov-agents-mcp")

    @server.list_tools()                    # server.py:48 · IDE 问"你有啥工具"
    async def handle_list_tools():
        return _list_tools()                # → tools.py 那 5 个 invoke_*

    @server.call_tool()                     # server.py:52 · IDE 说"调这个工具"
    async def handle_call_tool(name, arguments):
        try:
            return await _call_tool(name, arguments, client=pc)
        except McpProxyError as e:          # ← 错误不外抛，而是当文本返给 IDE 的 AI
            text = f"[{e.code}] {e.message}"
            if e.case_id:  text += f"\n(case_id: {e.case_id})"
            return [TextContent(type="text", text=text)]

    @server.list_resources()                # server.py:62 · 5 个 cases://
    async def handle_list_resources():
        return _list_static_resources()

    @server.read_resource()                 # server.py:70 · 读某条历史 case
    async def handle_read_resource(uri):
        try:
            return await _read_resource(str(uri), client=pc)
        except McpProxyError as e:
            return [TextResourceContents(uri=uri, mimeType="text/plain",
                                         text=f"[{e.code}] {e.message}")]
    return server, pc

resources 侧的 URI 解析（gov-agents-mcp/src/gov_agents_mcp/resources.py:56）展示了怎么把 cases://rca 这种 URI 变成一次 HTTP GET：

def _parse_uri(uri: str) -> tuple[str, str, str | None]:      # resources.py:56
    if uri.startswith("cases://"):                # 列表：cases://
        agent = uri.removeprefix("cases://").rstrip("/")
        if not agent or agent not in AGENT_NAMES:
            raise McpProxyError("NOT_FOUND", f"未知 agent {agent!r}...")
        return "cases", agent, None
    if uri.startswith("case://"):                 # 单条：case:///
        rest = uri.removeprefix("case://")
        parts = rest.split("/", 1)
        if len(parts) != 2 or not parts[0] or not parts[1]:
            raise McpProxyError("NOT_FOUND", f"case:// URI 格式应为 case:///...")
        agent, case_id = parts[0], parts[1]
        ...
        return "case", agent, case_id
    raise McpProxyError("NOT_FOUND", f"未识别的 URI scheme: {uri!r}")

cases://rca（复数）= 最近 50 条概览，case://rca/（单数）= 单条详情。IDE 的 AI 拿到某次调用的 case_id 后，可以直接 read case://rca/01HXXX 把详情读出来给你看，无需再发 tool call。

本框架有两个都叫"MCP"、架构却完全不同的东西，一次讲清避免踩坑：

apps/bmc-agent-mcp 用 FastMCP 暴露 4 个 tool，且直接 import 复用 bmc_agent 的图（不走 HTTP！）。看它的 tool 定义和那个著名的 stdio 坑（apps/bmc-agent-mcp/bmc_agent_mcp/server.py）：

mcp = FastMCP("bmc-agent")                          # server.py:37

@mcp.tool()                                         # server.py:71
def bmc_scan(project_path: str, bmc_parent_path: str = "") -> dict:
    """只扫描,returns 项目现状(便宜,几百 ms)。"""
    state = _initial_state(project_path, bmc_parent_path=bmc_parent_path)
    with redirect_stdout(sys.stderr):               # server.py:90 ← 关键 trick！
        result = scan_node(state)                   # 直接调 bmc_agent 的节点（内嵌，不走 HTTP）
    return {...}

顺带另外两组"名字像、职责不同"的东西（前几天讲过，这里汇总）：

• 两个 supervisor（Day 04）：图内 supervisor_node（选专家、真跑）vs 跨 agent router（选 agent、只建议）。
• 三层命名（Day 12/本日 L06）：注册表名(sre-rca) ≠ 路由名(rca) ≠ MCP tool 名(invoke_sre_rca)。

✋ 动手

# 1. CLI 入口 + 退出码翻译（本日核心）
sed -n '30,93p' packages/gov-agents-cli/src/gov_agents_cli/main.py

# 2. MCP schema 自动生成 + tool 转发
sed -n '27,75p' packages/gov-agents-mcp/src/gov_agents_mcp/schemas.py
cat packages/gov-agents-mcp/src/gov_agents_mcp/tools.py

# 3. PlatformClient 的 user_id 派生 + 错误归 4 类
sed -n '28,115p' packages/gov-agents-mcp/src/gov_agents_mcp/client.py

# 4. server 装配 + resources URI 解析
sed -n '36,85p'  packages/gov-agents-mcp/src/gov_agents_mcp/server.py
sed -n '56,102p' packages/gov-agents-mcp/src/gov_agents_mcp/resources.py

# 5. 另一个 MCP：内嵌 agent + stdout 重定向坑
sed -n '1,95p' apps/bmc-agent-mcp/bmc_agent_mcp/server.py