Monty
一个用 Rust 编写的极简、安全的 Python 解释器,供 AI 使用。
实验性 - 本项目仍处于开发阶段,尚未达到生产就绪状态。
一个用 Rust 编写的极简、安全的 Python 解释器,供 AI 使用。
Monty 避免了使用基于容器的完整沙箱来运行 LLM 生成代码所带来的成本、延迟、复杂性以及一般性的繁琐操作。
相反,它允许你在智能体中安全地运行由 LLM 编写的 Python 代码,启动时间以个位数微秒计,而非数百毫秒。
Monty 能够做到的:
- 运行合理的 Python 代码子集 - 足以让你的 agent 表达它想要做什么
- 完全阻止访问宿主环境:文件系统、环境变量和网络访问均通过开发者可控制的外部函数调用实现
- 调用宿主上的函数 - 仅限你授予其访问权限的函数
- 运行类型检查 - monty 支持完整的现代 Python 类型提示,并附带 ty 包含在单个二进制文件中以运行类型检查
- 在外部函数调用时快照为字节,这意味着你可以将解释器状态存储在文件或数据库中,并稍后恢复
- 启动极快(从代码到执行结果 <1μs),且运行时性能与 CPython 相似(通常介于快 5 倍到慢 5 倍之间)
- 可从 Rust、Python 或 Javascript 调用 - 因为 Monty 不依赖 cpython,你可以在任何能运行 Rust 的地方使用它
- 控制资源使用 - Monty 可以跟踪内存使用、栈深度和执行时间,并在超过预设限制时取消执行
- 收集 stdout 和 stderr 并将其返回给调用者
- 通过宿主上的异步或同步代码在宿主上运行异步或同步代码
- 使用标准库的一个小子集:
sys,os,typing,asyncio,re,datetime,json,dataclasses(即将推出)
Monty 不能 做的事情:
- 使用标准库的其余部分
- 使用第三方库(如 Pydantic),对外部 python 库的支持并非目标
- 定义类(支持即将推出)
- 使用 match 语句(同样,支持即将推出)
简而言之,Monty 功能极其有限,专为一个用例而设计:
运行由智能体编写的代码。
关于为何可能需要这样做的动机,请参阅:
- Cloudflare 的 Codemode
- Anthropic 的 Programmatic Tool Calling
- Anthropic 的 Code Execution with MCP
- Hugging Face 的 Smol Agents
用最简单的话来说,上述所有方案的理念是:如果要求 LLM 编写 Python(或 Javascript)代码,而不是依赖传统的工具调用,它们可以工作得更快、更便宜且更可靠。Monty 使得这成为可能,而无需沙箱的复杂性,也无需直接在主机上运行代码的风险。
注意: Monty 将(很快)用于在 Pydantic AI 中实现 codemode
用法
Monty 可以从 Python、JavaScript/TypeScript 或 Rust 中调用。
Python
安装方法:
uv add pydantic-monty
(或者 pip install pydantic-monty 给婴儿潮一代)
用法:
from typing import Any
import pydantic_monty
code = """
async def agent(prompt: str, messages: Messages):
while True:
print(f'messages so far: {messages}')
output = await call_llm(prompt, messages)
if isinstance(output, str):
return output
messages.extend(output)
await agent(prompt, [])
"""
type_definitions = """
from typing import Any
Messages = list[dict[str, Any]]
async def call_llm(prompt: str, messages: Messages) -> str | Messages:
raise NotImplementedError()
prompt: str = ''
"""
Messages = list[dict[str, Any]]
async def call_llm(prompt: str, messages: Messages) -> str | Messages:
if len(messages) < 2:
return [{'role': 'system', 'content': 'example response'}]
else:
return f'example output, message count {len(messages)}'
async def main():
async with pydantic_monty.AsyncMonty() as pool:
async with pool.checkout(
script_name='agent.py',
type_check=True,
type_check_stubs=type_definitions,
) as session:
output = await session.feed_run(
code,
inputs={'prompt': 'testing'},
external_lookup={'call_llm': call_llm},
)
print(output)
#> example output, message count 2
if __name__ == '__main__':
import asyncio
asyncio.run(main())
执行发生在 monty 个工作子进程的池中,因此即使由对抗性代码触发的内存
错误(栈溢出、分配器中止)也永远不会导致你的进程崩溃 —— 工作进程会终止,抛出 MontyCrashedError,并
被替换。此外,还有一个完全同步的 API:
import pydantic_monty
with pydantic_monty.Monty() as pool:
with pool.checkout() as session:
# session state persists between feed_run calls
session.feed_run('x = 21')
print(session.feed_run('x * 2'))
#> 42
JavaScript / TypeScript
安装方法:
npm install @pydantic/monty
JS 包是对 Python 包所使用的同一 Rust worker 池的原生 (napi) 绑定——该绑定和 monty worker 二进制文件通过平台特定的 npm 包分发:
import { Monty } from '@pydantic/monty'
await using pool = await Monty.create()
await using session = await pool.checkout()
// session state persists between feedRun calls
await session.feedRun('x = 21')
console.log(await session.feedRun('x * 2')) // 42
// external functions may be async
const result = await session.feedRun('await fetch_data()', {
externalLookup: { fetch_data: async () => 'data' },
})
对于浏览器(或任何无法使用子进程的环境),相同的包
在 @pydantic/monty/wasm
子路径下暴露了一个进程内 WebAssembly 构建(无崩溃隔离:沙箱崩溃即为主机崩溃)。
Rust
对于从 Rust 运行不受信任的代码,我们推荐
使用 monty-pool crate,而不是下面的进程内 API。
monty-pool 仅在 monty worker 子进程中运行代码,这提供了额外的保护:
由对抗性代码触发的崩溃(栈溢出、分配器中止)只会终止 worker ——
池会检测到死亡并替换 worker —— 并且父进程端的看门狗可以终止
超过硬性超时的 worker。它是上述 Python 和 JavaScript 包
所基于的同一引擎。请参阅 monty-pool README
以了解用法。
monty crate 本身提供了进程内解释器:
use monty::MontyRun;
use monty_types::{CompileOptions, ResourceTracker, MontyObject, PrintWriter, ResourceLimits};
let code = r#"
def fib(n):
if n <= 1:
return n
return fib(n - 1) + fib(n - 2)
fib(x)
"#;
let runner = MontyRun::new(code.to_owned(), "fib.py", vec!["x".to_owned()], CompileOptions::default()).unwrap();
let result = runner.run(vec![MontyObject::Int(10)], ResourceTracker::default(), PrintWriter::Stdout).unwrap();
assert_eq!(result, MontyObject::Int(55));
序列化
MontyRun 和 RunProgress 可以使用 dump() 和 load() 方法进行序列化:
use monty::MontyRun;
use monty_types::{CompileOptions, ResourceTracker, MontyObject, PrintWriter, ResourceLimits};
// Serialize parsed code
let runner = MontyRun::new("x + 1".to_owned(), "main.py", vec!["x".to_owned()], CompileOptions::default()).unwrap();
let bytes = runner.dump().unwrap();
// Later, restore and run
let runner2 = MontyRun::load(&bytes).unwrap();
let result = runner2.run(vec![MontyObject::Int(41)], ResourceTracker::default(), PrintWriter::Stdout).unwrap();
assert_eq!(result, MontyObject::Int(42));
工作进程中的内存限制
会话的 max_memory 由工作进程的分配器进行测量。当超过软限制后,解释器会报告一次优雅的 MemoryError;如果某次分配在检查点之间跳跃过大,则更高的硬限制会终止并替换该工作进程。
请参阅 limitations/resource_limits.md 了解超出限制如何呈现给宿主,以及 monty-alloc 了解子进程和 WebAssembly 工作进程所运行的分配器。
PydanticAI 集成
Monty 将驱动 Pydantic AI 中的代码模式。LLM 不再进行 顺序的工具调用,而是编写调用你的工具作为函数的 Python 代码,并由 Monty 安全地执行它。
import asyncio
import json
import logfire
from httpx import AsyncClient
from pydantic_ai import Agent, RunContext
from pydantic_ai.toolsets.code_mode import CodeModeToolset
from pydantic_ai.toolsets.function import FunctionToolset
from typing_extensions import TypedDict
logfire.configure()
logfire.instrument_pydantic_ai()
class LatLng(TypedDict):
lat: float
lng: float
weather_toolset: FunctionToolset[AsyncClient] = FunctionToolset()
@weather_toolset.tool
async def get_lat_lng(
ctx: RunContext[AsyncClient], location_description: str
) -> LatLng:
"""Get the latitude and longitude of a location."""
# NOTE: the response here will be random, and is not related to the location description.
r = await ctx.deps.get(
'https://demo-endpoints.pydantic.workers.dev/latlng',
params={'location': location_description},
)
r.raise_for_status()
return json.loads(r.content)
@weather_toolset.tool
async def get_temp(ctx: RunContext[AsyncClient], lat: float, lng: float) -> float:
"""Get the temp at a location."""
# NOTE: the responses here will be random, and are not related to the lat and lng.
r = await ctx.deps.get(
'https://demo-endpoints.pydantic.workers.dev/number',
params={'min': 10, 'max': 30},
)
r.raise_for_status()
return float(r.text)
@weather_toolset.tool
async def get_weather_description(
ctx: RunContext[AsyncClient], lat: float, lng: float
) -> str:
"""Get the weather description at a location."""
# NOTE: the responses here will be random, and are not related to the lat and lng.
r = await ctx.deps.get(
'https://demo-endpoints.pydantic.workers.dev/weather',
params={'lat': lat, 'lng': lng},
)
r.raise_for_status()
return r.text
agent = Agent(
'gateway/anthropic:claude-sonnet-4-5',
# toolsets=[weather_toolset],
toolsets=[CodeModeToolset(weather_toolset)],
deps_type=AsyncClient,
)
async def main():
async with AsyncClient() as client:
await agent.run('Compare the weather of London, Paris, and Tokyo.', deps=client)
if __name__ == '__main__':
asyncio.run(main())
社区绑定
替代方案
当人们看到 Monty 时,通常会有两种反应:
- 天哪,这解决了很多问题,我想要它。
- 为什么不用 X?
其中 X 是某种替代技术。奇怪的是,这些反应经常结合在一起,表明人们尚未找到适合他们的替代方案,但对于从头创建一个完整的 Python 实现竟然没有好的替代方案这一事实感到难以置信。
我将尝试梳理最明显的替代方案,以及为什么它们不适合我们的需求。
注意:所有这些技术都令人印象深刻且应用广泛,关于它们在特定用例中局限性的评论不应被视为批评。大多数这些解决方案在设计之初并未以提供 LLM 沙箱为目标,因此它们在这方面并不一定表现优异。
| 技术 | 语言完整性 | 安全性 | 启动延迟 | 开源 | 配置复杂度 | 文件挂载 | 快照 |
|---|---|---|---|---|---|---|---|
| Monty | 部分 | 严格 | 0.06ms | 免费 / 开源 | 简单 | 简单 | 简单 |
| Docker | 完整 | 良好 | 195ms | 免费 / 开源 | 中等 | 简单 | 中等 |
| Pyodide | 完整 | 差 | 2800ms | 免费 / 开源 | 中等 | 简单 | 困难 |
| starlark-rust | 非常有限 | 良好 | 1.7ms | 免费 / 开源 | 简单 | 不可用? | 不可能? |
| WASI / Wasmer | 部分,接近完整 | 严格 | 66ms | 免费 * | 中等 | 简单 | 中等 |
| 沙箱服务 | 完整 | 严格 | 1033ms | 非免费 | 中等 | 困难 | 中等 |
| YOLO Python | 完整 | 不存在 | 0.1ms / 30ms | 免费 / 开源 | 简单 | 简单 / 危险 | 困难 |
参见 ./scripts/startup_performance.py 用于计算启动性能数据的脚本。
以下是每行的详细信息:
Monty
- 语言完整性:尚无类(yet),标准库有限,无第三方库
- 安全性:显式控制文件系统、网络和环境访问,对执行时间和内存使用有严格限制
- 启动延迟:微秒级启动
- 配置复杂度:仅需
pip install pydantic-monty或npm install @pydantic/monty,下载大小约 4.5MB - 文件挂载:严格受控,参见 #85
- 快照:Monty 的暂停和恢复功能配合
dump()和load()使得暂停、恢复和分叉执行变得轻而易举
Docker
- 语言完整性:完整的 CPython,支持任意库
- 安全性:进程和文件系统隔离,网络策略,但存在容器逃逸风险,可实现内存限制
- 启动延迟:容器启动开销(实测约 195ms)
- 配置复杂度:需要 Docker 守护进程、容器镜像、编排,
python:3.14-alpine为 50MB - docker 无法从 PyPI 安装 - 文件挂载:卷挂载工作良好
- 快照:可通过 Temporal 等持久化执行方案实现,或通过对镜像进行快照并将其保存为 Docker 镜像来实现。
Pyodide
- 语言完整性:完整的 CPython 编译为 WASM,几乎所有库均可用
- 安全性:依赖浏览器/WASM 沙箱 - 并非为服务端隔离而设计,Python 代码可以在 JS 运行时中执行任意代码,只有 deno 允许隔离,而在 deno 中内存限制难以/无法强制执行
- 启动延迟:WASM 运行时加载缓慢(冷启动约 2800ms)
- 配置复杂度:需要加载 WASM 运行时,处理异步初始化,pyodide NPM 包约为 12MB,deno 约为 50MB - Pyodide 无法仅通过 PyPI 包调用
- 文件挂载:通过浏览器 API 实现虚拟文件系统
- 快照:据推测可以通过 Temporal 等持久化执行方案实现,但难度较大
starlark-rust
参见 starlark-rust.
- 语言完整性:配置语言,非 Python - 无类、异常、异步
- 安全性:设计上具有确定性和封闭性
- 启动延迟:像 Monty 一样嵌入在进程中运行,因此启动时间令人印象深刻
- 配置复杂度:可通过 starlark-pyo3 在 Python 中使用
- 文件挂载:据我所知,设计上不支持文件处理?
- 快照:据我所知,不可能?
WASI / Wasmer
通过 Wasmer 在 WebAssembly 中运行 Python。
- 语言完整性:完整的 CPython,纯 Python 外部包可通过挂载运行,带有 C 绑定的外部包无法运行
- 安全性:原则上 WebAssembly 应提供强沙箱保证。
- 启动延迟:wasmer python 包已 3 年未更新,且我未找到从 Python 调用 wasmer 中 Python 的文档,因此我通过 subprocess 调用。启动延迟为 66ms。
- 配置复杂度:wasmer 下载大小为 100mb,"python/python" 包为 50mb。
- FOSS:我将其标记为 "free *",因为成本为零,但并非所有内容似乎都是开源的。截至 2026-02-10,
python/pythonwasmer 包 包没有 readme,没有 license,没有 source 链接,也没有说明其构建方式,最近上传的版本显示大小为 "0B",尽管下载大小约为 ~50MB - Python 二进制的构建过程不清晰且不透明。(如果我在这里错了,请创建一个 issue 来纠正我) - 文件挂载:支持
- 快照:通过日志记录支持
沙箱服务
存在类似的挑战,使用 k8s 设置自己的沙箱环境时,配置复杂度更高,但网络延迟更低。
- 语言完整性:完整的 CPython,支持任意库
- 安全性:专业管理的容器隔离
- 启动延迟:网络往返时间和容器启动时间。我在伦敦使用 Daytona EU 测得约 1 秒的冷启动时间,Daytona 宣称延迟低于 90 毫秒,推测这是针对已存在容器的情况,不清楚是否包含网络延迟
- 开源软件(FOSS):按执行次数或计算时间计费,部分实现是开源的
- 设置复杂度:API 集成、认证令牌——对初创公司来说没问题,但对企业来说通常难以接受
- 文件挂载:通过 API 调用进行上传/下载
- 快照:使用 Temporal 等持久化执行方案可以实现,这些服务也提供了一些相关解决方案,我认为基于 Docker 容器
YOLO Python
直接通过 exec()(约 0.1 毫秒)或子进程(约 30 毫秒)运行 Python。
- 语言完整性:完整的 CPython,支持任意库
- 安全性:无——完整的文件系统、网络、环境变量、系统命令访问权限
- 启动延迟:
exec()接近零,子进程约 30 毫秒 - 设置复杂度:无
- 文件挂载:直接文件系统访问(这就是问题所在)
- 快照:使用 Temporal 等持久化执行方案可以实现
Pydantic 技术栈的一部分
Pydantic 技术栈包含您发布生产级 AI 智能体所需的一切:
- Pydantic AI - 类型安全的智能体框架
- Pydantic Logfire - AI 优先的全栈可观测性
- Logfire AI Gateway - 统一的 LLM 代理