)
1. 从零手搓一个能被 Agent 调用的 MCP 服务到底难在哪如果你正在学 Agent 开发大概率已经听过 MCPModel Context Protocol模型上下文协议这个词。简单说它是一套让大语言模型安全调用外部工具、数据源的标准化协议。你可以把它理解成「AI 世界的 USB 接口」只要你的服务按 MCP 规范暴露工具任何支持 MCP 的客户端都能即插即用不用为每个模型单独写适配层。这篇是 Agent 学习笔记系列的第三篇聚焦一个最小可跑通的场景用 Python FastMCP 从零写一个 MCP 服务注册两个工具写文件、查天气用 SSE 传输方式启动最后用客户端发起一次真实调用验证成功。适合已经会写 Python 函数、想搞明白 MCP 服务端到底长什么样的同学。我试过直接啃官方协议文档结果被 JSON-RPC 的消息格式绕晕。后来发现 FastMCP 这个框架把协议细节全封装了你只需要写普通函数加个装饰器它自动帮你生成工具描述、参数 schema、路由分发。所以这篇不讲协议底层只讲能跑起来的代码。整篇会覆盖环境依赖、服务端完整代码、SSE 与 STDIO 的取舍、客户端联调、以及几个我踩过的报错。跟着敲一遍你就能拥有一个自己的 MCP 服务后面接 Claude Code、Cline 或者自研 Agent 都行。2. 前置准备Python 环境、依赖清单与 TaoToken 统一通道先说环境。Python 建议 3.10 以上FastMCP 用到了较新的类型注解特性。依赖就三个包一条命令搞定pip install mcp requests python-dotenvmcp是官方 SDK里面包含 FastMCP 框架requests用来调外部 APIpython-dotenv负责从.env文件读密钥避免硬编码。这里插一句关于模型通道的事。MCP 服务本身只负责「提供工具」真正决定调用哪个工具的是背后的大模型。如果你本地还没配好可用的模型 API可以用 TaoToken 的统一 Key 通道它把多家模型的调用收敛成一个 Base URL 和一把 Key省得你为每个模型单独申请。配置入口在控制台Key 在 API Keys 页面生成控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite它的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的请求格式。也就是说你在客户端里把base_url指向它、api_key填上生成的 Key、model填对应模型 ID就能跑起来。这三件套Base URL Key Model ID是后面客户端联调的关键先记下来。再准备一个天气 API 的 Key。我用的是高德开放平台的天气接口免费额度够学习用。注册后在控制台创建一个「Web 服务」类型的 Key拿到一串字符。这个 Key 不要写进代码放到.env里。目录结构建议这样mcp-demo/ ├── mcp_server.py ├── mcp_client.py ├── .env └── .gitignore.gitignore里务必加一行.env防止密钥被提交。这是新手最容易犯的错我见过有人把带 Key 的仓库推到公开平台几分钟就被扫走刷爆额度。3. 可复制配置FastMCP 服务端完整代码与 SSE 启动先给完整代码再逐段拆。新建mcp_server.pyimport os import logging import requests from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP load_dotenv() logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) AMAP_API_KEY os.environ.get(AMAP_API_KEY, ) WRITE_DIR os.environ.get(WRITE_DIR, ./sandbox) os.makedirs(WRITE_DIR, exist_okTrue) mcp FastMCP(demo-server) mcp.tool() def write_file(filename: str, content: str) - str: 把内容写入指定文件filename 只能是文件名不能带路径 safe_name os.path.basename(filename) if not safe_name: return 错误文件名非法 target os.path.join(WRITE_DIR, safe_name) try: with open(target, w, encodingutf-8) as f: f.write(content) logger.info(f写入成功: {target}) return f文件 {safe_name} 写入成功 except Exception as e: logger.error(f写入失败: {e}) return f写入失败: {str(e)} mcp.tool() def get_weather(city: str) - str: 查询某个城市或区县的实时天气与预报 if not AMAP_API_KEY: return 错误未配置 AMAP_API_KEY if not city or len(city.strip()) 0: return 错误城市名不能为空 url ( https://restapi.amap.com/v3/weather/weatherInfo f?city{city}key{AMAP_API_KEY}extensionsall ) try: resp requests.get(url, timeout10) data resp.json() if data.get(status) ! 1: return f查询失败: {data.get(info)} return str(data.get(forecasts)) except Exception as e: return f请求异常: {str(e)} if __name__ __main__: mcp.settings.host 0.0.0.0 mcp.settings.port 8001 mcp.run(transportsse)配套的.envAMAP_API_KEY你的高德Key WRITE_DIR./sandbox启动命令python mcp_server.py看到类似Uvicorn running on http://0.0.0.0:8001的日志说明 SSE 服务起来了。此时访问http://127.0.0.1:8001/sse会保持一个长连接这就是 MCP 客户端要连的端点。几个设计点解释一下。FastMCP(demo-server)里的名字会出现在客户端工具列表里方便区分多个服务。mcp.tool()装饰器把普通函数变成 MCP 工具函数名就是工具名docstring 就是给模型看的工具说明——模型靠这段文字判断什么时候该调它所以写清楚很重要。参数的类型注解会被 FastMCP 转成 JSON Schema模型据此知道要传什么类型的值。write_file里我加了os.path.basename做路径清洗防止../../etc/passwd这种路径遍历。WRITE_DIR限定在沙箱目录即使被恶意调用也写不到系统关键位置。get_weather里做了 Key 检查、空值检查、超时设置和状态码判断异常统一返回字符串而不是抛出去这样模型能收到可读的错误信息。关于传输方式的选择这里展开说下。FastMCP 支持三种传输方式通信机制适用场景特点stdio标准输入输出管道本地 IDE 插件、CLI 工具低延迟、无需网络、进程紧耦合sseHTTP 长连接单向推送远程服务、跨机器调用支持网络访问、可多客户端streamable-http离散 HTTP 请求响应微服务、通用远程调用最通用、易调试、易负载均衡本地开发调试用 stdio 最省事客户端直接拉起进程通信。但如果你想让自己电脑上的 MCP 服务被局域网里另一台机器上的 Agent 调用就得用 SSE。这篇选 SSE 是因为它更接近真实部署形态而且能直观看到端口和连接。4. 验证请求用客户端发起一次真实工具调用服务端跑起来了怎么确认它真的能被调用写个最小客户端mcp_client.py。MCP 官方 SDK 提供了客户端类但为了让你看清 SSE 通信过程我用更直白的方式先拉工具列表再调一次天气工具。import asyncio from mcp.client.sse import sse_client from mcp import ClientSession async def main(): async with sse_client(http://127.0.0.1:8001/sse) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具:) for t in tools.tools: print(f - {t.name}: {t.description}) result await session.call_tool( get_weather, {city: 110000} ) print(\n天气调用结果:) print(result.content[0].text) result2 await session.call_tool( write_file, {filename: hello.txt, content: MCP 联调成功} ) print(\n写文件结果:) print(result2.content[0].text) if __name__ __main__: asyncio.run(main())运行python mcp_client.py正常输出类似可用工具: - write_file: 把内容写入指定文件... - get_weather: 查询某个城市或区县的实时天气与预报 天气调用结果: [{city: 北京市, casts: [...]}] 写文件结果: 文件 hello.txt 写入成功看到工具列表和两次调用都返回结果说明整条链路通了客户端通过 SSE 连上服务端initialize完成握手list_tools拿到工具清单call_tool触发实际函数执行。去./sandbox/hello.txt看一眼内容应该就是「MCP 联调成功」。这一步的意义在于你已经有了一个「模型可以调用的工具服务」。接下来只要在支持 MCP 的客户端里配置这个 SSE 端点模型就能自主决定调用get_weather还是write_file。比如在 Claude Code 或 Cline 里配置项通常长这样以 JSON 为例{ mcpServers: { demo-server: { url: http://127.0.0.1:8001/sse, transport: sse } } }如果你用的是需要模型 Key 的客户端把前面提到的三件套填进去Base URL 用https://taotoken.net/apiKey 用你在 API Keys 页面生成的Model ID 按文档选。这样模型侧和工具侧就都齐了。5. 本篇常见报错排查401、连接失败与工具不显示这一节按真实遇到的报错来。你照着敲大概率会撞上下面几个。报错一401 Unauthorized 或 invalid api key。这个通常不是 MCP 服务本身的错而是客户端调模型时 Key 不对。检查三处Key 有没有复制完整前后别带空格、Base URL 是不是https://taotoken.net/api注意结尾没有多余斜杠、Model ID 是否拼写正确。如果用的是环境变量确认.env被正确加载load_dotenv()要在读取os.environ之前调用。报错二Connection refused 或 local proxy failed。客户端连不上127.0.0.1:8001。先确认服务端进程还活着终端里有没有异常退出。再确认端口没被占用换8002试试。如果是跨机器调用mcp.settings.host要设成0.0.0.0而不是127.0.0.1否则只监听本机。防火墙也要放行对应端口。报错三Error reading choices 或返回结构解析失败。这类多半是模型返回格式和客户端预期不一致。检查你填的 Model ID 是否支持工具调用function calling有些纯对话模型不支持 tools 参数。另外确认客户端版本老版本对 MCP 的 SSE 支持可能不完整升级到较新版本。报错四工具列表为空list_tools 返回空数组。说明服务端没注册上工具。常见原因是装饰器写成了mcp.tool少了括号或者函数定义在mcp FastMCP()之前。还有一种情况是函数有语法错误导致模块导入失败但进程没崩工具自然没注册。看服务端启动日志有没有 traceback。报错五OAuth 相关报错。如果你接的远程 MCP 服务要求 OAuth 授权而本地客户端没配会卡在授权环节。本地自建服务一般用不到 OAuth遇到就检查是不是误配了需要鉴权的远程端点。自建服务保持无鉴权或简单 Token 即可。报错六写文件报 Permission denied。沙箱目录没创建成功或者当前用户对该目录没写权限。代码里os.makedirs(WRITE_DIR, exist_okTrue)应该能处理但如果WRITE_DIR指向了系统保护目录就会失败。换成项目内的相对路径最稳。排查思路统一是先看服务端日志再看客户端日志最后看网络连通性。MCP 的报错信息有时候比较隐晦养成在工具函数里打日志的习惯能省很多时间。6. 继续往下走把 MCP 服务接进你的 Agent 工作流到这里你已经有了一个能跑、能被调用、有基本安全防护的 MCP 服务。它现在只有两个工具但扩展方式极其简单——再写一个函数加mcp.tool()就行。比如加个读文件、查数据库、调内部 API 的工具模型立刻就能用上。如果你想让这个服务长期跑着给编码 Agent 用可以考虑把它注册到 Coding Plan 里统一管理配合前面说的统一 Key 通道模型侧和工具侧就都收敛到一套配置了Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite模型对话快速验证模型是否正常https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite最后留个实用技巧调试 MCP 工具时先用客户端脚本单独调call_tool确认函数逻辑没问题再交给模型去决策。因为模型调用失败时你很难分清是模型没选对工具还是工具本身报错。分开验证定位快得多。另外工具函数的 docstring 一定要写清楚「什么时候用」这是模型选工具的唯一依据比参数说明还重要。