在 E2B 沙箱中运行 Pydantic AI 编码代理
一个能执行 python -m unittest、还能改文件的编码代理确实好用。但如果这个代理就在你的笔记本上、以你的身份运行，~/.ssh 里放着你的 SSH 密钥，往上两级目录还有生产环境的 .env 文件——那这就是一份你没读条款就默认签下的风险协议。跑什么由代理自己决定，命令执行前也没人审核。可一个要等审核才动手的代理，也就称不上代理了。
现在，E2B 成了 Pydantic AI 的原生沙箱提供商。只要给代理的能力列表加上 E2BSandbox()，它执行的每条 shell 命令和每次文件操作都会跑在隔离的 E2B 微虚拟机（microVM）里，不再碰你的机器。你的环境里的任何东西都进不了沙箱。一次运行结束时，沙箱可以暂停；下次运行开始时再恢复，文件系统和内存原封不动。而且，工作区是通过一个不含凭据的引用（reference）来标识的，所以持久化工作流（durable workflow）能在 worker 崩溃后接着用它。
本文会走一遍 E2B cookbook 里的两个例子：一个是分两次运行的“先诊断、再修复”代理，两次运行之间沙箱会暂停；另一个是同样的任务放进 Temporal 工作流，但 worker 进程会在中途被杀掉。
一行代码，代理就离开了你的机器
先安装 harness，带上 E2B 和 Temporal 两个额外依赖，也就是下面代码片段用到的 provider 额外依赖（快速开始调用的是 Anthropic，cookbook 里的例子调用的是 OpenAI），再加上 Logfire 额外依赖，方便追踪运行过程。然后设置 E2B_API_KEY 和对应 provider 的密钥：
uv add "pydantic-ai-harness[e2b,anthropic,temporal]" "pydantic-ai-slim[openai,logfire]"
接着把沙箱和要用到它的工具一起交给代理：
import logfire
from pydantic_ai import Agent
from pydantic_ai_harness.coder import Coder
from pydantic_ai_harness.e2b_sandbox import E2BSandbox
logfire.configure()
logfire.instrument_pydantic_ai()
agent = Agent(
'anthropic:claude-opus-5-5',
capabilities=[E2BSandbox(), Coder()])
result = agent.run_sync(
'Clone https://github.com/pydantic/pydantic-ai and summarize how capabilities work.'
)
E2BSandbox 只负责执行。工具来自 Coder，或更细粒度的 Shell 和 FileSystem 能力，而这些工具现在跑在沙盒里。不需要写自定义 E2B 适配器，也不用构建模板。E2B 默认模板自带 Python 和 git，对本文来说够用。如果项目有自己的依赖，可以在自定义模板里安装，或者在代理启动前在沙盒里跑一遍 pip install——默认模板带 pip，但不带 uv。
首次运行之前，有几个默认设置值得先了解。沙盒默认存活一小时，除非你传入 sandbox_timeout；超时后 E2B 只会把它暂停，不会杀掉。默认允许联网，除非用 allow_internet_access=False 关掉。另外，一次运行结束时 Pydantic AI 不会杀掉沙盒：谁创建的沙盒，谁负责它的生命周期；忘了清理，就得一直为它付费。E2B 沙盒指南里讲了清理方法，包括 on_run_error 钩子，用于处理那些还没产出结果、拿不到引用就失败的运行。
诊断、暂停、恢复、修复
第一个 cookbook 示例让代理操作一个小型 Python CLI，用来汇总 JSONL 格式的 API 请求日志。它的测试没通过：CLI 统计服务器错误的方式不对，p95 延迟也算错了。代理会运行两轮。第一轮读代码和失败的测试，写出一份 diagnosis.md，但不碰源码。然后应用暂停沙盒，挂上一个全新的后端，第二轮利用保存下来的诊断和第一轮的对话历史修复 bug。
代理定义就是普通的 Pydantic AI 写法。唯一与 E2B 相关的部分是能力配置：
import logfire
from pydantic_ai import Agent
from pydantic_ai_harness.e2b_sandbox import E2BSandbox
from pydantic_ai_harness.filesystem import FileSystem
from pydantic_ai_harness.shell import Shell
logfire.configure()
logfire.instrument_pydantic_ai()
WORKDIR = '/home/user/log-cli'
agent = Agent(
'openai:gpt-4.1-mini',
name='log_cli_fixer',
instructions='只在当前沙箱项目里干活。用提供的工具；结论要通过跑测试来验证。',
capabilities=[
E2BSandbox(
working_dir=WORKDIR,
sandbox_timeout=600,
allow_internet_access=False,
),
Shell(tools=['run_command'], default_timeout=20, max_output_chars=6000),
FileSystem(tools=['read_file', 'write_file', 'edit_file']),
],
)
这个 cookbook 有意收窄了工具范围。Shell 只暴露 run_command，FileSystem 只给三个文件工具，而且因为演练环境用不着下载任何东西，所以直接关掉了联网。沙箱本身是一道边界，工具列表越窄，这道边界越容易想清楚。
Agent 启动前，应用会先把演练环境准备好：自己构造后端，在 Workspace 里把文件写进去。
from pydantic_ai.workspaces import Workspace
from pydantic_ai_harness.e2b_sandbox import E2BSandbox, E2BSandboxBackend
backend = E2BSandboxBackend(working_dir=WORKDIR, sandbox_timeout=600, allow_internet_access=False)
workspace = Workspace(backend)
for name in ('log_stats.py', 'test_log_stats.py', 'requests.jsonl'):
await workspace.write_bytes(name, (FIXTURE / name).read_bytes())
first = await agent.run(DIAGNOSE, workspace=backend, usage_limits=limits())
ref = first.workspace.ref
构造后端这一步不产生任何 I/O。第一次调用 write_bytes 才真正创建沙箱，从那时起，backend.ref 就能标识这个沙箱。运行结果在 first.workspace.ref 里带着同一个引用。
接下来就是暂停。应用从后端取出强类型的 e2b.AsyncSandbox，把它暂停，然后新建一个只知道引用的后端：
sandbox = await backend.get_sandbox()
assert await sandbox.pause()
attached = E2BSandbox(working_dir=WORKDIR, sandbox_timeout=600).backend(ref)
second = await agent.run(
FIX,
workspace=ref,
message_history=first.all_messages(),
usage_limits=limits(),
)
) assert second.workspace.ref == ref
第一次读取文件时，附加的后端就会恢复暂停的沙箱。E2B 会无限期保留暂停的沙箱，文件系统和内存都保留，所以 diagnosis.md 还在第一次运行留下的位置。第二次运行通过 workspace=ref 指定沙箱，用 message_history 带上对话历史。ref 或包含它的历史记录都能选中沙箱，但只有历史记录才携带代理已知的内容，示例代码两者都传了。
接下来这个示例做了我希望更多代理演示都做的事：检查代理的工作，而不是信任代理的总结。它从可信的宿主机 fixture 恢复测试文件（因为 shell 命令可以绕过文件工具的限制），确认输入日志没有被改动，然后自己运行测试。四个测试通过，CLI 打印出正确的计数，应用在 finally 块里销毁沙箱，并确认它已从账户中消失。
Created sandbox: <id>
Diagnosis saved. Model usage: ...
Paused sandbox: <id>
Resumed sandbox: <same id>; checksums unchanged before repair
...
Ran 4 tests ...
OK
CLI: {"requests": 4, "server_errors": 2, "p95_ms": 100}
清理验证通过： 脚本在暂停前后分别校验了沙箱身份和 SHA-256 校验和。相同的 id，相同的字节。
杀掉 worker，保留沙箱
暂停与恢复的例子证明了沙箱的生命周期长于一次运行。Temporal 的例子则证明，沙箱的生命周期也长于驱动 agent 的那个进程。任务还是同一个 log CLI。这一次，两次 agent 运行都放在一个 Temporal 工作流里，两次运行之间，demo 向 worker 发送 SIGKILL。没有 Python 的 finally，没有优雅关闭。一个新 worker 在同一个任务队列上启动，Temporal 重放已记录的历史，工作流在被杀掉的 worker 留下的沙箱里继续修复。能力列表里多了 TemporalDurability，其余部分就是第一个例子里的那个 agent。Cookbook 把它封装进一个共享的 make_agent() 辅助函数；完整写出来，再加上示例所用的 Temporal activity 设置，是这样的：
from datetime import timedelta
import logfire
from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import TemporalDurability
from pydantic_ai_harness.e2b_sandbox import E2BSandbox
from pydantic_ai_harness.filesystem import FileSystem
from pydantic_ai_harness.shell import Shell
from temporalio.common import RetryPolicy
logfire.configure()
logfire.instrument_pydantic_ai()
WORKDIR = '/home/user/log-cli'
agent = Agent(
'openai:gpt-4.1-mini',
name='log_cli_fixer',
instructions='Work only in this sandbox project. Use the supplied tools; verify claims by running tests.',
capabilities=[
E2BSandbox(working_dir=WORKDIR, sandbox_timeout=600, allow_internet_access=False),
Shell(tools=['run_command'], default_timeout=20, max_output_chars=6000),
FileSystem(tools=['read_file', 'write_file', 'edit_file']),
TemporalDurability(
activity_config={
'start_to_close_timeout': timedelta(seconds=45),
'retry_policy': RetryPolicy(maximum_attempts=2),
},
),
],
Agent 需要一个名字，而且它和 workflow 都必须在模块层级定义，这样 Temporal 才能注册这些 activity。两个 worker 进程导入同一个模块。workflow 把 sandbox 引用作为输入，这一点很关键：controller 创建并初始化一个 sandbox，然后交给 workflow 一个 WorkspaceRef，其中不含任何凭证。Temporal 会在任何 worker 接触 agent 之前记录这个输入，因此无论最终是哪个 worker 执行任务的后半段，它都知道该连接到哪个 sandbox。下面是 cookbook 中 workflow.py 的代码，只保留了关键部分。DIAGNOSE、FIX、fingerprints 和 limits 分别对应示例 project.py 中的提示词、校验和辅助函数以及 UsageLimits：
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from agent import agent
from project import DIAGNOSE, FIX, fingerprints, limits
from pydantic_ai.durable_exec.temporal import PydanticAIWorkflow
from pydantic_ai.workspaces import WorkspaceRef
@workflow.defn
class FixLogCLI(PydanticAIWorkflow):
__pydantic_ai_agents__ = (agent,)
def __init__(self):
self.checkpoint = None
self.proceed = False
@workflow.query
def diagnosis_checkpoint(self) -> dict | None:
return self.checkpoint
@workflow.signal
def continue_fix(self) -> None:
self.proceed = True
@workflow.run
async def run(self, ref: WorkspaceRef) -> dict:
first = await agent.run(DIAGNOSE, workspace=ref, usage_limits=limits())
saved = await fingerprints(first.workspace)
self.checkpoint = {'sandbox_id': ref.id, 'checksums': saved}
await workflow.wait_condition(lambda: self.proceed, timeout=180)
assert await fingerprints(first.workspace) == saved
second = await agent.run(
FIX,
workspace=ref,
message_history=first.all_messages(),
usage_limits=limits(),
) return { 'sandbox_id' : ref.id , 'summary' : second.output}
诊断和修复之间的 wait_condition 给崩溃制造了一个可复现的触发点。控制器轮询 diagnosis_checkpoint 查询，直到第一次运行已经保存了诊断结果，然后杀掉工作进程，启动另一个，再次轮询同一个查询，之后才发送 continue_fix。恢复的检查点与已死工作进程记录的那个一致。
Created sandbox: <id>
Worker started: pid=<first pid>
Durable checkpoint: <id>; diagnosis and files saved
Worker crashed: pid=<first pid>, exit=-9
Worker started: pid=<different pid>
Checkpoint recovered by pid=<different pid>; sandbox=<same id>
...
Ran 4 tests ...
OK
Exactly one sandbox; original workflow completed after worker restart
核对通过： 最后的资源检查这一行最值得先看。整个流程跑完后，账号里只多出一个 sandbox，无论运行中还是已暂停。如果恢复逻辑写得粗糙，就会多建一个 sandbox，并在空目录里从头重跑任务。当引用指向一个已经不存在的 sandbox 时，E2BSandbox 会抛出 WorkspaceUnavailableError，而不是悄悄新建一个来顶替。恢复要么接着诊断中断的地方继续，要么明确报错。cookbook 的 README 对这套方案能证明什么、不能证明什么写得很克制，这里值得再强调一遍：崩溃发生在空闲检查点，当时没有命令正在执行。如果 worker 在完成状态被记录之前消失，Temporal 会重新运行 activity，所以 shell 命令是至少一次，不是恰好一次。恢复不会回滚文件写入。写工具如果用“替换某个命名结果”而不是“追加”的方式，那么 activity 跑多少次，结果都一样。
追踪数据流向哪里
E2BSandbox 自己不产生任何 span。Pydantic AI 的埋点会把 sandbox 记在 agent run 的 span 上，字段是 pydantic_ai.workspace.provider 和 pydantic_ai.workspace.id，工具执行的每条命令、每次文件操作，都落在对应 tool call 的 span 里。把前面代码片段里那两行 logfire 配置加上，Pydantic Logfire 就能显示诊断那一轮和恢复那一轮、它们各自的 run_command，以及两轮里相同的 sandbox id，这样 trace 就能和 E2B 控制台对上。在 Temporal 示例里，两轮运行在同一个 workflow 下，所以崩溃看起来像是两个 activity 之间的断档，而不是两个不相干的任务。如果你还没接入 trace，Pydantic Logfire 有免费额度，足够把 cookbook 里的两个示例跑很多次。建个项目，跑一次 logfire auth，再加上代码片段里那两行，剩下的它都替你做好：每次模型请求、每条 run_command，以及把一次运行和 E2B 控制台对上的 sandbox id，全在一条 trace 里，还能用 SQL 查询。
选 E2B 还是 Modal，或者两个都要
Pydantic AI Harness 也提供 ModalSandbox，我们在 Harness Week 期间写过介绍。
两种做法都是把 agent 的 shell 放进一个你没亲手准备的隔离容器里，而且都只是能力清单中的一项。讲 Modal 的那篇侧重横向铺开：五百个容器，每个跑六分钟，测试套件跑完就销毁。这一篇讲的是留住沙箱：一次运行结束后把它暂停，记下引用，下周再恢复——克隆的代码、virtualenv，还有做了一半的分支都还在。跨越多次 agent 运行的编码会话，或者必须在自己 worker 挂掉后还能接着走的工作流，属于 E2B 这种形态；测试矩阵属于 Modal 那种形态。应用两者各用一摊，也没什么不可以。