数据截至 2026-08-17(上游 commit 771b20c93db8)

沙箱内服务:FastAPI 路由与上下文管理

30 秒导读: 每个 E2B 沙箱里跑着一台小小的 FastAPI 服务。它是「门面」——把客户端发来的 POST /execute HTTP 请求,翻译成对沙箱内 Jupyter 内核的操作,再把内核吐出的执行结果,一行一个 JSON 地流式回传。本章讲这台服务怎么开机、怎么按语言/上下文选路、怎么管理「上下文」(其实就是一个个 Jupyter 内核会话)。

本章是 SDK 客户端的「服务端对面」:客户端发出的那条流式 HTTP,落到这里被接住。至于接住之后内核层面到底怎么收发 Jupyter 消息、怎么把消息拼成结果,是最深的一层,放在内核桥讲;本章只到「调用 ws.execute(...)」这条边界为止。

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

一句话定义: 沙箱内的一台 FastAPI HTTP 服务,充当「客户端 HTTP」与「沙箱里的 Jupyter 内核」之间的翻译器 + 调度台。
它解决什么问题: 客户端(你机器上的 SDK)不能直接跟沙箱里的 Jupyter 内核说话——那是 WebSocket、是 Jupyter 私有消息协议。于是沙箱里放一台 HTTP 服务:客户端用普通 HTTP 说「跑这段代码」,由它去跟内核交涉,再把结果翻译回 HTTP 流。

它对外提供什么(路由):

路由	干什么
`POST /execute`	跑一段代码,流式返回结果
`POST /contexts`	新建一个上下文(= 一个独立内核)
`GET /contexts`	列出当前所有上下文
`POST /contexts/{id}/restart`	重启某个上下文的内核
`DELETE /contexts/{id}`	删掉某个上下文
`GET /health`	健康检查

一句话直觉/类比: 把它想成餐厅前台。客人(客户端)在前台点单(HTTP),前台不亲自炒菜,而是把单子递给后厨的某个灶台(Jupyter 内核),再把菜一道道端出来(流式结果)。「上下文」就是一个个灶台,可以开新灶、关灶、重开灶。

本节不出现代码。你只要记住:这是沙箱里接 HTTP、指挥内核、回吐结果的那台服务。

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

2.1 一张图:一次 `/execute` 的旅程

怎么读这张图:从左到右是数据流向;中间那台 FastAPI 服务是主角,它下面那张 websockets 表是它的「灶台名册」。

  客户端                沙箱内 FastAPI 服务 (main.py)               沙箱内 Jupyter
 ─────────           ────────────────────────────────           ──────────────
                     ┌──────────────────────────────┐
  POST /execute ───► │ post_execute                 │
  {code, language     │  1. 归一化语言 (js→javascript)│
   或 context_id}      │  2. 选路:找到对应的内核 WS    │
                     │         │                    │
                     │         ▼                    │
                     │   websockets 表(名册)        │
                     │   ┌────────────────────────┐ │      WebSocket
                     │   │ "default"   → py 内核   │─┼───────────────► [python 内核]
                     │   │ <ctx_id_py> → py 内核   │ │
                     │   │ <ctx_id_js> → js 内核   │─┼───────────────► [javascript 内核]
                     │   └────────────────────────┘ │
                     │         │                    │
                     │  3. ws.execute(code) ────────┼──► (见 03 章:内核桥)
                     │         │ 产出 pydantic 结果流 │
                     │         ▼                    │
  NDJSON 流 ◄────────┤ StreamingListJsonResponse    │
  一行一个 JSON        │  (stream.py:编码器)          │
                     └──────────────────────────────┘

2.2 部件一句话职责

部件	干什么	文件
`lifespan`	服务开机时预建 python/javascript 默认内核	`template/server/main.py:32`
各 `@app` 路由	接 HTTP,选路,调内核,回结果	`template/server/main.py:74-217`
`websockets` 表	「灶台名册」:context_id → 内核 WebSocket	`template/server/main.py:27`
`default_websockets`	「语言 → 默认 context_id」的映射,带按语言加锁	`template/server/main.py:28`
`create_context`	建 Jupyter 会话 + 连内核 WebSocket + 设 cwd	`template/server/contexts.py:37`
`normalize_language`	用户输入的语言别名归一(js→javascript)	`template/server/contexts.py:22`
`StreamingListJsonResponse`	把 pydantic 结果生成器编码成 NDJSON 流	`template/server/stream.py:9`
`LockedMap`	按 key 发锁,防并发重复建默认 context	`template/server/utils/locks.py:4`
`ContextWebSocket`	一个内核的连接封装(本章的边界,内部见 03 章)	`template/server/messaging.py:60`

