🚀 核心挑战：如何为复杂数据分析任务构建可扩展的代码沙箱工具？本文将以E2B沙箱为例，通过对比Low-Level与FastMCP两种MCP-Server实现方案，深入剖析：
- Resource/Tool/Prompt的高阶应用场景
- 数据分析coding任务的难点和解决方案
- FastMCP在原有mcp-server的基础上做了哪些开发简化
Cluade的经典Demo多是用low level构建，而在最新版本中推出了高级FastMCP，但是工程上的简化，意味着很多功能被隐藏，所以更适合从low level一步步走过来对MCP设计每一步都很清晰得高级使用用户，而非新手小白。
这里我选了基于E2B的coding沙箱来搭建MCP server，刚好为下一章我们手搓数据分析智能体做个铺垫，完整代码详见: DAAgent
Coding MCP
🔥 真实场景的复杂性：生产级数据分析智能体中的Coding工具远非Python REPL可胜任，核心痛点在于三类数据流的信息传递：
| 数据流类型 | 挑战场景 | 本方案解决思路 |
|---|---|---|
| code2code | 多段代码块变量传递 | 持久化存储(CSV/JSON) |
| code2llm | 执行上下文（stdout/error） | 结构化日志捕获 |
| code2human | 结果可视化 | Jupyter Notebook组装 |
那基于以上3点考量，我们要设计的coding MCP server就需要具备以下功能
- sandbox initialize & stop：沙箱创建和销毁
- upload file：数据分析场景往往依赖前置输出的csv数据文件
- execude code：最后才是coding工具默认拥有的代码执行工具
🔍 和其他的一些sandbox mcp相比在工具设计上有几点不同
| 常见功能 | 本方案实现方式 | 选择依据 |
|---|---|---|
| 库安装❌ | 利用kernle内核内嵌!pip | Coding一致性更好，很多library是在code运行过程中发现的，而非在coding之前制定 |
| 文件下载❌ | 捕捉所有多模态的执行结果直接返回 | 对输出文件的目录管理会更加灵活 |
| 单步代码执行❌ | 显式create/destroy沙箱支持多步coding | 复杂任务通过多步coding传递中间变量和持久化文件，同时避免多次pip的时间浪费 |
Low-Level Server
下面我们先用Low-level Server框架来设计Coding MCP。按前面的设计，我们分别需要initialize sandbox，stop sandbox，upload file，execute code这四个工具。
除了执行代码之外的前三个工具，它们都可以使用文本输出格式，他们传递给模型的信息相对简单就是是否成功创建、销毁沙箱、以及是否成功上传文件，以及对应的沙箱ID标识和文件上传目录。所以这三个工具我选择了文本格式的输出。代码执行工具这里，我选择了结构化输出，这样可以更有效的拆分代码执行后的各种不同日志类型，方面后面的coding步骤定位修复问题。
需要注意的是，构建MCP 工具时不仅要全面完整输出成功日志，对于报错也要有详细的日志输出，必要时需要返回traceback，否则模型无法获取完整上下文很难修正工具调用方式。
以下是code 功能层核心代码实现
# 工具函数定义 async def initialize_sandbox(timeout: int = Field(description='沙箱最大运行时长，单位为秒', default=1000)) -> str: """创建一个新的沙箱环境用于代码执行 """ global Active_Sandboxes session_id = str(uuid.uuid4()) logger.info(f'创建沙箱中：session_id = {session_id}') try: sandbox = Sandbox(api_key=os.getenv("E2B_API_KEY"), timeout=timeout) Active_Sandboxes[session_id] = sandbox except Exception as e: msg = f'Failed to initialize sandbox: {str(e)}' logger.warning(e) return msg msg = f"Sandbox initialized successfully with session ID: {session_id}" logger.info(msg) return msg async def close_sandbox(session_id: str = Field(description='需要关闭的沙箱id')) -> str: """所有代码运行完毕之后，关闭已有的沙箱环境 :param session_id: 沙箱唯一标识符 :return: 执行日志 """ global Active_Sandboxes if session_id in Active_Sandboxes: sandbox = Active_Sandboxes[session_id] try: sandbox.kill() except Exception as e: msg = f'Failed to close sandbox with session ID: {session_id}, {str(e)}' logger.warning(e) return msg msg = f"Sandbox close successfully with session ID: {session_id}" logger.info(msg) return msg else: msg = f"Sandbox failed to stop session ID: {session_id} not found." logger.warning(msg) return msg class CodeOutput(BaseModel): stdout: str stderr: str error: str traceback: str async def execute_code(session_id: str = Field(description='需要运行代码的目标沙箱id'), code: str = Field(description='待运行的代码')) -> CodeOutput: """在沙箱中运行代码并获取代码的所有返回结果，包括stdout、stderr、各类持久化输出、图片等等 :param session_id: 沙箱唯一标识符 :param code: 待执行的代码 :return: 代码执行返回结构的结构体 """ global Active_Sandboxes, Execution_DB if session_id not in Active_Sandboxes: msg = f'Sandbox with session ID : {session_id} not found' data = {'error': msg} else: try: sandbox = Active_Sandboxes[session_id] # sandbox会在server内提供流式的内容输出 execution = sandbox.run_code( code, on_stdout=lambda data: logger.info(data), on_stderr=lambda data: logger.info(data), on_error=lambda data: logger.info(data) ) data = CodeOutput( 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 '', ) # Store execution record in database execution_record = { 'timestamp': datetime.now().isoformat(), 'code': code, 'output': execution # 存储沙箱原始结果即可 } Execution_DB[session_id]['executions'].append(execution_record) logger.info(f"Stored execution record for session {session_id}") except Exception as e: error_msg = f"Code execution failed: {str(e)}" traceback_str = traceback.format_exc() logger.warning(f"Execution error: {error_msg}n{traceback_str}") data = CodeOutput( error=error_msg, traceback=traceback_str, stdout='', stderr='' ) execution_record = { 'timestamp': datetime.now().isoformat(), 'code': code, 'output': None } Execution_DB[session_id]['executions'].append(execution_record) logger.info(f"Stored error execution record for session {session_id}") return data async def upload_file(session_id: str = Field(description='待上传文件的目标沙箱id'), file_list: List = Field(description='上传文件列表，每个都是本地文件的绝对路径') ) -> str: """上传本地文件到当前正在执行的沙箱中 :param session_id: 沙箱唯一标识ID :param file_list: 待上传的本地文件列表，文件名称包含本地文件路径 :return: 执行日志 """ global Active_Sandboxes if session_id not in Active_Sandboxes: msg = f'Sandbox with session ID : {session_id} not found' logger.warning(msg) return msg else: 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 = f'Uploading {file} success' logger.info(msg) except Exception as e: msg = f'Uploading {file} failed:' + str(e) logger.warning(msg) return_msg += msg + 'n' return return_msg 下面是low level Server核心list_tool和call_tool的实现，需要用户显式定义每一个工具的schema和具体的调用方法, 以下我已经参考后面的FastMCP的高级封装在定义时做了一些简化。每个工具的名称和描述，直接来自函数名称和注释，不用再手工填写。
@server.list_tools() async def list_tools() -> list[Tool]: tools = [ Tool(name=initialize_sandbox.__name__, description=initialize_sandbox.__doc__, inputSchema={ "type": "object", "properties": { "timeout": { "type": "integer", "description": "沙箱最大运行时长，单位为秒" } }, "required": ["timeout"], }), Tool(name=close_sandbox.__name__, description=close_sandbox.__doc__, inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "需要关闭的沙箱id" } }, "required": ["session_id"], }), Tool(name=execute_code.__name__, description=execute_code.__doc__, inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "待运行代码的沙箱id" }, "code": { "type": "string", "description": '待运行的代码' } }, "required": ["session_id", "code"], }, outputSchema=CodeOutput.model_json_schema()), Tool(name=upload_file.__name__, description=upload_file.__doc__, inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "待上传文件的目标沙箱id" }, "file_list": { "type": "array", "description": "上传文件列表，每个都是本地文件的绝对路径" } }, "required": ["session_id", "file_list"] }) ] return tools @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]): """ call tool返回结构并不局限在Text， Audio 和 Image类型, 如果你定义复杂结构体，返回Any类型即可 """ try: match name: case initialize_sandbox.__name__: result = await initialize_sandbox(arguments['timeout']) result = [TextContent(type="text", text=result)] case close_sandbox.__name__: result = await close_sandbox(arguments['session_id']) result = [TextContent(type="text", text=result)] case upload_file.__name__: result = await upload_file(arguments['session_id'], arguments['file_list']) result = [TextContent(type="text", text=result)] case execute_code.__name__: result = await execute_code(arguments['session_id'], arguments['code']) result = result.dict() case _: raise ValueError('Error calling tool: input name do not match any tool name') return result except Exception as e: raise ValueError(f'Error 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()) 唯一在开发时让我有些困惑的在于call_tool的返回类型，因为需要兼容结构化输出和纯文本输出这两种类型，细看server的Type Hint会发现call_tool的输出类型是Sequence[ContentBlock] | dict[str, Any], 也就是结构化和非结构化（纯文本）的输出类型分别是List和Dict两种类型，会被解析到两个不同的字段content和structuredContent。
本质上是否以文本格式返回的争议点在于工具返回内容的使用方是工具？模型？人类？，纯文本默认了工具的使用方是模型，但其实很多场景下并不是，例如多步工具调用传递参数、工具作为程序控制器、编程场景、数据分析场景等等。于是官方代码后面才把StructuredContent加上，git上有不少讨论，感兴趣的可以去瞅瞅Bring back the concept of "toolResult" (non-chat result)
High-Level FastMCP
有了Low-Level API的基础FastMCP就好理解多了。以下是效果相同，基于FastMCP这个High Level API的MCP工具实现，code功能部分和以上相同就不重复了。
from mcp.server.fastmcp import FastMCP logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("e2b-sandbox") # 服务初始化 mcp = FastMCP("code-sandbox") Active_Sandboxes = {} load_dotenv() # 读取env文件中的E2B API Key @mcp.tool() async def initialize_sandbox(timeout: int = Field(description='沙箱最大运行市场，单位为秒', default=1000)) -> str: """创建一个新的沙箱环境用于代码执行 :return: str: 返回初始化日志包含启动沙箱id """ ... return msg @mcp.tool() async def close_sandbox(session_id: str = Field(description='需要关闭的沙箱id')) -> str: """所有代码运行完毕之后，关闭已有的沙箱环境 :return: 执行日志 """ ... return msg class CodeOutput(TypedDict): results: List stdout: str stderr: str error: str traceback: str @mcp.tool() async def execute_code(session_id: str = Field(description='需要运行代码的沙箱id'), code: str = Field(description='待运行的代码')) -> CodeOutput: """在沙箱中运行代码并获取代码的所有返回结果 :return: 代码执行返回结构体包含stdout，stderr，error """ ... return data @mcp.tool() async def upload_file(session_id: str = Field(description='上传到目标沙箱的唯一标识符'), file_list: List = Field(description='上传文件列表，每个都是本地文件的绝对路径') ) -> str: """上传本地文件到当前正在执行的沙箱中 :return: 执行日志 """ ... return return_msg if __name__ == '__main__': mcp.run(transport='stdio') 从中我们可以发现FastMCP在服务开发上做了以下几点简化
- 服务器创建和启动的简化：FastMCP可以像flask、FastAPI一样直接mcp.run启动
- 工具（资源、prompt）定义和调用的简化：无需显式构建list_tool和call_tool API，使用mcp.tool装饰器直接注册工具，背后的tool Manager会自动完成以上工具说明和工具调用方法的构建，resource和prompt也是同理。
- 上下文和日志的简化：看实现FastMCP抽象了Context对象来简化日志和进度管理，但这块我还没具体用过就不做展开。
⚙️ 简单说下FastMCP的架构核心：
- FastMCP引入了3个核心管理器Tool、Resource、Prompt Manger:每个manager内部都各自实现了注册（add_tool）、获取(list_tool)、调用（call_tool）的功能，共同维护服务内所有的tool、resource和prompt
class FastMCP: def __init__(self, ...): self._tool_manager = ToolManager(...) self._resource_manager = ResourceManager(...) self._prompt_manager = PromptManager(...) - FastMCP通过装饰器和自动类型推断简化注册：在装饰器注册工具后，通过typing和pydantic BaseModeln能自动完成工具输入参数和输出类型的解析直接生成list_tool的schema。并且对比low-level，FastMCP基于pydantic提供了更多参数验证的功能。
def tool(self, name=None, title=None, description=None): def decorator(fn): # 自动分析函数签名 func_metadata = func_metadata(fn) # 自动生成参数schema parameters = func_metadata.arg_model.model_json_schema() # 注册到工具管理器 self._tool_manager.add_tool(fn, name, title, description) return fn return decorator 考虑到FastMCP开发的便捷性，后面再进一步的MCP开发我们都会基于FastMCP开发，不再演示low-level server的实现（哈哈哈用low-level的耐心已耗尽）。
MCP用法脑暴
Prompt有啥用？
和工具不同，Prompt是用户手动选择，而非模型选择的。 本质上只是Server提供者给到的一些工具指令的最佳实践而已，mcp的使用者可以选择用或者不用。
不过一个有意思的使用方式，是在团队内部可以整一个prompt-mcp-server来把大家在平时开发中亲测好用的一些prompt累积起来，避免各自为战和重复劳动，哈哈准备搞起来~
Resource vs Tool
Tool和Prompt的用法相对常见，但是Resource的存在一度让我很难理解，我看到了包括以下的几种解释
- Get vs Post：Tool和Resource的区别就是Get和Post的区别，工具可以执行任务而Resource只能获取信息。谁说Tool不能用Get方法了？
- 和文件或数据库对接返回数据：Resource专用于和文件和数据库对接，用于返回模型需要的数据。Tool不是一样也能对接数据库和系统文件？
以上的说法并没说服我，看了看多server也只发现Resource和tool之前还有以下几点不同
- 订阅和推送：Resource可以实现订阅推送机制，从模型主动获取数据变更，到检测数据变更并推送给模型，但是这依赖客户端是否有类似的设计，需要客户端订阅resource变更
- 多模态数据类型：还是前面的思路tool的返回结果默认是传递给模型的，因此以文本作为主要格式，后面才加入了structureOutput，而resource则支持数据流、二进制等更多数据类型
个人感觉resource的使用特性还在摸索阶段，我还没太想明白哈哈哈，要是大家有好的思路以欢迎评论区留言~
在server里面我设计了一个jupyter resource，用于当整个任务完成后，把所有code和code执行结果打包成jupyter notebook回传给客户端，进行展示。这里我选择把jupyter notebook按照nb4的格式返回json字符串的思路，当然返回二进制流也是可以的。
server部分的代码如下，在生产场景一般这里会使用OSS等远程存储，直接把ipynb上传到云，在客户端再直接使用返回的链接去云上下载文件即可，这里咱都在本地就简化成回传文件信息了~
@mcp.resource("file://notebook/{session_id}.ipynb", mime_type='application/json') async def get_execution_history(session_id: str) -> str: """获取指定session的所有代码和代码执行历史记录，以Jupyter Meta Data形式返回 """ global Execution_DB cell_list = [] code_counter = 0 logger.info('get jupyter history') if session_id in Execution_DB: session_data = Execution_DB[session_id] executions = session_data.get('executions', []) for execution_record in executions: cell_list.append({ "cell_type": "code", "execution_count": code_counter, "metadata": {}, "source": execution_record['code'], "outputs": parse_nb(execution_record["output"]) }) code_counter += 1 notebook = update_notebook_data(cell_list) logger.info(f"Generated notebook for session {session_id}") # 返回notebook的JSON内容作为bytes，这样客户端会接收到真正的ipynb文件内容 return json.dumps(notebook,ensure_ascii=False) else: raise ValueError(f'Sandbox with session ID : {session_id} not found') MCP Client
说完服务端，咱最后来看下客户端，需要注意的是low-level mcp和high-level FastMCP接口也存在细微差异，简单说fastmcp返回值嵌套减少了一层，考虑后面都会以FastMCP为主，这里只展示FastMCP对应client的相关代码和结果
from fastmcp import Client from fastmcp.client.transports import StdioTransport # 使用Transport的原因为为了制定server.py脚本的路径 transport = StdioTransport( command="python", args=["-m", "src.servers.e2b_high_level.server"], env={"LOG_LEVEL": "DEBUG"} ) session = Client(transport) async with session: # List available prompts prompts = await session.list_prompts() print(f"Available prompts: {[p.name for p in prompts]}") # List available resources:静态resource resources = await session.list_resources() print(f"Available resources: {[r.uri for r in resources]}") # List available resource templates：动态resource templates = await session.list_resource_templates() print(f"Available templates: {[r.uriTemplate for r in templates]}") # List available tools tools = await session.list_tools() print(tools) print(f"Available tools: {[t.name for t in tools]}") # 初始化沙箱 result = await session.call_tool("initialize_sandbox", arguments={"timeout": 1000}) print(f"Tool result: {result.content[0].text}") # 实际上session_id应该由模型基于上文，在下一步execute code的工具调用中推理得到 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(f"Tool result: {result.content[0].text}") # 执行2步代码: 对于结构化输出，call_tool内部会根据tool的output schema，自动进行model validate解析 result = await session.call_tool("execute_code", arguments={"session_id": session_id, "code": test_code1}) print(f"Execute Code: {result.structured_content}") result = await session.call_tool("execute_code", arguments={"session_id": session_id, "code": test_code2}) print(f"Execute Code: {result.structured_content}") # 直接获取notebook文件内容 - 使用application/octet-stream notebook_resource = await session.read_resource(f'file://notebook/{session_id}.ipynb') with open( f'notebook_{session_id}.ipynb', 'w',encoding='UTF-8') as f: f.write(notebook_resource[0].text) # 关闭沙箱 result = await session.call_tool("close_sandbox", arguments={"session_id": session_id}) print(f"Tool result: {result.content[0].text}") 结果如下
使用资源获取的notebook文件如下
学习资源列表
想看更全的大模型论文·Agent·开源框架·AIGC应用 >> DecryPrompt
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/2.jpg&w=218&h=124&zc=1)
