跳到主要内容

数据截至 (上游 commit 771b20c93db8)

SDK 客户端:一次 run_code 如何变成流式 HTTP

30 秒导读: 你写 sandbox.run_code("x+1"),SDK 把它打包成一个 POST /execute 请求发进沙箱,以流的方式逐行读回来,一行一行解析累积成一个 Execution 对象(结果 / 日志 / 错误)。本章讲这一层——人最先碰到的那几个方法,以及其中一个 精妙决定:为什么它强制 HTTP/1.1。

本章是 E2B Code Interpreter 的最外层:SDK 客户端。它上面是你的 agent 代码, 下面是沙箱里跑着的 FastAPI 服务(见 02-server-routing.md)。 本章只讲客户端如何发请求、如何读流、如何把流拼成数据对象。

1. 这是什么(零基础也能懂)​

一句话定义: 这是一个 Python / JS 的库,让你把一段代码发到云端沙箱里执行, 并把执行产物(打印出来的日志、最后一行的返回值、画的图、报的错)取回来。

解决什么问题 / 给谁用: 假设你在写一个 AI agent,模型生成了一段 Python, 你不敢直接在自己机器上 exec() 它——万一它 rm -rf 呢?于是你把它扔进一个 一次性的隔离沙箱去跑,跑完把结果读回来喂给模型。E2B Code Interpreter 就是那个"扔进去、读回来"的客户端。

它能做什么:

  • run_code(code) —— 执行一段代码,拿回 Execution。
  • 指定 language(python / javascript / r / bash …)或指定一个已有 context。
  • 边执行边通过回调拿到实时输出(on_stdout / on_result …)。
  • 管理 code context(有状态的执行环境,类似一个 Jupyter 内核):创建 / 列出 / 删除 / 重启。

用起来什么样(取自 README.md:62-66,真实示例):

from e2b_code_interpreter import Sandbox

with Sandbox.create() as sandbox:
    sandbox.run_code("x = 1")          # 定义变量
    execution = sandbox.run_code("x+=1; x")   # 复用上一次的 x
    print(execution.text)              # -> "2"

注意第二次调用能读到第一次定义的 x——因为它们默认落在同一个持久 context 里。这正是"Code Interpreter"和"跑个一次性脚本"的区别:状态跨调用保留。

一句话直觉: 把 run_code 想成一个远程的 Jupyter 单元格执行。你敲一个 cell、 按 Shift+Enter,输出一行行冒出来,变量还留在内核里等下一个 cell 用。SDK 客户端就是 那个"按 Shift+Enter 并把输出流回来"的遥控器。

2. 顶层全景(它大概怎么转)​

怎么读这张图: 从左到右是一次 run_code 的生命周期。左边是你的进程, 中间那条粗线是流式 HTTP,右边是沙箱内的服务。

你的进程                          │  沙箱 (49999 端口)
                                  │
run_code(code, language/context)  │
   │                              │
   │ 组装 JSON 体                 │
   │ {code, context_id,           │
   │  language, env_vars}         │
   ▼                              │
_client.stream("POST",  ═════════════▶  FastAPI /execute
  /execute)             流式请求  │       │ 转发给 Jupyter 内核
   │                              │       ▼
   │  ◀═══════════════════════════════  一行一个 JSON:
   │   逐行 (iter_lines)          │       {"type":"stdout",...}
   ▼                              │       {"type":"result",...}
parse_output(execution, line)     │       {"type":"error",...}
   │ 按 type 分派 → 累积           │       {"type":"number_of_executions"}
   │ 顺带触发 on_stdout/on_result  │
   ▼                              │
return Execution(results, logs,   │
       error, execution_count)    │

部件一句话职责:

部件干什么在哪
Sandbox继承基础 e2b Sandbox,只加 code-interpreter 那几个方法code_interpreter_sync.py:34
_client一个专用 HTTP/1.1 httpx Client,发往沙箱内的 Jupyter 服务code_interpreter_sync.py:84
run_code组装请求、开流、逐行 parse、返回 Executioncode_interpreter_sync.py:172
parse_output把一行 JSON 按 type 分派进 Execution,并触发回调models.py:426
Execution / Result / Logs / ExecutionError结果数据模型models.py:342/102/297/63
Context一个执行环境的句柄(id / language / cwd)models.py:486
create/list/remove/restart code_contextcontext 的四个 REST 方法code_interpreter_sync.py:235/319/284/350

