Async Python for Agent Developers

Why Async for Agents?

Async Basics: async def and await

import asyncio

async def fetch_data(source: str) -> str:
    print(f'Starting fetch from {source}')
    await asyncio.sleep(1)  # Simulates a network call
    print(f'Finished fetch from {source}')
    return f'Data from {source}'

async def main():
    # Sequential: takes 2 seconds total
    result1 = await fetch_data('source-A')
    result2 = await fetch_data('source-B')
    print('Sequential results:', result1, result2)

asyncio.run(main())

# asyncio.run() starts the event loop and runs main()
# It is the entry point for async programs

The Event Loop

import asyncio

async def task_a():
    print('Task A: start')
    await asyncio.sleep(2)
    print('Task A: done')

async def task_b():
    print('Task B: start')
    await asyncio.sleep(1)
    print('Task B: done')

async def main():
    # asyncio.gather runs both tasks concurrently
    # Total time: ~2 seconds (not 3)
    await asyncio.gather(task_a(), task_b())
    print('Both tasks complete')

# Expected output order:
# Task A: start
# Task B: start
# Task B: done   <- after 1s
# Task A: done   <- after 2s
# Both tasks complete
asyncio.run(main())

Coroutines vs Threads

import asyncio
import threading
import time

# Thread approach: multiple OS threads
def thread_worker(name):
    print(f'Thread {name}: start')
    time.sleep(1)  # Blocks the thread
    print(f'Thread {name}: done')

threads = [threading.Thread(target=thread_worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print('---')

# Coroutine approach: single-threaded event loop
async def coro_worker(name):
    print(f'Coro {name}: start')
    await asyncio.sleep(1)  # Suspends, does NOT block other coroutines
    print(f'Coro {name}: done')

async def main():
    await asyncio.gather(*[coro_worker(i) for i in range(3)])

asyncio.run(main())
# Both approaches run 3 tasks in ~1 second total, but coroutines use one thread

asyncio.run() Entry Point

import asyncio

async def agent_main():
    print('Agent starting')
    # All agent async work goes here
    results = await asyncio.gather(
        asyncio.sleep(0.1),  # Simulated LLM call
        asyncio.sleep(0.1),  # Simulated DB query
    )
    print('Agent done')
    return 'complete'

# Run the agent
result = asyncio.run(agent_main())
print('Result:', result)

# WRONG: calling asyncio.run() inside an already-running event loop
# In Jupyter notebooks, use: await agent_main() directly
# Or: nest_asyncio.apply() then asyncio.run()

Common Mistake: Blocking in Async Context

import asyncio
import time
import httpx

# WRONG: blocks the event loop
async def bad_agent_step():
    time.sleep(2)          # Blocks all other coroutines
    # requests.get(url)    # Also blocks - do NOT use requests in async code
    return 'done'

# RIGHT: use async equivalents
async def good_agent_step():
    await asyncio.sleep(2)  # Suspends, other coroutines can run
    
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
    return response.text

# For CPU-intensive work: use run_in_executor
import concurrent.futures

async def cpu_intensive_step(data: str):
    loop = asyncio.get_event_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, expensive_cpu_fn, data)
    return result

def expensive_cpu_fn(data):
    # CPU-bound work runs in separate process
    return data.upper()

print('Blocking vs non-blocking patterns demonstrated')

Forgot await: Silent Bug

import asyncio

async def get_answer() -> str:
    await asyncio.sleep(0.1)
    return 'The answer is 42'

async def bad_call():
    result = get_answer()   # WRONG: forgot await
    print(type(result))     # <class 'coroutine'> - not a string!
    # Using result as a string here causes AttributeError or wrong behavior
    return result

async def good_call():
    result = await get_answer()  # CORRECT
    print(type(result))           # <class 'str'>
    return result

async def main():
    bad = await bad_call()
    print('Bad result:', bad)    # coroutine object, not the string
    
    good = await good_call()
    print('Good result:', good)  # 'The answer is 42'
    
    # Clean up the uncollected coroutine
    if asyncio.iscoroutine(bad):
        bad.close()

asyncio.run(main())

Nested Event Loop Problem

import asyncio

async def my_agent_coroutine():
    await asyncio.sleep(0.1)
    return 'done'

# In FastAPI or other async frameworks, the event loop is already running
# Use await directly in async endpoints:
async def fastapi_endpoint():
    # WRONG inside async context:
    # result = asyncio.run(my_agent_coroutine())  # RuntimeError!
    
    # CORRECT: just await
    result = await my_agent_coroutine()
    return result

# In Jupyter notebooks: install nest_asyncio
# import nest_asyncio
# nest_asyncio.apply()
# Then asyncio.run() works

# Detect if running in event loop:
def run_agent(coro):
    try:
        loop = asyncio.get_running_loop()
        # Already in async context
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor() as pool:
            future = pool.submit(asyncio.run, coro)
            return future.result()
    except RuntimeError:
        # No running loop
        return asyncio.run(coro)

print('Nested event loop solution defined')

asyncio.create_task

import asyncio

async def background_job(job_id: int) -> str:
    await asyncio.sleep(0.5)
    return f'Job {job_id} completed'

async def main():
    # Start all tasks without waiting
    task1 = asyncio.create_task(background_job(1))
    task2 = asyncio.create_task(background_job(2))
    task3 = asyncio.create_task(background_job(3))
    
    # Do other work while tasks run
    print('Tasks started, doing other work...')
    await asyncio.sleep(0.1)
    print('Other work done')
    
    # Now wait for all tasks
    results = await asyncio.gather(task1, task2, task3)
    print('All results:', results)
    
    # Or wait for the first to complete
    task_a = asyncio.create_task(background_job(4))
    task_b = asyncio.create_task(background_job(5))
    done, pending = await asyncio.wait([task_a, task_b], return_when=asyncio.FIRST_COMPLETED)
    for t in pending:
        t.cancel()  # Cancel remaining tasks
    print('First result:', done.pop().result())

asyncio.run(main())

Async Context Managers

import asyncio
import httpx

async def fetch_multiple_urls(urls: list) -> list:
    # async with ensures the client is properly closed
    async with httpx.AsyncClient(timeout=10.0) as client:
        # Fetch all URLs concurrently
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        results = []
        for url, response in zip(urls, responses):
            if isinstance(response, Exception):
                results.append({'url': url, 'error': str(response)})
            else:
                results.append({'url': url, 'status': response.status_code})
        return results

# Async generators for streaming
async def stream_agent_events():
    events = ['thinking', 'searching', 'generating', 'done']
    for event in events:
        await asyncio.sleep(0.2)  # Simulate event arrival
        yield event

async def consume_stream():
    async for event in stream_agent_events():
        print(f'Event: {event}')

asyncio.run(consume_stream())

Error Handling in Async Code

import asyncio

async def might_fail(task_id: int) -> str:
    await asyncio.sleep(0.1)
    if task_id == 2:
        raise ValueError(f'Task {task_id} failed')
    return f'Task {task_id} succeeded'

async def robust_gather():
    tasks = [might_fail(i) for i in range(1, 5)]
    
    # return_exceptions=True: exceptions are returned as values, not raised
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    successes = []
    failures = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            failures.append({'task': i + 1, 'error': str(result)})
        else:
            successes.append(result)
    
    print(f'Succeeded: {len(successes)}, Failed: {len(failures)}')
    print('Failures:', failures)
    return successes, failures

asyncio.run(robust_gather())

Knowledge Check: Async Python

Async Python Summary