How to Fix 'async event loop error in production' in AutoGen (Python)

If you’re seeing RuntimeError: This event loop is already running or RuntimeError: There is no current event loop in thread, you’re hitting a mismatch between AutoGen’s async APIs and the way your app is starting Python. In production, this usually shows up when AutoGen runs inside FastAPI, Celery, Jupyter, Streamlit, or any environment that already manages its own event loop.

The fix is usually not in AutoGen itself. It’s in how you call AssistantAgent, UserProxyAgent, or a_initiate_chat() from sync vs async code.

The Most Common Cause

The #1 cause is calling an async AutoGen method from synchronous code using asyncio.run() in a place where an event loop already exists.

Typical offenders:

• •FastAPI request handlers
• •Jupyter notebooks
• •Streamlit callbacks
• •Background workers that already run an asyncio loop

Here’s the broken pattern:

# broken.py
import asyncio
from autogen import AssistantAgent, UserProxyAgent

assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4o-mini", "api_key": "YOUR_KEY"},
)

user = UserProxyAgent(name="user")

def handle_request():
    # RuntimeError: asyncio.run() cannot be called from a running event loop
    result = asyncio.run(user.a_initiate_chat(assistant, message="Draft a policy summary"))
    return result

And here’s the fixed version:

# fixed.py
from autogen import AssistantAgent, UserProxyAgent

assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4o-mini", "api_key": "YOUR_KEY"},
)

user = UserProxyAgent(name="user")

async def handle_request():
    result = await user.a_initiate_chat(
        assistant,
        message="Draft a policy summary"
    )
    return result

If your framework expects sync handlers, push the async boundary up one level instead of wrapping every call with asyncio.run().

Other Possible Causes

1) Running AutoGen in a thread without setting an event loop

You’ll see errors like:

• •RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'

Broken:

import threading
import asyncio

def worker():
    loop = asyncio.get_event_loop()  # fails in many threads on Python 3.10+

Fixed:

import threading
import asyncio

def worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

If you use threads for request fan-out, create and bind the loop explicitly.

2) Mixing sync AutoGen APIs with async-only usage

AutoGen has both sync and async paths. If you call async methods but treat them like sync ones, you’ll get confusing failures later.

Broken:

# wrong: missing await
chat_result = user.a_initiate_chat(assistant, message="Check claim notes")
print(chat_result)

Fixed:

chat_result = await user.a_initiate_chat(assistant, message="Check claim notes")
print(chat_result)

If you are inside plain Python script code, wrap it once at the top level:

import asyncio

async def main():
    await user.a_initiate_chat(assistant, message="Check claim notes")

if __name__ == "__main__":
    asyncio.run(main())

3) Calling AutoGen from FastAPI sync endpoints

FastAPI supports both sync and async routes. If your route is sync and you try to drive async AutoGen code from it, you’ll eventually hit loop issues.

Broken:

from fastapi import FastAPI

app = FastAPI()

@app.get("/chat")
def chat():
    return asyncio.run(user.a_initiate_chat(assistant, message="Hello"))

Fixed:

from fastapi import FastAPI

app = FastAPI()

@app.get("/chat")
async def chat():
    return await user.a_initiate_chat(assistant, message="Hello")

Use async def all the way down for request paths that touch AutoGen.

4) Jupyter / notebook runtime conflicts

In notebooks, the kernel already runs an event loop. Calling asyncio.run() usually fails with:

• •RuntimeError: asyncio.run() cannot be called from a running event loop

Broken:

import asyncio

asyncio.run(user.a_initiate_chat(assistant, message="Summarize this document"))

Fixed:

await user.a_initiate_chat(assistant, message="Summarize this document")

If you need notebook support in development but production runs as a service, keep both paths separate and don’t copy notebook code into your app server.

How to Debug It

1. •

   Read the exact exception text

   • •This event loop is already running means nested loop usage.
   • •There is no current event loop in thread means thread context issue.
   • •Missing output with no exception often means you forgot await.
2. •

   Find where AutoGen is called

   • •Search for a_initiate_chat, asyncio.run, get_event_loop, and thread pools.
   • •Check whether the caller is sync or async.
   • •In FastAPI/Starlette/Django ASGI apps, assume there is already a running loop.
3. •

   Print the execution context

   import asyncio
   print("running:", asyncio.get_running_loop())
   

   If this raises inside your handler, you are not in an active async context. If it succeeds and you still use asyncio.run(), that’s your bug.

4. •

   Reduce to one minimal call path

   • •Remove Celery tasks, background threads, wrappers, and retries.
   • •Call one AutoGen method from one entrypoint.
   • •Reintroduce infrastructure until the failure returns.

Prevention

• •Use one rule: sync entrypoints call sync APIs; async entrypoints await async APIs.
• •In ASGI apps, make routes that touch AutoGen async def end to end.
• •In worker threads or task runners, create and bind an explicit event loop before calling async code.

If you want a stable production setup with AutoGen, treat the event loop as infrastructure. Don’t hide it behind helper functions and don’t let multiple frameworks fight over ownership of it.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.