MCP学习 - 其一 - MCP服务组件理解:从Resources、Prompts到Sampling的TaoToken实践
1. 先把 MCP 服务端四个组件摆到桌面上Resources、Prompts、Tools、Sampling 到底谁管什么第一次看 MCP 文档的人十有八九会在 Resources、Prompts、Tools、Sampling 这四个词之间打转。它们都挂在同一个 FastMCP 实例上写法也长得像但职责边界完全不同。我按自己的理解先给一个不绕弯的对照Tools 是模型能主动调用的“动作”Resources 是模型按 URI 去读的“数据接口”Prompts 是服务端预置好的“工作流模板”Sampling 则是服务端反过来请求客户端模型生成内容的“回调通道”。前三个是客户端向服务端要东西Sampling 是服务端向客户端要东西方向正好相反这一点想通了后面代码就顺了。为什么初次接触容易懵因为很多示例把四者混在一个文件里mcp.tool()、mcp.resource()、mcp.prompt()挨着写读者会以为它们只是装饰器名字不同。实际上它们对应 MCP 协议里不同的方法族tools/list、tools/call、resources/read、resources/templates/list、prompts/list、prompts/get以及 Sampling 触发的sampling/createMessage。你在客户端调哪个方法服务端就走哪条注册路径。这篇面向的是刚上手 MCP 服务端的人目标很具体用 FastMCP 在本地起一个最小可运行服务把 Resources 读取、Prompts 模板渲染、Sampling 回调三条链路分别跑通模型侧统一走 TaoToken 的 Key 和 API 通道避免在多个模型供应商之间来回切配置。读完你应该能自己判断某个需求该注册成 Tool 还是 Resource某个“让模型再生成一段”的逻辑该不该用 Sampling。先给一个判断口诀后面所有代码都围绕它展开Tool 解决“做事”Resource 解决“取数”Prompt 解决“套模板”Sampling 解决“服务端请模型帮忙生成”。适合谁看写过一点 Python、装过 FastMCP、但被MCP error -32602或 Sampling 没反应卡住的人。如果你还没装 FastMCP先执行pip install mcp本文不重复讲安装。下面从环境准备开始一步步把服务端和客户端都搭起来。2. TaoToken 前置准备统一 Key 与 API 通道让 Sampling 有模型可用Sampling 这条链路最容易“看起来通了其实没通”原因是它需要客户端侧真的有一个能响应sampling/createMessage的模型。很多本地示例用 Ollama 顶但一旦你想换成云端模型Base URL、Key、Model ID 三件套就得重新配。我的做法是统一走 TaoToken 的 API 通道服务端和客户端都指向同一个入口减少变量。先拿 Key。打开https://taotoken.net/api-keys登录后创建一个 API Key复制出来形如sk-开头的一串。这个 Key 同时用于模型对话和后续 Coding Plan 场景建议单独建一个给 MCP 实验用方便随时吊销。拿到 Key 后模型侧调用统一走https://taotoken.net/api注意这个地址不带任何查询参数。模型 ID 按你实际要用的填比如做 Sampling 验证时选一个响应快的小模型即可别一上来就用大模型回调链路调试阶段 token 消耗会很明显。如果你后面要长期跑编码类 Agent可以了解下 Coding Plan入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcc_plan它和按量调用是两条线实验阶段用按量就够。环境变量建议这样设服务端和客户端共用export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你的模型IDWindows PowerShell 用$env:TAOTOKEN_API_KEYsk-...。设完用echo $TAOTOKEN_API_KEY确认非空这一步别省后面 401 报错八成是这里没生效。关于接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc里有各语言的调用示例遇到参数不确定时对照一下比猜快。模型对话的在线调试入口在https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodels可以先用它确认 Key 和模型 ID 是通的再去调 MCP能省掉一半排障时间。这里强调一个边界TaoToken 是模型 API 通道不是编辑器替代品也不是 MCP 服务本身。MCP 服务端仍然跑在你本地TaoToken 只负责 Sampling 回调时那次模型生成。把这两层分清配置就不会乱。3. 可复制配置FastMCP 服务端四组件注册与 settings 片段这一节给完整可复制的服务端代码。文件名建议weather_server.py放在一个干净目录里。先看配置片段我用一个settings.json存模型侧参数方便客户端读取{ mcpServers: { WeatherAssistant: { url: http://localhost:9000/mcp, transport: streamable_http } }, model: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, model_id: 你的模型ID } }注意base_url写的是https://taotoken.net/api不带 UTM也不带/v1后缀具体路径由 SDK 拼接。api_key_env表示从环境变量读不要把 Key 明文写进文件。下面是服务端主体四个组件都注册上并修正了 URI 里下划线导致的-32602问题from mcp.server.fastmcp import FastMCP, Context from mcp.types import SamplingMessage, TextContent mcp FastMCP(WeatherAssistant, port9000) mcp.tool() def get_weather(city: str) - str: Return fake weather for a given city return fIts sunny and 30C in {city}. # 注意URI 协议部分用连字符不要用下划线 mcp.resource(weather-history://{city}) def get_weather_history(city: str) - str: Get historical weather data for a city return fHistorical weather for {city}: Sunny on Monday, Rainy on Tuesday. mcp.resource(greeting://{name}) def get_greeting(name: str) - str: Get a personalized greeting return fHello, {name}! mcp.resource(version://current) def get_version() - str: Get the version of the MCP server return 1.0.0 mcp.prompt() def weather_report_template(city: str, style: str friendly) - str: Generate a weather report prompt for a given city and style styles { friendly: Please write a warm, friendly weather report, formal: Please write a formal, professional weather report, casual: Please write a casual, relaxed weather report, } return ( f{styles.get(style, styles[friendly])} for {city} fbased on the following data: {get_weather(city)} fand {get_weather_history(city)}. ) mcp.tool() async def generate_weather_report(city: str, style: str friendly, ctx: Context None) - str: Generate a weather report using LLM sampling prompt ( fGenerate a weather report for {city} in a {style} style. fCurrent weather: {get_weather(city)}. fHistorical data: {get_weather_history(city)}. ) result await ctx.session.create_message( messages[ SamplingMessage( roleuser, contentTextContent(typetext, textprompt), ) ], max_tokens200, system_promptYou are a helpful weather assistant providing accurate and engaging weather reports., ) if result.content.type text: return result.content.text return str(result.content) if __name__ __main__: mcp.run(streamable-http)几个关键点单独说。第一mcp.resource(weather-history://{city})里的协议段必须是合法 URI 字符下划线_在部分解析器里会被拒报MCP error -32602: Invalid request parameters换成连字符-就正常这是实测踩过的坑。第二version://current这种固定资源不要写成version:带//更规范。第三Sampling 必须在async工具里用ctx.session.create_message同步函数里拿不到 session。启动服务python weather_server.py看到监听 9000 端口的日志就说明服务端起来了。此时 Resources、Prompts、Tools 已经注册完毕Sampling 的入口挂在generate_weather_report这个工具上等客户端来触发。4. 逐项验证Resources 读取、Prompts 渲染、Sampling 回调的成功结果服务端起来后别急着上 Agent先用最直接的方式逐条验证。我习惯用 MCP Inspector 或直接写客户端脚本这里给客户端脚本顺便把模型侧接到 TaoToken。先装依赖pip install langchain langchain-openai langchain-mcp-adapters客户端脚本client_test.pyimport asyncio import json import os from langchain_openai import ChatOpenAI from langchain_mcp_adapters.client import MultiServerMCPClient mcp_client MultiServerMCPClient( { WeatherAssistant: { url: http://localhost:9000/mcp, transport: streamable_http, } } ) model ChatOpenAI( modelos.environ[TAOTOKEN_MODEL], base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], temperature0, ) async def read_resource(uri: str) - dict: return await mcp_client.request(methodresources/read, params{uri: uri}) async def get_prompt(name: str, arguments: dict) - dict: return await mcp_client.request( methodprompts/get, params{name: name, arguments: arguments} ) async def list_templates() - dict: return await mcp_client.request(methodtemplates/list, params{}) async def list_prompts() - dict: return await mcp_client.request(methodprompts/list, params{}) async def main(): tools await mcp_client.get_tools() print(Available tools:, [t.name for t in tools]) resource_result await read_resource(weather-history://Shanghai) print(Resource result:, json.dumps(resource_result, indent2, ensure_asciiFalse)) templates await list_templates() print(Resource templates:, json.dumps(templates, indent2, ensure_asciiFalse)) prompt_result await get_prompt( weather_report_template, {city: Tokyo, style: casual} ) print(Prompt result:, json.dumps(prompt_result, indent2, ensure_asciiFalse)) prompt_text prompt_result[result][messages][0][content][text] llm_result await model.ainvoke(prompt_text) print(LLM result from prompt:, llm_result.content) prompts await list_prompts() print(Prompt templates:, json.dumps(prompts, indent2, ensure_asciiFalse)) if __name__ __main__: asyncio.run(main())运行python client_test.py预期看到四类输出。Resources 读取返回Historical weather for Shanghai: Sunny on Monday, Rainy on Tuesday.说明resources/read通了。templates/list返回weather-history://{city}和greeting://{name}两个模板说明 URI 模板注册正确。prompts/get返回渲染后的文本里面已经拼进了天气和历史数据说明 Prompts 模板渲染成功。最后model.ainvoke返回一段天气报告说明模型侧走 TaoToken 通道正常。Sampling 的验证要单独走工具调用。用 Agent 触发from langgraph.prebuilt import create_react_agent agent create_react_agent(model, tools) result await agent.ainvoke( {messages: [{role: user, content: Generate a formal weather report for Tokyo}]} ) print(Sampling result:, result)当 Agent 选中generate_weather_report时服务端会通过ctx.session.create_message向客户端发sampling/createMessage客户端用 TaoToken 的模型生成内容再回传。成功时你能在服务端日志看到 Sampling 请求客户端拿到最终报告文本。如果这一步没反应先确认工具确实被选中再看客户端是否实现了 Sampling 回调。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 对照调试 MCP 时遇到的报错就那么几类我按实际出现频率排一下每条给判断依据和动作。401 Unauthorized基本是 Key 问题。先echo $TAOTOKEN_API_KEY看是否为空再确认base_url写的是https://taotoken.net/api而不是别的。如果 Key 刚创建等几秒再试。还有一种情况是客户端和服务端用了不同的 KeySampling 回调时服务端侧没配 Key也会 401检查两边环境变量是否一致。local proxy failed通常出现在客户端连不上http://localhost:9000/mcp。先确认服务端进程还在curl http://localhost:9000/mcp看有没有响应。端口被占用也会这样换 9001 再试。如果用了容器注意 localhost 在容器里指向容器自身要换成宿主机地址。reading choices这类报错多出现在模型返回结构不符合预期时比如result.content.type不是text。Sampling 返回的内容类型要判断后再取代码里已经做了if result.content.type text分支。如果模型返回了非文本内容直接str(result.content)兜底别硬取.text。OAuth相关报错一般和 MCP 服务的鉴权配置有关。本地实验阶段建议先关掉 OAuth用最简单的 streamable-http 跑通链路等四组件都验证过再考虑加鉴权。如果确实需要检查客户端是否带了正确的 token以及服务端mcp.run的参数是否匹配。MCP error -32602: Invalid request parameters就是本文开头那个 URI 问题。把weather_history://改成weather-history://所有资源 URI 的协议段都别用下划线。改完重启服务端客户端重新连接。还有一个隐蔽的坑templates/list返回空。这通常是资源注册时 URI 模板写错比如漏了{city}占位符或者用了不被支持的字符。对照weather-history://{city}这个格式检查。排查顺序建议固定先确认服务端进程和端口再确认 Key 和 base_url然后单独测 Resources 和 Prompts不依赖模型最后测 Sampling。这样能把问题范围快速缩小到某一层。6. 继续往下走把四组件用顺手的几个实操建议跑通最小示例后真正影响效率的是怎么组织代码。我的习惯是把 Resources 按数据域分文件Prompts 单独放一个模块Tools 里只留动作逻辑Sampling 调用统一封一个 helper避免每个工具里重复写create_message。这样加新资源时不用动工具代码。URI 命名上协议段统一用连字符路径段用斜杠比如weather-history://{city}/monthly可读性和兼容性都好。Prompts 的参数尽量给默认值客户端调用时少传参减少prompts/get报参数缺失的概率。Sampling 的max_tokens别设太大调试阶段 200 够用设大了回调慢还费 token。system_prompt写清楚角色模型输出稳定性会好很多。如果发现 Sampling 偶尔超时先看客户端模型响应时间再考虑加超时重试。最后MCP 服务端和模型通道是两层TaoToken 只负责后者。把 Key、Base URL、Model ID 三件套固定成环境变量服务端和客户端共用换模型时只改一个地方。这套配置跑顺之后再去看更复杂的 Agent 场景会轻松很多。