进程内 MCP Server 搭建

基于 FC Agent Sandbox 和 Claude Agent SDK 的 MCP 服务搭建

本文是一份从 0 开始的搭建教程：把阿里云函数计算云沙箱（FC Agent Sandbox）的代码执行能力，封装为一组 MCP 工具，注册进 Claude Agent SDK 的 Agent 循环，让大模型在对话中自主调用沙箱执行代码。
读者需要具备 Python 基础；按顺序把各步代码保存到同一文件即可运行。

目标与原理

要解决什么问题

Claude Agent SDK 提供的是 Agent 循环：模型读到用户任务 → 决定调用某个工具 → 拿到工具结果 →决定下一步 → 直到给出最终答复。但模型开箱可用的只有读文件、搜索等本地工具——它没有地方执行代码。

我们的目标是补上这块能力：把「在阿里云沙箱里执行代码 / 传文件 / 取结果」包装成 MCP 工具，注册给模型。模型负责「想做什么」，沙箱负责「安全地做」，业务进程自身不运行任何生成代码。

进程内 MCP Server：最轻的服务形态

传统 MCP 服务是一个独立进程（stdio 或 HTTP/SSE 传输），需要手写 JSON-RPC 消息处理。
Claude Agent SDK 提供了第三种形态——进程内 SDK Server：

server = create_sdk_mcp_server(name="sandbox", version="1.0.0", tools=[...])

它返回一个配置对象，直接挂到 Agent 选项里。协议解析、与 CLI 子进程的通信全部由 SDK 完成，你只需要提供「工具清单」。
因此本文搭建的”MCP 服务”= 一组用 @tool 装饰器定义的 async 函数

• 一个 create_sdk_mcp_server 调用，不需要独立进程、端口或网络服务。

一次工具调用的完整链路

用户任务 ─▶ Claude（决策） ─▶ 发起工具调用 mcp__sandbox__run_code
                                    │
                                    ▼
                     SDK 调用你的 async handler（进程内）
                                    │
                                    ▼
                     handler 调用沙箱执行代码，拿到结果
                                    │
                                    ▼
                     结果以 {"content": [...], "is_error": ...} 回传模型
                                    │
                                    ▼
              模型读到结果（或错误+traceback）─▶ 修正重试 / 汇报结论

