基于MCP协议构建AI代码安全沙盒：原理、实现与工程实践

1. 项目概述：一个为AI模型安全执行代码的“沙盒”工具

最近在折腾AI应用开发，特别是那些能调用外部工具、执行代码的智能体（Agent）时，一个绕不开的核心问题就是：如何让AI安全地运行它生成的代码？这可不是杞人忧天。想象一下，你让一个AI助手帮你分析数据，它“聪明”地写了一段Python脚本，结果里面藏了个os.system('rm -rf /')或者尝试访问你的私人文件。轻则数据丢失，重则系统被黑。这就是为什么我们需要一个像“沙盒”一样的隔离环境。

今天要聊的这个项目Kanak03-star/mcp-safe-run，就是为解决这个问题而生的。它是一个基于Model Context Protocol (MCP)的服务器实现，核心目标是为AI模型提供一个安全、可控的代码执行环境。简单来说，它就像是给AI配了一个专属的、带监控的“编程实验室”，AI可以在里面尽情尝试运行代码，但所有危险操作都会被限制在实验室的围墙内，不会影响到外面的主机系统。

无论你是正在构建AI智能体的开发者，还是对AI安全运行机制感兴趣的研究者，这个项目都提供了一个非常具体且实用的参考方案。它不只是一个理论构想，而是包含了资源限制、网络隔离、文件系统隔离等具体实现细节的工程化项目。接下来，我会带你深入拆解它的设计思路、核心实现，并分享在复现和扩展这类安全沙盒时的实操要点与避坑经验。

2. 核心设计思路与架构拆解

2.1 为什么是MCP？协议层的安全赋能

要理解mcp-safe-run，首先得弄明白MCP是什么。Model Context Protocol 是由 Anthropic 提出的一种开放协议，旨在标准化AI模型与外部工具、数据源之间的交互方式。你可以把它想象成AI世界的“USB标准”或“HTTP协议”。它定义了一套清晰的请求-响应格式，让不同的AI模型（客户端）能够以一种统一、安全的方式调用各种后端服务（服务器）。

mcp-safe-run选择基于MCP来实现，是一个非常高明的设计决策，主要基于以下几点考量：

1. 标准化与互操作性：MCP正在成为AI工具生态的事实标准之一。基于MCP构建，意味着你的安全执行服务器可以轻松被任何兼容MCP的AI客户端（如Claude Desktop、各类AI应用框架）发现和调用，无需为每个客户端编写特定的适配器。
2. 关注点分离：MCP协议本身处理了工具发现、参数传递、结果返回等通信问题。mcp-safe-run可以专注于最核心的挑战——安全地执行不可信的代码，而不用分心去设计通信协议。
3. 安全性内置：MCP的传输层通常基于SSE或标准IO，本身就运行在本地或可信网络。这为沙盒提供了一个相对安全的“入口”，避免了复杂的网络认证问题。服务器只需要确保从MCP通道接收到的代码本身被执行时的安全。

所以，这个项目的整体架构可以概括为：一个MCP服务器，它对外暴露一个或多个“执行代码”的工具（Tools）。当AI客户端通过MCP调用这个工具时，服务器并非直接在主进程执行代码，而是将其派发到一个预先构建好的、高度隔离的容器或沙盒环境中运行，并严格监控其资源消耗和行为。

2.2 安全沙盒的四大核心支柱

一个可靠的代码安全执行环境，绝不仅仅是eval()外面加个try-catch。mcp-safe-run的设计必然围绕着以下几个核心安全维度展开，这也是我们分析其源码或自行实现时需要重点关注的：

1. 资源限制：防止代码耗尽系统资源。

   • CPU：限制代码运行所能使用的CPU时间或核心数。
   • 内存：设定硬性内存上限，一旦超出立即终止进程。
   • 运行时间：设置超时限制，防止无限循环或死锁。
   • 磁盘：限制其能写入的数据量。

2. 文件系统隔离：防止代码读取或篡改主机上的敏感文件。

   • 只读根目录：通常使用一个最小化的Linux镜像作为根文件系统，并且以只读方式挂载。
   • 临时工作目录：为每次代码执行提供一个独立的、可写的临时目录（如/tmp/workspace），执行结束后自动清理。这是代码读写文件的唯一场所。
   • 绑定挂载：有时需要提供只读的输入文件（如数据集）或允许写入特定输出文件，这可以通过精心控制的绑定挂载实现。

