Monty

The Monty class is the main interface for creating and running a sandboxed Python interpreter. It parses Python code on initialization and can execute it multiple times with different inputs, separating parsing cost from execution.

Constructor

Monty(
    code: str,
    *,
    script_name: str = 'main.py',
    inputs: list[str] | None = None,
    type_check: bool = False,
    type_check_stubs: str | None = None,
    dataclass_registry: list[type] | None = None,
) -> Monty

Create a new Monty interpreter by parsing the given code.

code

str

required

Python code to execute

script_name

str

default:"'main.py'"

Name used in tracebacks and error messages

inputs

list[str] | None

List of input variable names available in the code

type_check

bool

default:"False"

Whether to perform type checking on the code

type_check_stubs

str | None

Optional code to prepend before type checking, e.g. with input variable declarations or external function signatures

dataclass_registry

list[type] | None

Optional list of dataclass types to register for proper isinstance() support on output

Raises:

MontySyntaxError: If the code cannot be parsed
MontyTypingError: If type_check is True and type errors are found

Example

import pydantic_monty

# Simple code with no inputs
m = pydantic_monty.Monty('1 + 2')
print(m.run())
# Output: 3

# Code with input variables
m = pydantic_monty.Monty('x * y', inputs=['x', 'y'])
print(m.run(inputs={'x': 2, 'y': 3}))
# Output: 6

Methods

run()

def run(
    self,
    *,
    inputs: dict[str, Any] | None = None,
    limits: ResourceLimits | None = None,
    external_functions: dict[str, Callable[..., Any]] | None = None,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    os: Callable[[OsFunction, tuple[Any, ...]], Any] | None = None,
) -> Any

Execute the code and return the result. The GIL is released allowing parallel execution.

inputs

dict[str, Any] | None

Dict of input variable values (must match names from constructor)

limits

ResourceLimits | None

Optional resource limits configuration

external_functions

dict[str, Callable[..., Any]] | None

Dict of external function callbacks

print_callback

Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

Callable[[OsFunction, tuple[Any, ...]], Any] | None

Optional callback for OS calls. Called with (function_name, args) where function_name is like 'Path.exists' and args is a tuple of arguments. Must return the appropriate value for the OS function (e.g., bool for exists(), stat_result for stat()).

The result of the last expression in the code

Raises:

MontyRuntimeError: If the code raises an exception during execution

Example with External Functions

import pydantic_monty

# Code that calls an external function
m = pydantic_monty.Monty('double(x)', inputs=['x'])

# Provide the external function implementation
result = m.run(
    inputs={'x': 5},
    external_functions={'double': lambda x: x * 2}
)
print(result)
# Output: 10

Example with Resource Limits

import pydantic_monty

m = pydantic_monty.Monty('x + y', inputs=['x', 'y'])

limits = pydantic_monty.ResourceLimits(max_duration_secs=1.0)
result = m.run(inputs={'x': 1, 'y': 2}, limits=limits)
print(result)
# Output: 3

start()

def start(
    self,
    *,
    inputs: dict[str, Any] | None = None,
    limits: ResourceLimits | None = None,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
) -> FunctionSnapshot | NameLookupSnapshot | FutureSnapshot | MontyComplete

Start the code execution and return a progress object, or completion. This allows you to iteratively run code and pause/resume whenever an external function is called. The GIL is released allowing parallel execution.

inputs

dict[str, Any] | None

Dict of input variable values (must match names from constructor)

limits

ResourceLimits | None

Optional resource limits configuration

print_callback

Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

FunctionSnapshot if an external function call is pending
NameLookupSnapshot if more futures need to be resolved
FutureSnapshot if futures need to be resolved
MontyComplete if execution finished without external calls

Raises:

MontyRuntimeError: If the code raises an exception during execution

Example with Iterative Execution

import pydantic_monty

code = """
data = fetch(url)
len(data)
"""

m = pydantic_monty.Monty(code, inputs=['url'])

# Start execution - pauses when fetch() is called
result = m.start(inputs={'url': 'https://example.com'})

print(type(result))
# Output: <class 'pydantic_monty.FunctionSnapshot'>

print(result.function_name)
# Output: fetch

print(result.args)
# Output: ('https://example.com',)

# Perform the actual fetch, then resume with the result
result = result.resume(return_value='hello world')

print(type(result))
# Output: <class 'pydantic_monty.MontyComplete'>

print(result.output)
# Output: 11

type_check()

def type_check(self, prefix_code: str | None = None) -> None

Perform static type checking on the code. Analyzes the code for type errors without executing it. This uses a subset of Python’s type system supported by Monty.

prefix_code

str | None

Optional code to prepend before type checking, e.g. with input variable declarations or external function signatures

Raises:

MontyTypingError: If type errors are found. Use .display(format, color) on the exception to render the diagnostics in different formats.
RuntimeError: If the type checking infrastructure fails internally.

dump()

def dump(self) -> bytes

Serialize the Monty instance to a binary format. The serialized data can be stored and later restored with Monty.load(). This allows caching parsed code to avoid re-parsing on subsequent runs.

Bytes containing the serialized Monty instance

Raises:

ValueError: If serialization fails

Example

import pydantic_monty

# Serialize parsed code to avoid re-parsing
m = pydantic_monty.Monty('x + 1', inputs=['x'])
data = m.dump()

# Later, restore and run
m2 = pydantic_monty.Monty.load(data)
print(m2.run(inputs={'x': 41}))
# Output: 42

load() (static method)

@staticmethod
def load(
    data: bytes,
    *,
    dataclass_registry: list[type] | None = None,
) -> Monty

