🔌 第6章 动手写一个MCP
📄 一句话概括

一个能用的 Server 只要一百多行:工具表、存取、校验、三个实现、协议循环。 最该逐行看的是校验那一段——它是这类 Server 最容易写漏的地方。 还有一个坑不分语言:往 stdout 打一行日志,协议就废了。

📝 本章和前面几章的关系

上一节把环境备好了,这一节真写。 教材用的是官方 SDK; 这一节用纯标准库写,好处是能看清每一行在干什么,坏处是要自己写协议循环——也就三十九行。 写完这一个,再看任何框架都会觉得「它只是把这几十行包起来了」。

一、三条工具,先想清楚边界

note.add 写、note.list 读、note.search 读。两读一写。 写的那条要能被宿主识别出来,才好弹确认框——第 3 章那道闸就靠这个信息。 参数上给了 minimum 和 maximum:limit 是 1 到 50,默认 10。这不是装饰,服务端会照着它查。

三个工具的分工也值得看一眼:一个写、一个按时间列、一个按关键词过滤。 没有「删除」——不是忘了,是删了收不回来,先不给。 工具表就是契约。 你在里面写什么,模型就照着理解什么,宿主也照着它画输入框。

二、代码分成五块

五块各管一段:工具表 20 行、存取 10 行、校验 17 行、三个实现 21 行、协议循环 39 行。 合计 107 行,全文件 149 行,剩下的 42 行是常量和空行。 工具表是数据,不是代码。 它是一段 JSON,改它不需要动逻辑。

存取那 10 行就是读一个 json 文件、写一个 json 文件。没有数据库、没有 ORM。 这就是第 7 章要讲的「共享记忆」最朴素的样子。 协议循环 39 行:读一行、解析、分派、写一行、flush。flush 那一句最容易漏,漏了就是宿主一直等。

三、校验那一段,最容易写漏

演示里贴了完整的 check 函数:必填、类型、上下界,三样都查。 必填的判断是 k not in args or args[k] in (None, "")——注意后半句,传了空字符串也算缺。 类型的判断靠一张小表:string 对 str、integer 对 int、array 对 list、boolean 对 bool。

上下界只对整数查,所以前面要加 isinstance 判断,否则字符串跟数字比大小会抛异常。 少查一样,就多一类脏数据能进库。 而脏数据进来之后,受影响的不是这一条,是后面每一次读。

四、一个能毁掉一切的 print

这一段是本节最实用的。演示里往 stdout 打了一行 server started,客户端读到的第一行就是它。 它不是 JSON,解析失败,回 -32700 Parse error,一条正常报文都收不到。 stdout 是协议通道,只能走 JSON-RPC;日志一律走 stderr。

这条坑不分语言:Node 里是 console.log,Go 里是 fmt.Println,Python 里是 print。 写 MCP Server 的第一天就会遇到。 排查的时候记住:Server 一启动就不说话,先看它往 stdout 打了什么。

五、交给宿主的配置,以及在 Trae 里配一次

配置还是那三样:command、args、env。 args 的第一个元素必须是 -u,command 写绝对路径更稳——不靠 PATH。 教材里是在 Trae 里配的,步骤照原样记下来(本沙箱不执行): 打开 Trae 的设置,找到 MCP 那一栏,点添加,把上面那段 json 贴进去, 保存之后列表里会多出一个 notes,右边有个开关和工具数。

判据是三步:状态灯变绿、工具数显示 3、点开能看到 note.add 的描述。三样都对了才算接上。 接不上就看两处:一是把那段 json 单独跑一遍(command 加 args 能不能起进程), 二是看宿主自己的日志里有没有读到这份配置。

六、动手 Demo:从零写一个 notes-mcp

一个一百多行的 Server:三条工具、一个 json 文件存数据、一段手写校验、三十九行协议循环。 真起子进程跑一遍,看握手、工具表、三次调用、三种错。 最后故意加一行 print,看它怎么把协议毁掉。

