读完生态速览，你大概发现 30% 的需求是现成 server 满足不了的——尤其是自家业务 API和团队内部数据。本章手把手写一个：把公司内部 admin API 暴露成 MCP，让所有 harness 都能调用。

17.1 任务范围

做一个 MCP server 叫 mcp-tickets，让 Agent 能：

  • 列出内部工单系统里的 ticket（按状态）
  • 取一条 ticket 详情
  • 给 ticket 加一条 internal comment
  • 只读为主，写操作需要确认

我们用 Python 写（生态成熟，团队好上手）；TypeScript 版结构近似，章末给一行对比。

17.2 起手装环境

uv init mcp-tickets
cd mcp-tickets
uv add "mcp[cli]"
uv add httpx pydantic

mcp[cli] 是官方 Python SDK，含 stdio server 与 cli 工具。httpx 是要调内部 API，pydantic 做参数校验。

17.3 项目骨架

mcp-tickets/
├── pyproject.toml
├── src/
│   └── mcp_tickets/
│       ├── __init__.py
│       ├── __main__.py     ← entry: 启动 server
│       ├── server.py       ← MCP server 注册
│       ├── client.py       ← 内部 admin API client
│       └── schema.py       ← pydantic models
└── README.md

17.4 schema.py：TypedDict / pydantic

from pydantic import BaseModel, Field

class Ticket(BaseModel):
    id: int
    title: str
    status: str
    assignee: str | None = None

class CommentIn(BaseModel):
    ticket_id: int = Field(..., description="工单 ID")
    body: str = Field(..., min_length=1, max_length=2000,
                      description="评论内容（markdown）")

class ListArgs(BaseModel):
    status: str | None = Field(None, description="按状态过滤 open/closed")
    limit: int = Field(20, ge=1, le=100)

重点：建议永远用 pydantic 校验输入，不要在 MCP 输入处直接信任 dict。

17.5 client.py：内部 API 封装

import os, httpx
from .schema import Ticket

BASE = os.environ.get("TICKETS_BASE_URL", "https://tickets.internal/api")
TOKEN = os.environ["TICKETS_API_TOKEN"]   # 强制 env，不允许写死

class TicketsClient:
    def __init__(self):
        self._c = httpx.Client(base_url=BASE,
                                headers={"Authorization": f"Bearer {TOKEN}"},
                                timeout=10.0)

    def list(self, status: str | None = None, limit: int = 20) -> list[Ticket]:
        params = {"limit": limit}
        if status: params["status"] = status
        r = self._c.get("/tickets", params=params)
        r.raise_for_status()
        return [Ticket(**x) for x in r.json()["items"]]

    def get(self, ticket_id: int) -> Ticket:
        r = self._c.get(f"/tickets/{ticket_id}")
        r.raise_for_status()
        return Ticket(**r.json())

    def comment(self, ticket_id: int, body: str) -> dict:
        r = self._c.post(f"/tickets/{ticket_id}/comments",
                          json={"body": body})
        r.raise_for_status()
        return r.json()

两原则：1) 不在 client 层混入 MCP 概念；2) 写操作用一个明确的 .comment() 不要泛化 .mutate()。

17.6 server.py：MCP 注册

from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types
from .client import TicketsClient
from .schema import ListArgs, CommentIn

server = Server("mcp-tickets")
client = TicketsClient()

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="tickets.list",
            description="列出工单",
            inputSchema=ListArgs.model_json_schema(),
        ),
        types.Tool(
            name="tickets.get",
            description="取单条工单详情",
            inputSchema={
                "type":"object",
                "properties":{"ticket_id":{"type":"integer"}},
                "required":["ticket_id"],
            },
        ),
        types.Tool(
            name="tickets.comment",
            description="给工单加 internal comment",
            inputSchema=CommentIn.model_json_schema(),
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        if name == "tickets.list":
            args = ListArgs(**arguments)
            rows = client.list(status=args.status, limit=args.limit)
            return [types.TextContent(type="text",
                    text="\n".join(f"#{t.id} [{t.status}] {t.title}" for t in rows))]
        if name == "tickets.get":
            t = client.get(arguments["ticket_id"])
            return [types.TextContent(type="text",
                    text=f"#{t.id} {t.title}\n状态: {t.status}\n负责人: {t.assignee}")]
        if name == "tickets.comment":
            args = CommentIn(**arguments)
            r = client.comment(args.ticket_id, args.body)
            return [types.TextContent(type="text", text=f"commented: {r['id']}")]
        raise ValueError(f"unknown tool: {name}")
    except httpx.HTTPError as e:
        # 把 HTTP 异常当 text 返，不让 server crash
        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}")]