Deserialize a Monty instance from binary format.

data

bytes

required

The serialized Monty data from dump()

dataclass_registry

list[type] | None

Optional list of dataclass types to register for proper isinstance() support on output

A new Monty instance

Raises:

ValueError: If deserialization fails

register_dataclass()

def register_dataclass(self, cls: type) -> None

Register a dataclass type for proper isinstance() support on output. When a dataclass passes through Monty and is returned, it normally becomes an UnknownDataclass. By registering the original type, we can use it to instantiate a real instance of that dataclass.

cls

type

required

The dataclass type to register

Raises:

TypeError: If the argument is not a dataclass type

Async Helper Function

run_monty_async()

async def run_monty_async(
    monty_runner: Monty,
    *,
    inputs: dict[str, Any] | None = None,
    external_functions: dict[str, Callable[..., Any]] | None = None,
    limits: ResourceLimits | None = None,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    os: AbstractOS | None = None,
) -> Any

Run a Monty script with async external functions and optional OS access. This function provides a convenient way to run Monty code that uses both async external functions and filesystem operations.

monty_runner

Monty

required

The Monty instance to execute

inputs

dict[str, Any] | None

Dictionary of input variable values

external_functions

dict[str, Callable[..., Any]] | None

Dictionary of external functions (can be sync or async)

limits

ResourceLimits | None

Optional resource limits configuration

print_callback

Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

AbstractOS | None

Optional OS access handler for filesystem operations (e.g., OSAccess instance)

The output of the Monty script

Example

import asyncio
import pydantic_monty

async def async_fetch(url):
    # Simulated async operation
    await asyncio.sleep(0.1)
    return f"Data from {url}"

async def main():
    code = "result = fetch('https://example.com')"
    m = pydantic_monty.Monty(code)
    
    output = await pydantic_monty.run_monty_async(
        m,
        external_functions={'fetch': async_fetch}
    )
    print(output)

asyncio.run(main())

MontyRepl

The MontyRepl class provides an incremental, no-replay REPL (Read-Eval-Print Loop) session. Each feed() call compiles and executes only the provided snippet against preserved heap/global state, unlike traditional REPLs that replay all history.

create() (static method)

@staticmethod
def create(
    code: str,
    *,
    script_name: str = 'main.py',
    inputs: list[str] | None = None,
    start_inputs: dict[str, Any] | None = None,
    limits: ResourceLimits | None = None,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    dataclass_registry: list[type] | None = None,
) -> tuple[MontyRepl, Any]

Create a REPL session directly from source code. Returns a tuple of (repl, output) where output is the initial execution result.

code

str

required

Initial Python code to execute

script_name

str

default:"'main.py'"

Name used in tracebacks and error messages

inputs

list[str] | None

List of input variable names available in the code

start_inputs

dict[str, Any] | None

Initial input values for the session

limits

ResourceLimits | None

Optional resource limits configuration

print_callback

Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

dataclass_registry

list[type] | None

Optional list of dataclass types to register

A tuple containing the REPL session and the initial execution result

Raises:

MontySyntaxError: If the code cannot be parsed
MontyRuntimeError: If the code raises an exception during execution

script_name (property)

@property
def script_name(self) -> str

Returns the name of the script being executed.

feed()

def feed(
    self,
    code: str,
    *,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
) -> Any

Execute one incremental snippet and return its output. The snippet is compiled and executed against the current session state without replaying previous code.

code

str

required

Python code snippet to execute

print_callback

Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

The result of the code snippet

Raises:

MontyRuntimeError: If the code raises an exception during execution

Example

import pydantic_monty

# Create REPL with initial state
repl, initial = pydantic_monty.MontyRepl.create("x = 1")
print(initial)
# Output: None (assignment has no return value)

# Feed additional snippets
result = repl.feed("x + 10")
print(result)
# Output: 11

# State is preserved
result = repl.feed("x = x * 2; x")
print(result)
# Output: 2

dump()

def dump(self) -> bytes

Serialize the REPL session to bytes. The serialized data can be stored and later restored with MontyRepl.load().

Bytes containing the serialized REPL session

load() (static method)

@staticmethod
def load(
    data: bytes,
    *,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    dataclass_registry: list[type] | None = None,
) -> MontyRepl

Restore a REPL session from bytes.

data

bytes

required

The serialized REPL data from dump()

print_callback

Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

dataclass_registry

list[type] | None

Optional list of dataclass types to register

A new MontyRepl instance

Raises:

ValueError: If deserialization fails

Python

JavaScript/TypeScript

Rust

Constructor

Example

Methods

run()

Example with External Functions

Example with Resource Limits

start()

Example with Iterative Execution

type_check()

dump()

Example

load() (static method)

register_dataclass()

Async Helper Function

run_monty_async()

Example

MontyRepl

create() (static method)

script_name (property)

feed()

Example

dump()

load() (static method)

Python

JavaScript/TypeScript

Rust

Documentation Index

​Constructor

​Example

​Methods

​run()

​Example with External Functions

​Example with Resource Limits

​start()

​Example with Iterative Execution

​type_check()

​dump()

​Example

​load() (static method)

​register_dataclass()

​Async Helper Function

​run_monty_async()

​Example

​MontyRepl

​create() (static method)

​script_name (property)

​feed()

​Example

​dump()

​load() (static method)

Constructor

Example

Methods

run()

Example with External Functions

Example with Resource Limits

start()

Example with Iterative Execution

type_check()

dump()

Example

load() (static method)

register_dataclass()

Async Helper Function

run_monty_async()

Example

MontyRepl

create() (static method)

script_name (property)

feed()

Example

dump()

load() (static method)