2.3 核心概念:什么是「上下文(context)」

一个 context = 一个独立的 Jupyter 内核会话,有自己独立的变量空间。你在同一个 context 里跑 x = 1,下次跑 print(x) 还在;换一个 context 就互不相干。

在这台服务里,一个 context 对应:

一个 Jupyter session(session_id)
一个 Jupyter kernel(context_id 直接就是 kernel id,见 contexts.py:55)
一个 ContextWebSocket 对象(连接该内核的 WebSocket 封装)

服务用一张全局字典 websockets 把 context_id 映射到 ContextWebSocket——这是全章的中枢数据结构。

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

3.1 开机:`lifespan` 预建两个默认内核

它要解决的小问题: 第一次执行代码不该现场等着建内核(慢)。所以服务一开机就先把 python 和 javascript 的默认内核建好、连好、备着。

思路: FastAPI 的 lifespan 是「服务启动/关闭」的钩子。启动段里建默认 context,关闭段里挨个关掉所有 WebSocket。

真实实现: template/server/main.py:32-59,lifespan。

python_context = await create_context(
    client, websockets, "python", "/home/user"
)
default_websockets["python"] = python_context.id      # 语言→默认 ctx
websockets["default"] = websockets[python_context.id]  # 特殊别名 "default"

javascript_context = await create_context(
    client, websockets, "javascript", "/home/user"
)
default_websockets["javascript"] = javascript_context.id

关键细节 —— 两张表,别混淆:

表	键	值	作用
`websockets`	context_id(还有特殊键 `"default"`)	`ContextWebSocket`	真正拿来收发的连接名册
`default_websockets`	语言名(如 `"python"`)	context_id	「每种语言的默认 context 是哪个」

注意 main.py:42 那句 websockets["default"] = websockets[python_context.id]——它给 python 默认内核额外挂了个 "default" 别名。这样当请求既没给 language 也没给 context_id 时,可以直接取 websockets["default"](见 3.3)。python 因此享有这个「兜底默认」的特殊地位,javascript 没有。

启动段最后 yield(main.py:50);yield 之后的代码(main.py:53-56)是关闭时才跑的——挨个 await ws.close(),再关 httpx 客户端。

3.2 语言归一:`normalize_language` 与 `get_kernel_for_language`

它要解决的小问题: 用户可能写 js、ts、JavaScript (带空格大小写乱)。得先把这些归一成规范名,才能查表选路。

这里有两级归一,别搞混:

用户输入                normalize_language           get_kernel_for_language
"js" ─────────────────► "javascript" ──────────────► "javascript"
"ts" ─────────────────► "typescript" ──────────────► "javascript"  ← ts 复用 js 内核
"Python " ────────────► "python"     ──────────────► "python"
None ─────────────────► "python"(默认)

normalize_language(contexts.py:22-34):面向用户 API。小写去空格,js→javascript、ts→typescript,空值默认 python。归一后的名字会存进 context、返回给客户端。
get_kernel_for_language(contexts.py:15-19):面向内核。只做一件事:typescript → javascript。因为 TS 和 JS 共用同一个 Jupyter 内核(Deno/tslab 那套),但对外仍保留 typescript 这个语言标识。

一句话:normalize_language 决定「对外叫什么」,get_kernel_for_language 决定「实际启哪个内核」。

3.3 选路:`post_execute` 如何在 language 与 context_id 之间挑内核

它要解决的小问题: /execute 请求可以三种形态出现,得挑出正确的那个内核 WebSocket。

三种形态 + 互斥校验: 先看 main.py:83-87——context_id 和 language 不能同时给,否则 400。逻辑上分三路:

请求给了什么	走哪条路	结果
给了 `language`	懒创建/复用该语言的默认 context	用 `default_websockets[language]`
给了 `context_id`	直接用这个 context	用 `websockets[context_id]`
都没给	兜底	用 `websockets["default"]`(即 python)