关键细节：

  1. name="tickets.list" 命名空间化，避免跟其他 MCP 撞名
  2. 错误包成 TextContent 返回——不要 raise；raise 会击垮整个 stdio 通道
  3. model_json_schema() 让 model 兼当 MCP schema 源

17.7 main.py：启动 entrypoint

import asyncio
from .server import server

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_opts())

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

测试：

uv run mcp-tickets                              # 启动 stdio server
# 另一个终端
TICKETS_API_TOKEN=xxx uv run mcp-tickets        # 注入 secret

17.8 在 OpenCode 中接入

{
  "mcp": {
    "tickets": {
      "type": "stdio",
      "command": "uv",
      "args": ["run", "--project", "/path/to/mcp-tickets", "mcp-tickets"],
      "env": {
        "TICKETS_API_TOKEN": "${TICKETS_API_TOKEN}",
        "TICKETS_BASE_URL":  "https://tickets.internal/api"
      }
    }
  }
}

启动 OpenCode → 它自动 fork 这个 uv 子进程 → Agent 现在能列出工单 / 加评论了。

17.9 把它做成 resource：提供工单"实时知识"

Tools 是模型调，resources 是客户端拉。我们的 list 也可以做成 resource：

@server.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="tickets://open",
            name="Open tickets",
            description="当前所有 open 状态工单",
            mimeType="application/json",
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "tickets://open":
        import json
        rows = client.list(status="open")
        return json.dumps([r.model_dump() for r in rows], ensure_ascii=False)
    raise ValueError(f"unknown resource: {uri}")

Resource 与 Tool 的取舍：Robust 默认读的情况做 Resource；模型决定调的情况做 Tool。

17.10 TypeScript 版差异

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server({ name: "mcp-tickets", version: "0.1.0" },
  { capabilities: { tools: {}, resources: {} } });

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{ name: "tickets.list", description: "...", inputSchema: {...} }],
}));

结构完全对等 Python 版，只是 async/Promise 引入。

17.11 测试三件套

# 1. lint + type
uv run ruff check . && uv run basedpyright src/

# 2. 用官方 inspector 测协议
npx @modelcontextprotocol/inspector uv run mcp-tickets

# 3. 单元 test
uv add --dev pytest pytest-asyncio
uv run pytest tests/

mcp-inspector 是官方提供的 debug UI——能可视化看到 listing / calling 工具的整个包文流程。写 MCP server 头 10 分钟用它调试，胜过 print 1 小时。

17.12 实战注意清单

□ 所有 secret 走 env，不写死
□ 所有 tool 输入用 pydantic 校验
□ 所有异常 raise 之前包成 TextContent 返回
□ 所有写操作加 idempotency key（避免模型重复 call）
□ resource 走 read-only 视角，别在 read_resource 内调写 API
□ 文档化每个 tool：用途 + 输入 + 输出 + 错误
□ 给 server 写 CHANGELOG,标 protocolVersion 兼容
□ 高敏操作（如关闭工单）建议拆成单独 "destructive"工具并加 confirm

17.13 小结

  • Python uv + mcp[cli] 30 行起步
  • Tool / Resource 分工：模型调 = Tool，客户端拉 = Resource
  • 命名空间化 + 异常包成 TextContent 是稳定性的两条主线
  • 用 mcp-inspector 调试,不要 print

下一篇：《18 MCP 进阶：Resources / Prompts / Sampling》——把 MCP 的另外两个原语深入，并引入 sampling：让 server 反过来调模型。

Summary: 用 Python uv + mcp[cli] 30-60 行写一个能上生产的 tickets server；命名空间、异常包成 TextContent、用 mcp-inspector 调试是三件套。