三亩地 三亩地SAN MU DI · CODE DIARY
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天,欢迎挑你感兴趣的翻一翻。

MCP协议下AI Agent代码执行安全实践:Sidecar架构与安全档位设计

MCP协议下AI Agent代码执行安全实践:Sidecar架构与安全档位设计

1. 项目缘起:为什么我们需要一个“翻译官”来执行代码?

如果你正在尝试构建一个AI Agent,尤其是那种需要处理复杂任务、调用外部工具或执行代码的智能体,那么你很可能已经遇到了一个核心难题:如何让大语言模型(LLM)安全、可控地“动手”操作外部世界?直接让LLL去执行rm -rf /或者访问敏感API?这无异于打开潘多拉魔盒。于是,一个名为MCP(Model 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的世界里,一切皆资源。一个代码执行服务器主要暴露两种东西:

  1. 资源(Resources):通常是可供读取的上下文信息,例如当前工作目录列表、已安装的Python包列表、某个脚本文件的预览。这些是“只读”的,用于给LLM提供决策依据。
  2. 工具(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 Agent(LLM客户端),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( uri="env://info", name="Execution Environment Info", description="Information about the current code execution environment", mimeType="text/plain", ) ] @app.read_resource() async def read_resource(uri: str) -> str: """读取资源内容""" if uri == "env://info": info = f"Python {sys.version}\\n" info += f"Safety Gear: {SAFETY_GEAR}\\n" info += f"Allowed Languages: {ALLOWED_LANGUAGES}\\n" info += f"Work Dir: {os.getcwd()}\\n" return info raise ValueError(f"Unknown 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": f"Language '{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(f"Code 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": f"Unknown tool: {name}"}

3.3 实现安全执行层:隔离与限制

上面代码中的execute_code_safely函数是关键。我们不能直接exec(code),那太危险了。我们需要一个隔离环境。

async def execute_code_safely(code: str, language: str) -> dict[str, Any]: """在受限环境中执行代码""" with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) 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, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, # 限制进程组,便于后续终止整个进程树 preexec_fn=os.setsid if sys.platform != ‘win32’ else None ) try: # 设置超时(例如10秒) stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=10.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)}”}

这段代码的要点与踩坑点:

  1. 临时文件:将代码写入临时文件再执行,比直接exec更清晰,也便于处理多行代码和依赖。
  2. 子进程隔离:这是最基本的安全边界。子进程崩溃不会导致主服务器崩溃。
  3. 超时控制:防止无限循环或死锁代码。asyncio.wait_for是关键。
  4. 进程组终止(Unix):如果代码又启动了子进程,简单的process.terminate()可能杀不掉它们。使用os.killpg能终止整个进程树。这是我在处理一个启动后台线程的脚本时踩过的坑。
  5. 编码处理:使用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})...”, file=sys.stderr) await server.serve_forever() if __name__ == “__main__”: asyncio.run(main())

运行服务器:

python server.py

服务器现在正在标准输入/输出上监听MCP协议消息。你需要一个MCP客户端来连接它。例如,你可以配置Claude Desktop或Cursor编辑器来连接这个本地服务器。

4. 协议调试与客户端集成:让Agent真正“动”起来

服务器跑起来了,但如何验证它工作正常?如何让我们的Agent(LLM)使用它?

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.py

CLI启动后,它会列出服务器提供的所有资源和工具。你可以使用交互式命令来调用:

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 安全加固:构筑多层防线

  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

  2. 代码静态分析:在执行前,用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等
  3. 资源限额与监控:使用resource模块(Unix)或psutil在父进程中监控子进程的资源使用,防止内存泄漏或CPU耗尽攻击。

  4. 用户确认通道:在“用户确认”档位,需要有一个可靠的、防篡改的通道将确认请求送达真实用户(例如,通过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立即返回一个特殊的“等待确认”响应给Agent,Agent将这个状态以及一个唯一的operation_id呈现给用户界面。用户在前端点击确认后,UI直接向Sidecar的另一个端点发送确认信号,Sidecar再继续执行并异步通知Agent结果。这需要更复杂的事件驱动架构。

构建一个可靠、安全的MCP代码执行Sidecar,是一个在功能、安全与易用性之间不断权衡的过程。从理解协议的抽象层,到选择适合的安全档位,再到实现一个健壮的Sidecar服务,每一步都需要仔细考量。它不是一个简单的“执行代码”的API,而是一个为AI Agent赋予安全行动力的核心基础设施。通过今天的拆解,希望你能避开我踩过的那些坑,更顺畅地让你的Agent“动手”去做那些它本该擅长的事。记住,最强的安全不是把刀锁起来,而是设计一套好的规则,让刀在需要时能被安全地使用。MCP的档位设计,正是这一思想的体现。

← 返回列表