Skip to main content
Monty fully supports Python’s async/await syntax and provides helpers for running async code from both sync and async host environments.

​
Running Async Code

Monty can execute async Python code using the run_monty_async() helper function: 
import asyncio
import pydantic_monty
from pydantic_monty import run_monty_async

code = """
async def fetch_data():
    return "data"

await fetch_data()
"""

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(m)
    print(result)  # "data"

asyncio.run(main())

​
Async External Functions

You can provide async external functions that will be awaited automatically: 
import asyncio
from pydantic_monty import run_monty_async

code = "await fetch_data()"

async def fetch_data():
    await asyncio.sleep(0.1)
    return "async result"

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(
        m,
        external_functions={'fetch_data': fetch_data}
    )
    print(result)  # "async result"

asyncio.run(main())
run_monty_async() automatically handles both sync and async external functions. Sync functions are called directly, while async functions are awaited.

​
Mixed Sync and Async Functions

You can mix sync and async external functions in the same execution: 
import asyncio
from pydantic_monty import run_monty_async

code = """
sync_val = sync_func()
async_val = await async_func()
sync_val + async_val
"""

def sync_func():
    return 10

async def async_func():
    await asyncio.sleep(0.001)
    return 5

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(
        m,
        external_functions={
            'sync_func': sync_func,
            'async_func': async_func
        }
    )
    print(result)  # 15

asyncio.run(main())

​
Asyncio.gather for Concurrent Execution

Monty supports asyncio.gather() for running multiple async operations concurrently: 
import asyncio
from pydantic_monty import run_monty_async

code = """
import asyncio
await asyncio.gather(fetch_a(), fetch_b())
"""

async def fetch_a():
    await asyncio.sleep(0.01)
    return 'a'

async def fetch_b():
    await asyncio.sleep(0.005)
    return 'b'

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(
        m,
        external_functions={'fetch_a': fetch_a, 'fetch_b': fetch_b}
    )
    print(result)  # ['a', 'b']

asyncio.run(main())

​
Complex Async Patterns

Monty handles nested asyncio.gather() with external functions: 
import asyncio
from pydantic_monty import run_monty_async

code = """
import asyncio

async def get_city_weather(city_name: str):
    coords = await get_lat_lng(location_description=city_name)
    lat, lng = coords['lat'], coords['lng']
    temp_task = get_temp(lat=lat, lng=lng)
    desc_task = get_weather_description(lat=lat, lng=lng)
    temp, desc = await asyncio.gather(temp_task, desc_task)
    return {
        'city': city_name,
        'temp': temp,
        'description': desc
    }

async def main():
    cities = ['London', 'Paris', 'Tokyo']
    results = await asyncio.gather(*(get_city_weather(city) for city in cities))
    return results

await main()
"""

async def get_lat_lng(location_description: str):
    # Return mock coordinates
    coords = {
        'London': {'lat': 51.5, 'lng': -0.1},
        'Paris': {'lat': 48.9, 'lng': 2.3},
        'Tokyo': {'lat': 35.7, 'lng': 139.7},
    }
    return coords[location_description]

async def get_temp(lat: float, lng: float):
    # Return mock temperature
    return 20.0

async def get_weather_description(lat: float, lng: lng):
    # Return mock description
    return 'Sunny'

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(
        m,
        external_functions={
            'get_lat_lng': get_lat_lng,
            'get_temp': get_temp,
            'get_weather_description': get_weather_description,
        }
    )
    print(result)
    # [{'city': 'London', 'temp': 20.0, 'description': 'Sunny'}, ...]

asyncio.run(main())

​
Manual Async Handling with FutureSnapshot

For advanced use cases, you can manually handle async execution using FutureSnapshot: 
import pydantic_monty

code = "await foobar(1, 2)"
m = pydantic_monty.Monty(code)

# Start execution
progress = m.start()
assert isinstance(progress, pydantic_monty.FunctionSnapshot)
assert progress.function_name == 'foobar'

call_id = progress.call_id

# Resume with future (indicates async operation)
progress = progress.resume(future=...)
assert isinstance(progress, pydantic_monty.FutureSnapshot)

# Resume with return value for the pending call
result = progress.resume({call_id: {'return_value': 3}})
assert result.output == 3

​
Multiple Pending Futures

code = """
import asyncio
await asyncio.gather(foo(1), bar(2))
"""

m = pydantic_monty.Monty(code)

