pydantic-monty

Python bindings for the Monty sandboxed Python interpreter.

Execution always happens in a pool of monty worker subprocesses: a monty process can never be made fully crash-proof against memory errors (stack overflows, allocator aborts) triggered by adversarial input, so crash isolation is built in. A crashed worker raises MontyCrashedError and is replaced transparently — your process is never at risk.

Installation

bash

pip install pydantic-monty

pydantic-monty is a metapackage with no code of its own; it installs the two distributions that make up a working sandbox:

pydantic-monty-client — the pydantic_monty module you import (pool, sessions, value conversion)
pydantic-monty-runtime — the monty worker binary the pool spawns, shipped the same way uv and ruff ship their binaries

Install pydantic-monty-client on its own when the worker binary comes from somewhere else — a base image, a system package, a build of this repo — and point pydantic_monty at it via MONTY_BIN, binary_path=, or PATH.

Usage

Basic execution

python

from pydantic_monty import Monty

with Monty() as pool:
    with pool.checkout() as session:
        print(session.feed_run('1 + 2'))
        #> 3

Monty() is a pool of workers; pool.checkout() dedicates one worker to a REPL session. Session state persists across feed_run calls:

python

from pydantic_monty import Monty

with Monty() as pool:
    with pool.checkout() as session:
        session.feed_run('x = 40')
        print(session.feed_run('x + 2'))
        #> 42

Async

AsyncMonty is the asyncio counterpart: worker I/O runs off the event loop, and external functions may be coroutines.

python

import asyncio

from pydantic_monty import AsyncMonty


async def fetch(url: str) -> str:
    await asyncio.sleep(0.01)
    return f'contents of {url}'


async def main():
    async with AsyncMonty() as pool:
        async with pool.checkout() as session:
            result = await session.feed_run(
                "await fetch('https://example.com')",
                external_lookup={'fetch': fetch},
            )
    print(result)
    #> contents of https://example.com


asyncio.run(main())

Input variables and external lookup

python

from pydantic_monty import Monty

with Monty() as pool:
    with pool.checkout() as session:
        result = session.feed_run(
            'double(x) + y',
            inputs={'x': 5, 'y': 1},
            external_lookup={'double': lambda x: x * 2},
        )
    print(result)
    #> 11

Snapshots: pausing and resuming execution

feed_start is the suspendable counterpart of feed_run: instead of driving a snippet to completion, it hands control back at each external call, OS call, name lookup, or future resolution as a snapshot. You answer with snapshot.resume(...), which returns the next snapshot or a MontyComplete.

python

from pydantic_monty import FunctionSnapshot, Monty, MontyComplete

with Monty() as pool:
    with pool.checkout() as session:
        snapshot = session.feed_start('greet(name) + "!"', inputs={'name': 'Ada'})
        assert isinstance(snapshot, FunctionSnapshot)
        print(snapshot.function_name, snapshot.args)
        #> greet ('Ada',)
        result = snapshot.resume({'return_value': 'hello Ada'})
        assert isinstance(result, MontyComplete)
        print(result.output)
        #> hello Ada!

To iterate a snippet to completion without answering each suspension by hand, pass an external_lookup (and/or os) to feed_start and drive with snapshot.resume_auto(), which resolves each external call and name lookup from them automatically — the same resolution feed_run performs, but one step at a time so you can inspect or dump() each snapshot along the way:

python

from pydantic_monty import Monty, MontyComplete

with Monty() as pool:
    with pool.checkout() as session:
        snapshot = session.feed_start(
            'greet(name) + "!"',
            inputs={'name': 'Ada'},
            external_lookup={'greet': lambda n: f'hello {n}'},
        )
        while not isinstance(snapshot, MontyComplete):
            snapshot = snapshot.resume_auto()
        print(snapshot.output)
        #> hello Ada!

On AsyncMonty, external_lookup callables may be coroutine functions and resume_auto is awaitable (snapshot = await snapshot.resume_auto()); a coroutine external is awaited concurrently and settled via an AsyncFutureSnapshot.

snapshot.dump() serializes the paused worker to bytes; a fresh session's load_snapshot restores it and returns the snapshot to resume. This lets you checkpoint execution and continue it later, even in a different process:

python

from pydantic_monty import FunctionSnapshot, Monty, MontyComplete

