上篇我说了,花了一个月研究 MCP,发现它跟 Harness 是天然搭档。但说实话,光看协议文档是看不出来什么东西的——你得真的动手写一个。
我写第一个 MCP Server 的契机其实挺土的。有次我跑了一个 Agent 任务,结果文件到处乱丢——有在/tmp的,有在工作目录的,还有跑到~/.cache下面的。我当时就想,能不能让 Agent 读写文件的时候统一走一个"文件管家",别自己乱写?
然后我就开始搭 MCP Server 了。
先说环境准备。MCP 官方推荐用 Python 或 TypeScript 写 Server,我选了 Python——不是因为它多好,是因为我手头一堆 Python 脚本,懒得翻译。你需要的其实就三个东西:Python 3.10+、pip install mcp、以及一个文本编辑器。
笑死,我第一次跑pip install mcp的时候,还以为会装一堆依赖,结果就一个包——mcp本身不到 200KB。这个包封装了 MCP 协议的全部通信逻辑,包括 JSON-RPC 的消息格式、请求/响应的生命周期管理、以及stdio和SSE两种传输模式。你不需要自己写 JSON 解析,也不需要处理 WebSocket 连接,暴露一个函数就是"加一个 Tool"。
来看最小实现。我写了一个能读文件和写文件的 MCP Server,核心代码加注释不到 60 行:
from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio # 创建一个 Server 实例 server = Server("file-manager") # 注册一个 Tool:读文件 @server.list_tools() async def handle_list_tools() -> list: return [ { "name": "read_file", "description": "读取指定路径的文件内容", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"} }, "required": ["path"] } }, { "name": "write_file", "description": "写入内容到指定路径的文件", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"}, "content": {"type": "string", "description": "文件内容"} }, "required": ["path", "content"] } } ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list: if name == "read_file": with open(arguments["path"], "r") as f: content = f.read() return [{"type": "text", "text": content}] elif name == "write_file": with open(arguments["path"], "w") as f: f.write(arguments["content"]) return [{"type": "text", "text": "写入成功"}] async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, InitializationOptions( server_name="file-manager", server_version="0.1.0" ))跑起来就一行:python file_manager.py。然后这个 Server 就开始监听stdio,等着 MCP Client 来调它。
你可能会问,这跟写一个普通的 Python 脚本有什么区别?区别在于,它暴露的是标准化的接口。任何支持 MCP 协议的客户端——Claude Desktop、VS Code 的 Copilot、当然也包括 Harness——都能直接调这两个函数,不需要知道底层是怎么实现的。
但这里有个坑。我第一次跑通之后,兴冲冲地让 Harness 去调这个 Server,结果发现——MCP 协议默认走的是stdio,也就是子进程的标准输入输出流。这意味着 MCP Server 和 Client 必须在同一个进程树里。Harness 跑 Agent 的时候,Agent 本身是一个子进程,如果 Agent 再启一个 MCP Server 子进程,进程管理就变得很复杂。你关 Agent 的时候,MCP Server 可能没关掉,变成孤儿进程挂在系统里。
我查了一下,MCP 协议支持两种传输模式,除了stdio还有SSE(Server-Sent Events,服务端推送事件)。SSE模式把 MCP Server 跑成一个 HTTP 服务,客户端通过 HTTP 请求来调用工具。这样 Client 和 Server 就解耦了,各自独立管理生命周期。
我改了一下,把stdio换成SSE:
from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route sse = SseServerTransport("/messages") async def handle_sse(request): async with sse.connect_sse(request.scope, request.receive, request._send) as streams: await server.run(streams[0], streams[1], InitializationOptions( server_name="file-manager", server_version="0.1.0" )) app = Starlette(routes=[ Route("/sse", endpoint=handle_sse), Route("/messages", endpoint=sse.handle_post_message), ])跑起来变成uvicorn file_manager:app --port 8000。现在 Harness 那边配置一个 HTTP 工具的地址为http://localhost:8000/sse,就能调这个 MCP Server 了。
你别说,我在 Harness 里配好之后,对着 Agent 说了一句"帮我把/tmp/result.txt的内容读出来,然后加一行 'done' 再写回去"——Agent 真的自己调了read_file再调write_file,整个过程不到 3 秒。我盯着终端看了半天,感觉这玩意儿比我想象的实用得多。
当然,这个版本还有很多问题。比如没有路径安全校验,Agent 说read_file /etc/passwd它真的会读。也没有并发控制,同时来两个请求可能会写冲突。更没有错误重试,网络闪断一次就挂了。
但话说回来,60 行代码就能让 Harness 拥有一个"文件管理工具",而且这个工具不仅 Harness 能用,其他任何 MCP 客户端也能用——这个性价比,我觉得挺值的。
下一篇,咱们搞个数据库 MCP Server,让 Agent 自己查数据库、写报表。你平时写 Agent 的时候,最想让它能自动调什么工具?评论区说说,我考虑优先安排上。