(或者，`pip install pydantic-monty` 是给老古董们用的)

`pydantic-monty` 是一个元包，将 `pydantic-monty-client`（`pydantic_monty` 模块）与 `pydantic-monty-runtime`（`monty` 工作程序二进制文件）配对。如果二进制文件已经来自其他来源，则可以单独安装 `pydantic-monty-client`。

用法：```python
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())

### JavaScript / TypeScript

安装：```bash
npm install @pydantic/monty

对于浏览器（或任何无法使用子进程的环境），同一个包
在 `@pydantic/monty/wasm` 子路径下提供了进程内 WebAssembly 构建
（没有崩溃隔离：那里的沙箱崩溃即是主机崩溃）。

### Rust

对于在 Rust 中运行不受信任的代码，我们推荐使用
[`monty-pool`](https://crates.io/crates/monty-pool) crate，而不是下面提供的进程内 API。
`monty-pool` 仅在 `monty` 工作子进程中运行代码，这提供了额外保护：
由对抗性代码（栈溢出、分配器中止）触发的崩溃只会杀死工作进程 —
池会检测到该进程死亡并替换它 — 并且父端看门狗可以杀死
超过硬超时时间的工作进程。它与上述 Python 和 JavaScript 包所基于的
引擎相同。有关用法，请参阅 [monty-pool README](https://github.com/pydantic/monty/tree/main/crates/monty-pool)
以获取用法。

`monty` crate 本身提供了进程内解释器：```rust
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` 本身没有转储格式，但两者都实现了 `serde::Serialize`/`Deserialize`，因此宿主可以使用它已有的任何格式来序列化解析后的代码或暂停中的运行。

## Worker 中的内存限制

会话的 `max_memory` 由 worker 的分配器（allocator）衡量。解释器在超过软限制后会报告一个优雅的 `MemoryError`；如果某次分配在检查点之间跳跃过大，更高的硬限制会终止并替换该 worker。

有关超过限制后如何向宿主呈现的说明，请参阅 [`limitations/resource_limits.md`](https://github.com/pydantic/monty/blob/HEAD/limitations/resource_limits.md)；有关子进程和 WebAssembly worker 运行时所使用的分配器，请参阅 `monty-alloc`。

## PydanticAI 集成

Monty 将为 [Pydantic AI](https://github.com/pydantic/pydantic-ai) 中的代码模式（code-mode）提供支持。LLM 不再进行顺序的工具调用，而是编写 Python 代码，以函数形式调用你的工具，并由 Monty 安全地执行。```python test="skip"
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())