with Monty() as pool:
    with pool.checkout() as session:
        snapshot = session.feed_start(
            'fetch(url)', inputs={'url': 'https://example.com'}
        )
        blob = snapshot.dump()

    # later — restore into a fresh session and resume
    with pool.checkout() as session:
        snapshot = session.load_snapshot(blob)
        assert isinstance(snapshot, FunctionSnapshot)
        result = snapshot.resume({'return_value': 'page contents'})
        assert isinstance(result, MontyComplete)
        print(result.output)
        #> page contents

If the paused feed used filesystem mounts, re-supply the same ones to load_snapshot(blob, mount=...) — their host paths are not stored in the dump.

session.dump() between feeds serializes an idle session instead; restore it with session.load_session(blob) (which returns None) and keep feeding. Both load_session and load_snapshot are valid only on a fresh session, before any feed; using the wrong one for a dump's kind raises. AsyncMonty sessions expose the same feed_start / load_session / load_snapshot, with awaitable resume(...).

Resource limits

Limits are enforced inside the worker; the pool's request_timeout is a host-side backstop that kills a hung worker outright. An installed telemetry adapter invokes trusted Python SDK callbacks synchronously; enforcement is delayed while such a callback runs. max_duration_secs limits cumulative execution time — the clock runs only while the interpreter executes, never while suspended waiting on the host, and accumulates across feeds. The worker reports its execution time on every protocol turn, and sessions with the limit are additionally killed duration_limit_grace (1s, not currently configurable from Python) after the remaining budget expires, covering hangs the in-sandbox limit cannot catch (its check only runs at interpreter checkpoints).

python

from pydantic_monty import Monty, MontyRuntimeError

with Monty(request_timeout=10) as pool:
    with pool.checkout(limits={'max_duration_secs': 0.1}) as session:
        try:
            session.feed_run('while True:\n    pass')
        except MontyRuntimeError as exc:
            print(exc.display(format='type-msg').split(':')[0])
            #> TimeoutError

Type checking

Monty bundles ty: each fed snippet can be type-checked inside the worker before it runs, with successfully executed snippets accumulating into the checking context.

python

from pydantic_monty import Monty, MontyTypingError

with Monty() as pool:
    with pool.checkout(type_check=True) as session:
        try:
            session.feed_run("x: int = 'not an int'")
        except MontyTypingError as exc:
            print('invalid-assignment' in exc.display())
            #> True

type_check_format picks the rendering — ty's 'full' (the default: source snippet and carets), 'concise', 'azure', 'json', 'jsonlines', 'rdjson', 'pylint', 'gitlab' or 'github' — and type_check_color adds ANSI colour to 'full' and 'concise'. Both are checkout() arguments rather than display() arguments because the diagnostics are rendered inside the worker: ty's structured diagnostics resolve their spans against the type checker's database, so only the rendered text crosses the wire.

python

from pydantic_monty import Monty, MontyTypingError

with Monty() as pool:
    with pool.checkout(type_check=True, type_check_format='concise') as session:
        try:
            session.feed_run("x: int = 'not an int'")
        except MontyTypingError as exc:
            print(exc.display())
            """
            main.py:1:10: error[invalid-assignment] Object of type `Literal["not an int"]` is not assignable to `int`
            """

Crash isolation

python

from pydantic_monty import Monty, MontyCrashedError

hostile_code = '...'

with Monty() as pool:
    with pool.checkout() as session:
        try:
            session.feed_run(hostile_code)  # even a segfault is contained
        except MontyCrashedError:
            ...  # the worker died; the pool already replaced it

Observability

The Python Logfire integration instruments the pool through a private adapter hook. It propagates the active Python OTel context into each checkout, which becomes one session span with nested feed and suspension spans recording code, inputs, external calls, exceptions, and print output. Session dumps and restores are recorded by size only.

Logfire's Python SDK owns sampling, export credentials, resources, flushing, and shutdown. The Rust binding runs only an exporter-free processor pipeline; workers receive no credentials. Instrumentation is disabled unless an adapter is explicitly installed. Enabled instrumentation captures content, truncating large values at the telemetry attribute size limit.

See limitations/pool-architecture.md in the repository for the behavioural details of subprocess execution (host-side mounts, buffered print callbacks, session dumps).