一个完整的 MCP 调用：从工具发现到业务校验

让 planner 能请求 events，并不等于允许它向 kubectl 传任意参数。这里把一次查询跟到底：谁选择来源，谁固定命名空间，谁检查协议返回的业务错误。这三层中任意一层放宽，都会改变助手实际能做的事。

sequenceDiagram
Runtime->>MCP: initialize 并列举工具
Runtime->>MCP: evidence 指定来源
MCP->>API: 固定身份和对象范围查询
API-->>MCP: 返回对象或拒绝访问
MCP-->>Runtime: 有界证据或工具错误

stdio 模式下客户端启动子进程，与其建立 MCP 会话；工具日志应该去 stderr，stdout 留给协议。客户端 initialize → list_tools → call_tool，再区分协议成功与 result.isError。读会话不注册写工具，环境变量也只透传白名单；这样模型凭据不会随每次 kubectl 调用流出。

完整源码：opspilot/cluster/run.py · 第 15–29 行

SOURCES=["deployment","pods","events","nodes","logs","endpoints","metrics"]
async def rpc(name,args):
    write=name=="release_selector"
    keys={"PATH","HOME","KUBECONFIG","OPS_IN_CLUSTER","KUBERNETES_SERVICE_HOST","KUBERNETES_SERVICE_PORT","PROMETHEUS_URL","LANG","TMPDIR","OPS_WRITE_KUBECONFIG" if write else "OPS_READ_KUBECONFIG"}
    env={k:v for k,v in os.environ.items() if k in keys}
    env["OPS_TOOL_ROLE"]="write" if write else "read"
    params=StdioServerParameters(command=sys.executable,args=[str(ROOT/"tools_live.py")],env=env)
    async with stdio_client(params) as (r,w):
        async with ClientSession(r,w) as c:
            await c.initialize()
            if name not in {t.name for t in (await c.list_tools()).tools}:raise ValueError("tool unavailable")
            result=await c.call_tool(name,args)
    if result.isError:raise ValueError(result.content[0].text)
    return json.loads(result.content[0].text)
def call(name,**args):return asyncio.run(asyncio.wait_for(rpc(name,args),20))

服务端如何把“查询”限定到可审计范围

FastMCP 用注解生成接口类型；字符串是合法类型，并不意味着任意字符串都是合法来源。服务端对 pods、events、logs、nodes、metrics、deployment、endpoints 逐项映射到固定操作。限制 Pod 数、日志行数、字节数和超时；事件条件使用 involvedObject.uid，避免拿到同名旧 Pod 的事件。metrics 查询使用预先固定的 PromQL，不接受任意表达式。

完整源码：opspilot/cluster/tools_live.py · 第 20–39 行

def evidence(source: str) -> dict:
    """Read one bounded observation about pending-lab, without executing user commands."""
    if source == "deployment":
        d=json.loads(kube("get","deployment",NAME,"-o","json"))
        return {"uid":d["metadata"]["uid"],"generation":d["metadata"]["generation"],
                "selector":d["spec"]["template"]["spec"].get("nodeSelector",{}),"status":d.get("status",{})}
    if source == "pods":
        pods=json.loads(kube("get","pod","-l","app="+NAME,"-o","json"))["items"]
        return {"pods":[{"name":p["metadata"]["name"],"uid":p["metadata"]["uid"],"phase":p["status"]["phase"],
                         "conditions":p["status"].get("conditions",[]),"containers":p["status"].get("containerStatuses",[])} for p in pods[:5]]}
    if source in ("events","logs"):
        pods=evidence("pods")["pods"]; items=[]
        for p in pods[:3]:
            if source=="logs":
                items.append(kube("logs",p["name"],"--tail=20","--limit-bytes=2000") if p["phase"]=="Running" else "no running container")
            else:
                ev=json.loads(kube("get","events","--field-selector","involvedObject.uid="+p["uid"],"-o","json"))["items"]
                items.extend([{"reason":x["reason"],"message":x.get("message","")[:1000]} for x in ev[-8:]])
        return {"items":items}
    if source=="nodes":

实际启动客户端，检查正常和异常两条路径

从根目录创建环境；若已配好则跳过安装。server 由客户端自动拉起，无需另开一个永久监听端口。

uv venv --python 3.13 .venv
uv pip install --python .venv/bin/python -r opspilot/cluster/requirements.lock
export KUBECONFIG="$PWD/tmp/handbook.kubeconfig"
.venv/bin/python opspilot/cluster/probe.py

本轮 probe 返回实际 Deployment generation，再输出 unsupported evidence source rejected。连接超时先检查专用 context 是否可达；返回 forbidden 检查 observe 的 RoleBinding；“unknown evidence source”是业务校验，不应切换到更大权限绕过。Prometheus 缺 kube-state-metrics 时可能返回空向量，不代表集群没有 Pending。

完整源码：opspilot/cluster/probe.py · 第 5–18 行

async def main():
    observed = await asyncio.wait_for(rpc("evidence", {"source": "deployment"}), 20)
    print("deployment generation:", observed["generation"])
    try:
        await asyncio.wait_for(rpc("evidence", {"source": "not-a-source"}), 20)
    except ValueError as exc:
        if "unknown evidence source" not in str(exc):
            raise
        print("unsupported evidence source rejected")
    else:
        raise AssertionError("unexpected tool acceptance")

if __name__ == "__main__":
    asyncio.run(main())

工程取舍。 stdio 每次启动一个进程，隔离明确、启动成本可测。若改成长连接池，须分别管理读写身份、任务取消、并发请求与连接失效，不可把观察会话升级为通用执行器。写工具为何还要检查资源版本。