Pools
The execution surface of pydantic_monty: a Monty or AsyncMonty pool of worker subprocesses, checked out one
session at a time, plus the resource limits and print() sinks sessions are configured
with.
Install as pydantic-monty — see the
Python quickstart.
Sync context manager owning a pool of monty subprocess workers.
Monty processes can never be made fully crash-proof against memory errors
(stack overflow, allocator aborts), so execution always happens in worker
subprocesses: a crashed worker raises MontyCrashedError and is replaced
transparently — the host Python process is never at risk.
with Monty() as pool:
with pool.checkout() as session:
result = session.feed_run('1 + 1')
def __new__(
cls,
*,
binary_path: str | Path | None = None,
min_processes: int = 1,
max_processes: int | None = None,
checkout_timeout: float | None = None,
request_timeout: float | None = None,
max_checkouts_per_worker: int | None = None,
) -> Self
Configure a worker pool; the workers are spawned by with.
Path to the monty CLI binary. When omitted it is
resolved from the MONTY_BIN environment variable, the
environment’s scripts directory (where the pydantic-monty-runtime
dependency installs it), or PATH.
min_processes : int Default: 1
Workers spawned eagerly and kept warm.
Cap on live workers (defaults to the CPU count); checkouts beyond it wait for a worker to be returned.
Seconds checkout() waits for a free worker
before raising TimeoutError. None waits forever.
Per-turn parent-side deadline in seconds — a worker
that exceeds it is killed and the call raises MontyCrashedError
with timed_out=True. Trusted synchronous telemetry callbacks
delay enforcement while they run. Backstops sandbox limits.
Recycle a worker after this many sessions.
def checkout(
*,
script_name: str = 'main.py',
limits: ResourceLimits | None = None,
type_check: bool = False,
type_check_stubs: str | None = None,
type_check_format: TypeCheckFormat | None = None,
type_check_color: bool = False,
assert_message_annotations: bool | int = ...,
) -> MontySession
Prepare a REPL session served by a dedicated worker.
The worker is checked out of the pool by with on the returned
session and returned to the pool when the with block exits.
script_name : str Default: 'main.py'
Name used in tracebacks and error messages.
limits : ResourceLimits | None Default: None
Resource limits enforced inside the worker.
type_check : bool Default: False
Type-check each fed snippet before executing it; each successfully executed snippet is appended to the accumulated context used for type-checking subsequent snippets.
Stub declarations made available to type checking.
type_check_format : TypeCheckFormat | None Default: None
How MontyTypingError diagnostics are rendered;
None (the default) means 'full'. Chosen here rather than on
the error because the checker’s structured diagnostics never
leave the worker.
type_check_color : bool Default: False
Render diagnostics with ANSI colour escapes; only
'full' and 'concise' carry colour.
Give failed assert statements
pytest-style introspected messages, e.g.
AssertionError: assert 2 == 5 — a deliberate divergence from
CPython’s empty AssertionError. On by default; set to False
to restore CPython’s behavior, or to an int >= 1 to customize
the per-operand repr truncation length (default 120 bytes).
A REPL session running in a dedicated monty subprocess worker.
Obtained from Monty.checkout() and used as a context manager. Session
state (globals, functions) persists across feed_run calls within the
session.
OS process id of this session’s worker (diagnostics/tests).
None when no worker is attached or a turn is currently in flight
on another thread (the getter never blocks on a running turn).
def feed_run(
code: str,
*,
inputs: dict[str, Any] | None = None,
external_lookup: dict[str, Any] | None = None,
print_callback: Callable[[Literal['stdout', 'stderr'], str], None] | CollectStreams | CollectString | None = None,
mount: MountDir | list[MountDir] | None = None,
os: Callable[[OsFunction, tuple[Any, ...], dict[str, Any]], Any] | AbstractOS | None = None,
skip_type_check: bool = False,
) -> Any
Execute one snippet in the worker and return its result.
Blocks the calling thread (with the GIL released) while the worker
runs; external functions, the os fallback, and print callbacks are
invoked in this process. Async external functions are not supported
here — use AsyncMonty.
code : str
The Python snippet to execute; its trailing expression value (if any) is converted to a Python object and returned.
Values eagerly bound as globals before the snippet runs — every entry is converted and bound once, whether or not it is referenced.
Host values resolving names the snippet leaves
undefined, lazily and on demand: a callable entry becomes a host
function the sandbox can call, any other value is converted and
returned directly when the name is read, and an absent name
raises NameError. The lazy counterpart to inputs; a name
present in both is served by the eager inputs binding.
print_callback : Callable[[Literal[‘stdout’, ‘stderr’], str], None] | CollectStreams | CollectString | None Default: None
Receives the sandbox’s print() output as
(stream, text), or a CollectStreams / CollectString
collector. Defaults to the host process stdout/stderr.
Host directories mounted into the sandbox for this feed.
Serviced by the pool on the host side — 'overlay' writes
live in the pool’s per-feed mount table and are discarded when
the feed ends.
os : Callable[[OsFunction, tuple[Any, …], dict[str, Any]], Any] | AbstractOS | None Default: None
Fallback handler for OS calls (e.g. filesystem access) not
covered by a mount, invoked as (function_name, args, kwargs),
or an AbstractOS instance.
skip_type_check : bool Default: False
Skip type checking for this feed even when the
session was checked out with type_check=True.
MontyRuntimeError— The code raised an exception (session survives).MontyTypingError— Type checking rejected the snippet (session survives).MontyCrashedError— The worker process died or hitrequest_timeout; the session is lost but the pool replaces the worker.
def feed_start(
code: str,
*,
inputs: dict[str, Any] | None = None,
external_lookup: dict[str, Any] | None = None,
print_callback: PrintCallback | None = None,
mount: MountDir | list[MountDir] | None = None,
os: OsHandler | None = None,
skip_type_check: bool = False,
) -> SyncSnapshot
Start a snippet and return a snapshot at each external call, OS call, name lookup, or future resolution instead of driving to completion.
Answer the snapshot with snapshot.resume(...), which returns the next
snapshot or a MontyComplete. Alternatively, supply external_lookup
(and/or os) and drive the whole snippet with snapshot.resume_auto(),
which answers each suspension from them automatically:
snapshot = session.feed_start(code, external_lookup={'fetch': fetch})
while not isinstance(snapshot, MontyComplete):
snapshot = snapshot.resume_auto()
Unlike feed_run, external_lookup is not consulted during this
initial drive — external calls and name lookups are still surfaced as
snapshots; it is only captured for later resume_auto() calls.
Use snapshot.dump() to checkpoint the worker mid-execution and
load_snapshot to restore it.
code : str
The Python snippet to execute; its trailing expression value
(if any) is the MontyComplete.output when the feed completes.
Values eagerly bound as globals before the snippet runs — every entry is converted and bound once, whether or not it is referenced.
Host functions and values, by name, that
resume_auto() resolves external calls and undefined names
against (as in feed_run). Captured for resume_auto(); not
used by a plain resume(...).
print_callback : PrintCallback | None Default: None
Receives the sandbox’s print() output as
(stream, text), or a CollectStreams / CollectString
collector. Defaults to the host process stdout/stderr.
Host directories mounted into the sandbox for the whole feed
(there is no mount= on resume). 'overlay' writes live in
the pool’s per-feed mount table and are discarded when the feed
ends.
Fallback handler for OS calls not covered by a mount, invoked
as (function_name, args, kwargs), or an AbstractOS instance.
Consulted only by resume_auto() — feed_start always surfaces
OS calls as snapshots.
skip_type_check : bool Default: False
Skip type checking for this feed even when the
session was checked out with type_check=True.
def load_session(state: bytes) -> None
Restore a session between feeds.
This method should take data from session.dump() taken when no block of
code is running (i.e. between feeds).
Use load_snapshot for a dump taken mid-execution.
The dump restores its own script_name /
limits / type-check state (the checkout() config for those is not
applied). The class-instance store starts empty — it is host state and
never part of a dump, so restored ClassInstance values fall back to
MontyClassProxy stand-ins and method calls on them fail. Raises if
the dump is actually a suspended snapshot.
def load_snapshot(
state: bytes,
*,
mount: MountDir | list[MountDir] | None = None,
print_callback: PrintCallback | None = None,
external_lookup: dict[str, Any] | None = None,
os: OsHandler | None = None,
) -> SyncSnapshot
Restore a snapshot generated while a block of code is running (e.g.
after feed_start) and return the re-announced snapshot to resume.
Use load_session for a dump taken between feeds.
Valid only on a fresh session, before any feed or load; raises
RuntimeError otherwise. The dump restores its own script_name /
limits / type-check state (the checkout() config for those is not
applied). The class-instance store starts empty — it is host state and
never part of a dump, so restored ClassInstance values fall back to
MontyClassProxy stand-ins and method calls on them fail. mount
re-establishes the suspended feed’s mounts, which are never part of the
dump — pass the same mounts the original feed used, or its filesystem
calls degrade into unhandled OS calls. 'overlay' writes made before
the dump are not preserved (the restored overlay starts empty). Raises
if the dump is actually an idle session.
external_lookup / os are captured for resume_auto(), exactly as on
feed_start. One caveat applies to a restored snapshot: a restored
FutureSnapshot’s pending coroutines are gone (they lived in the
previous process), so resume_auto() on it raises — resolve it manually
with resume({call_id: ...}).
def dump() -> bytes
Serialize the worker’s session state (idle or suspended) to opaque bytes using monty’s existing dump format. The session stays usable.
def install_dependencies(requirements: list[str]) -> None
Install third-party Python packages into the session, making them
importable by subsequent feed_run calls. Session-scoped and
repeatable; an empty list is a no-op.
Only supported by an embedded-CPython worker.
Against the pure-Monty sandbox worker, or on a uv install failure
(the error carries uv’s stderr), raises MontyRuntimeError; the
session stays usable. Bounded by the pool’s request_timeout, so raise
it for large dependency sets.
Requirements are PEP 508 strings, e.g. ["httpx>=0.27", "numpy"].
Dependencies a script declares inline via PEP 723 (# /// script) are
installed automatically on feed_run and need no call here.
Async context manager owning a pool of monty subprocess workers.
The async counterpart of Monty: worker I/O runs off the event loop, and
external functions may be coroutines.
async with AsyncMonty() as pool:
async with pool.checkout() as session:
result = await session.feed_run('1 + 1')
def __new__(
cls,
*,
binary_path: str | Path | None = None,
min_processes: int = 1,
max_processes: int | None = None,
checkout_timeout: float | None = None,
request_timeout: float | None = None,
max_checkouts_per_worker: int | None = None,
) -> Self
Configure a worker pool; the workers are spawned by async with.
Arguments are identical to Monty.
def checkout(
*,
script_name: str = 'main.py',
limits: ResourceLimits | None = None,
type_check: bool = False,
type_check_stubs: str | None = None,
type_check_format: TypeCheckFormat | None = None,
type_check_color: bool = False,
assert_message_annotations: bool | int = ...,
) -> AsyncMontySession
Prepare a REPL session served by a dedicated worker.
The worker is checked out of the pool by async with on the returned
session and returned to the pool when the async with block exits.
Arguments are identical to Monty.checkout.
A REPL session running in a dedicated monty subprocess worker.
Obtained from AsyncMonty.checkout() and used as an async context
manager. Session state (globals, functions) persists across
feed_run calls within the session.
OS process id of this session’s worker (diagnostics/tests).
None when no worker is attached or a turn is currently in flight
on another thread (the getter never blocks on a running turn).
@async
def feed_run(
code: str,
*,
inputs: dict[str, Any] | None = None,
external_lookup: dict[str, Any] | None = None,
print_callback: Callable[[Literal['stdout', 'stderr'], str], None] | CollectStreams | CollectString | None = None,
mount: MountDir | list[MountDir] | None = None,
os: Callable[[OsFunction, tuple[Any, ...], dict[str, Any]], Any] | AbstractOS | None = None,
skip_type_check: bool = False,
) -> Any
Execute one snippet in the worker and return its result.
Worker I/O runs off the event loop; external functions (the callable
entries in external_lookup) may be coroutines, awaited concurrently.
See MontySession.feed_run for the shared error types.
code : str
The Python snippet to execute; its trailing expression value (if any) is converted to a Python object and returned.
Values eagerly bound as globals before the snippet runs — every entry is converted and bound once, whether or not it is referenced.
Host values resolving names the snippet leaves
undefined, lazily and on demand: a callable entry (sync or a
coroutine function) becomes a host function the sandbox can call,
any other value is converted and returned directly when the name
is read, and an absent name raises NameError. The lazy
counterpart to inputs; a name present in both is served by the
eager inputs binding.
print_callback : Callable[[Literal[‘stdout’, ‘stderr’], str], None] | CollectStreams | CollectString | None Default: None
Receives the sandbox’s print() output as
(stream, text), or a CollectStreams / CollectString
collector. Defaults to the host process stdout/stderr.
Host directories mounted into the sandbox for this feed.
Serviced by the pool on the host side — 'overlay' writes
live in the pool’s per-feed mount table and are discarded when
the feed ends.
os : Callable[[OsFunction, tuple[Any, …], dict[str, Any]], Any] | AbstractOS | None Default: None
Fallback handler for OS calls (e.g. filesystem access) not
covered by a mount, invoked as (function_name, args, kwargs),
or an AbstractOS instance.
skip_type_check : bool Default: False
Skip type checking for this feed even when the
session was checked out with type_check=True.
@async
def feed_start(
code: str,
*,
inputs: dict[str, Any] | None = None,
external_lookup: dict[str, Any] | None = None,
print_callback: PrintCallback | None = None,
mount: MountDir | list[MountDir] | None = None,
os: OsHandler | None = None,
skip_type_check: bool = False,
) -> AsyncSnapshot
Async counterpart of MontySession.feed_start: resolves to a snapshot
(whose resume(...) / resume_auto() is awaitable) or a
MontyComplete.
As in the sync version, external_lookup (and os) are captured for
await snapshot.resume_auto() rather than consulted during this initial
drive. A coroutine external answered by resume_auto() is awaited
concurrently: it yields an AsyncFutureSnapshot whose resume_auto()
settles the pending coroutines.
code : str
The Python snippet to execute; its trailing expression value
(if any) is the MontyComplete.output when the feed completes.
Values eagerly bound as globals before the snippet runs — every entry is converted and bound once, whether or not it is referenced.
Host functions and values, by name, that
resume_auto() resolves external calls and undefined names
against (as in feed_run). Callables may be coroutine
functions. Captured for resume_auto(); not used by a plain
resume(...).
print_callback : PrintCallback | None Default: None
Receives the sandbox’s print() output as
(stream, text), or a CollectStreams / CollectString
collector. Defaults to the host process stdout/stderr.
Host directories mounted into the sandbox for the whole feed
(there is no mount= on resume). 'overlay' writes live in
the pool’s per-feed mount table and are discarded when the feed
ends.
Fallback handler for OS calls not covered by a mount, invoked
as (function_name, args, kwargs), or an AbstractOS instance.
Consulted only by resume_auto() — feed_start always surfaces
OS calls as snapshots.
skip_type_check : bool Default: False
Skip type checking for this feed even when the
session was checked out with type_check=True.
@async
def load_session(state: bytes) -> None
Async counterpart of MontySession.load_session: restore a session between feeds.
@async
def load_snapshot(
state: bytes,
*,
mount: MountDir | list[MountDir] | None = None,
print_callback: PrintCallback | None = None,
external_lookup: dict[str, Any] | None = None,
os: OsHandler | None = None,
) -> AsyncSnapshot
Async counterpart of MontySession.load_snapshot.
Restore a snapshot generated while a block of code is running (e.g.
after feed_start) and return the re-announced snapshot to resume.
external_lookup / os are captured for resume_auto(), with the same
restored-snapshot caveats as the sync method (a restored FutureSnapshot
cannot be driven with resume_auto() — its pending coroutines are gone).
@async
def dump() -> bytes
Serialize the worker’s session state (idle or suspended) to opaque bytes using monty’s existing dump format. The session stays usable.
@async
def install_dependencies(requirements: list[str]) -> None
Async counterpart of MontySession.install_dependencies: install
third-party packages into the session (off the event loop) so later
feed_run calls can import them. Session-scoped and repeatable; an
empty list is a no-op.
Only supported by an embedded-CPython worker. Against the pure-Monty
sandbox worker, or on a uv install failure, raises
MontyRuntimeError; the session stays usable. PEP 723 inline
dependencies are installed automatically on feed_run.
Bases: TypedDict
Configuration for resource limits during code execution.
All limits are optional. Omit a key — or set it to None explicitly —
to disable that limit, with one exception: max_recursion_depth cannot
be disabled, and omitting it leaves the 1000-frame default in place.
Maximum execution time in seconds.
Maximum heap memory in bytes.
Run garbage collection every N allocations.
Maximum function call stack depth (default: 1000).
Collect printed output as (stream, text) tuples.
Defaults to a 10 MiB cap. Pass max_bytes=None to disable (trusted hosts).
Exceeding the cap fails the feed with MontyRuntimeError wrapping a
MemoryError. Not covered by ResourceLimits.max_memory.
The cap includes a fixed per-entry overhead (many tiny fragments).
Collected output so far.
Type: list[tuple[Literal[‘stdout’, ‘stderr’], str]]
Collect printed output as one concatenated string.
Defaults to a 10 MiB cap. Pass max_bytes=None to disable (trusted hosts).
Exceeding the cap fails the feed with MontyRuntimeError wrapping a
MemoryError. Not covered by ResourceLimits.max_memory.
Collected output so far.
Type: str
How MontyTypingError diagnostics are rendered — ty’s diagnostic formats.
Picked by checkout(type_check_format=...), not on the raised error: the type
checker runs inside the worker and its structured diagnostics never leave it,
so only the already-rendered text crosses the wire.
Type: TypeAlias Default: Literal['full', 'concise', 'azure', 'json', 'jsonlines', 'rdjson', 'pylint', 'gitlab', 'github']
Print sink accepted by feed_run / feed_start / load_snapshot.
Type: TypeAlias Default: Callable[[Literal['stdout', 'stderr'], str], None] | CollectStreams | CollectString