ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

解密prompt系列60. MCP实战:从Low-Level到FastMCP的搭建演进与TaoToken统一接入

解密prompt系列60. MCP实战:从Low-Level到FastMCP的搭建演进与TaoToken统一接入 1. 从 Low-Level 手写协议到 FastMCP为什么我要把 E2B 沙箱包成 MCP ServerMCPModel Context Protocol是让大模型调用外部工具的一套标准协议而 MCP Server 就是这套协议的服务端实现。你可以把它理解成给模型装了一个USB 接口模型不需要知道沙箱怎么创建、文件怎么上传只要按协议发一个tools/callServer 就把活干了。适合谁适合手里有一堆 Python 脚本、想让 Claude Desktop、Cline、Codex 这类客户端直接调用又不想每次手动复制粘贴代码的人。我这次的目标很具体把 E2B 沙箱封装成一个 coding MCP Server提供四个工具——initialize_sandbox、close_sandbox、upload_file、execute_code。为什么不用现成的 Python REPL因为生产级数据分析任务里代码执行只是最后一环前面还有数据文件上传、多步执行中间变量传递、执行历史打包成 Jupyter Notebook 回传。这些用裸 REPL 做不了。搭建路径我走了两遍先用 Low-Level Server 手写list_tools和call_tool把协议每一层都摸清楚再换成 FastMCP 用装饰器重写代码量砍掉一半。最后把整个服务接入 TaoToken 统一 Key/API 通道这样客户端侧只需要配一个 Base URL 和一个 Key不用在多个厂商之间来回切。这篇文章交付三样东西可复制的 FastMCP 配置片段、Low-Level 对照代码、端到端验证步骤。你跟着走完能从知道 MCP 是什么到自己跑通一个带沙箱的 Server。2. TaoToken 前置准备统一 Key 与 API 通道怎么配在写 Server 之前先把模型侧的通道打通。TaoToken 在这里扮演的角色是统一入口你的 MCP 客户端比如 Cline、Claude Code、Codex需要调用模型来驱动工具如果每个客户端都单独配一家厂商的 Key管理起来很乱。TaoToken 提供统一的 Base URL 和 API Key客户端侧只认这一套。先拿 Key。打开 https://taotoken.net/api-keys 登录后创建一个 API Key复制出来形如sk-xxxxxxxx。这个 Key 后面要填到客户端的配置里不要提交到 Git。Base URL 用https://taotoken.net/api注意这里不加任何查询参数。模型 ID 按你实际要用的填比如claude-sonnet-4-5或gpt-4o具体以控制台模型列表为准。如果你用的是 Claude Code配置走环境变量或 settings 文件如果用 Cline走 MCP 的 JSON 配置。下面给一份 Cline 的 MCP 配置片段路径是cline_mcp_settings.json三件套Base URL Key Model ID都在里面{ mcpServers: { e2b-coding: { command: python, args: [-m, src.servers.e2b_high_level.server], env: { E2B_API_KEY: your_e2b_key, LOG_LEVEL: INFO } } } }注意这份配置里env只放了 E2B 的 Key因为 MCP Server 本身不直接调模型模型调用发生在客户端侧。客户端侧的模型通道配置单独走 TaoToken{ apiProvider: openai, openAiBaseUrl: https://taotoken.net/api, openAiApiKey: sk-你的TaoTokenKey, openAiModelId: claude-sonnet-4-5 }如果你用 Codex配置写在~/.codex/auth.json结构类似把base_url指向https://taotoken.net/apiapi_key填 TaoToken 的 Key。Claude Code 则在settings.json里配env段的ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY。这里有个容易踩的点MCP Server 的env和客户端的模型配置是两套东西别混在一起。Server 的env只放工具运行需要的密钥比如 E2B客户端的模型配置放 TaoToken 的 Key。分清楚之后后面排障会省很多事。3. 可复制配置FastMCP 版 E2B 沙箱 Server 完整实现这一节是全文技术核心。先给 FastMCP 的完整实现再给 Low-Level 对照最后说清楚 FastMCP 到底简化了什么。先装依赖pip install mcp e2b-code-interpreter python-dotenv pydanticFastMCP 版 Server 代码文件放src/servers/e2b_high_level/server.pyimport os import uuid import logging import json from datetime import datetime from typing import List from dotenv import load_dotenv from pydantic import Field from mcp.server.fastmcp import FastMCP from e2b_code_interpreter import Sandbox logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) logger logging.getLogger(e2b-sandbox) mcp FastMCP(code-sandbox) Active_Sandboxes {} Execution_DB {} load_dotenv() mcp.tool() async def initialize_sandbox( timeout: int Field(description沙箱最大运行时长单位为秒, default1000) ) - str: 创建一个新的沙箱环境用于代码执行 global Active_Sandboxes session_id str(uuid.uuid4()) logger.info(f创建沙箱中session_id {session_id}) try: sandbox Sandbox(api_keyos.getenv(E2B_API_KEY), timeouttimeout) Active_Sandboxes[session_id] sandbox Execution_DB[session_id] {executions: []} except Exception as e: msg fFailed to initialize sandbox: {str(e)} logger.warning(msg) return msg msg fSandbox initialized successfully with session ID: {session_id} logger.info(msg) return msg mcp.tool() async def close_sandbox( session_id: str Field(description需要关闭的沙箱id) ) - str: 所有代码运行完毕之后关闭已有的沙箱环境 global Active_Sandboxes if session_id in Active_Sandboxes: sandbox Active_Sandboxes[session_id] try: sandbox.kill() except Exception as e: msg fFailed to close sandbox with session ID: {session_id}, {str(e)} logger.warning(msg) return msg msg fSandbox close successfully with session ID: {session_id} logger.info(msg) return msg msg fSandbox failed to stop session ID: {session_id} not found. logger.warning(msg) return msg mcp.tool() async def upload_file( session_id: str Field(description待上传文件的目标沙箱id), file_list: List Field(description上传文件列表每个都是本地文件的绝对路径) ) - str: 上传本地文件到当前正在执行的沙箱中 global Active_Sandboxes if session_id not in Active_Sandboxes: msg fSandbox with session ID : {session_id} not found logger.warning(msg) return msg sandbox Active_Sandboxes[session_id] return_msg for file in file_list: try: with open(file, rb) as f: sandbox.files.write(file, f.read()) msg fUploading {file} success logger.info(msg) except Exception as e: msg fUploading {file} failed: str(e) logger.warning(msg) return_msg msg \n return return_msg mcp.tool() async def execute_code( session_id: str Field(description需要运行代码的目标沙箱id), code: str Field(description待运行的代码) ) - dict: 在沙箱中运行代码并获取代码的所有返回结果 global Active_Sandboxes, Execution_DB if session_id not in Active_Sandboxes: return {error: fSandbox with session ID : {session_id} not found} try: sandbox Active_Sandboxes[session_id] execution sandbox.run_code( code, on_stdoutlambda data: logger.info(data), on_stderrlambda data: logger.info(data), on_errorlambda data: logger.info(data) ) data { stdout: .join(execution.logs.stdout), stderr: .join(execution.logs.stderr), error: str(execution.error) if execution.error else , traceback: execution.error.traceback if execution.error else , } Execution_DB[session_id][executions].append({ timestamp: datetime.now().isoformat(), code: code, output: execution }) return data except Exception as e: import traceback as tb error_msg fCode execution failed: {str(e)} traceback_str tb.format_exc() logger.warning(fExecution error: {error_msg}\n{traceback_str}) return {error: error_msg, traceback: traceback_str, stdout: , stderr: } if __name__ __main__: mcp.run(transportstdio)启动方式python -m src.servers.e2b_high_level.server现在给 Low-Level 对照。Low-Level 版需要手写list_tools和call_tool每个工具的 schema 都要手动填from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent from pydantic import BaseModel server Server(code-sandbox) class CodeOutput(BaseModel): stdout: str stderr: str error: str traceback: str server.list_tools() async def list_tools() - list[Tool]: return [ Tool( nameinitialize_sandbox.__name__, descriptioninitialize_sandbox.__doc__, inputSchema{ type: object, properties: { timeout: {type: integer, description: 沙箱最大运行时长单位为秒} }, required: [timeout], }), Tool( nameexecute_code.__name__, descriptionexecute_code.__doc__, inputSchema{ type: object, properties: { session_id: {type: string, description: 待运行代码的沙箱id}, code: {type: string, description: 待运行的代码} }, required: [session_id, code], }, outputSchemaCodeOutput.model_json_schema()), ] server.call_tool() async def call_tool(name: str, arguments: dict): try: match name: case initialize_sandbox.__name__: result await initialize_sandbox(arguments[timeout]) return [TextContent(typetext, textresult)] case execute_code.__name__: result await execute_code(arguments[session_id], arguments[code]) return result.dict() case _: raise ValueError(Error calling tool: input name do not match any tool name) except Exception as e: raise ValueError(fError calling tool: {str(e)}) async def main(): options server.create_initialization_options() async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, options) if __name__ __main__: import asyncio asyncio.run(main())对比下来FastMCP 简化了三件事。第一服务启动Low-Level 要手写stdio_server和server.runFastMCP 一行mcp.run(transportstdio)。第二工具注册Low-Level 要手填inputSchema的每个字段FastMCP 用mcp.tool()装饰器加Field描述schema 自动从函数签名和类型注解生成。第三返回值处理Low-Level 的call_tool返回类型是Sequence[ContentBlock] | dict[str, Any]文本和结构化输出走两个不同字段FastMCP 内部帮你做了这层适配。有个细节值得说Low-Level 里call_tool的返回类型设计本质是在区分工具结果给谁用。纯文本默认是给模型看的结构化输出是给程序看的。FastMCP 把这层判断藏起来了方便但也意味着你对协议的理解会浅一层。所以我建议先读一遍 Low-Level 代码再用 FastMCP 写业务。4. 端到端验证从 list_tools 到 notebook 回传的完整请求Server 写完了得验证它真的能跑。客户端用 FastMCP 自带的Client文件放tests/test_client.pyimport os import asyncio from fastmcp import Client from fastmcp.client.transports import StdioTransport transport StdioTransport( commandpython, args[-m, src.servers.e2b_high_level.server], env{LOG_LEVEL: DEBUG} ) async def main(): session Client(transport) async with session: tools await session.list_tools() print(fAvailable tools: {[t.name for t in tools]}) result await session.call_tool(initialize_sandbox, arguments{timeout: 1000}) print(fTool result: {result.content[0].text}) session_id result.content[0].text.split(:)[1].strip() current_dir os.path.dirname(os.path.abspath(__file__)) result await session.call_tool(upload_file, arguments{ session_id: session_id, file_list: [os.path.join(current_dir, tests, fund_information.csv)] }) print(fUpload result: {result.content[0].text}) code1 import pandas as pd\ndf pd.read_csv(fund_information.csv)\nprint(df.head()) result await session.call_tool(execute_code, arguments{ session_id: session_id, code: code1 }) print(fExecute Code: {result.structured_content}) result await session.call_tool(close_sandbox, arguments{session_id: session_id}) print(fClose result: {result.content[0].text}) asyncio.run(main())跑起来python tests/test_client.py预期输出分四段。第一段Available tools列出四个工具名。第二段Sandbox initialized successfully with session ID: xxxx-xxxx这行里的 UUID 就是后面所有调用的session_id。第三段Upload result显示Uploading .../fund_information.csv success。第四段Execute Code返回结构化字典stdout里是 DataFrame 的前几行stderr和error为空。如果你在 Server 里加了 Jupyter Resource还能多一步调用read_resource(file://notebook/{session_id}.ipynb)把返回的 JSON 字符串写成本地.ipynb文件用 Jupyter 打开就能看到所有执行过的代码和输出。这一步在生产场景里通常会把 ipynb 传到对象存储客户端拿链接下载本地测试就直接写文件。验证通过的标准很简单四个工具全部返回预期结果execute_code的structured_content里stdout有内容、error为空。如果stdout是空的但代码没报错检查一下sandbox.run_code的on_stdout回调有没有正确拼接。5. 本篇常见错排查401、local proxy failed、reading choices 逐个拆这一节按真实报错来。第一个401 Unauthorized。这个通常出现在客户端调模型时不是 MCP Server 本身。检查 TaoToken 的 Key 有没有填对Base URL 是不是https://taotoken.net/api注意不要多加/v1或斜杠。如果 Key 是从控制台复制的确认没有多余空格。还有一种情况是 Key 过期或被删去 https://taotoken.net/api-keys 重新生成一个。第二个local proxy failed或connection refused。这个多半是 MCP Server 启动失败客户端连不上 stdio。先手动跑python -m src.servers.e2b_high_level.server看有没有 import 错误。常见的是e2b_code_interpreter没装或者E2B_API_KEY没设。如果 Server 能启动但客户端报这个错检查cline_mcp_settings.json里的args路径对不对-m后面的模块名要和实际目录结构一致。第三个reading choices或choices field missing。这是模型返回格式不对通常发生在客户端把非 OpenAI 兼容的响应当 OpenAI 格式解析。确认 TaoToken 的 Base URL 走的是兼容接口模型 ID 填的是控制台里列出的可用模型。如果用的是 Claude 系列有些客户端需要把 provider 设成anthropic而不是openai但 Base URL 仍然指向 TaoToken。第四个OAuth相关报错。如果你在 Claude Code 里看到 OAuth 失败检查settings.json里的ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY是否都指向 TaoToken。Claude Code 默认走 Anthropic 官方 OAuth改成自定义 Base URL 后要确保 Key 是 TaoToken 的不是 Anthropic 的。第五个Sandbox with session ID not found。这是session_id传错了或者沙箱已经被close_sandbox关掉。每次initialize_sandbox都会生成新 UUID多步执行时要把上一步返回的 ID 传给下一步。如果模型自己推理session_id容易截断或拼错建议在工具描述里强调必须使用上一步返回的完整 ID。第六个execute_code返回空stdout。检查代码里有没有printE2B 的run_code只捕获标准输出和标准错误不捕获表达式的返回值。如果你写df.head()而不print(df.head())stdout就是空的。改成print(df.head())即可。6. 语义一致 CTA把这条通道用起来Server 跑通之后下一步是把它接到你日常的编码流程里。如果你主要用 Claude Code 做长期编码建议走 Coding Plan把模型通道和工具调用统一管理https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。如果你只是想先验证模型对话能不能通用模型对话页面发一条测试消息https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。需要新建或轮换 Key 的时候去 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。接入细节和参数说明在文档里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。我自己的用法是E2B 沙箱 Server 常驻在本地Cline 里配好 MCP模型通道走 TaoToken。这样换模型的时候只改一个 Model IDServer 和工具链完全不用动。踩过的坑是早期把 E2B Key 和 TaoToken Key 混在一个 env 里排障时花了半小时才分清哪套配置管哪件事。分开之后401 就查客户端配置沙箱报错就查 Server 的 env定位快很多。
返回列表