主线走一遍(高层):

  1. 你调 run_code(code, language="python")。
  2. 客户端组装一个 JSON 体,打开一个 POST /execute 的流。
  3. 沙箱内服务把代码丢给 Jupyter 内核,内核每产生一点输出,就往流里写一行 JSON。
  4. 客户端 iter_lines() 逐行读,每行交给 parse_output 累加进 Execution。
  5. 流结束,返回填满的 Execution。

它继承而非重写。 Sandbox 不重新实现文件系统、命令执行、沙箱生命周期 ——那些全在基础 e2b SDK 里(from e2b import Sandbox as BaseSandbox, code_interpreter_sync.py:6)。本库只加一层 run_code + context 管理。 所以 sandbox.files、sandbox.commands、sandbox.set_timeout 这些你都还能用, 它们来自父类。

3. 核心原理(逐个机制,由浅入深)​

3.1 run_code:组装 → 开流 → 逐行累积​

它要解决的小问题: 代码执行是慢且渐进的——一段代码可能先打印几行、 再画个图、再报个错。如果等它全跑完再一次性返回,你就拿不到实时反馈。所以要流式。

思路: 发一个流式 POST;服务端每有一点输出就写一行 NDJSON(每行一个独立 JSON 对象);客户端逐行读、逐行解析,顺带把每行喂给可选回调。

请求体只有四个字段(code_interpreter_sync.py:210-215):

字段含义
code要执行的代码
context_id在哪个 context 里跑(None = 用该语言的默认 context)
language用哪种语言(None = python)
env_vars本次执行注入的环境变量