理解这条链路，后面每一步的职责就清楚了：
[第 1 步](#第 1 步：封装沙箱会话——为工具提供真实执行能力)做最底层的沙箱执行；
[第 2 步](#第 2 步：定义 MCP 工具——把沙箱能力翻译成模型可调用的接口)把执行能力翻译成模型可调用的工具接口；
[第 3 步](#第 3 步：组装 MCP Server——给工具集一个名字)把工具集组装成 Server；
[第 4~5 步](#第 4 步：编写系统提示词——教模型正确使用你的服务)把 Server 接进 Agent 循环并告诉模型怎么用；
[第 6 步](#第 6 步：运行与验证)运行验证。

环境准备

安装依赖

pip install claude-agent-sdk e2b-code-interpreter

• claude-agent-sdk：Agent 循环 + 进程内 MCP Server 机制。官方 wheel 自带 Claude Code CLI二进制（Windows/Linux/macOS x64），无需另装 Node.js。
• e2b-code-interpreter：阿里云沙箱官方 Python SDK（E2B 兼容），负责创建沙箱、执行代码、读写沙箱文件。安装时会一并装上其依赖 e2b。

准备两组凭证

搭建这个服务需要两组互不相关的凭证：

组	变量	用途
沙箱凭证	E2B_API_KEY / E2B_API_URL / E2B_DOMAIN	创建与访问阿里云沙箱（在函数计算控制台创建 API Key）
模型凭证	ANTHROPIC_API_KEY（官方直连）或 ANTHROPIC_BASE_URL + ANTHROPIC_AUTH_TOKEN（中转网关）	Claude Agent SDK 调用大模型

两组凭证的消费者不同：沙箱凭证给你的 Python 进程用（创建沙箱）；模型凭证注入 SDK拉起的 CLI 子进程用。模型凭证绝不能进入沙箱——沙箱里跑的是不可信代码。

写入 .env

在项目根目录创建 .env（并将其加入 .gitignore，凭证永不进代码库与日志）：

# 阿里云沙箱三要素（<region> 需与账号开通地域一致）
E2B_API_KEY=your-e2b-api-key
E2B_API_URL=https://api.cn-shanghai.e2b.fc.aliyuncs.com
E2B_DOMAIN=cn-shanghai.e2b.fc.aliyuncs.com

# 模型凭证——中转网关方式（二选一）
ANTHROPIC_BASE_URL=https://your-gateway.example.com
ANTHROPIC_AUTH_TOKEN=sk-your-token
CLAUDE_MODEL=your-model-name

# 模型凭证——官方直连方式（与上面二选一）
# ANTHROPIC_API_KEY=sk-ant-your-key

CLAUDE_MODEL 为什么建议显式设置：CLI 子进程会读取本机 ~/.claude.json 里保存的”上次使用的模型”，SDK 的 setting_sources=[] 选项管不到这个文件。若本机残留过网关不支持的模型名，请求会直接 400。显式指定模型（经 options.model 传入）优先级最高，一律显式指定最稳妥。

用一个简单的加载函数把 .env 读进环境变量（SDK 不自动读取 .env）：

import os
from pathlib import Path

def load_env(path: str = ".env") -> None:
    """把 .env 的键值对写入进程环境变量（已存在的进程变量优先，不覆盖）。"""
    file = Path(path)
    if not file.is_file():
        return
    for line in file.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        value = value.strip().strip("'\"")
        os.environ.setdefault(key.strip(), value)

第 1 步：封装沙箱会话——为工具提供真实执行能力

这一步的作用：MCP 工具只是”接口”，真正干活的是沙箱。先写一个沙箱会话封装类，解决三个工程问题，后面所有工具都只调它：

1. 异步桥接：e2b SDK 是同步阻塞的（一次 HTTP 请求直到执行完成才返回），而 MCP 工具的handler 是 async 函数。直接调用会卡死整个事件循环——必须用 asyncio.to_thread 把阻塞调用丢进线程池。
2. 操作串行化：一个 Code Context（变量空间）同时只能跑一段代码，e2b 的并发行为也没有承诺。用一把 asyncio.Lock 把所有沙箱操作串行化，行为可预期。
3. 懒创建与确定性销毁：沙箱按存活时间计费。纯聊天轮不该花钱——首次真正执行代码才创建；结束后必须 kill()，放在 finally 里保证异常路径也不泄漏。

import asyncio
from e2b_code_interpreter import Sandbox

class SandboxBox:
    """沙箱会话封装：懒创建 + 串行锁 + 异步桥接。"""

    def __init__(self) -> None:
        self._sandbox: Sandbox | None = None
        self._context = None
        self._lock = asyncio.Lock()          # 串行化所有沙箱操作
        self._closed = False

    async def _ensure_started(self) -> Sandbox:
        """懒创建：首次调用时才真正创建沙箱与代码上下文。"""
        if self._closed:
            raise RuntimeError("SandboxBox 已关闭")
        if self._sandbox is None:
            def _create():
                sandbox = Sandbox.create(
                    template="code-interpreter-v1",   # 官方代码解释器模板
                    api_key=os.environ["E2B_API_KEY"],
                    api_url=os.environ["E2B_API_URL"],
                    domain=os.environ["E2B_DOMAIN"],
                    timeout=900,                      # 空闲 15 分钟自动回收（兜底）
                )
                # Code Context：独立变量空间，多次执行共享变量与导入
                context = sandbox.create_code_context(cwd="/tmp", language="python")
                return sandbox, context
            # 同步创建放线程池执行，不阻塞事件循环
            self._sandbox, self._context = await asyncio.to_thread(_create)
        return self._sandbox

    async def run_code(self, code: str, timeout: int = 25):
        """执行 python 代码（变量跨调用保留），返回 e2b 执行结果对象。"""
        async with self._lock:
            sandbox = await self._ensure_started()
            return await asyncio.to_thread(
                sandbox.run_code, code, context=self._context, timeout=timeout
            )

    async def write_file(self, path: str, data: str | bytes) -> None:
        """把数据写入沙箱文件（输入数据用）。"""
        async with self._lock:
            sandbox = await self._ensure_started()
            await asyncio.to_thread(sandbox.files.write, path, data)

    async def read_file(self, path: str, binary: bool = False):
        """读取沙箱文件；binary=True 返回 bytes（图片等产物用）。"""
        async with self._lock:
            sandbox = await self._ensure_started()
            if binary:
                return await asyncio.to_thread(sandbox.files.read, path, format="bytes")
            return await asyncio.to_thread(sandbox.files.read, path)

    async def close(self) -> None:
        """销毁沙箱（幂等，可安全重复调用）。"""
        async with self._lock:
            self._closed = True
            sandbox, self._sandbox, self._context = self._sandbox, None, None
        if sandbox is not None:
            await asyncio.to_thread(sandbox.kill)

再写一个结果格式化函数——e2b 返回的执行对象字段很多，直接 dump 给模型既浪费 token又不利阅读。组装成固定分节文本，模型定位信息最快：

def format_result(execution) -> tuple[str, bool]:
    """把 e2b 执行结果组装为分节文本，返回 (文本, 是否失败)。"""
    ok = execution.error is None
    sections = [f"[success] {ok}"]
    stdout = "".join(execution.logs.stdout).strip()   # logs.stdout 是日志片段列表
    stderr = "".join(execution.logs.stderr).strip()
    if stdout:
        sections.append("[stdout]\n" + stdout)
    if stderr:
        sections.append("[stderr]\n" + stderr)
    if execution.text is not None:                    # 最后一个表达式的值
        sections.append(f"[result] {execution.text}")
    if not ok:
        err = execution.error
        sections.append(f"[error] {err.value or err.name}")
        if err.traceback:                             # traceback 是自我修正的关键
            sections.append("[traceback]\n" + err.traceback.strip())
    return "\n".join(sections), not ok

至此沙箱后端就绪。注意这一步里没有任何 Claude SDK 的东西——它是一块可独立复用的能力，MCP 工具只是它的一层”翻译”。

第 2 步：定义 MCP 工具——把沙箱能力翻译成模型可调用的接口

这一步的作用：用 @tool 装饰器把一个 async 函数登记为 MCP 工具。模型在对话中能”看到”的只有工具名、描述和参数 schema——这三样写得好坏，直接决定模型用不用它、怎么传参、会不会选错工具。

@tool 的四个要素

from claude_agent_sdk import tool

@tool(name, description, input_schema)
async def my_tool(args: dict) -> dict:
    ...

要素	作用	写法要点
name	模型调用时使用的标识	动词或动宾结构，语义直白（run_code 而非 rc）
description	模型判断”何时该用这个工具”的唯一依据	写清用途 + 硬性约束（超时上限、文件路径约定）。约束写进描述，模型才会遵守
input_schema	完整 JSON Schema，控制参数名/类型/必填/取值范围	必须用完整 schema 并显式给 required 列表——简单 dict 会被 SDK 视为所有键必填，这是常见坑
handler	真正执行的 async 函数	入参是 dict（模型按 schema 生成的实参）；返回 {"content": [...], "is_error": bool}

完整示例：run_code 工具

RUN_CODE_DESCRIPTION = (
    "在阿里云云沙箱（云端隔离环境）中执行 python 代码，返回结构化分节结果"
    "（[success]/[stdout]/[stderr]/[result]/[error]/[traceback]）。"
    "变量与导入在同一会话中跨调用保留，适合分多步增量完成任务。"
    "硬性约束：单次执行不超过 25 秒（平台约 30 秒硬限制），长任务必须拆分为多次调用。"
)

RUN_CODE_SCHEMA = {
    "type": "object",
    "properties": {
        "code": {"type": "string", "description": "要执行的 python 代码"},
        "timeout": {
            "type": "integer", "minimum": 1, "maximum": 25,
            "description": "执行超时（秒），默认 25，上限 25",
        },
    },
    "required": ["code"],          # timeout 是可选参数——只有显式列进 required 才必填
}

def make_tools(box: SandboxBox) -> list:
    """构建工具集。handler 是闭包，捕获共享的沙箱会话 box。"""

    @tool("run_code", RUN_CODE_DESCRIPTION, RUN_CODE_SCHEMA)
    async def run_code(args: dict) -> dict:
        try:
            execution = await box.run_code(
                args["code"], timeout=args.get("timeout") or 25
            )
            text, is_error = format_result(execution)
            return {"content": [{"type": "text", "text": text}], "is_error": is_error}
        except Exception as exc:   # 沙箱创建失败等基础设施异常
            return {
                "content": [{"type": "text",
                             "text": f"[tool_error] {type(exc).__name__}: {exc}"}],
                "is_error": True,
            }

    @tool("write_file", ..., ...)   # schema/描述同理，此处从略
    async def write_file(args: dict) -> dict:
        ...

    return [run_code, write_file]

错误的两条通道——自我修正闭环的关键

handler 的返回结构里 is_error 是一个语义强烈的开关，务必区分两种失败：

1. 代码执行失败（除零、NameError……）：这是给模型看的正常业务结果。带上[error] 摘要与 [traceback] 尾部、is_error=True 回传——模型读到确切原因后，Agent 循环会驱动它修正代码自动重试，不需要人工介入。这是整个服务最核心的机制。
2. 基础设施异常（沙箱创建失败、网络断开……）：由 handler 的 except 兜底，同样以is_error 文本回传，并在文本里引导模型”这是环境问题，停止重试并告知用户”。

两条通道都回传文本而不是抛异常——抛异常会中断 Agent 循环，模型失去修正机会。

按同样的模式可以继续扩展工具：read_file（读沙箱文本）、read_image（读图片并以image 块回传，见 9.2）、list_files、download_file（产物落盘本地）等。工具数量没有硬限制，但每个工具的 description 都要写清”何时用我”。

第 3 步：组装 MCP Server——给工具集一个名字

这一步的作用：把工具列表组装成一个命名 Server。名字不是装饰——它决定工具的全局调用名，也决定后面权限配置怎么写。

from claude_agent_sdk import create_sdk_mcp_server

def build_server(box: SandboxBox) -> dict:
    return create_sdk_mcp_server(
        name="sandbox",        # Server 名
        version="1.0.0",
        tools=make_tools(box),
    )

命名规则：工具在模型眼中的全名是 mcp__{server名}__{工具名}。上面注册的 run_code全名即 mcp__sandbox__run_code。这个前缀在两处会用到：

• 自动批准：allowed_tools=["mcp__sandbox__*"] 一条通配规则放行全部沙箱工具，无需逐个罗列；
• 日志审计：Agent 循环打印的每次工具调用都带全名，一眼区分”模型调了沙箱”还是”模型调了本地工具”。

第 4 步：编写系统提示词——教模型正确使用你的服务

这一步的作用：工具描述只回答”单个工具怎么用”，系统提示词回答”整体怎么协作”——执行环境是什么、哪些红线不能碰、产物放哪、失败怎么办。MCP 服务搭建中这步常被忽略，但它直接决定模型的行为质量。最少要写清四件事：

SYSTEM_PROMPT = """\
你是一名编程与数据分析助手，接入了一个阿里云云沙箱（云端隔离执行环境）。

# 执行环境
- 所有代码一律通过沙箱工具在云端执行，业务进程不运行任何生成代码。
- 沙箱是懒创建的：首次调用沙箱工具时自动创建，同一会话内复用，python 变量跨调用保留。

# 硬性约束
1. 单次 run_code 执行不超过 25 秒（平台网关约 30 秒硬限制）。长任务必须拆分为
   多次调用，可借助变量保留分步计算。
2. 数据与产物文件统一放 /tmp。
3. 图表用 plt.savefig('/tmp/xxx.png') 保存（禁止 plt.show()），需要确认时用
   read_image 查看。
4. 执行失败时，[error] 与 [traceback] 给出了确切原因，修正后重试；连续失败
   3 次以上停止重试并向用户说明原因。
"""

第 5 步：组装 Agent 选项并启动循环

这一步的作用：ClaudeAgentOptions 是服务的总装配台——MCP Server 挂载、权限策略、模型凭证、提示词，全部在这里汇合。逐字段解释：

import claude_agent_sdk as sdk

def build_options(box: SandboxBox) -> sdk.ClaudeAgentOptions:
    return sdk.ClaudeAgentOptions(
        # 核心：挂载我们的 MCP 服务
        mcp_servers={"sandbox": build_server(box)},

        # 权限三件套（无人值守自动化的关键）
        allowed_tools=["mcp__sandbox__*"],   # 通配放行全部沙箱工具，免人工确认
        permission_mode="bypassPermissions", # 无人值守模式，工具调用不弹审批
        setting_sources=[],                  # 不加载本机用户级 Claude Code 设置
        strict_mcp_config=True,              # 忽略外部 .mcp.json 注入的 MCP 服务

        # 模型凭证：只注入 CLI 子进程环境（不进沙箱、不进日志）
        env={
            "ANTHROPIC_AUTH_TOKEN": os.environ["ANTHROPIC_AUTH_TOKEN"],
            "ANTHROPIC_BASE_URL": os.environ["ANTHROPIC_BASE_URL"],
        },
        model=os.environ.get("CLAUDE_MODEL"),

        # 行为参数
        system_prompt=SYSTEM_PROMPT,
        max_turns=30,                        # 工具调用轮次上限，防失控循环刷 token
    )

各字段的取舍理由：

字段	作用	不设置的后果
mcp_servers	挂载进程内 MCP 服务	模型没有沙箱工具可用
allowed_tools + permission_mode	沙箱工具自动批准、无人值守	每次工具调用挂起等待人工确认，自动化中断
setting_sources=[] / strict_mcp_config=True	隔离本机残留配置与外部 MCP 注入	用户级设置/外部 .mcp.json 混入未知工具，行为不可控
env	模型凭证进 CLI 子进程	CLI 子进程拿不到网关凭证，无法调模型
max_turns	循环轮次上限	模型陷入死循环时无限消耗 token 与沙箱时间

如果还想给模型本地能力，用 tools=["Read", "Glob", "Grep"] 白名单只开放只读内建工具（Bash/Edit/Write 一律不进白名单），必要时再配 disallowed_tools=["Read(.env)"]这类 deny 规则禁读凭证文件——deny 规则在包括 bypassPermissions 在内的所有权限模式下都生效。最小化原则：模型不需要的能力就不给它。

消费消息流

query() 返回一个异步生成器，流式产出三类消息。业务侧通常关心两种：

async def run_task(box: SandboxBox, prompt: str) -> None:
    options = build_options(box)
    async for message in sdk.query(prompt=prompt, options=options):
        if isinstance(message, sdk.AssistantMessage):
            for block in message.content:
                if isinstance(block, sdk.TextBlock) and block.text.strip():
                    print(f"[Claude] {block.text.strip()}")          # 模型的叙述
                elif isinstance(block, sdk.ToolUseBlock):
                    print(f"[工具] {block.name}({block.input})")     # 模型发起了工具调用
        elif isinstance(message, sdk.ResultMessage):
            print(f"[完成] subtype={message.subtype} "
                  f"轮次={message.num_turns} 成本=${message.total_cost_usd or 0:.4f}")

• AssistantMessage.content 是块列表：TextBlock（模型的话）与 ToolUseBlock（工具调用请求：工具名 + 参数）。工具的实际执行发生在 SDK 内部（它调你的 handler），你在循环里看到 ToolUseBlock 只是”记录”这次调用。
• ResultMessage 是整轮任务的终报：subtype（success / error_max_turns 等）、最终文本 result、成本 total_cost_usd、轮次 num_turns。

第 6 步：运行与验证

把入口串起来——注意 finally 里关闭沙箱，这是防泄漏的最后防线：

async def main() -> None:
    load_env()
    box = SandboxBox()
    try:
        await run_task(box, "在沙箱中计算 1 到 100 的和，并用高斯公式验证结果一致。")
    finally:
        await box.close()      # 无论成功失败，沙箱确定性销毁

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

预期输出形态（模型实际行为，措辞会略有差异）：

[Claude] 我先用循环求和计算...
[工具] mcp__sandbox__run_code({'code': 'total = sum(range(1, 101))\nprint(total)', ...})
[Claude] 再用高斯公式验证...
[工具] mcp__sandbox__run_code({'code': 'n = 100\nassert n*(n+1)//2 == total\nprint("一致")', ...})
[Claude] 两种方法都是 5050，结果一致。
[完成] subtype=success 轮次=3 成本=$0.0312

验收清单（一条不满足就回头查对应步骤）：

• 出现 mcp__sandbox__run_code 调用 → 第 3~5 步的挂载与批准链路通了
• 工具结果被模型正确引用（如复述 5050）→ 第 4 步的结果格式化有效
• 故意给一个会报错的任务，模型能读 traceback 后自动修正重试 → 第 4.3 节闭环生效
• 运行结束后沙箱列表为空 → 第 3 步的销毁逻辑生效
• 全程日志无明文凭证 → 第 2 步的凭证纪律生效

进阶要点（实测经验）

30 秒网关限制：超时护栏要三处同设

阿里云沙箱的单次 run_code HTTP 请求存在约 30 秒的网关硬限制——超时直接被切断（502 + SDK 抛 TimeoutException），timeout 参数调大也突破不了。护栏要在三处同时设：

1. handler 默认 timeout=25（留 5 秒余量）；
2. schema 里 "maximum": 25——让模型从源头就不传超限值；
3. 系统提示词写明”长任务拆分多次调用，python 变量会保留”——给模型一条可行的出路，而不只是禁令。

image 内容块：让模型”看见”沙箱里生成的图表

MCP 工具结果不只能是文本。返回 base64 图片块，模型即可直接视觉读取：

import base64

@tool("read_image", "读取沙箱图片文件（png/jpg，≤5MB），你能直接看到图", PATH_SCHEMA)
async def read_image(args: dict) -> dict:
    data = await box.read_file(args["path"], binary=True)
    return {"content": [{
        "type": "image",
        "data": base64.b64encode(data).decode("ascii"),
        "mimeType": "image/png",
    }]}

沙箱环境通常没有显示设备，matplotlib 的 plt.show() 是死路；配合提示词约定”先 savefig('/tmp/x.png') 再 read_image”，就形成了「画图 → 看图 → 修图」的视觉闭环。

上下文保护：一切回传给模型的东西都要截断

工具结果会全部进入模型上下文。一次 print 出 50 万字符，上下文当场被冲爆。原则：文本结果一律截断（如 2 万字符）并附截断标记；图片限大小（如 5MB）；traceback 只回传尾部若干行（头部是框架噪声，最后一行才是原因）。

安全边界回顾

• 模型凭证只进 CLI 子进程（options.env），沙箱凭证只在你的 Python 进程——两者都不进沙箱环境变量、不进日志；
• 沙箱里跑的是不可信代码：本地内建工具只开放只读白名单，deny 规则禁读 .env；
• 沙箱空闲超时（如 900 秒）作为兜底，finally 主动销毁是主路径。

中转网关的适配细节

用 Anthropic 兼容中转网关（而非官方直连）时，三个实测结论：

• 网关支持的模型集合与官方不同，报 400 model ... is not supported 就换模型名（部分网关不提供 /v1/models 列表，可用最小请求逐个探测）；
• 必须显式设置 CLAUDE_MODEL（见 2.3 节）覆盖 ~/.claude.json 的残留默认模型；
• CLI 后台的会话命名功能可能用网关不支持的小模型，产生 unrecognized_model 警告——
  不影响主流程，可忽略。

常见问题排查

现象	原因	处理
400 model ... is not supported	网关不支持该模型名（常来自 ~/.claude.json 残留）	显式设置 CLAUDE_MODEL（9.5 节）
工具调用挂起不返回	permission_mode 未设 bypass，或工具不在 allowed_tools	补 allowed_tools=["mcp__sandbox__*"] + permission_mode="bypassPermissions"
模型说”我没有执行代码的工具”	Server 未挂载或 strict_mcp_config 把它过滤了	检查 mcp_servers 键名与 Server 名一致
执行约 30 秒后 502 / TimeoutException	撞上网关硬限制	9.1 节三处护栏 + 提示词引导拆分
上下文爆炸 / 响应变慢	工具结果未截断	9.3 节
沙箱越跑越多 / 计费不止	kill() 未在 finally 中调用	第 8 步入口模式
沙箱创建 401/403	E2B_API_KEY 无效或 E2B_API_URL/E2B_DOMAIN 地域与账号资源不一致	核对三要素与地域