Skip to main content
Iterative execution allows you to pause Monty execution at each external function call, giving you complete control over when and how those functions are executed.

​
Why Iterative Execution?

Instead of providing all external functions upfront with run(), iterative execution lets you:
  • Inspect function calls before executing them (useful for logging, security, or rate limiting)
  • Handle async operations manually
  • Serialize and resume execution across process boundaries
  • Implement custom execution strategies (e.g., batching, caching)

​
Basic Flow

Use start() to begin execution and resume() to continue after each external function call:
1

Start 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'})
2

Inspect the function call

print(type(result))  # <class 'pydantic_monty.FunctionSnapshot'>
print(result.function_name)  # 'fetch'
print(result.args)  # ('https://example.com',)
print(result.kwargs)  # {}
3

Resume with return value

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

print(type(result))  # <class 'pydantic_monty.MontyComplete'>
print(result.output)  # 11

​
Snapshot Types

Iterative execution returns different types depending on the execution state:

​
FunctionSnapshot

Returned when execution pauses at an external function call: 
m = pydantic_monty.Monty('func(1, x="hello")')
progress = m.start()

if isinstance(progress, pydantic_monty.FunctionSnapshot):
    print(progress.script_name)    # 'main.py'
    print(progress.function_name)  # 'func'
    print(progress.args)           # (1,)
    print(progress.kwargs)         # {'x': 'hello'}

​
MontyComplete

Returned when execution finishes: 
m = pydantic_monty.Monty('42')
result = m.start()

if isinstance(result, pydantic_monty.MontyComplete):
    print(result.output)  # 42

​
FutureSnapshot

Returned when async code is waiting for futures to complete (Python only): 
code = "await foobar(1, 2)"
m = pydantic_monty.Monty(code)

progress = m.start()
assert isinstance(progress, pydantic_monty.FunctionSnapshot)

progress = progress.resume(future=...)
assert isinstance(progress, pydantic_monty.FutureSnapshot)

result = progress.resume({progress.pending_call_ids[0]: {'return_value': 3}})
assert result.output == 3

​
Handling Multiple External Calls

When code makes multiple external function calls, resume each one in sequence: 
m = pydantic_monty.Monty('a() + b()')

# First call to a()
progress = m.start()
assert progress.function_name == 'a'

# Resume with return value for a()
progress = progress.resume(return_value=10)
assert progress.function_name == 'b'

# Resume with return value for b()
result = progress.resume(return_value=5)
assert result.output == 15

​
Loop Pattern for Multiple Calls

m = pydantic_monty.Monty('c() + c() + c()')

call_count = 0
progress = m.start()

while isinstance(progress, pydantic_monty.FunctionSnapshot):
    print(f"Calling: {progress.function_name}")
    call_count += 1
    progress = progress.resume(return_value=call_count)

print(progress.output)  # 6 (1 + 2 + 3)

​
Resuming with Exceptions

You can resume with an exception instead of a return value: 
code = """
try:
    result = external_func()
except ValueError:
    caught = True
caught
"""

m = pydantic_monty.Monty(code)
progress = m.start()

# Resume with an exception
result = progress.resume(exception=ValueError('test error'))
print(result.output)  # True
If you resume with an exception that isn’t caught by the Monty code, it will propagate to your host code wrapped in a MontyRuntimeError.

​
Serialization

Both Monty and snapshot types can be serialized to bytes: 
# 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}))  # 42
Serialization enables powerful patterns like saving execution state to a database and resuming later, or distributing work across multiple processes.

​
Resume Constraints

A FunctionSnapshot can only be resumed once. Attempting to call resume() twice will raise a RuntimeError.
m = pydantic_monty.Monty('func()')
progress = m.start()

# First resume succeeds
progress.resume(return_value=1)

# Second resume raises RuntimeError
try:
    progress.resume(return_value=2)
except RuntimeError as e:
    print(e)  # 'Progress already resumed'

​
Name Lookups

In addition to FunctionSnapshot, you may receive a NameLookupSnapshot when code references an undefined variable: 
m = pydantic_monty.Monty('x = foo; x')
p = m.start()

if isinstance(p, pydantic_monty.NameLookupSnapshot):
    # Resume by providing the value for 'foo'
    p2 = p.resume(value=42)
    print(p2.output)  # 42
This feature is useful for implementing dynamic variable resolution or lazy imports.

​
JavaScript Example

import { Monty, MontySnapshot, MontyComplete } from '@pydantic/monty'

const m = new Monty('fetch_data(url)', { inputs: ['url'] })

let progress = m.start({ inputs: { url: 'https://api.example.com' } })

if (progress instanceof MontySnapshot) {
  console.log(`Calling: ${progress.functionName}(${progress.args})`)
  
  // Perform actual fetch
  const data = await fetch(progress.args[0])
  const text = await data.text()
  
  // Resume execution
  progress = progress.resume({ returnValue: text })
}

if (progress instanceof MontyComplete) {
  console.log(`Result: ${progress.output}`)
}

​
Best Practices

  1. Always check types: Use isinstance() (Python) or instanceof (JavaScript) to determine the snapshot type
  2. Handle all paths: Your code should handle both FunctionSnapshot and MontyComplete returns
  3. Log function calls: Iterative execution is perfect for audit logs and debugging
  4. Serialize strategically: Only serialize when needed (e.g., for long-running operations or persistence)
  5. Don’t reuse snapshots: Each snapshot can only be resumed once

​
Common Patterns

​
Function Call Logger

def execute_with_logging(monty: pydantic_monty.Monty, **kwargs):
    progress = monty.start(**kwargs)
    
    while isinstance(progress, pydantic_monty.FunctionSnapshot):
        print(f"[LOG] Calling {progress.function_name}{progress.args}")
        
        # Execute the function somehow
        result = execute_function(progress.function_name, progress.args, progress.kwargs)
        progress = progress.resume(return_value=result)
    
    return progress.output

​
Rate Limiting

import time

def execute_with_rate_limit(monty: pydantic_monty.Monty, delay: float = 1.0):
    progress = monty.start()
    
    while isinstance(progress, pydantic_monty.FunctionSnapshot):
        time.sleep(delay)  # Rate limit
        result = call_external_function(progress.function_name, progress.args)
        progress = progress.resume(return_value=result)
    
    return progress.output

​
Next Steps

Async Execution

Learn about async external functions and run_monty_async()

Error Handling

Handle exceptions and errors in Monty code

Build docs developers (and LLMs) love

Get started for freeTalk to us