pydantic/monty
GitHub: pydantic/monty
一个用 Rust 编写的极简安全 Python 解释器，专为 AI Agent 安全执行 LLM 生成的代码而设计，无需沙箱即可实现微秒级启动。
Stars: 5868 | Forks: 223
Monty
A minimal, secure Python interpreter written in Rust for use by AI.
**实验性** - 本项目仍在开发中，尚未达到生产就绪状态。
一个用 Rust 编写的、极简且安全的 Python 解释器，专为 AI 设计。
Monty 避免了使用基于完整容器的沙箱来运行 LLM 生成的代码所带来的成本、延迟、复杂性和各种麻烦。
相反，它允许你安全地运行由 LLM 编写并嵌入到你的 Agent 中的 Python 代码，其启动时间以个位数微秒计，而非数百毫秒。
Monty **能**做什么：
* 运行合理的 Python 代码子集 —— 足以让你的 Agent 表达它想做的事情
* 完全阻止对主机环境的访问：文件系统、环境变量和网络访问都是通过开发者可以控制的外部函数调用实现的
* 调用主机上的函数 —— 仅限于你授权它访问的函数
* 运行类型检查 —— monty 支持完整的现代 Python 类型提示，并在单个二进制文件中内置了 [ty](https://docs.astral.sh/ty/) 以运行类型检查
* 在外部函数调用时快照为字节，这意味着你可以将解释器状态存储在文件或数据库中，以便稍后恢复
* 启动极快（从代码到执行结果不到 1μs），运行时性能与 CPython 相当（通常在快 5 倍到慢 5 倍之间）
* 可从 Rust、Python 或 JavaScript 调用 —— 因为 Monty 不依赖 cpython，你可以在任何能运行 Rust 的地方使用它
* 控制资源使用 —— Monty 可以跟踪内存使用、分配、堆栈深度和执行时间，并在超过预设限制时取消执行
* 收集 stdout 和 stderr 并将其返回给调用者
* 通过主机上的 async 或 sync 代码运行 async 或 sync 代码
Monty **不能**做什么：
* 使用标准库（少数选定模块除外：`sys`, `typing`, `asyncio`, `dataclasses`（即将支持），`json`（即将支持））
* 使用第三方库（如 Pydantic），支持外部 Python 库不是目标
* 定义类（支持即将推出）
* 使用 match 语句（同样，支持即将推出）
简而言之，Monty 极其有限，专为**一**个用例设计：
**运行由 Agent 编写的代码。**
关于为什么要这样做的动机，请参阅：
* Cloudflare 的 [Codemode](https://blog.cloudflare.com/code-mode/)
* Anthropic 的 [Programmatic Tool Calling](https://platform.claude.com/docs/en/agents-and-tools/tool-use/programmatic-tool-calling)
* Anthropic 的 [Code Execution with MCP](https://www.anthropic.com/engineering/code-execution-with-mcp)
* Hugging Face 的 [Smol Agents](https://github.com/huggingface/smolagents)
简而言之，上述所有的核心思想是：如果让 LLM 编写 Python（或 JavaScript）代码，而不是依赖传统的工具调用，它们可以工作得更快、更便宜、更可靠。Monty 使这成为可能，且无需沙箱的复杂性，也没有直接在主机上运行代码的风险。
**注意：** Monty 将（很快）用于在 [Pydantic AI](https://github.com/pydantic/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 = ''
"""
m = pydantic_monty.Monty(
code,
inputs=['prompt'],
external_functions=['call_llm'],
script_name='agent.py',
type_check=True,
type_check_stubs=type_definitions,
)
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():
output = await pydantic_monty.run_monty_async(
m,
inputs={'prompt': 'testing'},
external_functions={'call_llm': call_llm},
)
print(output)
#> example output, message count 2
if __name__ == '__main__':
import asyncio
asyncio.run(main())
```
#### 使用外部函数的迭代执行
使用 `start()` 和 `resume()` 迭代地处理外部函数调用，
让你可以控制每一次调用：
```
import pydantic_monty
code = """
data = fetch(url)
len(data)
"""
m = pydantic_monty.Monty(code, inputs=['url'], external_functions=['fetch'])
# 开始执行 - 当调用 fetch() 时暂停
result = m.start(inputs={'url': 'https://example.com'})
print(type(result))
#>
print(result.function_name) # fetch
#> fetch
print(result.args)
#> ('https://example.com',)
# 执行实际的 fetch，然后携带结果恢复
result = result.resume(return_value='hello world')
print(type(result))
#>
print(result.output)
#> 11
```
#### 序列化
`Monty` 和 `MontySnapshot` 都可以序列化为字节并在稍后恢复。
这允许缓存解析后的代码或跨进程边界暂停执行：
```
import pydantic_monty
# 序列化解析后的代码以避免重复解析
m = pydantic_monty.Monty('x + 1', inputs=['x'])
data = m.dump()
# 稍后，恢复并运行
m2 = pydantic_monty.Monty.load(data)
print(m2.run(inputs={'x': 41}))
#> 42
# 序列化执行中途的状态
m = pydantic_monty.Monty('fetch(url)', inputs=['url'], external_functions=['fetch'])
progress = m.start(inputs={'url': 'https://example.com'})
state = progress.dump()
# 稍后，恢复并继续（例如，在不同的进程中）
progress2 = pydantic_monty.MontySnapshot.load(state)
result = progress2.resume(return_value='response data')
print(result.output)
#> response data
```
### Rust
```
use monty::{MontyRun, MontyObject, NoLimitTracker, PrintWriter};
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()], vec![]).unwrap();
let result = runner.run(vec![MontyObject::Int(10)], NoLimitTracker, &mut PrintWriter::Stdout).unwrap();
assert_eq!(result, MontyObject::Int(55));
```
#### 序列化
`MontyRun` 和 `RunProgress` 可以使用 `dump()` 和 `load()` 方法进行序列化：
```
use monty::{MontyRun, MontyObject, NoLimitTracker, PrintWriter};
// Serialize parsed code
let runner = MontyRun::new("x + 1".to_owned(), "main.py", vec!["x".to_owned()], vec![]).unwrap();
let bytes = runner.dump().unwrap();
// Later, restore and run
let runner2 = MontyRun::load(&bytes).unwrap();
let result = runner2.run(vec![MontyObject::Int(41)], NoLimitTracker, &mut PrintWriter::Stdout).unwrap();
assert_eq!(result, MontyObject::Int(42));
```
## PydanticAI 集成
Monty 将为 [Pydantic AI](https://github.com/pydantic/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 时，通常会有两种反应：
1. 天哪，这解决了很多问题，我想要。
2. 为什么不用 X？
其中 X 是某种替代技术。奇怪的是，这些反应通常是结合在一起的，这表明人们还没有找到适合他们的替代方案，但又难以置信竟然没有好的替代方案，以至于需要从头开始创建一个完整的 Python 实现。
我将尝试介绍最明显的替代方案，以及为什么它们不适合我们的需求。
注意：所有这些技术都令人印象深刻并有着广泛的用途，这些关于它们在我们用例中局限性的评论不应被视为批评。大多数这些解决方案在构思时并没有提供 LLM 沙箱的目标，这就是为什么它们在这方面不一定擅长。
| 技术 | 语言完整度 | 安全性 | 启动延迟 | FOSS | 设置复杂度 | 文件挂载 | 快照 |
|--------------------|-----------------------|--------------|----------------|------------|------------------|----------------|--------------|
| Monty | partial | strict | 0.06ms | free / OSS | easy | easy | easy |
| Docker | full | good | 195ms | free / OSS | intermediate | easy | intermediate |
| Pyodide | full | poor | 2800ms | free / OSS | intermediate | easy | hard |
| starlark-rust | very limited | good | 1.7ms | free / OSS | easy | not available? | impossible? |
| WASI / Wasmer | partial, almost full | strict | 66ms | free * | intermediate | easy | intermediate |
| sandboxing service | full | strict | 1033ms | not free | intermediate | hard | intermediate |
| YOLO Python | full | non-existent | 0.1ms / 30ms | free / OSS | easy | easy / scary | hard |
请参阅 [./scripts/startup_performance.py](scripts/startup_performance.py) 获取用于计算启动性能数据的脚本。
以下是每一行的详细信息：
### Monty
- **语言完整度**：尚无类（暂时）、有限的标准库、无第三方库
- **安全性**：显式控制的文件系统、网络和环境访问，对执行时间和内存使用有严格限制
- **启动延迟**：以微秒级启动
- **设置复杂度**：只需 `pip install pydantic-monty` 或 `npm install @pydantic/monty`，下载约 4.5MB
- **文件挂载**：严格控制，参见 [#85](https://github.com/pydantic/monty/pull/85)
- **快照**：Monty 通过 `dump()` 和 `load()` 的暂停和恢复功能，使得暂停、恢复和分叉执行变得轻而易举
### Docker
- **语言完整度**：完整的 CPython 和任意库
- **安全性**：进程和文件系统隔离，网络策略，但存在容器逃逸，内存限制是可能的
- **启动延迟**：容器启动开销（测量值约 195ms）
- **设置复杂度**：需要 Docker 守护进程、容器镜像、编排，`python:3.14-alpine` 为 50MB —— docker 无法从 PyPI 安装
- **文件挂载**：卷挂载工作良好
- **快照**：可以通过 Temporal 等持久执行解决方案实现，或者通过快照镜像并将其保存为 Docker 镜像
### Pyodide
- **语言完整度**：编译为 WASM 的完整 CPython，几乎所有库都可用
- **安全性**：依赖浏览器/WASM 沙箱 —— 不是为服务器端隔离设计的，Python 代码可以在 JS 运行时中运行任意代码，只有 deno 允许隔离，但在 deno 中很难/不可能强制执行内存限制
- **启动延迟**：WASM 运行时加载缓慢（冷启动约 2800ms）
- **设置复杂度**：需要加载 WASM 运行时，处理异步初始化，pyodide NPM 包约 12MB，deno 约 50MB —— Pyodide 不能仅通过 PyPI 包调用
- **文件挂载**：通过浏览器 API 的虚拟文件系统
- **快照**：推测可以通过 Temporal 等持久执行解决方案实现，但很难
### starlark-rust
参见 [starlark-rust](https://github.com/facebook/starlark-rust)。
- **语言完整度**：配置语言，不是 Python —— 无类、异常、async
- **安全性**：设计上确定性和密封
- **启动延迟**：像 Monty 一样在进程中嵌入运行，因此启动时间令人印象深刻
- **设置复杂度**：可通过 [starlark-pyo3](https://github.com/inducer/starlark-pyo3) 在 python 中使用
- **文件挂载**：据我所知设计上没有文件处理？
- **快照**：据我所知不可能？
### WASI / Wasmer
通过 [Wasmer](https://wasmer.io/) 在 WebAssembly 中运行 Python。
- **语言完整度**：完整的 CPython，纯 Python 外部包通过挂载工作，带有 C 绑定的外部包无法工作
- **安全性**：原则上 WebAssembly 应该提供强大的沙箱保证。
- **启动延迟**：[wasmer](https://pypi.org/project/wasmer/) python 包已经 3 年没有更新了，我找不到关于在 Python 中从 wasmer 调用 Python 的文档，所以我通过子进程调用它。启动延迟为 66ms。
- **设置复杂度**：wasmer 下载量为 100mb，"python/python" 包为 50mb。
- **FOSS**：我将其标记为 "free *" 是因为成本为零，但似乎并非所有内容都是开源的。截至 2026-02-10，[`python/python` wasmer package](https://wasmer.io/python/python) 包没有 readme、许可证、源链接，也没有说明它是如何构建的，最近上传的版本显示大小为 "0B"，尽管下载量约为 50MB —— Python 二进制文件的构建过程不清晰且不透明。（如果我错了，请创建一个 issue 来纠正我）
- **文件挂载**：支持
- **快照**：通过日志（journaling）支持
### 沙箱服务
如 [Daytona](https://daytona.io)、[E2B](https://e2b.dev)、[Modal](https://modal.com) 等服务。
也存在类似的挑战，设置复杂度更高，但在使用 k8s 设置你自己的沙箱时，网络延迟更低。
- **语言完整度**：完整的 CPython 和任意库
- **安全性**：专业管理的容器隔离
- **启动延迟**：网络往返和容器启动时间。我在伦敦访问 Daytona EU 的冷启动时间约为 1s，Daytona 宣称延迟低于 90ms，推测这是针对现有容器的，不清楚是否包括网络延迟
- **FOSS**：按执行或计算时间付费，一些实现是开源的
- **设置复杂度**：API 集成，认证令牌 —— 对初创公司没问题，但对大企业来说通常行不通
- **文件挂载**：通过 API 调用上传/下载
- **快照**：可以通过 Temporal 等持久执行解决方案实现，而且这些服务也提供一些解决方案，我认为基于 docker 容器
### YOLO Python
通过 `exec()`（约 0.1ms）或子进程（约 30ms）直接运行 Python。
- **语言完整度**：完整的 CPython 和任意库
- **安全性**：无 —— 完整的文件系统、网络、环境变量、系统命令访问权限
- **启动延迟**：`exec()` 接近零，子进程约 30ms
- **设置复杂度**：无
- **文件挂载**：直接文件系统访问（这就是问题所在）
- **快照**：可以通过 Temporal 等持久执行解决方案实现标签：Agent, AI工具, AST解析, DLL 劫持, DNS 反向解析, LLM, Pydantic, Python解释器, Rust, Unmanaged PE, 人工智能, 代码执行, 代码生成, 可视化界面, 大语言模型, 安全, 嵌入式, 微秒级启动, 沙箱, 渗透测试工具, 用户模式Hook绕过, 网络流量审计, 超时处理, 逆向工具, 通知系统, 通知系统