原理演示(极简版,# 示意,非源码,把核心想法演出来):

# 示意,非源码:run_code 的骨架
with client.stream("POST", url, json=body, timeout=(...)) as response:
    if err := extract_exception(response):   # 先看 HTTP 状态,4xx/5xx 直接抛
        raise err
    execution = Execution()                  # 空壳,准备被逐行填满
    for line in response.iter_lines():       # 一行一个 JSON
        parse_output(execution, line, ...)   # 按 type 累加进 execution
    return execution                         # 填满后返回

真实实现: 见 code_interpreter_sync.py:207-250,run_code 用 self._client.stream("POST", f"{self._jupyter_url}/execute", ...),先 extract_exception(response) 检查非 2xx(models.py:404),再 new Execution(), 然后 for line in response.iter_lines(): parse_output(...)。异步版对称: async with ... .stream(...) + async for line in response.aiter_lines() (code_interpreter_async.py:212-245)。

重点看: Execution 是先造一个空对象,再在循环里被副作用式填满的—— parse_output 直接往 execution.results / execution.logs append。这就是为什么 无论流多长,返回的都是同一个逐渐长大的对象。

3.2 language vs context:两个 overload,互斥​

它要解决的小问题: 你要么说"用 python 跑"(交给系统选默认 context),要么说 "在我这个具体 context 里跑"。这两件事不能同时说——一个 context 已经绑定了语言。

怎么表达: Python 侧用两个 @overload 声明,让类型检查器和 IDE 看到两种合法签名 (code_interpreter_sync.py:103 收 language,:133 收 context),真正的实现体 两个参数都收(:167),然后在运行时手动查互斥:

# code_interpreter_sync.py:182-185
if language and context:
    raise InvalidArgumentException(
        "You can provide context or language, but not both at the same time."
    )

JS 侧同构:两个 overload 签名(sandbox.ts:160、:182)+ 实现体里 if (opts?.context && opts?.language) throw new InvalidArgumentError(...) (sandbox.ts:199-203)。

关键细节: overload 是给人和工具看的门面,互斥的真正强制在运行时那句 if language and context。光有 overload 挡不住有人把两个都传进来。

3.3 双超时:执行超时 vs 请求超时(分相超时)​

它要解决的小问题: 一次流式执行有两种"卡住":一种是连接/发请求这一步卡住 (网络问题),一种是代码本身跑太久(死循环)。这两种要能分别配、分别报错。

Python 侧的巧法——httpx 的分相超时(code_interpreter_sync.py:228-232):

timeout=(
    httpx.Timeout(timeout, connect=request_timeout)
    if timeout is not None
    else httpx.Timeout(None)
)

httpx.Timeout 的第一个位置参数管 read/write/pool 三相,connect= 单独管建连接。这里把 执行超时 timeout 设为默认相位,只让建连接用 request_timeout。含义:

相位用哪个超时管的是
connectrequest_timeout建连接要多久
read/write/pooltimeout(执行超时)等下一块流数据、写请求体、从连接池取连接

为什么不给每个相位都塞 request_timeout: SDK 底层的 pyqwest 传输会把各相位超时 坍缩成「取最长」的单一整请求 deadline(code_interpreter_sync.py:217-227 的注释)—— write/pool 位带上 request_timeout 就把下限抬到 max(timeout, request_timeout),用户传 更短的 timeout 会被静默忽略;这也是为了和 JS SDK「按 timeout 定长 abort」对齐。

为什么默认相位就等于执行超时: 流式读的时候,客户端在等服务端写下一行。 代码在跑、还没输出 = read 迟迟不返回。所以"读超时"天然就是"执行超时"。

分开报错(code_interpreter_sync.py:251-254):httpx.ReadTimeout(read 位触发) → format_execution_timeout_error();其它 httpx.TimeoutException → format_request_timeout_error()。两条错误消息分别提示改 timeout 还是 request_timeout(exceptions.py:4-13)。

timeout=0 的特例: timeout = None if timeout == 0 else (timeout or DEFAULT_TIMEOUT) (code_interpreter_sync.py:192)——传 0 显式表示"不限执行时长",否则默认 DEFAULT_TIMEOUT = 300 秒(constants.py:3)。

JS 侧没有分相超时,改用两个 setTimeout + AbortController (sandbox.ts:205-262):先起一个 reqTimer(请求超时),拿到响应就 clearTimeout(reqTimer)、改起一个 bodyTimer(执行超时,默认 DEFAULT_TIMEOUT_MS = 60_000,consts.ts:1)。两个计时器都调 controller.abort()。注意 JS 的默认执行超时是 60 秒,Python 是 300 秒——两侧数值不一致。

3.4 鉴权 header:两个 token​

它要解决的小问题: 请求要证明"我有权访问这个沙箱"。这里有两级令牌。

两个 header 都在每个方法里手动拼(code_interpreter_sync.py:197-205):

Header来自作用(inferred,基于命名)
X-Access-Tokenself._envd_access_token访问沙箱内 envd(沙箱守护进程)的令牌
E2B-Traffic-Access-Tokenself.traffic_access_token流量层的访问令牌

两个属性都来自基础 e2b SDK(本库源码里没有定义,只是读取),仅在存在时才加进 header。JS 侧同样(sandbox.ts:222-227),但注意加的顺序相反——这不影响语义。

3.5 强制 HTTP/1.1:让"断连"能真正取消长执行(本章精妙点)​

这是本章最值得带走的一个设计。

它要解决的小问题: 你开了一个流式 /execute 跑一段长代码,中途你不想要了、 断开了客户端。你希望这一断,能让沙箱那边停下正在跑的代码。

为什么会失效: 基础 SDK 的共享 transport 现在默认 http2=True。HTTP/2 会把 多个请求复用到同一条 TCP 连接上(多路复用)。于是客户端取消一个请求时,只是 取消了那一条 HTTP/2 流——底层 TCP 连接还开着。服务端察觉不到你断了, 代码继续在沙箱里空跑。

解法: 给 Jupyter 请求用一个专用的 HTTP/1.1 transport。HTTP/1.1 是 一条 TCP 连接对一个请求的 1:1 映射,客户端一断,就是一个真实的 TCP close, 服务端立刻感知,长执行随之可被可靠取消。

真实代码就是这个 _client property 和它上面那段长注释 (code_interpreter_sync.py:69-84):

@property
def _client(self) -> Client:
    # ...(注释解释 http2 多路复用为何让断连察觉不到)...
    # Forcing HTTP/1.1 here keeps the 1:1 mapping between TCP
    # connection and request, so client disconnects propagate to the
    # server as a TCP close and long-running executions can be
    # cancelled reliably.
    return Client(transport=get_transport(self.connection_config, http2=False))

关键就是 get_transport(..., http2=False)。异步版一模一样,并多一句注释说明 transport 按事件循环缓存(code_interpreter_async.py:71-90)。

JS 侧靠什么达到同样效果: fetch 用 AbortController 的 signal (sandbox.ts:239),abort 直接触发底层 socket 关闭,不涉及 http2 多路复用问题; 另加 keepalive: true(:237)。所以 JS 不需要"强制 HTTP/1.1"这一手。

一句话记住: 协议选择是取消语义的一部分——HTTP/2 的多路复用会把"我断了" 稀释成"只是一条流没了",而长执行的取消依赖一个真实的连接关闭信号。

3.6 判断沙箱是不是被杀了(_handle_connection_error)​

它要解决的小问题: 流式读到一半突然断了(ReadError / RemoteProtocolError)。 到底是网络抖了一下,还是沙箱被杀了?两种要给用户不同的错误。

思路: 断连后回头问一句"沙箱还活着吗",据此决定报什么错 (code_interpreter_sync.py:86-101):

def _handle_connection_error(self, err):
    try:
        running = self.is_running()
    except Exception:
        return                     # 连状态都查不到 → 别乱下结论,让调用方抛原始错
    if not running:
        raise format_sandbox_killed_error() from err   # 确认死了 → 明确报"沙箱被杀"

调用点在 run_code 的 except (httpx.ReadError, httpx.RemoteProtocolError) (code_interpreter_sync.py:255-257):先调 _handle_connection_error, 它要么抛"sandbox killed"、要么返回让外层 raise 原始错误。

关键细节(一个诚实的容错): 如果状态检查本身失败了,它选择不下结论 (直接 return),而不是武断地说"沙箱没了"——注释明确写了这一点。JS 侧同构: handleRequestError 里 (await this.isRunning().catch(() => true)) === false ——检查失败时 .catch(() => true) 假设"还活着",从而回退到普通请求超时处理 (sandbox.ts:449-464)。判断"连接是否被关闭"由 isConnectionClosedError 按运行时(Bun/Deno 的 code、Node undici 包在 cause 里)递归识别(utils.ts:31-50)。

4. 数据模型:一行 JSON 如何落进 Execution​

4.1 parse_output 按 type 分派​

流里每一行是一个 JSON,带一个 type 字段。_parse_output 先 data.pop("type"), 再 switch(models.py:452-483):

type落到哪触发的回调
resultexecution.results.append(Result(**data))on_result(result)
stdoutexecution.logs.stdout.append(text)on_stdout(OutputMessage(...))
stderrexecution.logs.stderr.append(text)on_stderr(OutputMessage(...))
errorexecution.error = ExecutionError(...)on_error(error)
number_of_executionsexecution.execution_count = ...(无回调)

parse_output 是同步薄封装(models.py:426);异步版 async_parse_output (models.py:437)会 inspect.isawaitable 检测回调返回值,是 awaitable 就 await ——所以回调可同步可异步。JS 的 parseOutput(messaging.ts:323-375)是同一张 分派表,回调统一 await。

一个细节: on_result 等回调的返回值被当作可能的 awaitable 传出 (_parse_output 里是 return on_result(result),models.py:467),这正是为了让 异步包装层能决定要不要 await。

4.2 Execution / Result / Logs / ExecutionError / Context​

Execution(models.py:342)——一次执行的全部产物:

  • results: List[Result] —— 结果列表(最后一行的返回值 + 所有 display() 调用,如图表)。
  • logs: Logs —— stdout / stderr 两个字符串列表(models.py:297)。
  • error: Optional[ExecutionError] —— 出错时的 name / value / traceback(models.py:63)。
  • execution_count —— 这是第几次执行(来自 number_of_executions)。
  • .text 属性 —— 遍历 results 找 is_main_result 那个的文本(models.py:373-382)。

Result(models.py:102)——一个结果的多种表示:同一个结果可以同时有 text / html / png / svg / chart / data 等等字段,谁有值取决于对象能被 IPython 显示成什么。formats() 列出所有可用表示(models.py:172)。chart 字段会 被 _deserialize_chart 反序列化成结构化图表(models.py:162-168)。这些"富结果"的 产生细节见 04-rich-results.md。

Context(models.py:486)——一个执行环境的句柄:id / language / cwd, 由 from_json 从服务端响应构造(models.py:510-516)。

序列化辅助: serialize_results(models.py:323)把 results 转成可 JSON 化的 dict 列表(chart 特殊处理成 .to_dict());Execution.to_json / Logs.to_json / ExecutionError.to_json 各自给出 JSON 表示。HTTP 状态到异常的映射由 format_exception 负责:404→NotFoundException、502→TimeoutException(带"可能是沙箱超时"提示)、 其余→SandboxException(models.py:412-423)。

4.3 JS 侧的读流:readLines​

Python 靠 httpx 的 iter_lines() 天然按行切;JS 的 fetch 只给你一个字节流, 所以本库自己写了按行切分——readLines 是个 async generator,维护一个 buffer, TextDecoder 解码每个 chunk 后,反复找 \n 切出完整行 yield 出去,流结束时把残余 buffer 也吐出(utils.ts:52-84)。这就是 JS 版对 Python iter_lines 的对称补齐。

5. Context 的四个 REST 方法​

run_code 是流式的特例;context 管理是四个普通(非流)REST 调用,结构高度重复: 拼 header → 发请求 → extract_exception 检查状态 → 解析 → 统一的 except TimeoutException / (ReadError, RemoteProtocolError) 兜底。

方法HTTP端点源码
create_code_contextPOST/contexts(body: language, cwd)code_interpreter_sync.py:259
list_code_contextsGET/contextscode_interpreter_sync.py:347
remove_code_contextDELETE/contexts/{id}code_interpreter_sync.py:310
restart_code_contextPOST/contexts/{id}/restartcode_interpreter_sync.py:380

remove / restart 都接受 Union[Context, str],内部 context.id if isinstance(...) 统一成 id(code_interpreter_sync.py:321、:361)。JS 侧四个方法对称 (sandbox.ts:293/366/332/403),用 connectionConfig.getSignal(requestTimeoutMs) 做超时,没有流,所以也不涉及 HTTP/1.1 那套。

注意 create_code_context 的 docstring 说 request_timeout 单位是"毫秒", 但代码把它直接传给 httpx(单位秒)——docstring 与 run_code(明确写"seconds") 不一致(code_interpreter_sync.py:270 vs :127)。以秒为准更安全。

6. Python 与 JS 的对称一览​

两侧是同一套设计的两个实现,差异集中在语言原生的 HTTP 与流机制上:

关切PythonJS
HTTP 客户端httpx Client / AsyncClient,强制 http2=False原生 fetch + AbortController
取消长执行靠 HTTP/1.1 让断连 = TCP close靠 signal abort 关 socket
按行读流iter_lines() / aiter_lines()自写 readLines generator(utils.ts:52)
双超时httpx.Timeout 分相(默认相位=执行超时,connect= 请求超时)两个 setTimeout(reqTimer / bodyTimer)
默认执行超时300s(constants.py:3)60s(consts.ts:1)
互斥/沙箱死检测InvalidArgumentException / _handle_connection_errorInvalidArgumentError / handleRequestError

7. 边界与局限​

  • JS 与 Python 默认执行超时不同(60s vs 300s)——跨语言迁移代码时会踩到。
  • 强制 HTTP/1.1 只针对 Jupyter 请求;基础 SDK 其它请求仍可能走 http2。取消语义 的保证只覆盖 _client 发出的这些调用。
  • 沙箱死检测是"事后询问":断连后再问 is_running(),若这一步也失败就放弃判断、 抛原始错误——即代码选择宁可不报"沙箱被杀"也不误报。
  • create_code_context 的 request_timeout docstring 单位("毫秒")与实现(秒)不符。
  • 本章只到客户端边界。代码真正怎么在沙箱里跑、Jupyter 消息协议、图表抽取,分别见 02-server-routing.md、03-kernel-messaging.md、 04-rich-results.md;这台机器怎么造出来见 05-template-runtime.md。

8. 代码地图(导航索引)​

主题文件符号
Sandbox 继承基础 SDKpython/e2b_code_interpreter/code_interpreter_sync.pySandbox(BaseSandbox)
run_code 组装+开流+累积python/e2b_code_interpreter/code_interpreter_sync.pyrun_code
强制 HTTP/1.1 的客户端python/e2b_code_interpreter/code_interpreter_sync.py_client, get_transport(..., http2=False)
沙箱被杀检测python/e2b_code_interpreter/code_interpreter_sync.py_handle_connection_error
context 四方法python/e2b_code_interpreter/code_interpreter_sync.pycreate_code_context, list_code_contexts, remove_code_context, restart_code_context
异步对称实现python/e2b_code_interpreter/code_interpreter_async.pyrun_code(aiter_lines), _client(AsyncClient)
按 type 分派python/e2b_code_interpreter/models.pyparse_output, _parse_output, async_parse_output
结果数据模型python/e2b_code_interpreter/models.pyExecution, Result, Logs, ExecutionError, Context
HTTP 错误映射 / 序列化python/e2b_code_interpreter/models.pyformat_exception, extract_exception, serialize_results
常量python/e2b_code_interpreter/constants.pyDEFAULT_TEMPLATE, JUPYTER_PORT, DEFAULT_TIMEOUT
超时/沙箱死错误消息python/e2b_code_interpreter/exceptions.pyformat_execution_timeout_error, format_request_timeout_error, format_sandbox_killed_error
JS 客户端js/src/sandbox.tsSandbox, runCode, handleRequestError
JS 分派 + 数据模型js/src/messaging.tsparseOutput, Execution, Result, ExecutionError
JS 按行读流 / 断连识别js/src/utils.tsreadLines, isConnectionClosedError, formatExecutionTimeoutError
JS 常量js/src/consts.tsDEFAULT_TIMEOUT_MS, JUPYTER_PORT