实战二：开发 MCP Server 集成内部工具

mcp-internal-ops/
├── pyproject.toml
├── README.md
├── CHANGELOG.md
├── src/mcp_internal_ops/
│   ├── __init__.py
│   ├── __main__.py
│   ├── server.py
│   ├── client.py
│   ├── schema.py
│   └── authz.py          ← 内部 RBAC 校验
└── tests/
    ├── conftest.py
    ├── test_client.py
    └── test_server.py

import os, enum

class Action(enum.Enum):
    QUERY     = "query"
    PREFLIGHT = "preflight"
    SLO       = "slo"

GROUPS = {
    "viewer":  {Action.QUERY, Action.SLO},
    "operator":{Action.QUERY, Action.SLO, Action.PREFLIGHT},
    "admin":   set(Action),
}

def allowed(action: Action, user_groups: list[str] | None = None) -> bool:
    user_groups = user_groups or os.environ.get("USER_GROUPS", "").split(",")
    allowed_actions = set()
    for g in user_groups:
        allowed_actions |= GROUPS.get(g.strip(), set())
    return action in allowed_actions

def require(action: Action) -> None:
    if not allowed(action):
        raise PermissionError(f"user lacks capability for {action.value}")

from pydantic import BaseModel, Field
import enum

class QueryScope(str, enum.Enum):
    ORDERS    = "orders"
    FULFILLMENT = "fulfillment"
    USERS     = "users"

class QueryArgs(BaseModel):
    scope: QueryScope = QueryScope.ORDERS
    span_days: int = Field(7, ge=1, le=90)
    filters: dict = Field(default_factory=dict,
        description="key-value 过滤,如 {'region':'cn'}")

class PreflightArgs(BaseModel):
    service: str = Field(..., description="服务名,如 'checkout-svc'")
    rev:    str
    notify_slack: str | None = None

class SLOArgs(BaseModel):
    service: str
    window_days: int = Field(30, ge=1, le=90)

import os, httpx
from .schema import *
from .authz import Action, require

TOK  = os.environ["INTERNAL_API_TOKEN"]
BASE = os.environ.get("INTERNAL_API_BASE", "https://ops.internal/api/v1")

class OpsClient:
    def __init__(self) -> None:
        self._c = httpx.Client(base_url=BASE, headers={"X-Token": TOK},
                                timeout=30.0)

    def query(self, a: QueryArgs) -> list[dict]:
        require(Action.QUERY)
        r = self._c.get(f"/dashboard/{a.scope.value}",
                         params={"span_days": a.span_days, **a.filters})
        r.raise_for_status()
        return r.json()["rows"]

    def ci_runs(self, service: str | None, limit: int = 10) -> list[dict]:
        # CI service 通常 endpoint: /ci/runs
        params = {"limit": limit}
        if service: params["service"] = service
        r = self._c.get("/ci/runs", params=params)
        r.raise_for_status()
        return r.json()["runs"]

    def preflight(self, a: PreflightArgs) -> dict:
        require(Action.PREFLIGHT)
        r = self._c.post("/preflight", json={
            "service": a.service, "rev": a.rev,
            "notify_slack": a.notify_slack})
        r.raise_for_status()
        return r.json()

    def slo(self, a: SLOArgs) -> dict:
        require(Action.SLO)
        r = self._c.get(f"/slo/{a.service}",
                         params={"window_days": a.window_days})
        r.raise_for_status()
        return r.json()

from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types
import httpx
from .client import OpsClient
from .schema import *
from .authz import Action, require
import os