3. 网络隔离：防止代码进行未经授权的网络访问。

   • 禁用网络：最简单的策略是完全禁用容器的网络命名空间，使其无法访问任何外部网络。这对于纯计算任务足够。
   • 受限网络：如果任务需要（如下载公开数据包），可以配置一个白名单，只允许访问特定的、安全的域名或IP地址。
   • 用户命名空间隔离：在容器内以非root用户（如nobody）运行代码，即使代码存在漏洞，其权限也被限制在容器内部。

4. 系统调用过滤：这是最后一道，也是最深层的防线。通过Seccomp等机制，可以禁止代码使用危险的系统调用，例如clone（创建新进程）、mount（挂载文件系统）、ptrace（调试其他进程）等。

mcp-safe-run的具体实现，就是在MCP服务器的框架下，集成上述一个或多个隔离机制（很可能是基于Docker或gVisor等容器运行时），来构建这个安全沙盒。

3. 关键技术实现与选型分析

3.1 容器运行时选型：Docker vs. gVisor vs. Firecracker

实现沙盒，首选的底层技术就是容器。但容器技术也有不同层次的选择，各有优劣。

Docker (基于 runc)：

• 优点：生态成熟，使用广泛，文档丰富。通过配置docker run的参数可以方便地实现资源限制 (--memory,--cpus)、文件系统隔离 (--read-only,-v绑定卷)、网络隔离 (--network none) 和用户隔离 (--user)。
• 缺点：安全性相对较弱。尽管有命名空间隔离，但容器与主机共享同一个内核。一个内核漏洞可能被利用来“逃逸”出容器。对于运行完全不可信的代码，这存在理论风险。
• 适用场景：运行来源相对可信、但需要隔离的代码（如用户提交的竞赛代码、数据分析脚本）。mcp-safe-run如果追求快速实现和稳定，很可能采用此方案。

gVisor：

• 优点：谷歌开源的容器沙盒，它提供了一个用Go语言实现的、在用户空间运行的“替身内核”。容器内的系统调用会被拦截并由这个替身内核处理，再以安全的方式转发给主机内核。这大大减少了攻击面，安全性显著高于Docker。
• 缺点：性能有一定开销，兼容性略差（某些依赖特定内核特性的程序可能无法运行）。
• 适用场景：对安全性要求极高的多租户环境，运行不受信任的代码。如果mcp-safe-run定位是面向公开服务或极高安全需求，可能会考虑集成gVisor。

Firecracker：

• 优点：AWS开源的微型虚拟机管理器，基于KVM。每个沙盒都是一个完整的、极轻量的微型VM，拥有独立的内核。安全性是三者中最高的，达到了虚拟化级别的隔离。
• 缺点：启动速度比容器慢（虽然已优化到毫秒级），内存开销稍大，管理更复杂。
• 适用场景：需要最强安全隔离且对冷启动时间不极度敏感的场景，如Serverless函数服务。

实操心得：对于个人项目或内部工具，从Docker开始是最务实的选择。它的安全配置（如设置--security-opt seccomp=unconfined来禁用危险系统调用）对于防御大多数非恶意代码已经足够。先跑起来，再根据实际威胁模型决定是否需要升级到gVisor。

3.2 MCP服务器实现剖析

MCP服务器通常有两种实现方式：SSE服务器或Stdio服务器。mcp-safe-run很可能采用Stdio模式，因为它更简单，更适合作为命令行工具集成。

一个典型的Stdio MCP服务器工作流程如下：

1. 初始化：服务器启动，通过标准输入输出与客户端通信。
2. 工具列表：客户端发送initialize请求，服务器返回其提供的工具列表。例如，mcp-safe-run可能提供一个名为execute_code的工具。
3. 工具调用：AI客户端（如Claude）在需要时，会发送tools/call请求，指定工具名（execute_code）和参数（如{"language": "python", "code": "print('hello')"}）。
4. 安全执行：服务器收到请求后，核心逻辑在此触发。它不会直接执行代码，而是： a. 生成一个唯一的执行ID（Session ID）。 b. 准备一个隔离环境（如启动一个配置好的Docker容器）。 c. 将用户代码写入容器内的临时文件。 d. 在容器内使用对应的解释器（如python3）运行该文件，同时严格监控资源（时间、内存）。 e. 捕获容器的标准输出、标准错误和退出码。
5. 结果返回：将捕获到的输出、错误信息以及执行状态（成功、超时、内存溢出）封装成MCP规定的格式，通过标准输出返回给客户端。
6. 清理：无论执行成功与否，都必须销毁对应的容器，清理所有临时资源，防止资源泄漏。