language 路的精髓 —— 懒创建 + 按语言加锁。 看 main.py:90-105:

if exec_request.language:
    language = normalize_language(exec_request.language)

    async with await default_websockets.get_lock(language):   # 按语言拿锁
        context_id = default_websockets.get(language)
        if not context_id:                                    # 该语言还没默认 ctx
            context = await create_context(client, websockets, language, "/home/user")
            context_id = context.id
            default_websockets[language] = context_id         # 记下来,下次直接用

为什么要锁?假设两个请求几乎同时到,都要 r 语言、而 r 的默认 context 还没建。没锁的话两个请求都发现 context_id 为空,于是建了两个内核——浪费且状态分裂。锁保证「检查是否存在 + 建」这一步串行:第一个建完写进表,第二个拿到锁时已经能查到,直接复用。

选中 WebSocket 后: main.py:110-119 把 context_id 换成 ws;查不到就 404。最后 main.py:121-127 调 ws.execute(...),把返回的结果生成器交给 StreamingListJsonResponse。注意它顺手把请求头里的 X-Access-Token 透传给 execute(main.py:125)。

3.4 按语言加锁的机制:`LockedMap`

它要解决的小问题: 3.3 里那个 get_lock(language) 怎么做到「同一个 key 拿到同一把锁、不同 key 互不阻塞」?

思路: 一个继承自 dict 的类,内部再存一张 key → asyncio.Lock 的表。要锁某个 key 时,先用一把「元锁」保护「取/建这个 key 的锁」这一步,再把该 key 的锁返回给调用方去 async with。

真实实现: template/server/utils/locks.py:4-18,LockedMap.get_lock。

async def get_lock(self, key):
    await self._map_lock.acquire()          # 元锁:保护下面建锁这步
    if key not in self._locks:
        self._locks[key] = asyncio.Lock()   # 该 key 第一次出现,建一把锁
    lock = self._locks[key]
    self._map_lock.release()
    return lock                             # 返回该 key 专属的锁

这就是「细粒度锁」:python 的锁和 javascript 的锁是两把,建 python 默认 context 时不会挡住 javascript 的请求。元锁 _map_lock 只在「找/建这把 key 锁」的极短瞬间持有。

3.5 建上下文:`create_context` 的四步

它要解决的小问题: 「建一个 context」不是一个动作,而是一串:向 Jupyter 要一个会话 → 拿到内核 id → 连上它的 WebSocket → 把工作目录设成 cwd。

四步走(template/server/contexts.py:37-72):

① POST /api/sessions ──► Jupyter 建 session+kernel,回 {id, kernel:{id}}
        │                        contexts.py:46
        ▼
② 拆出 session_id 与 context_id(=kernel id)
        │                        contexts.py:54-55
        ▼
③ new ContextWebSocket(...) → await ws.connect()   连上内核 WS
        │                        contexts.py:59-60
        │   并登记:websockets[context_id] = ws
        ▼
④ await ws.change_current_directory(cwd, language)  在内核里 cd 到 cwd
                                 contexts.py:65

第①步细节: 发给 Jupyter 的请求体(contexts.py:38-43)里,kernel.name 用的是 get_kernel_for_language(language)——所以 typescript 在这里落成 javascript 内核。path/name 用随机 UUID 占位。

第④步细节: context 建好后不能就完事,还要让内核的工作目录落在用户指定的 cwd(默认 /home/user)。change_current_directory 是往内核里真的跑一段 cd 代码(具体每种语言跑什么,属于 03 章内核桥)。若这步抛 ExecutionError,返回 500「Failed to set working directory」(contexts.py:66-70)。

返回: 一个 Context pydantic 模型 {id, language, cwd}(contexts.py:72),其中 language 是归一后的对外名(如 typescript),而非内核名。

3.6 上下文的增删查与重启

这几个路由都围绕同一张 websockets 表做「增删查」,并且都要在操作内核(调 Jupyter REST)和同步本地表之间保持一致。

新建 POST /contexts(main.py:130-140): 归一语言、取 cwd(默认 /home/user),调 create_context。它复用的正是 3.5 那套。

列出 GET /contexts(main.py:143-155): 遍历 websockets,对每个 ws 拼一个 Context。注意过滤掉 "default" 这个键(main.py:154 的 if key != "default")——因为 "default" 只是 python 内核的别名,不是一个独立 context,列出来会重复。

