08b · 动手写一个 MCP Server

import json
import sys

VERSION = "2025-06-18"
TOOLS = [{"name": "word_count", "description": "统计 text 的字符数，上限 20000 字符。",
          "inputSchema": {"type": "object", "required": ["text"],
                          "additionalProperties": False,
                          "properties": {"text": {"type": "string", "maxLength": 20000}}}}]

def error(rid, code, message):
    return {"jsonrpc": "2.0", "id": rid, "error": {"code": code, "message": message}}

def handle(req):
    if not isinstance(req, dict) or req.get("jsonrpc") != "2.0" or not isinstance(req.get("method"), str):
        return error(None, -32600, "Invalid Request")
    if "id" not in req:  # 通知不回复；完整生命周期交给 SDK
        return None
    rid, method = req["id"], req["method"]
    if isinstance(rid, bool) or not isinstance(rid, (str, int)):
        return error(None, -32600, "Invalid request id")
    params = req.get("params", {})
    if not isinstance(params, dict):
        return error(rid, -32602, "params must be an object")
    if method == "initialize":
        if params.get("protocolVersion") != VERSION:
            return error(rid, -32602, "This teaching subset only supports " + VERSION)
        result = {"protocolVersion": VERSION, "capabilities": {"tools": {}},
                  "serverInfo": {"name": "bare-notes", "version": "0.2.0"}}
    elif method == "tools/list":
        result = {"tools": TOOLS}
    elif method == "tools/call":
        args = params.get("arguments", {})
        if params.get("name") != "word_count":
            return error(rid, -32602, "Unknown tool")
        if not isinstance(args, dict) or set(args) != {"text"} or not isinstance(args["text"], str):
            return error(rid, -32602, "Expected only a string text argument")
        if len(args["text"]) > 20000:
            return error(rid, -32602, "text too long")
        result = {"isError": False, "content": [{"type": "text", "text": str(len(args["text"]))}]}
    else:
        return error(rid, -32601, "Method not found")
    return {"jsonrpc": "2.0", "id": rid, "result": result}

if __name__ == "__main__":
    print("读取 stdin；本示例是旧版协议教学子集", file=sys.stderr)
    for line in sys.stdin:
        if not line.strip():
            continue
        try:
            result = handle(json.loads(line))
        except json.JSONDecodeError:
            result = error(None, -32700, "Parse error")
        if result is not None:
            print(json.dumps(result, ensure_ascii=False, allow_nan=False), flush=True)

# 一个完整可跑的 stdio MCP Server
from mcp.server import MCPServer

mcp = MCPServer("notes")

@mcp.tool()
def word_count(text: str) -> int:
    """数一段文本有多少个字符。"""     # ⭐ 这句 docstring 就是模型看到的 description
    return len(text)

if __name__ == "__main__":
    mcp.run()                           # 默认 stdio

{"name": "word_count", "description": "数一段文本有多少个字符。",
 "inputSchema": {"type": "object", "required": ["text"],
                 "properties": {"text": {"title": "Text", "type": "string"}}},
 "outputSchema": {"type": "object", "properties": {"result": {"type": "integer"}}}}

# 三类东西各写一个，完整可跑
from mcp.server import MCPServer

mcp = MCPServer("notes")
NOTES: dict[str, str] = {}

@mcp.tool()                                   # ① 工具：模型自己决定要不要调
def save_note(title: str, body: str) -> str:
    """把一条笔记存起来。title 是唯一键，同名会覆盖。"""
    NOTES[title] = body
    return f"已保存《{title}》，共 {len(body)} 字"

@mcp.resource("note://{title}")               # ② 资源：像文件一样被读，用 URI 寻址
def read_note(title: str) -> str:
    """按标题读一条笔记的正文。"""
    return NOTES.get(title, "(没有这条笔记)")

@mcp.prompt()                                 # ③ 提示模板：用户从菜单里挑
def review(title: str) -> str:
    """生成一个「审阅这条笔记」的提示模板。"""
    return f"请审阅笔记《{title}》，指出三处可以更清楚的地方。"

if __name__ == "__main__":
    mcp.run()

# 🧩 骨架：和上一段是同一个文件，只有最后这一行不一样
mcp.run(transport="streamable-http", host="127.0.0.1", port=8931)

no_doc    -> description = ''
read_file -> description = '读一个文件的内容。path 必须是绝对路径。'

{"mcpServers": {
  "notes": {"command": "python",
            "args": ["C:/work/notes_server.py"],
            "env": {"NOTES_DB": "C:/work/notes.sqlite"}}}}

claude mcp add my-server -e API_KEY=xxx -- npx my-mcp-server
claude mcp add --transport http sentry https://mcp.sentry.dev/mcp
claude mcp list        # 列出来并做健康检查，实测每行结尾是 ✔ Connected

# 量一遍：一个工具定义在线上占多少字符
import asyncio, json
from mcp.server import MCPServer

mcp = MCPServer("issues")

@mcp.tool()
def search_issues(query: str, project: str, assignee: str = "", state: str = "open",
                  limit: int = 20) -> str:
    """在项目里按关键词搜索工单。
    query 是搜索词；project 是项目 key（如 ENG）；assignee 传用户名可只看某人的；
    state 取 open / closed / all；limit 是最多返回几条，默认 20，上限 100。
    返回每条工单的短索引、标题、状态、负责人。"""
    return "（这里返回搜索结果）"

async def main():
    t = (await mcp.list_tools())[0]
    s = json.dumps({"name": t.name, "description": t.description,
                    "inputSchema": t.input_schema}, ensure_ascii=False)
    print("一个工具的定义占", len(s), "个字符；20 个就是", len(s) * 20)

asyncio.run(main())

# 同一个失败，两种写法 —— 完整可跑
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer("t")

@mcp.tool()
def read_file_bad(path: str) -> str:
    """读文件（❌ 错误示范：抛一个普通异常）。"""
    raise ValueError("path 必须是绝对路径，你传的是 " + path)

@mcp.tool()
def read_file_good(path: str) -> str:
    """读文件（✅ 正确示范：抛 ToolError）。"""
    raise ToolError("path 必须是绝对路径。收到的是 '" + path + "'。"
                    "请改成以盘符或 / 开头的完整路径，例如 /data/a.txt，然后重试。")

if __name__ == "__main__":
    mcp.run()