1. 项目缘起为什么我们需要一个“翻译官”来执行代码如果你正在尝试构建一个AI Agent尤其是那种需要处理复杂任务、调用外部工具或执行代码的智能体那么你很可能已经遇到了一个核心难题如何让大语言模型LLM安全、可控地“动手”操作外部世界直接让LLL去执行rm -rf /或者访问敏感API这无异于打开潘多拉魔盒。于是一个名为MCPModel Context Protocol的协议逐渐进入了开发者的视野而其中Code Execution代码执行能力无疑是MCP皇冠上最闪亮也最危险的一颗明珠。最近在尝试为我的一个数据分析Agent增加自动编写并运行Python脚本来处理Excel文件的功能时我深刻体会到了这种“危险”与“必要”的矛盾。Agent能生成完美的pandas代码但如何让它安全地运行最初我尝试了简单的subprocess调用结果立刻在权限控制和环境隔离上栽了跟头。直到我系统地研究了MCP协议特别是其关于代码执行的规范才找到了一个相对优雅的解决方案Sidecar边车模式。这就像给Agent配了一个专业的“翻译官”兼“保镖”所有危险的“动手”指令都必须通过这个中间人在严格划定的“安全区”内执行。今天我们就来彻底拆解MCP协议下的代码执行。这不仅仅是调用一个API那么简单它涉及到协议设计哲学、安全档位Gear的权衡以及Sidecar架构的实战部署。我会结合一个具体的Python Sidecar实现案例手把手带你走过从协议理解到代码落地的全过程并分享我在调试和安全性加固上踩过的那些坑。2. 深入MCP协议Code Execution的核心机制与安全哲学MCP协议本质上是一套标准化的“对话”规则它定义了LLM客户端与各种工具、数据源服务器之间如何通信。你可以把它想象成USB协议只要设备服务器遵循USB标准就能被电脑LLM客户端识别并使用无论这个设备是U盘、键盘还是摄像头。Code Execution Server就是这样一个特殊的“USB设备”——一个能运行代码的设备。2.1 协议基础资源Resources、工具Tools与调用在MCP的世界里一切皆资源。一个代码执行服务器主要暴露两种东西资源Resources通常是可供读取的上下文信息例如当前工作目录列表、已安装的Python包列表、某个脚本文件的预览。这些是“只读”的用于给LLM提供决策依据。工具Tools这是核心代表可执行的操作。对于代码执行最关键的Tool就是execute。一个典型的调用流程是这样的LLM如Claude Code“用户想分析这个CSV文件我需要用pandas。先看看当前环境有没有pandas。”LLM调用MCP Client向Code Execution Server请求一个名为list_packages的工具如果存在或者读取一个代表环境信息的资源。Code Execution Server执行pip list或检查虚拟环境将结果格式化后返回。LLM“好的有pandas。现在生成代码df pd.read_csv(‘data.csv’)。需要执行它。”LLM调用MCP Client调用execute工具输入参数为{“code”: “import pandas as pd\\ndf pd.read_csv(‘data.csv’)\\nprint(df.head())”, “language”: “python”}。Code Execution Server在安全隔离的环境中执行这段代码捕获标准输出、标准错误和返回值。Code Execution Server将执行结果{“stdout”: “…”, “stderr”: “”, “exit_code”: 0}返回给LLM。这个过程完全由协议标准化LLM不需要知道服务器是在Docker容器里、沙箱里还是直接在本机执行的它只关心输入和输出。这种解耦带来了巨大的灵活性。2.2 安全档位Gears从“游乐场”到“手术室”MCP协议最精妙的设计之一就是为代码执行定义了不同的安全档位Gears。这就像汽车的变速箱不同的路况信任等级使用不同的档位。协议草案中通常定义了几个级别只读Read-Only最低档位。服务器只能提供信息资源不能执行任何工具。适用于展示环境状态、文件结构等。受限执行Restricted Execution中档位。可以执行代码但有严格限制。例如只能使用预定义的安全库如纯Python的数学、字符串处理库禁止访问网络、文件系统、子进程。这就像一个沙盒游乐场。用户确认User-Confirmed Execution高危操作档位。当服务器收到执行网络请求、安装包、写入特定目录等危险指令时会暂停并生成一个请求必须由终端用户人明确批准后指令才会继续。这引入了“人在回路”的监督机制。完全信任Full Trust最高档位。服务器拥有与运行进程相同的权限可以执行任何操作。这通常仅用于高度可控的内部环境或者由用户完全知晓风险后手动开启。这相当于把手术刀交给了Agent。为什么档位设计如此重要因为它将安全决策从技术实现细节中抽象了出来变成了一个可配置的策略。作为Agent开发者你可以根据使用场景例如教育演示 vs. 生产数据分析来配置Sidecar运行在哪个档位而不需要修改Agent的核心逻辑。在我的项目中我始终让Sidecar运行在“用户确认”档位任何尝试安装包或访问外部网络的操作都会弹出一个简洁的命令行确认提示这成功阻止了好几次Agent因误解指令而试图pip install一个不存在的包的情况。3. Sidecar架构实战构建一个Python代码执行守护进程理解了协议和档位我们开始动手。Sidecar模式是一种架构模式指将一个辅助进程与主应用部署在一起就像摩托车的边车为主应用提供额外的能力。在这里主应用是我们的AI AgentLLM客户端Sidecar就是我们的MCP Code Execution Server。我们将构建一个用Python实现的、支持多档位的MCP服务器。这里假设你已经有一定的Python和网络编程基础。3.1 项目初始化与依赖选择首先我们不需要从零实现MCP的底层通信。官方和社区已经提供了优秀的SDK。这里我们使用mcp这个Python库它极大地简化了服务器和客户端的开发。# 创建项目目录并初始化虚拟环境 mkdir mcp-code-executor cd mcp-code-executor python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install mcp # 为了安全执行代码我们使用一个轻量级沙箱例如 py-sandbox 或自定义容器。 # 这里为了演示我们先使用简单的子进程但会加上超时和资源限制。 pip install psutil # 用于监控和控制子进程资源3.2 定义工具与资源暴露可控的能力我们创建一个server.py文件开始构建服务器。核心是定义execute工具并根据档位决定其行为。import asyncio import subprocess import sys import tempfile import os import psutil from typing import Any, List from mcp import Server, types # 初始化MCP服务器 app Server(python-code-executor) # 模拟一个简单的安全策略配置 SAFETY_GEAR user_confirmed # 可配置read_only, restricted, user_confirmed, full_trust ALLOWED_LANGUAGES [python, bash] RESTRICTED_MODULES [os, subprocess, socket, requests] # 受限模式下禁止导入 app.list_resources() async def list_resources() - List[types.Resource]: 列出可用资源例如环境信息 return [ types.Resource( urienv://info, nameExecution Environment Info, descriptionInformation about the current code execution environment, mimeTypetext/plain, ) ] app.read_resource() async def read_resource(uri: str) - str: 读取资源内容 if uri env://info: info fPython {sys.version}\\n info fSafety Gear: {SAFETY_GEAR}\\n info fAllowed Languages: {ALLOWED_LANGUAGES}\\n info fWork Dir: {os.getcwd()}\\n return info raise ValueError(fUnknown resource: {uri}) def should_require_confirmation(code: str, language: str) - bool: 一个简单的启发式函数判断本次执行是否需要用户确认 danger_patterns [ import os, import subprocess, import socket, __import__, eval(, exec(, pip install, curl , wget , rm -rf, open(, write(, ] lower_code code.lower() for pattern in danger_patterns: if pattern in lower_code: return True return False app.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) - dict[str, Any]: 处理工具调用核心是execute if name execute: code arguments.get(code, ) language arguments.get(language, python).lower() if language not in ALLOWED_LANGUAGES: return {error: fLanguage {language} is not allowed.} # --- 安全档位检查 --- if SAFETY_GEAR read_only: return {error: Server is in read-only mode. Code execution disabled.} is_dangerous should_require_confirmation(code, language) if SAFETY_GEAR restricted and is_dangerous: return {error: Restricted mode: Potentially dangerous operation blocked.} # 用户确认档位如果需要确认则返回一个待确认的请求而不是直接执行。 # 在实际实现中这里应该通过一个回调或消息队列通知前端/用户。 if SAFETY_GEAR user_confirmed and is_dangerous: # 这里简化处理直接阻塞等待一个模拟的“用户输入”。 # 真实场景应返回一个pending状态由另一个通道处理确认。 print(f⚠️ DANGEROUS OPERATION DETECTED:) print(fCode Snippet: {code[:200]}...) confirm input(Type YES to confirm execution: ) if confirm.strip().upper() ! YES: return {error: User denied the dangerous operation.} # 用户确认后继续执行 # --- 安全执行代码 --- return await execute_code_safely(code, language) else: return {error: fUnknown tool: {name}}3.3 实现安全执行层隔离与限制上面代码中的execute_code_safely函数是关键。我们不能直接exec(code)那太危险了。我们需要一个隔离环境。async def execute_code_safely(code: str, language: str) - dict[str, Any]: 在受限环境中执行代码 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code) script_path f.name try: # 使用子进程执行便于控制超时和资源 env os.environ.copy() # 可以在这里设置环境变量比如PYTHONPATH为空以限制模块导入 # env[‘PYTHONPATH’] ‘’ process await asyncio.create_subprocess_exec( sys.executable, script_path, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, envenv, # 限制进程组便于后续终止整个进程树 preexec_fnos.setsid if sys.platform ! ‘win32’ else None ) try: # 设置超时例如10秒 stdout, stderr await asyncio.wait_for(process.communicate(), timeout10.0) exit_code process.returncode except asyncio.TimeoutError: # 超时强制终止整个进程组 if sys.platform ! ‘win32’: os.killpg(os.getpgid(process.pid), signal.SIGKILL) else: process.kill() await process.wait() return { “stdout”: “”, “stderr”: “Execution timed out after 10 seconds.”, “exit_code”: -1 } finally: # 清理临时文件 os.unlink(script_path) return { “stdout”: stdout.decode(‘utf-8’, errors‘ignore’).strip(), “stderr”: stderr.decode(‘utf-8’, errors‘ignore’).strip(), “exit_code”: exit_code } except Exception as e: return {“error”: f“Failed to execute code: {str(e)}”}这段代码的要点与踩坑点临时文件将代码写入临时文件再执行比直接exec更清晰也便于处理多行代码和依赖。子进程隔离这是最基本的安全边界。子进程崩溃不会导致主服务器崩溃。超时控制防止无限循环或死锁代码。asyncio.wait_for是关键。进程组终止Unix如果代码又启动了子进程简单的process.terminate()可能杀不掉它们。使用os.killpg能终止整个进程树。这是我在处理一个启动后台线程的脚本时踩过的坑。编码处理使用errors‘ignore’避免非UTF-8输出导致解码崩溃。注意这只是一个基础演示远未达到生产级安全。真正的安全执行需要更强大的沙箱如Docker容器为每次执行启动一个全新的、网络受限的容器。gVisor / Firecracker提供更轻量级、更安全的微虚拟机隔离。seccomp-bpf / AppArmor在Linux上使用内核特性限制系统调用。 选择哪种方案取决于你的安全要求和性能开销的平衡。3.4 运行与测试Sidecar服务器在server.py末尾添加async def main(): # 初始化服务器使用stdio传输便于与MCP客户端如Claude Desktop通信 async with await app.create_stdin_stdout_server() as server: print(“Python Code Execution MCP Server started (Gear: {SAFETY_GEAR})...”, filesys.stderr) await server.serve_forever() if __name__ “__main__”: asyncio.run(main())运行服务器python server.py服务器现在正在标准输入/输出上监听MCP协议消息。你需要一个MCP客户端来连接它。例如你可以配置Claude Desktop或Cursor编辑器来连接这个本地服务器。4. 协议调试与客户端集成让Agent真正“动”起来服务器跑起来了但如何验证它工作正常如何让我们的AgentLLM使用它4.1 使用MCP CLI进行手动测试首先我们可以使用MCP官方工具modelcontextprotocol/cli进行手动测试这比直接对接LLM客户端要方便得多。# 全局安装MCP CLI (需要Node.js) npm install -g modelcontextprotocol/cli # 在一个新的终端使用CLI连接我们的Python服务器 # 这里我们通过stdio传输CLI会启动我们的server.py进程 mcp dev python /path/to/your/server.pyCLI启动后它会列出服务器提供的所有资源和工具。你可以使用交互式命令来调用mcp tools.execute ? code import sys; print(“Hello from MCP!”); print(f“Python: {sys.version}”) ? language python如果一切正常你将看到执行输出的JSON结果。这步调试至关重要能确保你的服务器协议实现是正确的。4.2 集成到AI Agent以LangChain为例假设你的Agent使用LangChain框架。你需要一个MCP集成工具来连接我们的Sidecar。# 首先安装必要的库。LangChain对MCP的支持可能在社区库中。 # 这里我们假设使用一个通用的MCP客户端如 mcp 库的客户端功能。 from mcp import Client import asyncio async def run_agent_with_code_executor(): # 创建MCP客户端并连接到我们的Sidecar服务器进程 # 注意这里演示的是进程间通信实际可能需要根据SDK调整 async with Client() as client: # 启动服务器子进程生产环境可能作为独立服务运行 transport StdioTransport([sys.executable, “server.py”]) await client.connect(transport) # 1. 让Agent先了解环境读取资源 resources await client.list_resources() env_info await client.read_resource(“env://info”) print(“Environment Info:”, env_info.contents) # 2. Agent根据任务生成代码 task “Calculate the factorial of 10” # 这里简化实际由LLM生成代码 generated_code “““ import math result math.factorial(10) print(f“Factorial of 10 is: {result}”) “““ # 3. Agent调用工具执行代码 result await client.call_tool( “execute”, arguments{“code”: generated_code, “language”: “python”} ) print(“Execution Result:”, result) # 处理结果继续Agent的后续步骤... if result.get(“exit_code”) 0: analysis f“The code ran successfully. Output: {result[‘stdout’]}” else: analysis f“Code execution failed. Error: {result[‘stderr’]}” # ... 将analysis送回LLM进行后续推理 # 在Agent的主循环中调用 asyncio.run(run_agent_with_code_executor())集成中的关键点连接管理Sidecar服务器可以是常驻进程Agent启动时连接。要处理好连接断开重连。错误处理MCP调用可能失败网络、服务器错误Agent需要有降级策略例如提示用户“代码执行功能暂时不可用”。上下文管理多次执行的代码之间是否有状态通常每次执行应该是独立的新鲜子进程。如果需要有状态会话如定义一个函数后续调用服务器需要实现更复杂的会话管理这大大增加了安全复杂度一般不建议。5. 生产环境考量安全、性能与可观测性将这样一个Sidecar投入生产环境远不止写好协议逻辑那么简单。5.1 安全加固构筑多层防线运行时隔离如前所述使用Docker是底线。每个执行请求在一个新的、网络隔离的容器中进行并限制CPU、内存。# Dockerfile.executor FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY server.py . CMD [“python”, “server.py”]启动时docker run --rm --network none --memory“100m” --cpus“0.5” -i your-image代码静态分析在执行前用AST抽象语法树解析代码进行白名单/黑名单检查。禁止导入危险模块、访问特殊属性__builtins__、使用某些语法如eval。import ast class DangerousVisitor(ast.NodeVisitor): def visit_Import(self, node): for alias in node.names: if alias.name in RESTRICTED_MODULES: raise SecurityError(f“Import of restricted module ‘{alias.name}’ is not allowed.”) # ... 检查其他节点类型如Call, Attribute等资源限额与监控使用resource模块Unix或psutil在父进程中监控子进程的资源使用防止内存泄漏或CPU耗尽攻击。用户确认通道在“用户确认”档位需要有一个可靠的、防篡改的通道将确认请求送达真实用户例如通过WebSocket推送到前端界面而不是简单的命令行输入。5.2 性能优化应对高并发如果Agent需要频繁执行小段代码频繁创建销毁Docker容器开销巨大。连接池保持一定数量的“热”容器实例处理完请求后清理内部状态如删除生成的文件而非销毁容器供下一个请求使用。异步处理确保服务器是异步的如我们使用asyncio能够同时处理多个执行请求。将耗时的代码执行放在线程池中运行避免阻塞事件循环。结果缓存对于纯函数式的、确定性的代码片段如相同的输入计算哈希可以考虑缓存执行结果但要注意缓存可能带来的副作用和安全问题。5.3 可观测性与日志完善的日志是排查问题的生命线。结构化日志记录每一次工具调用的请求参数可脱敏、执行时长、资源使用、退出码、安全决策如“因包含import os被阻止”。审计追踪将每段执行的代码、执行者哪个用户/会话、时间戳、结果持久化到审计日志中满足合规要求。指标监控暴露Prometheus指标如mcp_execution_requests_totalmcp_execution_duration_secondsmcp_execution_errors_total便于监控服务健康度。6. 避坑指南从协议细节到部署陷阱在开发和部署MCP Code Execution Sidecar的过程中我遇到了不少预料之外的问题。坑一协议版本兼容性MCP协议本身还在演进中。我最初基于一个较早的草案实现结果与最新版的Claude Desktop无法通信。教训始终关注官方协议仓库如github.com/modelcontextprotocol/specification的更新并在服务器初始化时明确声明支持的协议版本。坑二标准输入/输出的缓冲与死锁在子进程执行代码时如果代码产生了大量输出而未及时读取可能会导致管道缓冲区填满进而使子进程阻塞。解决方案使用asyncio.create_subprocess_exec并配合communicate()方法它会自动处理读写。对于需要交互式输入的程序极少在Agent场景需要则需要更复杂的asyncio流处理。坑三Sidecar的生命周期管理Sidecar是随Agent启动而启动还是作为独立服务如果Agent崩溃Sidecar是否要随之终止我采用了独立服务健康检查的模式。将Sidecar作为独立的守护进程运行Agent通过本地Socket或HTTP连接它。Agent定期发送心跳Sidecar也暴露健康检查端点。这样Agent可以重启而不影响已提交的长时间任务虽然不推荐长时间任务也便于多个Agent实例共享一个Sidecar池。坑四“用户确认”的体验断层当Sidecar等待用户确认时整个Agent对话线程会被阻塞。这对于需要连续交互的聊天体验是毁灭性的。优化方案实现异步确认。当遇到需要确认的操作时Sidecar立即返回一个特殊的“等待确认”响应给AgentAgent将这个状态以及一个唯一的operation_id呈现给用户界面。用户在前端点击确认后UI直接向Sidecar的另一个端点发送确认信号Sidecar再继续执行并异步通知Agent结果。这需要更复杂的事件驱动架构。构建一个可靠、安全的MCP代码执行Sidecar是一个在功能、安全与易用性之间不断权衡的过程。从理解协议的抽象层到选择适合的安全档位再到实现一个健壮的Sidecar服务每一步都需要仔细考量。它不是一个简单的“执行代码”的API而是一个为AI Agent赋予安全行动力的核心基础设施。通过今天的拆解希望你能避开我踩过的那些坑更顺畅地让你的Agent“动手”去做那些它本该擅长的事。记住最强的安全不是把刀锁起来而是设计一套好的规则让刀在需要时能被安全地使用。MCP的档位设计正是这一思想的体现。
1. 项目缘起为什么我们需要一个“翻译官”来执行代码如果你正在尝试构建一个AI Agent尤其是那种需要处理复杂任务、调用外部工具或执行代码的智能体那么你很可能已经遇到了一个核心难题如何让大语言模型LLM安全、可控地“动手”操作外部世界直接让LLL去执行rm -rf /或者访问敏感API这无异于打开潘多拉魔盒。于是一个名为MCPModel Context Protocol的协议逐渐进入了开发者的视野而其中Code Execution代码执行能力无疑是MCP皇冠上最闪亮也最危险的一颗明珠。最近在尝试为我的一个数据分析Agent增加自动编写并运行Python脚本来处理Excel文件的功能时我深刻体会到了这种“危险”与“必要”的矛盾。Agent能生成完美的pandas代码但如何让它安全地运行最初我尝试了简单的subprocess调用结果立刻在权限控制和环境隔离上栽了跟头。直到我系统地研究了MCP协议特别是其关于代码执行的规范才找到了一个相对优雅的解决方案Sidecar边车模式。这就像给Agent配了一个专业的“翻译官”兼“保镖”所有危险的“动手”指令都必须通过这个中间人在严格划定的“安全区”内执行。今天我们就来彻底拆解MCP协议下的代码执行。这不仅仅是调用一个API那么简单它涉及到协议设计哲学、安全档位Gear的权衡以及Sidecar架构的实战部署。我会结合一个具体的Python Sidecar实现案例手把手带你走过从协议理解到代码落地的全过程并分享我在调试和安全性加固上踩过的那些坑。2. 深入MCP协议Code Execution的核心机制与安全哲学MCP协议本质上是一套标准化的“对话”规则它定义了LLM客户端与各种工具、数据源服务器之间如何通信。你可以把它想象成USB协议只要设备服务器遵循USB标准就能被电脑LLM客户端识别并使用无论这个设备是U盘、键盘还是摄像头。Code Execution Server就是这样一个特殊的“USB设备”——一个能运行代码的设备。2.1 协议基础资源Resources、工具Tools与调用在MCP的世界里一切皆资源。一个代码执行服务器主要暴露两种东西资源Resources通常是可供读取的上下文信息例如当前工作目录列表、已安装的Python包列表、某个脚本文件的预览。这些是“只读”的用于给LLM提供决策依据。工具Tools这是核心代表可执行的操作。对于代码执行最关键的Tool就是execute。一个典型的调用流程是这样的LLM如Claude Code“用户想分析这个CSV文件我需要用pandas。先看看当前环境有没有pandas。”LLM调用MCP Client向Code Execution Server请求一个名为list_packages的工具如果存在或者读取一个代表环境信息的资源。Code Execution Server执行pip list或检查虚拟环境将结果格式化后返回。LLM“好的有pandas。现在生成代码df pd.read_csv(‘data.csv’)。需要执行它。”LLM调用MCP Client调用execute工具输入参数为{“code”: “import pandas as pd\\ndf pd.read_csv(‘data.csv’)\\nprint(df.head())”, “language”: “python”}。Code Execution Server在安全隔离的环境中执行这段代码捕获标准输出、标准错误和返回值。Code Execution Server将执行结果{“stdout”: “…”, “stderr”: “”, “exit_code”: 0}返回给LLM。这个过程完全由协议标准化LLM不需要知道服务器是在Docker容器里、沙箱里还是直接在本机执行的它只关心输入和输出。这种解耦带来了巨大的灵活性。2.2 安全档位Gears从“游乐场”到“手术室”MCP协议最精妙的设计之一就是为代码执行定义了不同的安全档位Gears。这就像汽车的变速箱不同的路况信任等级使用不同的档位。协议草案中通常定义了几个级别只读Read-Only最低档位。服务器只能提供信息资源不能执行任何工具。适用于展示环境状态、文件结构等。受限执行Restricted Execution中档位。可以执行代码但有严格限制。例如只能使用预定义的安全库如纯Python的数学、字符串处理库禁止访问网络、文件系统、子进程。这就像一个沙盒游乐场。用户确认User-Confirmed Execution高危操作档位。当服务器收到执行网络请求、安装包、写入特定目录等危险指令时会暂停并生成一个请求必须由终端用户人明确批准后指令才会继续。这引入了“人在回路”的监督机制。完全信任Full Trust最高档位。服务器拥有与运行进程相同的权限可以执行任何操作。这通常仅用于高度可控的内部环境或者由用户完全知晓风险后手动开启。这相当于把手术刀交给了Agent。为什么档位设计如此重要因为它将安全决策从技术实现细节中抽象了出来变成了一个可配置的策略。作为Agent开发者你可以根据使用场景例如教育演示 vs. 生产数据分析来配置Sidecar运行在哪个档位而不需要修改Agent的核心逻辑。在我的项目中我始终让Sidecar运行在“用户确认”档位任何尝试安装包或访问外部网络的操作都会弹出一个简洁的命令行确认提示这成功阻止了好几次Agent因误解指令而试图pip install一个不存在的包的情况。3. Sidecar架构实战构建一个Python代码执行守护进程理解了协议和档位我们开始动手。Sidecar模式是一种架构模式指将一个辅助进程与主应用部署在一起就像摩托车的边车为主应用提供额外的能力。在这里主应用是我们的AI AgentLLM客户端Sidecar就是我们的MCP Code Execution Server。我们将构建一个用Python实现的、支持多档位的MCP服务器。这里假设你已经有一定的Python和网络编程基础。3.1 项目初始化与依赖选择首先我们不需要从零实现MCP的底层通信。官方和社区已经提供了优秀的SDK。这里我们使用mcp这个Python库它极大地简化了服务器和客户端的开发。# 创建项目目录并初始化虚拟环境 mkdir mcp-code-executor cd mcp-code-executor python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install mcp # 为了安全执行代码我们使用一个轻量级沙箱例如 py-sandbox 或自定义容器。 # 这里为了演示我们先使用简单的子进程但会加上超时和资源限制。 pip install psutil # 用于监控和控制子进程资源3.2 定义工具与资源暴露可控的能力我们创建一个server.py文件开始构建服务器。核心是定义execute工具并根据档位决定其行为。import asyncio import subprocess import sys import tempfile import os import psutil from typing import Any, List from mcp import Server, types # 初始化MCP服务器 app Server(python-code-executor) # 模拟一个简单的安全策略配置 SAFETY_GEAR user_confirmed # 可配置read_only, restricted, user_confirmed, full_trust ALLOWED_LANGUAGES [python, bash] RESTRICTED_MODULES [os, subprocess, socket, requests] # 受限模式下禁止导入 app.list_resources() async def list_resources() - List[types.Resource]: 列出可用资源例如环境信息 return [ types.Resource( urienv://info, nameExecution Environment Info, descriptionInformation about the current code execution environment, mimeTypetext/plain, ) ] app.read_resource() async def read_resource(uri: str) - str: 读取资源内容 if uri env://info: info fPython {sys.version}\\n info fSafety Gear: {SAFETY_GEAR}\\n info fAllowed Languages: {ALLOWED_LANGUAGES}\\n info fWork Dir: {os.getcwd()}\\n return info raise ValueError(fUnknown resource: {uri}) def should_require_confirmation(code: str, language: str) - bool: 一个简单的启发式函数判断本次执行是否需要用户确认 danger_patterns [ import os, import subprocess, import socket, __import__, eval(, exec(, pip install, curl , wget , rm -rf, open(, write(, ] lower_code code.lower() for pattern in danger_patterns: if pattern in lower_code: return True return False app.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) - dict[str, Any]: 处理工具调用核心是execute if name execute: code arguments.get(code, ) language arguments.get(language, python).lower() if language not in ALLOWED_LANGUAGES: return {error: fLanguage {language} is not allowed.} # --- 安全档位检查 --- if SAFETY_GEAR read_only: return {error: Server is in read-only mode. Code execution disabled.} is_dangerous should_require_confirmation(code, language) if SAFETY_GEAR restricted and is_dangerous: return {error: Restricted mode: Potentially dangerous operation blocked.} # 用户确认档位如果需要确认则返回一个待确认的请求而不是直接执行。 # 在实际实现中这里应该通过一个回调或消息队列通知前端/用户。 if SAFETY_GEAR user_confirmed and is_dangerous: # 这里简化处理直接阻塞等待一个模拟的“用户输入”。 # 真实场景应返回一个pending状态由另一个通道处理确认。 print(f⚠️ DANGEROUS OPERATION DETECTED:) print(fCode Snippet: {code[:200]}...) confirm input(Type YES to confirm execution: ) if confirm.strip().upper() ! YES: return {error: User denied the dangerous operation.} # 用户确认后继续执行 # --- 安全执行代码 --- return await execute_code_safely(code, language) else: return {error: fUnknown tool: {name}}3.3 实现安全执行层隔离与限制上面代码中的execute_code_safely函数是关键。我们不能直接exec(code)那太危险了。我们需要一个隔离环境。async def execute_code_safely(code: str, language: str) - dict[str, Any]: 在受限环境中执行代码 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code) script_path f.name try: # 使用子进程执行便于控制超时和资源 env os.environ.copy() # 可以在这里设置环境变量比如PYTHONPATH为空以限制模块导入 # env[‘PYTHONPATH’] ‘’ process await asyncio.create_subprocess_exec( sys.executable, script_path, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, envenv, # 限制进程组便于后续终止整个进程树 preexec_fnos.setsid if sys.platform ! ‘win32’ else None ) try: # 设置超时例如10秒 stdout, stderr await asyncio.wait_for(process.communicate(), timeout10.0) exit_code process.returncode except asyncio.TimeoutError: # 超时强制终止整个进程组 if sys.platform ! ‘win32’: os.killpg(os.getpgid(process.pid), signal.SIGKILL) else: process.kill() await process.wait() return { “stdout”: “”, “stderr”: “Execution timed out after 10 seconds.”, “exit_code”: -1 } finally: # 清理临时文件 os.unlink(script_path) return { “stdout”: stdout.decode(‘utf-8’, errors‘ignore’).strip(), “stderr”: stderr.decode(‘utf-8’, errors‘ignore’).strip(), “exit_code”: exit_code } except Exception as e: return {“error”: f“Failed to execute code: {str(e)}”}这段代码的要点与踩坑点临时文件将代码写入临时文件再执行比直接exec更清晰也便于处理多行代码和依赖。子进程隔离这是最基本的安全边界。子进程崩溃不会导致主服务器崩溃。超时控制防止无限循环或死锁代码。asyncio.wait_for是关键。进程组终止Unix如果代码又启动了子进程简单的process.terminate()可能杀不掉它们。使用os.killpg能终止整个进程树。这是我在处理一个启动后台线程的脚本时踩过的坑。编码处理使用errors‘ignore’避免非UTF-8输出导致解码崩溃。注意这只是一个基础演示远未达到生产级安全。真正的安全执行需要更强大的沙箱如Docker容器为每次执行启动一个全新的、网络受限的容器。gVisor / Firecracker提供更轻量级、更安全的微虚拟机隔离。seccomp-bpf / AppArmor在Linux上使用内核特性限制系统调用。 选择哪种方案取决于你的安全要求和性能开销的平衡。3.4 运行与测试Sidecar服务器在server.py末尾添加async def main(): # 初始化服务器使用stdio传输便于与MCP客户端如Claude Desktop通信 async with await app.create_stdin_stdout_server() as server: print(“Python Code Execution MCP Server started (Gear: {SAFETY_GEAR})...”, filesys.stderr) await server.serve_forever() if __name__ “__main__”: asyncio.run(main())运行服务器python server.py服务器现在正在标准输入/输出上监听MCP协议消息。你需要一个MCP客户端来连接它。例如你可以配置Claude Desktop或Cursor编辑器来连接这个本地服务器。4. 协议调试与客户端集成让Agent真正“动”起来服务器跑起来了但如何验证它工作正常如何让我们的AgentLLM使用它4.1 使用MCP CLI进行手动测试首先我们可以使用MCP官方工具modelcontextprotocol/cli进行手动测试这比直接对接LLM客户端要方便得多。# 全局安装MCP CLI (需要Node.js) npm install -g modelcontextprotocol/cli # 在一个新的终端使用CLI连接我们的Python服务器 # 这里我们通过stdio传输CLI会启动我们的server.py进程 mcp dev python /path/to/your/server.pyCLI启动后它会列出服务器提供的所有资源和工具。你可以使用交互式命令来调用mcp tools.execute ? code import sys; print(“Hello from MCP!”); print(f“Python: {sys.version}”) ? language python如果一切正常你将看到执行输出的JSON结果。这步调试至关重要能确保你的服务器协议实现是正确的。4.2 集成到AI Agent以LangChain为例假设你的Agent使用LangChain框架。你需要一个MCP集成工具来连接我们的Sidecar。# 首先安装必要的库。LangChain对MCP的支持可能在社区库中。 # 这里我们假设使用一个通用的MCP客户端如 mcp 库的客户端功能。 from mcp import Client import asyncio async def run_agent_with_code_executor(): # 创建MCP客户端并连接到我们的Sidecar服务器进程 # 注意这里演示的是进程间通信实际可能需要根据SDK调整 async with Client() as client: # 启动服务器子进程生产环境可能作为独立服务运行 transport StdioTransport([sys.executable, “server.py”]) await client.connect(transport) # 1. 让Agent先了解环境读取资源 resources await client.list_resources() env_info await client.read_resource(“env://info”) print(“Environment Info:”, env_info.contents) # 2. Agent根据任务生成代码 task “Calculate the factorial of 10” # 这里简化实际由LLM生成代码 generated_code “““ import math result math.factorial(10) print(f“Factorial of 10 is: {result}”) “““ # 3. Agent调用工具执行代码 result await client.call_tool( “execute”, arguments{“code”: generated_code, “language”: “python”} ) print(“Execution Result:”, result) # 处理结果继续Agent的后续步骤... if result.get(“exit_code”) 0: analysis f“The code ran successfully. Output: {result[‘stdout’]}” else: analysis f“Code execution failed. Error: {result[‘stderr’]}” # ... 将analysis送回LLM进行后续推理 # 在Agent的主循环中调用 asyncio.run(run_agent_with_code_executor())集成中的关键点连接管理Sidecar服务器可以是常驻进程Agent启动时连接。要处理好连接断开重连。错误处理MCP调用可能失败网络、服务器错误Agent需要有降级策略例如提示用户“代码执行功能暂时不可用”。上下文管理多次执行的代码之间是否有状态通常每次执行应该是独立的新鲜子进程。如果需要有状态会话如定义一个函数后续调用服务器需要实现更复杂的会话管理这大大增加了安全复杂度一般不建议。5. 生产环境考量安全、性能与可观测性将这样一个Sidecar投入生产环境远不止写好协议逻辑那么简单。5.1 安全加固构筑多层防线运行时隔离如前所述使用Docker是底线。每个执行请求在一个新的、网络隔离的容器中进行并限制CPU、内存。# Dockerfile.executor FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY server.py . CMD [“python”, “server.py”]启动时docker run --rm --network none --memory“100m” --cpus“0.5” -i your-image代码静态分析在执行前用AST抽象语法树解析代码进行白名单/黑名单检查。禁止导入危险模块、访问特殊属性__builtins__、使用某些语法如eval。import ast class DangerousVisitor(ast.NodeVisitor): def visit_Import(self, node): for alias in node.names: if alias.name in RESTRICTED_MODULES: raise SecurityError(f“Import of restricted module ‘{alias.name}’ is not allowed.”) # ... 检查其他节点类型如Call, Attribute等资源限额与监控使用resource模块Unix或psutil在父进程中监控子进程的资源使用防止内存泄漏或CPU耗尽攻击。用户确认通道在“用户确认”档位需要有一个可靠的、防篡改的通道将确认请求送达真实用户例如通过WebSocket推送到前端界面而不是简单的命令行输入。5.2 性能优化应对高并发如果Agent需要频繁执行小段代码频繁创建销毁Docker容器开销巨大。连接池保持一定数量的“热”容器实例处理完请求后清理内部状态如删除生成的文件而非销毁容器供下一个请求使用。异步处理确保服务器是异步的如我们使用asyncio能够同时处理多个执行请求。将耗时的代码执行放在线程池中运行避免阻塞事件循环。结果缓存对于纯函数式的、确定性的代码片段如相同的输入计算哈希可以考虑缓存执行结果但要注意缓存可能带来的副作用和安全问题。5.3 可观测性与日志完善的日志是排查问题的生命线。结构化日志记录每一次工具调用的请求参数可脱敏、执行时长、资源使用、退出码、安全决策如“因包含import os被阻止”。审计追踪将每段执行的代码、执行者哪个用户/会话、时间戳、结果持久化到审计日志中满足合规要求。指标监控暴露Prometheus指标如mcp_execution_requests_totalmcp_execution_duration_secondsmcp_execution_errors_total便于监控服务健康度。6. 避坑指南从协议细节到部署陷阱在开发和部署MCP Code Execution Sidecar的过程中我遇到了不少预料之外的问题。坑一协议版本兼容性MCP协议本身还在演进中。我最初基于一个较早的草案实现结果与最新版的Claude Desktop无法通信。教训始终关注官方协议仓库如github.com/modelcontextprotocol/specification的更新并在服务器初始化时明确声明支持的协议版本。坑二标准输入/输出的缓冲与死锁在子进程执行代码时如果代码产生了大量输出而未及时读取可能会导致管道缓冲区填满进而使子进程阻塞。解决方案使用asyncio.create_subprocess_exec并配合communicate()方法它会自动处理读写。对于需要交互式输入的程序极少在Agent场景需要则需要更复杂的asyncio流处理。坑三Sidecar的生命周期管理Sidecar是随Agent启动而启动还是作为独立服务如果Agent崩溃Sidecar是否要随之终止我采用了独立服务健康检查的模式。将Sidecar作为独立的守护进程运行Agent通过本地Socket或HTTP连接它。Agent定期发送心跳Sidecar也暴露健康检查端点。这样Agent可以重启而不影响已提交的长时间任务虽然不推荐长时间任务也便于多个Agent实例共享一个Sidecar池。坑四“用户确认”的体验断层当Sidecar等待用户确认时整个Agent对话线程会被阻塞。这对于需要连续交互的聊天体验是毁灭性的。优化方案实现异步确认。当遇到需要确认的操作时Sidecar立即返回一个特殊的“等待确认”响应给AgentAgent将这个状态以及一个唯一的operation_id呈现给用户界面。用户在前端点击确认后UI直接向Sidecar的另一个端点发送确认信号Sidecar再继续执行并异步通知Agent结果。这需要更复杂的事件驱动架构。构建一个可靠、安全的MCP代码执行Sidecar是一个在功能、安全与易用性之间不断权衡的过程。从理解协议的抽象层到选择适合的安全档位再到实现一个健壮的Sidecar服务每一步都需要仔细考量。它不是一个简单的“执行代码”的API而是一个为AI Agent赋予安全行动力的核心基础设施。通过今天的拆解希望你能避开我踩过的那些坑更顺畅地让你的Agent“动手”去做那些它本该擅长的事。记住最强的安全不是把刀锁起来而是设计一套好的规则让刀在需要时能被安全地使用。MCP的档位设计正是这一思想的体现。
)