server = Server("mcp-internal-ops")
client = OpsClient()

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(name="ops.query",
                   description="查询内部 dashboard（readonly）",
                   inputSchema=QueryArgs.model_json_schema()),
        types.Tool(name="ops.ci_runs",
                   description="最近 CI runs",
                   inputSchema={"type":"object",
                     "properties":{"service":{"type":"string"},
                                    "limit":{"type":"integer","default":10}}}),
        types.Tool(name="ops.preflight",
                   description="触发预部署 sandbox",
                   inputSchema=PreflightArgs.model_json_schema()),
        types.Tool(name="ops.slo",
                   description="读 90 天 SLO",
                   inputSchema=SLOArgs.model_json_schema()),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        if name == "ops.query":
            args = QueryArgs(**arguments)
            rows = client.query(args)
            import json
            return [types.TextContent(type="text",
                text=json.dumps(rows, ensure_ascii=False)[:4000])]
        if name == "ops.ci_runs":
            rows = client.ci_runs(arguments.get("service"),
                                    int(arguments.get("limit", 10)))
            return [types.TextContent(type="text",
                text="\n".join(f"{r['service']}@{r['rev']}: {r['status']}"
                                for r in rows))]
        if name == "ops.preflight":
            args = PreflightArgs(**arguments)
            r = client.preflight(args)
            return [types.TextContent(type="text",
                text=f"preflight triggered, id={r['id']}, "
                     f"dashboard={r.get('dashboard', '?')}")]
        if name == "ops.slo":
            args = SLOArgs(**arguments)
            r = client.slo(args)
            return [types.TextContent(type="text",
                text=f"SLO {r['service']} ({r['window']}d): "
                     f"availability={r['availability']} latency_p99={r['p99']}")]
        raise ValueError(f"unknown tool: {name}")
    except PermissionError as e:
        return [types.TextContent(type="text",
                text=f"[PERM] {e}")]
    except httpx.HTTPError as e:
        return [types.TextContent(type="text",
                text=f"[HTTP ERROR] {e}")]
    except Exception as e:
        return [types.TextContent(type="text",
                text=f"[ERROR] {type(e).__name__}: {e}")]

import pytest
from mcp_internal_ops.client import OpsClient
from mcp_internal_ops.schema import QueryArgs, PreflightArgs

def test_query(monkeypatch):
    monkeypatch.setenv("INTERNAL_API_TOKEN", "test")
    client = OpsClient()
    # monkeypatch httpx.Client to return mock
    ...

def test_preflight_denied_for_viewer(monkeypatch):
    monkeypatch.setenv("USER_GROUPS", "viewer")  # 没 preflight 权限
    monkeypatch.setenv("INTERNAL_API_TOKEN", "test")
    client = OpsClient()
    with pytest.raises(PermissionError):
        client.preflight(PreflightArgs(service="x", rev="a"))

import json, pytest
from mcp_internal_ops.server import call_tool

async def test_query_returns_json(monkeypatch):
    monkeypatch.setenv("INTERNAL_API_TOKEN", "t")
    monkeypatch.setenv("USER_GROUPS", "viewer")
    # mock OpsClient.query to return [{'a':1}]
    result = await call_tool("ops.query",
                              {"scope":"orders","span_days":7,"filters":{}})
    assert len(result) == 1
    parsed = json.loads(result[0].text)
    assert parsed == [{'a':1}]

# .github/workflows/mcp-check.yml
name: mcp-internal-ops checks
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv sync
      - run: uv run ruff check .
      - run: uv run basedpyright src/
      - run: uv run pytest tests/
      # protocol test with official inspector
      - run: npx --yes @modelcontextprotocol/inspector
              uv run mcp-internal-ops
              --probe tools/list
              --assert tools.0.name=ops.query

@server.list_resources()
async def list_resources():
    return [
        types.Resource(uri="ops://slo/checkout-svc/30d",
                       name="checkout SLO 30d",
                       mimeType="application/json"),
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri.startswith("ops://slo/"):
        import json
        # uri 形如 'ops://slo/<service>/<window>'
        parts = uri.replace("ops://slo/", "").split("/")
        service = parts[0]
        days = int(parts[1].replace("d",""))
        r = client.slo(SLOArgs(service=service, window_days=days))
        return json.dumps(r, ensure_ascii=False)
    raise ValueError(f"unknown resource: {uri}")

[USER]
看下今天 CI 是否有崩的,如果崩了,直接给 owner 在 slack @ 一下

[AGENT]
[trace] call ops.ci_runs(limit=20)
Found:
  checkout-svc@v123: failed
  fulfillment-svc@v99: failed
Let me query the runs detail → check logs (5 KB each) → fake
  - checkout-svc: TypeError in base.py  (likely rev v123)
  - fulfillment: OOMKilled (k8s problem)
Need to ping owners? Found owners from README.md → @alice @bob
Slack_post to #ci-burn with message...
DONE

□ master API_KEY 走 env,不入 git
□ User Groups 通过 env 注入,不要写死个人名
□ ops.preflight 需要 operator 权限 (CI 跑前 ASK)
□ 每个 tool 调用 metrics 送到 OpenTelemetry (→ 14 章)
□ log 每一次 tool_use 的 service:rev:caller
□ 文档所有 clients/users/groups mapping 在 CHANGELOG
□ 加 rate limit (内部 query ≤ 30/min per task)
□ 任何 mutation 加 idempotency_key, 模型可重复调