Code Interpreter MCP Server

一个支持负载均衡的 Python 代码执行服务，通过 MCP (Model Context Protocol) 协议提供 Jupyter 沙箱环境。

功能特性

• ✅ 负载均衡：支持多个后端端口，自动分配请求
• ✅ 会话持久化：同一 session_id 的代码在同一环境中执行，保持状态
• ✅ 自动重试：端口错误时自动切换到其他可用端口
• ✅ 图像支持：自动捕获 matplotlib 图表，支持初始化图像加载
• ✅ MCP 协议：标准化的工具调用接口

系统架构

┌─────────────┐
│  MCP Client │
└──────┬──────┘
       │ MCP Protocol (stdio)
       ▼
┌─────────────────────────────┐
│  Code Interpreter MCP       │
│  (Load Balancer)            │
│  - Port Selection           │
│  - Session Management       │
│  - Error Handling           │
└──────┬──────────────────────┘
       │ HTTP POST
       ▼
┌─────────────────────────────┐
│  Jupyter Sandbox Servers    │
│  Port 18901, 18902, 18903..│
│  Host: 8.134.183.168        │
└─────────────────────────────┘

HTTP API 说明

后端服务接口

负载均衡器会向后端 Jupyter 沙箱服务器发送 HTTP POST 请求：

端点: http://8.134.183.168:{port}/jupyter_sandbox

默认端口: 18901, 18902, 18903, 18904

请求格式:

{
    "session_id": "my_session_123",
    "code": "print('Hello World')\nimport matplotlib.pyplot as plt\nplt.plot([1,2,3])",
    "timeout": 30
}

响应格式:

{
    "status": "success",
    "execution_time": 0.123,
    "output": {
        "stdout": "Hello World\n",
        "stderr": "",
        "images": ["base64_encoded_png_data..."]
    }
}

错误响应:

{
    "status": "error",
    "error": "NameError: name 'undefined_var' is not defined",
    "output": {
        "stdout": "",
        "stderr": "Traceback...",
        "images": []
    }
}

安装

pip install mcp httpx Pillow

配置

编辑 code_interpreter_mcp.py 修改配置：

balancer = CodeInterpreterLoadBalancer(
    host="8.134.183.168",  # 后端服务器地址
    ports=[18901, 18902, 18903, 18904]  # 可用端口列表
)

使用方法

1. 启动 MCP 服务器

python code_interpreter_mcp.py

服务器将通过 stdio 等待 MCP 客户端连接。

2. 在代码中使用（完整示例）

