用底层API造MCP：工具、资源、提示词三原语一网打尽

mkdir mcp-status-board && cd mcp-status-board
uv init --name mcp-status-board          # 或手动建 pyproject.toml
uv add "mcp[cli]>=1.2"                   # 装官方 Python SDK（含 CLI 辅助）

python -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]>=1.2"

mcp version                                # 能打印版本号即装通

import asyncio
import json
import os
from pathlib import Path
from typing import Any

import mcp.server.stdio
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl

app = Server("status-board")

STORE = Path(os.environ.get("BOARD_STORE", "board.json"))

# ---------- 工具：模型控制 ----------
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="add_task",
            description="向看板新增一个任务，返回新任务 ID",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string", "description": "任务标题"},
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                        "default": "medium",
                    },
                },
                "required": ["title"],
            },
        ),
        types.Tool(
            name="transition",
            description="把任务流转到下一状态：todo -> doing -> done",
            inputSchema={
                "type": "object",
                "properties": {
                    "task_id": {"type": "string"},
                    "to": {"type": "string", "enum": ["todo", "doing", "done"]},
                },
                "required": ["task_id", "to"],
            },
        ),
        types.Tool(
            name="summary",
            description="统计各状态的任务数量，返回简洁文本",
            inputSchema={"type": "object", "properties": {}},
        ),
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
    board = _load()
    if name == "add_task":
        tid = f"T{len(board['tasks']) + 1:03d}"
        board["tasks"].append({
            "id": tid, "title": arguments["title"],
            "priority": arguments.get("priority", "medium"),
            "status": "todo",
        })
        _save(board)
        return [types.TextContent(type="text", text=f"已新增 {tid}: {arguments['title']}")]
    if name == "transition":
        for t in board["tasks"]:
            if t["id"] == arguments["task_id"]:
                t["status"] = arguments["to"]
                _save(board)
                return [types.TextContent(type="text", text=f"{t['id']} -> {t['status']}")]
        raise ValueError(f"找不到任务 {arguments['task_id']}")
    if name == "summary":
        from collections import Counter
        c = Counter(t["status"] for t in board["tasks"])
        return [types.TextContent(type="text", text=json.dumps(dict(c), ensure_ascii=False))]
    raise ValueError(f"未知工具: {name}")

def _load() -> dict:
    if not STORE.exists():
        return {"tasks": []}
    return json.loads(STORE.read_text("utf-8"))

def _save(board: dict) -> None:
    STORE.write_text(json.dumps(board, ensure_ascii=False, indent=2), "utf-8")

# ---------- 资源：应用控制 ----------
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    board = _load()
    tasks = board["tasks"]
    return [
        types.Resource(
            uri=AnyUrl("board://tasks/all"),
            name="全部任务",
            description="看板上所有任务的原始 JSON",
            mimeType="application/json",
        ),
        types.Resource(
            uri=AnyUrl("board://config"),
            name="看板配置",
            description="状态流转规则与版本",
            mimeType="application/json",
        ),
        *[
            types.Resource(
                uri=AnyUrl(f"board://task/{t['id']}"),
                name=t["title"],
                description=f"{t['id']} [{t['status']}/{t['priority']}]",
                mimeType="application/json",
            )
            for t in tasks
        ],
    ]


@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    s = str(uri)
    if s == "board://tasks/all":
        return json.dumps(_load()["tasks"], ensure_ascii=False)
    if s == "board://config":
        return json.dumps({"flow": ["todo", "doing", "done"], "version": "1.0"})
    if s.startswith("board://task/"):
        tid = s.split("/")[-1]
        for t in _load()["tasks"]:
            if t["id"] == tid:
                return json.dumps(t, ensure_ascii=False)
    raise ValueError(f"未知资源: {s}")

# ---------- 提示词：用户控制 ----------
@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="standup",
            description="生成每日站会汇报，自动汇总看板状态",
            arguments=[],
        ),
        types.Prompt(
            name="triage",
            description="找出一批任务中的阻塞项与高优项",
            arguments=[
                types.PromptArgument(
                    name="count", description="最多列出几条", required=False,
                )
            ],
        ),
    ]


@app.get_prompt()
async def get_prompt(
    name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
    board = _load()
    tasks = board["tasks"]
    if name == "standup":
        doing = [t for t in tasks if t["status"] == "doing"]
        todo = [t for t in tasks if t["status"] == "todo"]
        body = (
            f"## 昨日/进行中\n"
            + "\n".join(f"- {t['id']} {t['title']}" for t in doing)
            + f"\n## 今日计划\n"
            + "\n".join(f"- {t['id']} {t['title']}" for t in todo[:3])
        )
    elif name == "triage":
        n = int(arguments.get("count", "5")) if arguments else 5
        top = sorted(
            [t for t in tasks if t["status"] != "done"],
            key=lambda t: (t["status"] != "doing", t["priority"] != "high"),
        )[:n]
        body = "请逐条分析以下任务是否阻塞，并给出流转建议：\n" + "\n".join(
            f"- {t['id']} [{t['status']}/{t['priority']}] {t['title']}" for t in top
        )
    else:
        raise ValueError(f"未知提示词: {name}")
    return types.GetPromptResult(
        description=f"Prompt {name}",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(type="text", text=body),
            )
        ],
    )

async def main() -> None:
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options(
                notification_options=NotificationOptions(),
            ),
        )


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

python -c "import server" && echo OK
# 更进一步：用 SDK 自带的 inspector
mcp dev server.py

claude mcp add status-board -- uv run --directory "$(pwd)" server.py
# 或纯 python：claude mcp add status-board -- python server.py

{
  "mcpServers": {
    "status-board": {
      "command": "uv",
      "args": ["run", "server.py"]
    }
  }
}

mcp-status-board/
├── pyproject.toml        # 依赖 mcp[cli]>=1.2
├── server.py             # 三原语 + 主循环
└── board.json            # 运行后自动生成的存储文件

flowchart TD
    subgraph Host["Claude Code（宿主）"]
        C["MCP Client"]
    end
    subgraph Srv["status-board Server（你造的）"]
        H["initialize 握手\n声明 capabilities"]
        T["tools/list → tools/call"]
        R["resources/list → resources/read"]
        P["prompts/list → prompts/get"]
        ST["board.json 存储"]
    end
    U["用户"] -->|"输入 /standup"| Host
    Host -->|"模型决定"| T
    Host -->|"应用注入"| R
    Host -->|"用户选中"| P
    C <-->|"JSON-RPC over stdio"| H
    T --> ST
    R --> ST
    P --> ST
    H --> T
    H --> R
    H --> P