重启 POST /contexts/{id}/restart(main.py:158-191):

查表拿 ws ──► 记下 session_id ──► ws.close() 关旧连接
   │                                   │
   │        POST /api/kernels/{id}/restart   (Jupyter 重启内核)
   │                                   │  main.py:173-175
   ▼                                   ▼
 失败→404                       新建 ContextWebSocket(同 id/session/lang/cwd)
                                       │  main.py:182-187
                                       ▼
                                 await ws.connect()  重新连上
                                 websockets[id] = ws  覆盖回表

要点:重启用的是同一个 context_id/session_id,只是内核被 Jupyter 重置了变量空间,连接对象换成新的。Jupyter 侧调 /api/kernels/{id}/restart;restart 失败返回 500(main.py:176-180)。

删除 DELETE /contexts/{id}(main.py:194-217): 先 ws.close()(裹在 try/except: pass 里,main.py:205-208——关连接失败不该挡住删除),再调 Jupyter DELETE /api/kernels/{id} 真删内核(main.py:210),最后从本地表 del websockets[context_id](main.py:217)。删内核失败返回 500。

一张表看清「本地表 ↔ Jupyter REST」的对应:

路由	本地 `websockets` 表动作	打给 Jupyter 的调用
`POST /contexts`	加一项(在 `create_context` 内)	`POST /api/sessions`
`GET /contexts`	只读遍历(排除 `"default"`)	无
`POST /{id}/restart`	换成新 `ContextWebSocket`	`POST /api/kernels/{id}/restart`
`DELETE /{id}`	`del` 该项	`DELETE /api/kernels/{id}`

3.7 回吐结果:`StreamingListJsonResponse` 与 NDJSON

它要解决的小问题: ws.execute(...) 产出的是一个异步生成器,一路吐出 pydantic 模型(stdout、result、error……,类型见下)。怎么把它变成 HTTP 能流式传的字节流,让客户端边收边解析?

思路: 不等全部结果攒齐再一次性返回,而是每产出一个,就编码成一行 JSON 立刻发出——即 NDJSON(Newline-Delimited JSON,每行一个独立 JSON 对象)。最后补一行「执行结束」标记,让客户端知道到头了。

真实实现: template/server/stream.py:34-40,_encoded_async_generator。

async def _encoded_async_generator(self, async_generator):
    async for item in async_generator:
        yield f"{json.dumps(jsonable_encoder(item))}\n"   # 每个 pydantic 模型 → 一行 JSON
    yield '{"type": "end_of_execution"}\n'                # 收尾标记

jsonable_encoder(FastAPI 提供)把 pydantic 模型转成可 JSON 化的 dict,json.dumps 序列化,末尾 \n 分行。这个生成器被喂给 Starlette 的 StreamingResponse(stream.py:26-32),于是就成了一条流。

流里每行长什么样(type 字段来自 api/models/output.py:7 的 OutputType):

`type`	含义	模型来源
`stdout` / `stderr`	标准输出/错误的一段文字	`api/models/logs.py`
`result`	一个富结果(文本/图/表…,见 04 章)	`api/models/result.py:13`
`error`	执行报错(name/value/traceback)	`api/models/error.py`
`number_of_executions`	内核执行计数	`api/models/output.py:29`
`end_of_execution`	由本编码器补的收尾行	`stream.py:40`

客户端(01 章)正是靠逐行读 + 按 type 分派来重建结果的;end_of_execution 这行是它停止读取的信号。

3.8 CORS 全开

main.py:64-69 给应用挂了 CORSMiddleware,allow_origins/allow_methods/allow_headers 全是 ["*"]。因为这台服务只跑在沙箱内部、由 E2B 网关暴露,不直接面向公网多租户,所以放开跨域简化了 SDK/浏览器直连。

4. 巧妙之处(可借鉴的技术)

「语言 → 默认 context」懒创建 + 细粒度锁。 不预建所有语言,只预建 py/js;其余语言首次用到时才建,且用 LockedMap 按语言加锁避免并发重复建(main.py:93 + locks.py:4)。省资源又并发安全。
一张表两种键。 websockets 既用 context_id 当键,又给 python 内核塞了个 "default" 别名键(main.py:42),让「不指定语言」的请求零成本兜底;代价只是 GET /contexts 时记得过滤掉这个别名(main.py:154)。
对外名 vs 内核名分离。 typescript 对用户是一等语言,对内核却复用 javascript(get_kernel_for_language,contexts.py:15)。命名分离让「共用运行时」这个实现细节不泄露到 API。
流式而非缓冲。 结果一产出就 NDJSON 逐行吐(stream.py:38),配合 end_of_execution 收尾,客户端可以边跑边看输出,长任务体验好。