# 这是一个高度简化的伪代码逻辑，展示MCP服务器处理 `execute_code` 的核心环节 async def handle_execute_code(language, code): session_id = generate_uuid() # 1. 准备沙盒环境 container_spec = { 'image': f'python:3.11-slim', # 根据语言选择镜像 'command': ['sleep', 'infinity'], # 先保持容器运行 'mem_limit': '256m', 'cpus': '0.5', 'read_only': True, 'network_disabled': True, 'user': 'nobody', 'tmpfs': {'/tmp/workspace': 'rw,noexec,nosuid,size=64m'} # 提供可写的临时空间 } container = await docker_client.containers.create(**container_spec) await container.start() try: # 2. 将代码写入容器内的临时文件 code_file_path = f'/tmp/workspace/{session_id}.py' await container.put_archive('/tmp/workspace', create_tar_archive(code_file_path, code)) # 3. 在容器内执行代码，并设置超时 exec_cmd = ['python3', code_file_path] exec_result = await container.exec_run(exec_cmd, user='nobody', demux=True, socket=False) stdout, stderr = exec_result.output exit_code = exec_result.exit_code # 4. 处理结果（简化，实际需处理超时、内存溢出等） if exit_code == 0: return {"status": "success", "stdout": stdout.decode(), "stderr": stderr.decode()} else: return {"status": "error", "stdout": stdout.decode(), "stderr": stderr.decode(), "exit_code": exit_code} except asyncio.TimeoutError: return {"status": "timeout"} except Exception as e: return {"status": "system_error", "message": str(e)} finally: # 5. 强制停止并清理容器 await container.stop() await container.remove(force=True)

3.3 多语言支持与依赖管理

一个实用的安全运行环境，不可能只支持Python。mcp-safe-run的理想形态是支持主流脚本语言，如 Python、JavaScript (Node.js)、Bash，甚至 Rust/Go 的脚本模式。

实现多语言支持的关键点：

1. 基础镜像选择：为每种语言准备一个最小化的、安全的官方镜像（如python:slim、node:alpine）。镜像越小，攻击面越小，启动越快。
2. 依赖注入：AI写的代码常常需要第三方库。沙盒需要能处理pip install或npm install。
   • 策略A（安全优先）：完全禁止网络安装。提供一个预装了常用科学计算库（如numpy, pandas）的基础镜像。AI只能使用这些库。
   • 策略B（灵活优先）：允许在沙盒内进行网络安装，但必须严格超时，并且可以考虑使用国内镜像源加速。这是一个巨大的安全权衡！需要评估pip install过程中执行setup.py的风险。
   • 策略C（折中）：允许用户（开发者）预先定义一个“允许列表”（Allow List），只有列表内的包及其特定版本可以被安装。这需要维护一个包数据库。

注意事项：依赖管理是安全沙盒中最棘手的部分之一。我个人的经验是，对于内部工具，采用策略A（固定基础镜像）最为稳妥。你可以构建一个自己的Docker镜像，里面预装好项目常用的所有库。这样虽然不灵活，但绝对安全，且执行环境一致。如果AI请求的库不在镜像中，直接返回错误，提示用户“该环境不支持此库”。

4. 从零构建一个简易的MCP安全执行服务器

理解了原理，我们动手实现一个简化版的safe-run服务器。我们将使用 Python、Docker SDK 和mcp库来构建。

4.1 环境准备与依赖安装

首先，确保你的开发环境满足以下条件：

1. 安装并运行 Docker Engine。
2. Python 3.10+。

创建项目目录并安装依赖：

mkdir mcp-safe-run-demo && cd mcp-safe-run-demo python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows pip install docker mcp

docker包是 Docker 的 Python SDK，用于控制容器。mcp是社区维护的 Python MCP 协议实现库，能帮我们处理协议通信的细节。

