mcp-code-execution-mode
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseMCP Code Execution Mode
MCP代码执行模式
What This Does
功能介绍
This MCP server solves the "token bloat" problem when connecting LLMs to multiple MCP servers. Instead of loading 30,000+ tokens of tool schemas into every prompt, it exposes a single tool that executes Python code in rootless containers. The LLM discovers and calls other MCP tools programmatically, reducing context overhead by 95%+.
run_pythonKey benefits:
- Constant ~200 token overhead regardless of server count
- Discovery-first: Query schemas only when needed
- Universal proxying: Works with any stdio MCP server
- Production security: Rootless containers, no network, read-only filesystem
- Persistent sessions: Variables and MCP clients survive across calls
这款MCP服务器解决了将LLM连接到多个MCP服务器时的「token冗余」问题。它无需在每个提示词中加载30000+ token的工具 schema,而是暴露一个单一的工具,在无根容器中执行Python代码。LLM可通过编程方式发现并调用其他MCP工具,将上下文开销降低95%以上。
run_python核心优势:
- 恒定约200 token开销,不受服务器数量影响
- 优先发现机制:仅在需要时查询schema
- 通用代理:兼容任何标准输入输出的MCP服务器
- 生产级安全:无根容器、无网络、只读文件系统
- 持久会话:变量和MCP客户端可在多次调用间保留
Installation
安装步骤
Prerequisites
前置依赖
-
Container runtime (choose one):bash
# Podman (recommended) brew install podman podman machine init podman machine start # Docker Desktop (alternative) # Download from docker.com -
Python 3.11+:bash
python3 --version # Must be 3.11+
-
容器运行时(二选一):bash
# Podman(推荐) brew install podman podman machine init podman machine start # Docker Desktop(备选) # 从docker.com下载 -
Python 3.11+:bash
python3 --version # 版本必须为3.11+
Install via pip
通过pip安装
bash
pip install mcp-code-execution-modebash
pip install mcp-code-execution-modeInstall from source
从源码安装
bash
git clone https://github.com/elusznik/mcp-server-code-execution-mode.git
cd mcp-server-code-execution-mode
pip install -e .bash
git clone https://github.com/elusznik/mcp-server-code-execution-mode.git
cd mcp-server-code-execution-mode
pip install -e .Configuration
配置说明
Claude Desktop Setup
Claude Desktop 设置
Add to :
~/Library/Application Support/Claude/claude_desktop_config.jsonjson
{
"mcpServers": {
"code-execution": {
"command": "python",
"args": ["-m", "mcp_code_execution_mode"],
"env": {
"MCP_BRIDGE_RUNTIME": "podman",
"MCP_BRIDGE_IMAGE": "ghcr.io/elusznik/mcp-code-execution-mode:latest",
"MCP_BRIDGE_OUTPUT_MODE": "compact"
}
}
}
}添加配置到:
~/Library/Application Support/Claude/claude_desktop_config.jsonjson
{
"mcpServers": {
"code-execution": {
"command": "python",
"args": ["-m", "mcp_code_execution_mode"],
"env": {
"MCP_BRIDGE_RUNTIME": "podman",
"MCP_BRIDGE_IMAGE": "ghcr.io/elusznik/mcp-code-execution-mode:latest",
"MCP_BRIDGE_OUTPUT_MODE": "compact"
}
}
}
}Environment Variables
环境变量
| Variable | Default | Description |
|---|---|---|
| Auto-detect | |
| | Container image |
| | |
| | Execution timeout (seconds) |
| | Container memory limit |
| | Keep variables between calls |
| 变量名 | 默认值 | 描述 |
|---|---|---|
| 自动检测 | |
| | 容器镜像 |
| | |
| | 执行超时时间(秒) |
| | 容器内存限制 |
| | 保留调用间的变量 |
Proxying Other MCP Servers
代理其他MCP服务器
To give the agent access to other MCP servers (e.g., filesystem, GitHub), configure them in the same :
claude_desktop_config.jsonjson
{
"mcpServers": {
"code-execution": {
"command": "python",
"args": ["-m", "mcp_code_execution_mode"]
},
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/username/Documents"]
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
}
}
}
}The bridge auto-discovers these servers at runtime. No manual catalog needed.
要让Agent访问其他MCP服务器(如文件系统、GitHub),在同一个中配置:
claude_desktop_config.jsonjson
{
"mcpServers": {
"code-execution": {
"command": "python",
"args": ["-m", "mcp_code_execution_mode"]
},
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/username/Documents"]
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
}
}
}
}桥接器会在运行时自动发现这些服务器,无需手动维护目录。
Core API
核心API
Discovery Functions
发现函数
python
from mcp import runtimepython
from mcp import runtimeList available MCP servers
列出可用的MCP服务器
servers = await runtime.discovered_servers()
servers = await runtime.discovered_servers()
Returns: ["filesystem", "github", "slack"]
返回结果: ["filesystem", "github", "slack"]
Get tool schemas for a specific server
获取指定服务器的工具schema
docs = await runtime.query_tool_docs("github")
docs = await runtime.query_tool_docs("github")
Returns: {"tools": [{"name": "create_issue", "description": "...", ...}]}
返回结果: {"tools": [{"name": "create_issue", "description": "...", ...}]}
Fuzzy search across all servers
在所有服务器中模糊搜索
matches = await runtime.search_tool_docs("list files", limit=5)
matches = await runtime.search_tool_docs("list files", limit=5)
Returns: [{"server": "filesystem", "tool": "list_directory", "description": "..."}]
返回结果: [{"server": "filesystem", "tool": "list_directory", "description": "..."}]
undefinedundefinedCalling MCP Tools
调用MCP工具
Dynamic lookup:
python
from mcp import mcp_servers动态查找:
python
from mcp import mcp_serversCall a tool
调用工具
result = await mcp_servers["github"].create_issue(
repo="owner/repo",
title="Bug report",
body="Description here"
)
**Attribute access:**
```python
from mcp import mcp_github
result = await mcp_github.create_issue(
repo="owner/repo",
title="Feature request",
body="Add dark mode"
)Module import (explicit):
python
from mcp.servers.github import create_issue
result = await create_issue(
repo="owner/repo",
title="Enhancement",
body="Improve performance"
)result = await mcp_servers["github"].create_issue(
repo="owner/repo",
title="Bug report",
body="Description here"
)
**属性访问:**
```python
from mcp import mcp_github
result = await mcp_github.create_issue(
repo="owner/repo",
title="Feature request",
body="Add dark mode"
)模块导入(显式):
python
from mcp.servers.github import create_issue
result = await create_issue(
repo="owner/repo",
title="Enhancement",
body="Improve performance"
)Session Persistence
会话持久化
Variables persist across calls in the same session:
python
undefined同一会话中的变量会在多次调用间保留:
python
undefinedFirst call
第一次调用
import pandas as pd
df = pd.DataFrame({"col": [1, 2, 3]})
df.to_csv("/tmp/data.csv", index=False)
```pythonimport pandas as pd
df = pd.DataFrame({"col": [1, 2, 3]})
df.to_csv("/tmp/data.csv", index=False)
```pythonSecond call (same session)
第二次调用(同一会话)
import pandas as pd
df = pd.read_csv("/tmp/data.csv")
print(df.sum()) # Works! File still exists
undefinedimport pandas as pd
df = pd.read_csv("/tmp/data.csv")
print(df.sum()) # 可正常运行!文件仍存在
undefinedCommon Patterns
常见模式
Pattern 1: Discovery → Execution
模式1:发现 → 执行
python
from mcp import runtime, mcp_serverspython
from mcp import runtime, mcp_serversStep 1: Find tools related to "calendar"
步骤1:查找与"calendar"相关的工具
matches = await runtime.search_tool_docs("calendar events", limit=3)
matches = await runtime.search_tool_docs("calendar events", limit=3)
Step 2: Load full schema for the first match
步骤2:加载第一个匹配项的完整schema
server_name = matches[0]["server"]
docs = await runtime.query_tool_docs(server_name)
server_name = matches[0]["server"]
docs = await runtime.query_tool_docs(server_name)
Step 3: Call the tool
步骤3:调用工具
result = await mcp_servers[server_name].list_events(
start_date="2025-01-01",
end_date="2025-01-31"
)
print(result)
undefinedresult = await mcp_servers[server_name].list_events(
start_date="2025-01-01",
end_date="2025-01-31"
)
print(result)
undefinedPattern 2: Data Analysis Workflow
模式2:数据分析工作流
python
import pandas as pd
import matplotlib.pyplot as plt
from mcp import mcp_filesystempython
import pandas as pd
import matplotlib.pyplot as plt
from mcp import mcp_filesystemRead data from filesystem server
从文件系统服务器读取数据
csv_content = await mcp_filesystem.read_file(path="/data/sales.csv")
csv_content = await mcp_filesystem.read_file(path="/data/sales.csv")
Parse and analyze
解析与分析
df = pd.read_csv(pd.io.common.StringIO(csv_content))
monthly = df.groupby("month")["revenue"].sum()
df = pd.read_csv(pd.io.common.StringIO(csv_content))
monthly = df.groupby("month")["revenue"].sum()
Generate chart
生成图表
plt.bar(monthly.index, monthly.values)
plt.title("Monthly Revenue")
plt.savefig("/tmp/revenue.png")
plt.bar(monthly.index, monthly.values)
plt.title("Monthly Revenue")
plt.savefig("/tmp/revenue.png")
Write back
写回数据
with open("/tmp/revenue.png", "rb") as f:
await mcp_filesystem.write_file(
path="/reports/revenue.png",
content=f.read()
)
print(f"Analyzed {len(df)} records, saved chart")
undefinedwith open("/tmp/revenue.png", "rb") as f:
await mcp_filesystem.write_file(
path="/reports/revenue.png",
content=f.read()
)
print(f"已分析{len(df)}条记录,图表已保存")
undefinedPattern 3: Multi-Server Orchestration
模式3:多服务器编排
python
from mcp import mcp_github, mcp_slackpython
from mcp import mcp_github, mcp_slackGet open issues
获取未解决的Issue
issues = await mcp_github.list_issues(
repo="myorg/myrepo",
state="open",
labels=["bug"]
)
issues = await mcp_github.list_issues(
repo="myorg/myrepo",
state="open",
labels=["bug"]
)
Post summary to Slack
在Slack发布摘要
await mcp_slack.post_message(
channel="#engineering",
text=f"📊 {len(issues)} open bugs:\n" +
"\n".join(f"• {i['title']}" for i in issues[:5])
)
print(f"Posted {len(issues)} issues to Slack")
undefinedawait mcp_slack.post_message(
channel="#engineering",
text=f"📊 {len(issues)}个未解决Bug:\n" +
"\n".join(f"• {i['title']}" for i in issues[:5])
)
print(f"已在Slack发布{len(issues)}个Issue")
undefinedPattern 4: Error Handling & Retries
模式4:错误处理与重试
python
from mcp import mcp_servers
import asyncio
async def safe_call(server, tool, **kwargs):
for attempt in range(3):
try:
return await mcp_servers[server].__getattr__(tool)(**kwargs)
except Exception as e:
if attempt == 2:
raise
await asyncio.sleep(2 ** attempt)
result = await safe_call(
"github",
"create_issue",
repo="owner/repo",
title="Test",
body="Retry logic"
)python
from mcp import mcp_servers
import asyncio
async def safe_call(server, tool, **kwargs):
for attempt in range(3):
try:
return await mcp_servers[server].__getattr__(tool)(**kwargs)
except Exception as e:
if attempt == 2:
raise
await asyncio.sleep(2 ** attempt)
result = await safe_call(
"github",
"create_issue",
repo="owner/repo",
title="Test",
body="Retry logic"
)Pattern 5: Bash Commands
模式5:Bash命令
The sandbox includes common CLI tools:
python
import subprocess沙箱包含常用CLI工具:
python
import subprocessList files
列出文件
result = subprocess.run(["ls", "-lh", "/tmp"], capture_output=True, text=True)
print(result.stdout)
result = subprocess.run(["ls", "-lh", "/tmp"], capture_output=True, text=True)
print(result.stdout)
Parse JSON with jq
使用jq解析JSON
json_data = '{"name": "test", "count": 42}'
result = subprocess.run(
["jq", ".count"],
input=json_data,
capture_output=True,
text=True
)
print(f"Count: {result.stdout.strip()}")
undefinedjson_data = '{"name": "test", "count": 42}'
result = subprocess.run(
["jq", ".count"],
input=json_data,
capture_output=True,
text=True
)
print(f"Count: {result.stdout.strip()}")
undefinedTroubleshooting
故障排除
Error: "No container runtime available"
错误:"No container runtime available"
Cause: Podman/Docker not installed or not running.
Fix:
bash
undefined原因: Podman/Docker未安装或未运行。
解决方法:
bash
undefinedCheck status
检查状态
podman machine list
podman machine start
podman machine list
podman machine start
Or switch to Docker
或切换到Docker
export MCP_BRIDGE_RUNTIME=docker
undefinedexport MCP_BRIDGE_RUNTIME=docker
undefinedError: "Image pull failed"
错误:"Image pull failed"
Cause: Network issues or image not found.
Fix:
bash
undefined原因: 网络问题或镜像不存在。
解决方法:
bash
undefinedPre-pull the image
预拉取镜像
podman pull ghcr.io/elusznik/mcp-code-execution-mode:latest
podman pull ghcr.io/elusznik/mcp-code-execution-mode:latest
Or build locally
或本地构建
git clone https://github.com/elusznik/mcp-server-code-execution-mode.git
cd mcp-server-code-execution-mode
podman build -t mcp-code-execution:local -f Containerfile .
git clone https://github.com/elusznik/mcp-server-code-execution-mode.git
cd mcp-server-code-execution-mode
podman build -t mcp-code-execution:local -f Containerfile .
Update config to use local image
更新配置使用本地镜像
export MCP_BRIDGE_IMAGE=mcp-code-execution:local
undefinedexport MCP_BRIDGE_IMAGE=mcp-code-execution:local
undefinedError: "Tool not found in server X"
错误:"Tool not found in server X"
Cause: Tool name mismatch or server not configured.
Fix:
python
from mcp import runtime原因: 工具名称不匹配或服务器未配置。
解决方法:
python
from mcp import runtimeCheck what's actually available
检查实际可用的工具
docs = await runtime.query_tool_docs("github")
print([t["name"] for t in docs["tools"]])
docs = await runtime.query_tool_docs("github")
print([t["name"] for t in docs["tools"]])
Use exact name from output
使用输出中的准确名称
await mcp_github.create_or_update_file(...) # Not create_file
undefinedawait mcp_github.create_or_update_file(...) # 而非create_file
undefinedVariables Not Persisting
变量未持久化
Cause: Session restarted (happens on bridge reload).
Fix: Store critical data in files:
python
import pickle原因: 会话已重启(桥接器重载时会发生)。
解决方法: 将关键数据存储到文件中:
python
import pickleSave state
保存状态
state = {"counter": 42, "data": [1, 2, 3]}
with open("/tmp/state.pkl", "wb") as f:
pickle.dump(state, f)
state = {"counter": 42, "data": [1, 2, 3]}
with open("/tmp/state.pkl", "wb") as f:
pickle.dump(state, f)
Restore in next call
在下一次调用中恢复
with open("/tmp/state.pkl", "rb") as f:
state = pickle.load(f)
undefinedwith open("/tmp/state.pkl", "rb") as f:
state = pickle.load(f)
undefinedTimeout Errors
超时错误
Cause: Long-running computation exceeds 120s default.
Fix:
bash
undefined原因: 长时间计算超出默认120秒限制。
解决方法:
bash
undefinedIncrease timeout
增加超时时间
export MCP_BRIDGE_TIMEOUT=300
Or break work into chunks:
```pythonexport MCP_BRIDGE_TIMEOUT=300
或拆分任务为多个批次:
```pythonBad: process 1M rows in one call
不推荐:一次处理100万行
df = pd.read_csv("huge.csv") # Times out
df = pd.read_csv("huge.csv") # 会超时
Good: process in batches
推荐:分批次处理
for chunk in pd.read_csv("huge.csv", chunksize=10000):
process(chunk)
undefinedfor chunk in pd.read_csv("huge.csv", chunksize=10000):
process(chunk)
undefinedPermission Denied in Container
容器内权限被拒绝
Cause: Trying to write to read-only filesystem.
Fix: Use for temporary files:
/tmppython
undefined原因: 尝试写入只读文件系统。
解决方法: 使用存储临时文件:
/tmppython
undefinedBad
不推荐
with open("/data/output.txt", "w") as f: # Read-only
f.write("data")
with open("/data/output.txt", "w") as f: # 只读路径
f.write("data")
Good
推荐
with open("/tmp/output.txt", "w") as f: # Writable
f.write("data")
undefinedwith open("/tmp/output.txt", "w") as f: # 可写路径
f.write("data")
undefinedAdvanced Configuration
高级配置
Custom Container Image
自定义容器镜像
Build an image with extra dependencies:
dockerfile
FROM ghcr.io/elusznik/mcp-code-execution-mode:latest
RUN pip install --no-cache-dir \
scikit-learn \
seaborn \
sqlalchemybash
podman build -t mcp-custom:latest .
export MCP_BRIDGE_IMAGE=mcp-custom:latest构建包含额外依赖的镜像:
dockerfile
FROM ghcr.io/elusznik/mcp-code-execution-mode:latest
RUN pip install --no-cache-dir \
scikit-learn \
seaborn \
sqlalchemybash
podman build -t mcp-custom:latest .
export MCP_BRIDGE_IMAGE=mcp-custom:latestResource Limits
资源限制
bash
undefinedbash
undefinedIncrease memory for ML workloads
为机器学习工作负载增加内存
export MCP_BRIDGE_MEMORY_LIMIT=2g
export MCP_BRIDGE_MEMORY_LIMIT=2g
CPU quota (% of one core)
CPU配额(单个核心的百分比)
export MCP_BRIDGE_CPU_QUOTA=50000 # 50%
export MCP_BRIDGE_CPU_QUOTA=50000 # 50%
Max processes
最大进程数
export MCP_BRIDGE_PIDS_LIMIT=200
undefinedexport MCP_BRIDGE_PIDS_LIMIT=200
undefinedSecurity Hardening
安全加固
The bridge already runs rootless with:
- (no capabilities)
--cap-drop=ALL - (immutable root)
--read-only --security-opt=no-new-privileges- (no internet)
--network=none
For even stricter isolation:
bash
undefined桥接器已默认以无根模式运行,包含以下安全配置:
- (无特殊权限)
--cap-drop=ALL - (根文件系统只读)
--read-only --security-opt=no-new-privileges- (无网络连接)
--network=none
如需更严格的隔离:
bash
undefinedUse SELinux labels (Fedora/RHEL)
使用SELinux标签(Fedora/RHEL)
export MCP_BRIDGE_SECURITY_OPT="label=type:container_runtime_t"
export MCP_BRIDGE_SECURITY_OPT="label=type:container_runtime_t"
Disable session persistence
禁用会话持久化
export MCP_BRIDGE_SESSION_PERSIST=false
undefinedexport MCP_BRIDGE_SESSION_PERSIST=false
undefinedBest Practices
最佳实践
-
Use discovery before calling: Alwaysor
search_tool_docs()first to avoid guessing tool names.query_tool_docs() -
Handle errors gracefully: MCP servers can fail. Wrap calls in try/except and provide fallback logic.
-
Minimize round-trips: Write loops and conditionals in Python instead of asking the LLM to orchestrate multiple calls.
-
Persist critical state: Save important data tofiles. Variables persist within a session but not across bridge restarts.
/tmp/ -
Test locally first: Runstandalone to verify configuration before integrating with Claude.
python -m mcp_code_execution_mode
-
调用前先发现:始终先使用或
search_tool_docs(),避免猜测工具名称。query_tool_docs() -
优雅处理错误:MCP服务器可能故障,将调用包裹在try/except中并提供回退逻辑。
-
减少往返次数:在Python中编写循环和条件判断,而非让LLM编排多次调用。
-
持久化关键状态:将重要数据保存到文件中。变量在会话内持久,但桥接器重载后会丢失。
/tmp/ -
先本地测试:先独立运行验证配置,再集成到Claude中。
python -m mcp_code_execution_mode
Example: End-to-End Workflow
示例:端到端工作流
python
from mcp import runtime, mcp_github, mcp_slack
import pandas as pdpython
from mcp import runtime, mcp_github, mcp_slack
import pandas as pd1. Discover GitHub tools
1. 发现GitHub工具
servers = await runtime.discovered_servers()
if "github" not in servers:
raise ValueError("GitHub MCP server not configured")
servers = await runtime.discovered_servers()
if "github" not in servers:
raise ValueError("未配置GitHub MCP服务器")
2. Fetch issues
2. 获取Issue
issues = await mcp_github.list_issues(
repo="myorg/myrepo",
state="all",
since="2025-01-01"
)
issues = await mcp_github.list_issues(
repo="myorg/myrepo",
state="all",
since="2025-01-01"
)
3. Analyze with pandas
3. 使用pandas分析
df = pd.DataFrame(issues)
df["created"] = pd.to_datetime(df["created_at"])
monthly_counts = df.groupby(df["created"].dt.to_period("M")).size()
df = pd.DataFrame(issues)
df["created"] = pd.to_datetime(df["created_at"])
monthly_counts = df.groupby(df["created"].dt.to_period("M")).size()
4. Format report
4. 格式化报告
report = "📈 Issue Report\n\n"
for month, count in monthly_counts.items():
report += f"{month}: {count} issues\n"
report = "📈 Issue报告\n\n"
for month, count in monthly_counts.items():
report += f"{month}: {count}个Issue\n"
5. Post to Slack
5. 发布到Slack
if "slack" in servers:
await mcp_slack.post_message(
channel="#engineering",
text=report
)
print("✅ Report posted to Slack")
else:
print(report)
This workflow demonstrates discovery, data fetching, analysis, and multi-server orchestration—all in a single, token-efficient Python execution.if "slack" in servers:
await mcp_slack.post_message(
channel="#engineering",
text=report
)
print("✅ 报告已发布到Slack")
else:
print(report)
该工作流展示了发现、数据获取、分析和多服务器编排——全部在一次高效token利用的Python执行中完成。