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

What the error means

async event loop error in LangGraph usually means you tried to run an async graph from the wrong execution context. In practice, it shows up when you call asyncio.run() inside an already-running loop, mix sync and async graph APIs, or invoke graph execution from a framework that already owns the event loop.

The exact traceback often includes one of these:

• •RuntimeError: asyncio.run() cannot be called from a running event loop
• •RuntimeError: This event loop is already running
• •RuntimeError: Event loop is closed

The Most Common Cause

The #1 cause is wrapping LangGraph async execution with asyncio.run() inside code that is already async.

This happens a lot in:

• •FastAPI endpoints
• •Jupyter notebooks
• •Streamlit callbacks
• •Any app using async def handlers

Broken vs fixed pattern

# BROKEN
import asyncio
from langgraph.graph import StateGraph, START, END

async def call_graph(graph, state):
    # This will fail if you're already inside an event loop
    result = asyncio.run(graph.ainvoke(state))
    return result


# FIXED
import asyncio
from langgraph.graph import StateGraph, START, END

async def call_graph(graph, state):
    # Correct: await the coroutine in the current loop
    result = await graph.ainvoke(state)
    return result

If you are in sync code, use the sync API:

# Sync entrypoint
result = graph.invoke(state)

If you are in async code, use the async API:

# Async entrypoint
result = await graph.ainvoke(state)

That distinction matters. LangGraph exposes both invoke() and ainvoke() for a reason.

Other Possible Causes

1) Mixing sync and async nodes incorrectly

If one node returns a coroutine but your graph is being executed with invoke(), you can get event-loop-related failures.

# BROKEN: async node used with sync execution path
async def fetch_user(state):
    return {"user": "alice"}

graph.invoke({"input": "hi"})  # wrong if graph expects async execution


# FIXED: use ainvoke for async graphs
result = await graph.ainvoke({"input": "hi"})

Rule of thumb:

• •invoke() for fully synchronous graphs
• •ainvoke() when any node is async

2) Running inside Jupyter/IPython with asyncio.run()

Jupyter already runs an event loop. Calling asyncio.run() there triggers:

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

# BROKEN in notebook cells
import asyncio

result = asyncio.run(graph.ainvoke({"input": "hello"}))


# FIXED in notebook cells
result = await graph.ainvoke({"input": "hello"})

If you must stay sync in notebooks, keep the entire flow synchronous and use graph.invoke().

3) Using LangGraph inside FastAPI sync endpoints

FastAPI sync routes run in a threadpool, but your app may still end up calling async code incorrectly through wrappers.

# BROKEN
from fastapi import FastAPI

app = FastAPI()

@app.get("/chat")
def chat():
    return asyncio.run(graph.ainvoke({"input": "hi"}))


# FIXED
from fastapi import FastAPI

app = FastAPI()

@app.get("/chat")
async def chat():
    return await graph.ainvoke({"input": "hi"})

If your route is declared with def, make sure you are not forcing async execution through nested loops.

4) Closing the loop too early in worker scripts

Some scripts manually create and close loops around long-lived graph operations. That can lead to:

• •RuntimeError: Event loop is closed

# BROKEN
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    result = loop.run_until_complete(graph.ainvoke({"input": "hi"}))
finally:
    loop.close()


# FIXED
async def main():
    result = await graph.ainvoke({"input": "hi"})
    print(result)

asyncio.run(main())

For production services, let your framework manage the lifecycle unless you have a strong reason not to.

How to Debug It

1. •

   Read the exact traceback

   • •If you see asyncio.run() cannot be called from a running event loop, look for nested asyncio.run().
   • •If you see This event loop is already running, check for notebook or framework context.
   • •If you see Event loop is closed, inspect shutdown logic.
2. •

   Check which API you are calling

   • •Use graph.invoke() only from sync code.
   • •Use await graph.ainvoke() only from async code.
   • •Do not wrap one inside the other unless you know exactly why.
3. •

   Inspect your runtime

   • •Jupyter/IPython: always assume a loop exists.
   • •FastAPI/Starlette: prefer async route handlers.
   • •Celery/CLI scripts: keep them fully sync or fully async, not mixed.
4. •

   Search for hidden wrappers

   • •Look for utility functions that call:
     • •asyncio.run(...)
     • •loop.run_until_complete(...)
     • •custom event-loop setup/teardown
   • •The bug is often one layer above where the traceback points.

Prevention

• •Pick one execution model per code path:
  • •all-sync with invoke()
  • •all-async with ainvoke()
• •In web apps, make route handlers async if they call LangGraph asynchronously.
• •In notebooks and REPLs, prefer direct await instead of creating new loops.

If you want fewer runtime surprises, standardize this rule across the team:

• •Never call asyncio.run() from application code that might already be inside an event loop

That single rule prevents most LangGraph event-loop failures before they ship.

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.