TOOLS = [
    {'name': 'note.add',
     'description': '新增一条笔记,返回它的编号',
     'inputSchema': schema({'text': prop('string', '笔记正文'),
                            'tags': prop('array', '标签,可空')}, ['text'])
     },
    {'name': 'note.list',
     'description': '按时间倒序列出最近的笔记',
     'inputSchema': schema({'limit': dict(prop('integer', '要几条,默认 10'),
                                           minimum=1, maximum=50)}, [])
     },
    {'name': 'note.search',
     'description': '在笔记正文里搜关键词,返回命中的',
     'inputSchema': schema({'keyword': prop('string', '关键词')}, ['keyword'])
     },
]


def check(sch, args):
    bad = []
    for k in sch.get('required', []):
        if k not in args or args[k] in (None, ''):
            bad.append('缺必填 ' + k)
    for k, p in sch.get('properties', {}).items():
        if k not in args:
            continue
        if p['type'] == 'integer' and not isinstance(args[k], int):
            bad.append(k + ' 应为 integer')
        if 'maximum' in p and isinstance(args[k], int) \
                and args[k] > p['maximum']:
            bad.append(k + ' 不能大于 %d' % p['maximum'])
    return bad


def main():
    for line in sys.stdin:                 # 一行一条报文
        msg = json.loads(line)
        resp = handle(msg)                 # 分派到具体的工具
        sys.stdout.write(json.dumps(resp, ensure_ascii=False) + '\n')
        sys.stdout.flush()                 # 少这一行,宿主就一直等

实跑输出:

--------------------------------------------------------------------
从零写一个 notes-mcp
--------------------------------------------------------------------


  ① 三条工具,先想清楚边界
  --------------------------------------------------------------------

| 工具 | 读还是写 | 参数 | 返回什么 |
|---|---|---|---|
| note.add | 写 | text 必填,tags 可选 | 返回新编号 |
| note.list | 读 | limit 1 到 50,默认 10 | 倒序 |
| note.search | 读 | keyword 必填 | 正文里包含就算命中 |

  三条工具两读一写。写的那条【要能被宿主识别出来】,才好弹确认框。
  参数上给了 minimum 和 maximum——不是装饰,服务端会照着它查。

  ② 代码分成四块
  --------------------------------------------------------------------

| 块 | 干什么 | 行数 |
|---|---|---:|
| 工具表 TOOLS | 三条工具的 name、description、inputSchema | 20 |
| 存取 load 与 save | 读写同目录下的 notes.json | 10 |
| 校验 check | 必填、类型、上下界 | 17 |
| 三个实现 | 加一条、列几条、按关键词过滤 | 21 |
| 协议循环 main | 读一行、分派、写一行 | 39 |

  全文件 149 行,五块加起来 107 行,剩下的是常量和空行。

