橙心橙怡1. 项目背景与核心需求
在AI编程辅助工具日益普及的当下，代码沙箱作为隔离执行环境的核心组件，其重要性愈发凸显。这次实战源于我在实际工作中遇到的几个痛点：现有沙箱服务（如E2B）在中文环境支持、自定义包管理等方面存在局限，而商业解决方案又往往缺乏灵活性。于是决定基于字节开源的TRAE框架，从零构建一个支持MCP协议的Python代码沙箱系统。
这个项目的核心目标有三个：
- 实现安全的代码隔离执行环境，支持Python代码的实时执行与结果捕获
- 提供完善的周边功能：文件管理、依赖安装、工作目录隔离等
- 通过FastAPI-MCP封装为标准化的AI工具服务，便于大模型调用
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 沙箱架构设计与关键技术选型
2.1 整体架构设计
沙箱系统采用分层架构设计：
code复制应用层(FastAPI接口)
↓
服务层(Sandbox核心类)
↓
执行层(Jupyter Kernel)
↓
隔离层(工作目录+虚拟环境)
关键设计决策：
-
选择Jupyter Kernel作为执行后端，而非直接使用subprocess，原因在于：
- 原生支持代码交互式执行
- 完善的执行状态管理机制
- 丰富的输出类型处理（文本/图像/表格等）
-
工作目录隔离方案：
- 每个沙箱实例分配独立的工作目录
- 使用虚拟环境隔离Python依赖
- 采用绝对路径校验防止目录穿越攻击
2.2 核心类实现细节
沙箱核心类的主要属性和方法：
python复制class Sandbox:
def __init__(self, sandbox_id: str, work_dir: str, venv_dir: str):
self.sandbox_id = sandbox_id # 沙箱唯一标识
self.work_dir = work_dir # 工作目录路径
self.venv_dir = venv_dir # 虚拟环境路径
self.kernel_manager = None # Jupyter内核管理器
self.kernel_client = None # 内核通信客户端
self.last_execute_id = 0 # 执行ID计数器
# 主要方法
def execute_code(self, code: str) -> Dict: ...
def upload_file(self, file: UploadFile) -> str: ...
def get_files(self) -> List[Dict]: ...
def install_package(self, package_name: str) -> Dict: ...
def shutdown(self): ...
关键技巧：在
__init__方法中完成环境初始化工作，包括：
- 内核启动
- 基础包安装检查
- 中文字体预装
- 工作目录准备
3. 核心功能实现详解
3.1 代码执行模块
代码执行流程的核心逻辑：
python复制def execute_code(self, code: str) -> Dict:
msg_id = self.kernel_client.execute(code)
stdout, stderr, results = [], [], []
while True:
msg = self.kernel_client.get_iopub_msg()
msg_type = msg['header']['msg_type']
if msg_type == 'stream':
# 处理标准输出/错误
content = msg['content']
if content['name'] == 'stdout':
stdout.append(content['text'])
else:
stderr.append(content['text'])
elif msg_type == 'execute_result':
# 处理执行结果
data = msg['content']['data']
for data_type, data_value in data.items():
results.append({
"type": data_type,
"data": data_value
})
elif msg_type == 'status' and msg['content']['execution_state'] == 'idle':
break
return {
'stdout': stdout,
'stderr': stderr,
'results': results
}
注意事项：
- 使用
ansi_escape清理控制台颜色代码- 设置合理的超时时间（示例中为3600秒）
- 对traceback信息进行格式化处理
3.2 文件管理模块
文件上传的安全实现要点：
python复制def upload_file(self, file: UploadFile, file_path: str = None) -> str:
# 路径安全检查
if file_path and '../' in file_path:
raise HTTPException(status_code=403, detail="Invalid path")
save_path = os.path.join(self.work_dir, file_path or file.filename)
# 创建父目录
os.makedirs(os.path.dirname(save_path), exist_ok=True)
# 保存文件
with open(save_path, "wb") as f:
shutil.copyfileobj(file.file, f)
return save_path
文件获取的安全检查：
python复制def get_file_path(self, file_path: str) -> str:
full_path = os.path.abspath(os.path.join(self.work_dir, file_path))
# 防止目录穿越攻击
if not full_path.startswith(os.path.abspath(self.work_dir)):
raise HTTPException(status_code=403, detail="File access denied")
return full_path
3.3 环境隔离方案
虚拟环境管理的关键实现：
python复制def _install_basic_packages(self):
pip_path = os.path.join(self.venv_dir, 'bin', 'pip')
python_exe = os.path.join(self.venv_dir, 'bin', 'python')
# 检查ipykernel是否已安装
check_result = subprocess.run(
[python_exe, '-c', 'import ipykernel'],
capture_output=True,
text=True
)
if check_result.returncode != 0:
subprocess.check_call([pip_path, 'install', 'ipykernel'])
包安装接口：
python复制def install_package(self, package_name: str) -> Dict:
pip_path = os.path.join(self.venv_dir, 'bin', 'pip')
try:
result = subprocess.run(
[pip_path, 'install', package_name],
timeout=60,
capture_output=True,
text=True
)
return {
'success': result.returncode == 0,
'stdout': result.stdout,
'stderr': result.stderr
}
except subprocess.TimeoutExpired:
return {
'success': False,
'message': 'Installation timeout'
}
4. FastAPI-MCP服务化实战
4.1 基础接口实现
示例路由定义：
python复制@app.post("/sandboxes")
async def create_sandbox():
sandbox_id = str(uuid.uuid4())
work_dir = os.path.join(WORK_DIR_BASE, sandbox_id)
venv_dir = os.path.join(VENV_DIR_BASE, sandbox_id)
os.makedirs(work_dir, exist_ok=True)
os.makedirs(venv_dir, exist_ok=True)
sandbox = Sandbox(sandbox_id, work_dir, venv_dir)
SANDBOXES[sandbox_id] = sandbox
return {"sandbox_id": sandbox_id}
@app.post("/sandboxes/{sandbox_id}/execute")
async def execute_code(sandbox_id: str, code: CodeRequest):
sandbox = SANDBOXES.get(sandbox_id)
if not sandbox:
raise HTTPException(status_code=404, detail="Sandbox not found")
return sandbox.execute_code(code.code)
4.2 MCP服务适配
使用FastAPI-MCP的三步集成：
python复制from fastapi_mcp import FastApiMCP
# 在所有路由定义之后添加
mcp = FastApiMCP(app)
mcp.mount_http()
工具命名优化技巧：
python复制@app.post("/sandboxes/{sandbox_id}/execute",
operation_id="execute_python_code")
async def execute_code(sandbox_id: str, code: CodeRequest):
...
4.3 接口文档增强
为提升大模型使用效果，建议添加详细的OpenAPI描述：
python复制@app.post("/sandboxes/{sandbox_id}/execute",
summary="Execute Python code in sandbox",
description="""Execute Python code in isolated environment.
Returns stdout, stderr and execution results.""",
response_description="Execution results",
operation_id="execute_python_code")
async def execute_code(sandbox_id: str, code: CodeRequest):
...
5. 实战经验与避坑指南
5.1 中文支持方案
中文字体处理的最佳实践：
- 将字体文件打包到项目资源中
- 沙箱初始化时复制到工作目录
- 通过代码注册字体：
python复制font_code = """
import matplotlib as mpl
font_path = '/sandbox/SimHei.ttf'
if os.path.exists(font_path):
mpl.font_manager.fontManager.addfont(font_path)
mpl.rcParams['font.sans-serif'] = ['SimHei']
"""
5.2 常见问题排查
-
内核启动失败：
- 检查虚拟环境中的ipykernel是否安装正确
- 验证Python解释器路径配置
-
文件权限问题：
- 确保工作目录有写权限
- 检查selinux/apparmor等安全模块配置
-
包安装超时：
- 配置国内pip镜像源
- 设置合理的超时时间（默认60秒）
5.3 性能优化建议
- 沙箱池预初始化：提前创建若干沙箱实例备用
- 依赖预装：根据业务场景预装常用包
- 资源监控：添加内存/CPU使用监控，防止资源耗尽
6. 扩展功能实现
6.1 超时控制机制
代码执行超时实现方案：
python复制from concurrent.futures import ThreadPoolExecutor, TimeoutError
def execute_with_timeout(sandbox: Sandbox, code: str, timeout: int):
with ThreadPoolExecutor() as executor:
future = executor.submit(sandbox.execute_code, code)
try:
return future.result(timeout=timeout)
except TimeoutError:
sandbox.shutdown()
return {"error": "Execution timeout"}
6.2 会话持久化
沙箱状态保存方案：
python复制def save_sandbox_state(sandbox: Sandbox):
state = {
'files': [],
'installed_packages': get_installed_packages(sandbox.venv_dir)
}
for root, _, files in os.walk(sandbox.work_dir):
for file in files:
file_path = os.path.join(root, file)
state['files'].append({
'path': os.path.relpath(file_path, sandbox.work_dir),
'content': open(file_path, 'r').read()
})
return state
6.3 安全增强措施
-
代码安全检查：
python复制def validate_code_safety(code: str): forbidden_keywords = ['os.system', 'subprocess.Popen', 'open('/etc/'] for keyword in forbidden_keywords: if keyword in code: raise HTTPException(status_code=400, detail="Dangerous code detected") -
资源限制：
- 使用cgroups限制CPU/内存
- 设置文件大小上限
7. 项目部署与测试
7.1 生产环境部署
推荐部署架构：
code复制Nginx (负载均衡)
↓
Gunicorn (FastAPI Worker)
↓
Sandbox Service
↓
Docker/Kubernetes (资源隔离)
关键配置参数：
python复制# Gunicorn配置示例
workers = 4
worker_class = 'uvicorn.workers.UvicornWorker'
timeout = 120
7.2 测试方案设计
-
单元测试重点：
- 代码执行功能
- 文件操作边界条件
- 错误处理逻辑
-
集成测试场景：
- 连续创建/销毁沙箱
- 大文件上传下载
- 长时间运行任务
-
压力测试指标：
- 并发沙箱数量
- 平均响应时间
- 资源占用率
8. 项目演进方向
- 多语言支持：扩展支持Node.js、Go等语言
- 可视化增强：集成Jupyter Notebook界面
- 智能补全：结合大模型提供代码建议
- 协作功能：支持多人协作编辑
这个项目的完整代码已在GitHub开源，包含了更详细的文档和示例。在实际使用中，这套方案已经稳定支持了我们的AI编程辅助系统超过6个月，日均处理代码执行请求超过5000次。最大的收获是认识到：好的工具设计应该像空气一样，使用时感觉不到存在，但缺了它立刻就会察觉不适。