5. 边界与局限(诚实)

状态全在内存。 websockets/default_websockets 是进程内全局字典(main.py:27-28)。服务重启即全丢——这在沙箱语境里可接受(沙箱本身就是短命的),但不是持久化设计。
强绑定本地 Jupyter。 所有内核操作都打到 http://localhost:8888(consts.py:1,JUPYTER_BASE_URL),假设同机跑着 Jupyter。搬离沙箱环境不成立。
create_context 的错误返回类型不一致(inferred)。 正常返回 Context,但 cwd 设置失败时返回的是 PlainTextResponse(contexts.py:67-70)——而 lifespan 里直接拿返回值当 Context 用(main.py:41)。启动阶段若 cwd 设置失败,行为可能不如预期。这是从类型不一致推断,非代码显式声明。
CORS 全开 只在「沙箱内部服务」这一前提下安全(见 3.8);别把这份代码当成公网服务模板照抄。

6. 横向对比 / 与邻章的关系

上游 01-sdk-client.md:客户端如何把一次 run_code 变成这条流式 HTTP 请求、又如何逐行解析本章吐出的 NDJSON。
下游 03-kernel-messaging.md:本章止于 ws.execute(...) 和 ws.change_current_directory(...);ContextWebSocket 内部怎么按 Jupyter 消息协议收发、怎么把消息拼成本章 3.7 表里那些 type,是全项目最深的一章。
结果模型 04-rich-results.md:3.7 里 result 那行的富内容(DataFrame、图表、多种 MIME)怎么从 IPython 抽出来。

7. 代码地图(导航索引)

主题	文件路径	符号
服务开机/关机、预建默认内核	`template/server/main.py`	`lifespan`
全局连接名册 / 语言默认表	`template/server/main.py`	`websockets`、`default_websockets`
执行路由 + 选路 + 懒创建	`template/server/main.py`	`post_execute`
新建上下文路由	`template/server/main.py`	`post_contexts`
列出上下文(过滤 default)	`template/server/main.py`	`get_contexts`
重启上下文	`template/server/main.py`	`restart_context`
删除上下文	`template/server/main.py`	`remove_context`
CORS 中间件	`template/server/main.py`	`CORSMiddleware`(64-69 行)
建 session→连 WS→设 cwd	`template/server/contexts.py`	`create_context`
用户语言别名归一	`template/server/contexts.py`	`normalize_language`
内核名映射(ts→js)	`template/server/contexts.py`	`get_kernel_for_language`
NDJSON 流式响应	`template/server/stream.py`	`StreamingListJsonResponse`、`_encoded_async_generator`
按 key 细粒度加锁	`template/server/utils/locks.py`	`LockedMap`、`get_lock`
Jupyter 基址常量	`template/server/consts.py`	`JUPYTER_BASE_URL`
请求/响应模型	`template/server/api/models/`	`ExecutionRequest`、`CreateContext`、`Context`、`Result`、`OutputType`、`EnvVars`
内核连接封装(边界,详见 03)	`template/server/messaging.py`	`ContextWebSocket`(`execute`、`connect`、`close`、`change_current_directory`)

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

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

2.1 一张图:一次 /execute 的旅程​

2.2 部件一句话职责​

2.3 核心概念:什么是「上下文(context)」​

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

3.1 开机:lifespan 预建两个默认内核​

3.2 语言归一:normalize_language 与 get_kernel_for_language​

3.3 选路:post_execute 如何在 language 与 context_id 之间挑内核​

3.4 按语言加锁的机制:LockedMap​

3.5 建上下文:create_context 的四步​

3.6 上下文的增删查与重启​

3.7 回吐结果:StreamingListJsonResponse 与 NDJSON​

3.8 CORS 全开​

4. 巧妙之处(可借鉴的技术)​

5. 边界与局限(诚实)​

6. 横向对比 / 与邻章的关系​

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