# Start and collect call IDs
progress = m.start()
foo_id = progress.call_id
progress = progress.resume(future=...)

bar_id = progress.call_id
progress = progress.resume(future=...)

# Now we have a FutureSnapshot with multiple pending calls
assert isinstance(progress, pydantic_monty.FutureSnapshot)
assert set(progress.pending_call_ids) == {foo_id, bar_id}

# Resume all at once
result = progress.resume({
    foo_id: {'return_value': 3},
    bar_id: {'return_value': 4}
})
assert result.output == [3, 4]
Most users should use run_monty_async() instead of manually handling FutureSnapshot. Manual handling is only needed for advanced patterns like custom async schedulers.

​
Async with OS Access

run_monty_async() supports OS access for file operations: 
import asyncio
from pydantic_monty import run_monty_async, OSAccess, MemoryFile

fs = OSAccess([MemoryFile('/data.txt', content='test data')])

async def process(text: str) -> str:
    return text.upper()

code = """
from pathlib import Path
content = Path('/data.txt').read_text()
await process(content)
"""

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(
        m,
        external_functions={'process': process},
        os=fs
    )
    print(result)  # 'TEST DATA'

asyncio.run(main())

​
Error Handling in Async Code

Exceptions from async external functions propagate normally: 
import asyncio
from pydantic_monty import run_monty_async
import pydantic_monty

code = "await async_fail()"

async def async_fail():
    await asyncio.sleep(0.001)
    raise RuntimeError('async error')

m = pydantic_monty.Monty(code)

async def main():
    try:
        await run_monty_async(
            m,
            external_functions={'async_fail': async_fail}
        )
    except pydantic_monty.MontyRuntimeError as e:
        inner = e.exception()
        print(type(inner))  # <class 'RuntimeError'>
        print(inner.args[0])  # 'async error'

asyncio.run(main())

​
Exception Handling in Async Code

Async exceptions can be caught with try/except: 
code = """
try:
    await fail()
except ValueError:
    caught = True
caught
"""

async def fail():
    raise ValueError('caught error')

m = pydantic_monty.Monty(code)

async def main():
    result = await run_monty_async(
        m,
        external_functions={'fail': fail}
    )
    print(result)  # True

asyncio.run(main())

​
JavaScript Async Example

import { Monty, runMontyAsync } from '@pydantic/monty'

const code = `
import asyncio

async def fetch_all():
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3)
    )
    return results

await fetch_all()
`

const m = new Monty(code)

const result = await runMontyAsync(m, {
  externalFunctions: {
    fetch_user: async (id: number) => {
      const response = await fetch(`https://api.example.com/users/${id}`)
      return response.json()
    }
  }
})

console.log(result)

​
Performance Considerations

1

Use asyncio.gather for concurrency

When you need to call multiple external functions, use asyncio.gather() to run them concurrently instead of sequentially:
# Sequential (slow)
a = await fetch_a()
b = await fetch_b()

# Concurrent (fast)
a, b = await asyncio.gather(fetch_a(), fetch_b())
2

Prefer async for I/O

Use async external functions for I/O-bound operations (network, file, database):
async def fetch_from_api(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()
3

Keep sync for CPU-bound work

Use sync functions for CPU-bound operations that don’t benefit from async:
def compute_hash(data: str) -> str:
    return hashlib.sha256(data.encode()).hexdigest()
Async functions have overhead. Only use async when you have actual I/O operations that benefit from concurrency.

​
Best Practices

  1. Use run_monty_async() for simplicity: It handles both sync and async functions automatically
  2. Provide type stubs for async functions: Help type checking understand async function signatures
  3. Handle exceptions properly: Async exceptions propagate the same way as sync exceptions
  4. Use asyncio.gather() wisely: Only for independent operations that can run concurrently
  5. Test error paths: Make sure your error handling works for both sync and async paths

​
When to Use Async

Use async execution when:
  • Your external functions perform I/O (network requests, file operations, database queries)
  • You need to call multiple external functions concurrently
  • You’re building a web service or API that uses async frameworks (FastAPI, aiohttp)
Stick with sync when:
  • Your code has no external function calls
  • External functions are CPU-bound (computation, parsing)
  • You’re building a simple script or tool

​
Next Steps

Iterative Execution

Learn about manual async handling with FutureSnapshot

Error Handling

Handle exceptions in async code

Build docs developers (and LLMs) love

Get started for freeTalk to us