最值得逐行看的是校验这一段,它是这类 Server 最容易写漏的地方:
  def check(sch, args):
      bad = []
      for k in sch.get("required", []):
          if k not in args or args[k] in (None, ""):
              bad.append("缺必填 " + k)
      ty = {"string": str, "integer": int, "array": list, "boolean": bool}
      for k, p in sch.get("properties", {}).items():
          if k not in args:
              continue
          t = p["type"]
          if t in ty and not isinstance(args[k], ty[t]):
              bad.append(k + " 应为 " + t)
          if "minimum" in p and isinstance(args[k], int) and args[k] < p["minimum"]:
              bad.append(k + " 不能小于 %d" % p["minimum"])
          if "maximum" in p and isinstance(args[k], int) and args[k] > p["maximum"]:
              bad.append(k + " 不能大于 %d" % p["maximum"])
      return bad
  这一段把必填、类型、上下界全查了。少查一样,就多一类脏数据能进库。

  ③ 真跑:握手、列表、三个工具
  --------------------------------------------------------------------
  serverInfo = {"name": "notes-mcp", "version": "0.1.0"}
  tools/list 回来 3 条:note.add、note.list、note.search

  note.add     {"id": 1, "total": 1}
  note.add     {"id": 2, "total": 2}
  note.list    {"count": 2, "items": [{"id": 2, "text": "tool 描述会进模型上下文"}, {…
  note.search  {"count": 1, "items": [{"id": 1, "text": "stdio 就是两根管道"}]}

  ④ 三种错,错在三个地方
  --------------------------------------------------------------------

| 调用 | 回来的 code | data |
|---|---|---|
| limit=99 | -32602 | limit 不能大于 50 |
| 工具名不存在 | -32602 | 没有工具 note.nope |
| tags 传了字符串 | -32602 | tags 应为 array |

  三个都是 -32602,因为都是「参数不合法」这个大类。
  【具体哪里不对,写在 data 里】——这一栏是给人看的,别省。

  ⑤ 一个能毁掉一切的 print
  --------------------------------------------------------------------
  客户端读到的第一行:server started
  能解析成 JSON:不是
  -> Parse error,一条正常报文都没收到
  Server 里只加了一行 print,往 stdout 打了句 server started,协议就废了。
  【stdout 是协议通道,只能走 JSON-RPC;日志一律走 stderr。】
  这条坑不分语言,写 MCP Server 的第一天就会遇到。

  ⑥ 交给宿主的配置
  --------------------------------------------------------------------
    {
      "mcpServers": {
        "notes": {
          "command": "python",
          "args": [
            "-u",
            ".../notes-mcp/src/server.py"
          ],
          "env": {
            "PYTHONIOENCODING": "utf-8"
          }
        }
      }
    }
  command 写 python 也行,但写解释器的绝对路径更稳——不靠 PATH。
  args 的第一个元素必须是 -u。
  【这两条是 Windows 上最省事的两个动作】,第 4 章第 3 节讲过为什么。
  ! 到这里,一个能用的 Server 就有了。它没有框架、没有依赖、一百多行。

①第一张表三行,最后一列写清返回什么。这一栏在真机上就是模型判断「调完能拿到什么」的依据。 ②第二张表五行的行数是真算出来的:20、10、17、21、39。加起来 107,剩下的 42 行是常量和空行。 ③第三段那段 check 函数值得抄下来。三个 if 各管一类错:缺必填、类型不对、超范围。 少一个就多一类脏数据。 ④第四段是本节最值钱的坑:一行 print 让整条连接作废。 注意它报的是 -32700,不是参数错——因为连 JSON 都不是。 ⑤最后那段配置里 command 写的是 python,正文里补了「写绝对路径更稳」。 这两条是 Windows 上最省事的两个动作。

30 秒实验

  1. 给这个 Server 加一条 note.del,参数是 id。想想它该不该默认需要确认。
  2. 把 check 里的上下界判断去掉 isinstance,传一个字符串进去,看会抛什么异常。
  3. 在协议循环里故意漏掉 flush,看客户端会等多久。

常见误区

  1. 「Server 要用官方 SDK 才能写。」一百多行标准库就够。SDK 是省事,不是门槛。
  2. 「日志用 print 就行。」print 走 stdout,那是协议通道。日志一律走 stderr。
  3. 「参数校验写在描述里,模型会遵守。」模型不会严格执行描述。约束进 schema,服务端再查一遍。
自测三问(点开对答案) 1. 一个最小可用的 Server 由哪几块组成? —— 工具表、存取、校验、工具实现、协议循环。标准库里一百多行能写完。 2. 往 stdout 打一行日志会怎样? —— 客户端读到的第一行不是 JSON,解析失败回 -32700,整条连接作废。日志要走 stderr。 3. 校验里最容易漏的是哪一样? —— 上下界,而且要先判断是不是整数再比大小,否则字符串跟数字比较会抛异常。

Server 有了,还缺一半:谁去连它、谁把工具表拼成话交给模型。下一节写一个 Client,把这条回路闭上。

相关笔记