import asyncio
import base64
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def use_code_interpreter():
    # 连接到 MCP 服务器
    server_params = StdioServerParameters(
        command="python",
        args=["code_interpreter_mcp.py"],
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 示例 1: 简单计算
            result = await session.call_tool(
                "execute_python",
                {
                    "session_id": "demo_session",
                    "code": """
import numpy as np
data = np.random.rand(10)
print(f"Mean: {data.mean():.3f}")
print(f"Sum: {data.sum():.3f}")
"""
                }
            )
            print("执行结果:", result)
            
            # 示例 2: 绘制图表
            result = await session.call_tool(
                "execute_python",
                {
                    "session_id": "demo_session",
                    "code": """
import matplotlib.pyplot as plt
plt.figure(figsize=(8, 6))
plt.plot(data, marker='o')
plt.title('Random Data')
plt.show()
"""
                }
            )
            # result.content 包含生成的图片（base64）
            
            # 示例 3: 使用初始化图像
            with open("input.jpg", "rb") as f:
                img_base64 = base64.b64encode(f.read()).decode()
            
            result = await session.call_tool(
                "execute_python",
                {
                    "session_id": "image_session",
                    "code": """
# image_0 已自动加载为 PIL.Image 对象
import matplotlib.pyplot as plt
plt.imshow(image_0)
plt.title(f'Image size: {image_0.size}')
plt.show()
""",
                    "initialization_images": [img_base64]
                }
            )

asyncio.run(use_code_interpreter())

3. 高级用法：完整的 AI 视觉流水线

参考提供的 PeopleDetectionPipeline 示例，展示了如何：

1. 连接 MCP 服务器

server_params = StdioServerParameters(
    command="python",
    args=["./code_interpreter_mcp.py"],
)

self.stdio_context = stdio_client(server_params)
read, write = await self.stdio_context.__aenter__()

self.mcp_session = ClientSession(read, write)
await self.mcp_session.__aenter__()
await self.mcp_session.initialize()

2. 执行代码并处理结果

result = await self.mcp_session.call_tool(
    "execute_python",
    {
        "session_id": session_id,
        "code": generated_code,
        "timeout": 60,
        "initialization_images": [base64_image]
    }
)

# 解析返回的内容
for content in result.content:
    if content.type == "text":
        # 处理文本输出
        pass
    elif content.type == "image":
        # 处理生成的图片 (base64)
        image_data = content.data

3. 会话管理

# 同一 session_id 保持状态
await session.call_tool("execute_python", {
    "session_id": "my_work",
    "code": "x = [1, 2, 3]"
})

# 后续调用可以访问 x
await session.call_tool("execute_python", {
    "session_id": "my_work",
    "code": "print(sum(x))"  # 输出: 6
})

MCP 工具接口

execute_python

执行 Python 代码的工具。

输入参数:

• session_id (string, required): 会话标识符，相同 ID 共享状态
• code (string, required): 要执行的 Python 代码
• timeout (integer, optional): 超时时间（秒），默认 30
• initialization_images (array of strings, optional): Base64 编码的图像，将作为 image_0, image_1 等变量加载

返回内容:

• TextContent: 执行状态、stdout、stderr
• ImageContent: matplotlib 生成的图表（自动捕获）

示例返回:

**Execution Status**: success
**Time**: 0.156s

**stdout:**

Mean: 0.523 Sum: 5.234

[ImageContent: base64_png_data...]

负载均衡机制

端口选择策略

1. 会话绑定: 同一 session_id 优先使用上次的端口
2. 负载均衡: 选择当前活跃会话最少的端口
3. 错误恢复: 端口连续失败 3 次后标记为 ERROR 状态
4. 自动重置: 所有端口都 ERROR 时自动重置为 IDLE

状态监控

status = balancer.get_status()
# {
#     "ports": {
#         18901: {
#             "status": "idle",
#             "active_sessions": 0,
#             "error_count": 0,
#             "last_used": "2025-12-04T10:30:00"
#         },
#         ...
#     },
#     "total_sessions": 5
# }

错误处理

MCP 服务器返回的错误格式：

**Execution Status**: error

**stderr:**

HTTPStatusError: 500 Internal Server Error

**Error**: Connection timeout after 30s

常见错误：

• Connection refused: 后端服务器未启动
• Timeout: 代码执行超时
• NameError: 代码语法错误
• Port ERROR: 所有端口都不可用

最佳实践

1. 会话命名: 使用描述性的 session_id，如 user123_task_20251204
2. 超时设置: 复杂计算适当增加 timeout
3. 图像传递: 使用 initialization_images 而不是在代码中加载文件
4. 错误处理: 检查 result.content 中的 error 信息
5. 资源清理: 使用 async with 确保连接正确关闭

性能指标

• 单次代码执行延迟: 100-500ms（取决于代码复杂度）
• 并发支持: 4 个后端端口 × N 个会话
• 图像传输: 支持最大 10MB base64 图像
• 会话持久化: 会话在后端服务器重启前一直有效

故障排查

问题: MCP 连接失败

# 检查 Python 环境
python --version
pip list | grep mcp

# 手动启动测试
python code_interpreter_mcp.py

问题: 后端服务器不可用

# 测试后端连接
curl -X POST http://8.134.183.168:18901/jupyter_sandbox \
  -H "Content-Type: application/json" \
  -d '{"session_id":"test","code":"print(123)","timeout":10}'

问题: 代码执行超时

• 增加 timeout 参数
• 优化代码性能
• 检查是否有无限循环