4.2 核心服务器代码实现

我们创建一个server.py文件，实现一个支持 Python 代码执行的 MCP 服务器。

# server.py import asyncio import json import uuid import tarfile import io from typing import Any, Dict, List import docker from mcp import Server, StdioServerParameters from mcp.types import Tool, TextContent class SafeCodeExecutionServer: def __init__(self): # 初始化Docker客户端和MCP服务器 self.docker_client = docker.from_env() self.mcp_server = Server() # 注册工具和处理函数 self.mcp_server.tool_handler.register_tool( Tool( name="execute_python", description="Execute Python code in a secure sandbox. Returns stdout, stderr, and exit status.", inputSchema={ "type": "object", "properties": { "code": {"type": "string", "description": "The Python code to execute."}, "timeout_seconds": {"type": "integer", "description": "Maximum execution time in seconds.", "default": 30} }, "required": ["code"] } ), self.handle_execute_python ) async def handle_execute_python(self, arguments: Dict[str, Any]) -> List[TextContent]: """处理执行Python代码的请求""" code = arguments.get("code", "") timeout = arguments.get("timeout_seconds", 30) session_id = str(uuid.uuid4())[:8] print(f"[{session_id}] Received code execution request, timeout: {timeout}s") # 1. 创建并启动一个安全的容器 # 使用官方slim镜像，减少体积和攻击面 container = self.docker_client.containers.create( image='python:3.11-slim', command=['sleep', 'infinity'], # 保持容器运行，等待执行命令 mem_limit='256m', # 内存限制256MB cpuset_cpus='0', # 限制使用第一个CPU核心 network_disabled=True, # 禁用网络 read_only=True, # 根文件系统只读 user='nobody', # 以非特权用户运行 tmpfs={'/tmp/workspace': 'rw,noexec,nosuid,size=64m'}, # 提供可写的临时空间，禁止执行，禁止suid name=f'safe-run-{session_id}', detach=True ) container.start() result_text = "" try: # 2. 将代码写入容器的临时文件 # 创建一个包含代码文件的tar内存流 file_content = code.encode('utf-8') tar_stream = io.BytesIO() with tarfile.open(fileobj=tar_stream, mode='w') as tar: info = tarfile.TarInfo(name=f'code_{session_id}.py') info.size = len(file_content) tar.addfile(info, io.BytesIO(file_content)) tar_stream.seek(0) container.put_archive('/tmp/workspace', tar_stream.read()) # 3. 在容器内执行代码，并设置超时 exec_cmd = f"timeout {timeout} python3 /tmp/workspace/code_{session_id}.py" exec_result = container.exec_run( cmd=['sh', '-c', exec_cmd], user='nobody', workdir='/tmp/workspace', demux=True # 分离stdout和stderr ) exit_code, (stdout_bytes, stderr_bytes) = exec_result.exit_code, exec_result.output stdout = stdout_bytes.decode('utf-8', errors='ignore') if stdout_bytes else "" stderr = stderr_bytes.decode('utf-8', errors='ignore') if stderr_bytes else "" # 4. 解析结果 if exit_code == 124: # timeout命令返回124表示超时 status = "TIMEOUT" result_text = f"Execution timed out after {timeout} seconds." elif exit_code == 137: # SIGKILL，通常是内存超限被Docker杀死 status = "MEMORY_LIMIT_EXCEEDED" result_text = "Process was terminated for exceeding memory limit (256MB)." elif exit_code != 0: status = "RUNTIME_ERROR" result_text = f"Execution failed with exit code {exit_code}.\nStdout:\n{stdout}\nStderr:\n{stderr}" else: status = "SUCCESS" result_text = f"Execution succeeded.\nStdout:\n{stdout}" if stderr: result_text += f"\nStderr:\n{stderr}" print(f"[{session_id}] Execution finished with status: {status}") except docker.errors.APIError as e: status = "DOCKER_ERROR" result_text = f"Docker API error: {str(e)}" print(f"[{session_id}] Docker error: {e}") except Exception as e: status = "SYSTEM_ERROR" result_text = f"Unexpected system error: {str(e)}" print(f"[{session_id}] System error: {e}") finally: # 5. 无论如何，都尝试清理容器 try: container.stop(timeout=1) container.remove() print(f"[{session_id}] Container cleaned up.") except: pass # 忽略清理错误 # 6. 按照MCP格式返回结果 return [TextContent(type="text", text=result_text)] async def run(self): """启动MCP服务器（Stdio模式）""" server_params = StdioServerParameters() async with self.mcp_server.run_stdio(server_params) as (read_stream, write_stream): print("MCP Safe Run Server started (Stdio mode). Waiting for requests...") await asyncio.Future() # 永久运行 if __name__ == "__main__": server = SafeCodeExecutionServer() asyncio.run(server.run())

4.3 配置与运行

为了让MCP客户端（如Claude Desktop）发现我们的服务器，需要创建一个配置文件。

创建一个claude_desktop_config.json（假设用于Claude Desktop）：

{ "mcpServers": { "safe-run": { "command": "python", "args": ["/ABSOLUTE/PATH/TO/YOUR/mcp-safe-run-demo/server.py"], "env": {} } } }

将/ABSOLUTE/PATH/TO/YOUR/替换为你项目server.py的绝对路径。

然后，将Claude Desktop的配置指向这个文件（具体方法参考Claude Desktop文档）。重启Claude Desktop后，它就能识别到我们的execute_python工具了。

现在，你可以在Claude中尝试说：“请用execute_python工具计算一下1到100的和。” Claude会调用我们的服务器，在安全的沙盒中运行sum(range(1, 101))并将结果返回。

5. 生产级考量与常见问题排查

我们上面实现的是一个极简的演示版本。要用于生产或更严肃的场景，还需要考虑很多增强点。

5.1 性能优化与资源管理

• 容器池化：频繁创建销毁容器开销大。可以维护一个“温暖”的容器池，收到请求时分配一个空闲容器，用完后重置（清理/tmp/workspace）而非销毁，供下次使用。这能极大提升并发处理能力。
• 资源动态调整：根据代码的语言和复杂度动态分配资源。一个简单的print语句不需要256MB内存，而一个机器学习任务可能需要更多。可以通过分析代码（如导入的库）或让调用方指定资源需求来实现。
• 结果缓存：对于完全相同的代码输入，可以直接返回缓存的结果，避免重复执行。注意缓存要有失效策略。

5.2 增强安全性与审计

• 系统调用过滤：在Docker中可以使用自定义的Seccomp配置文件，进一步限制容器内可用的系统调用。

  # docker-compose.yml 或 create_container 参数示例 security_opt: - seccomp:./security-profile.json

• 行为监控：除了资源，还可以监控进程树、文件操作序列等。集成像falco这样的运行时安全工具，可以检测异常行为（如挖矿、暴力破解）。
• 完整的审计日志：记录每一次执行的元数据：会话ID、请求来源（客户端）、代码哈希、执行参数、资源使用量、开始结束时间、退出状态。这对于调试、计费和事后安全分析至关重要。

5.3 常见问题与排查技巧

在实际运行中，你肯定会遇到各种问题。下面是一个速查表：

5.4 扩展方向：从单机到集群

当单个服务器的负载过高时，就需要考虑集群化部署。

1. 任务队列：引入Redis或RabbitMQ作为任务队列。MCP服务器变为“调度器”，只接收请求并将其放入队列。多个独立的“工作节点”（Worker）从队列中取出任务，在各自的Docker环境中执行，并将结果回传。这实现了水平扩展。
2. 健康检查与负载均衡：工作节点定期上报心跳和负载（CPU、内存、队列长度）。调度器根据负载将新任务分发给最空闲的节点。
3. 共享存储：如果任务需要处理大型输入文件或产生大型输出，需要共享存储（如NFS、S3）让所有工作节点都能访问。这时，文件路径的传递和清理逻辑会变得更复杂。

构建一个mcp-safe-run这样的项目，远不止是写一个调用Docker的脚本。它涉及协议理解、安全工程、资源管理和系统设计。从最简单的原型开始，逐步迭代，针对遇到的具体问题去增强相应的模块，是驾驭这类复杂系统的不二法门。这个项目为我们提供了一个绝佳的蓝图，展示了如何将前沿的AI协议与经典的隔离技术结合，去解决AI时代一个实实在在的安全痛点。