Async Bridge - PraisonAI

The async bridge lets your tools and callbacks move between sync and async without crashing the event loop.

from praisonaiagents import Agent

agent = Agent(name="fetcher", instructions="Fetch URLs safely from sync or async tools.")
agent.start("Summarise https://example.com")

The user calls a sync tool that needs async I/O; the bridge runs the coroutine safely or surfaces a clear error if called from a running loop.

Quick Start

From a sync tool

Use run_coroutine_from_any_context to call async code from a sync tool:

from praisonaiagents import Agent
from praisonaiagents.utils.async_bridge import run_coroutine_from_any_context
import httpx

async def _fetch(url: str) -> str:
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).text[:500]

def fetch_sync(url: str) -> str:
    """Sync tool that safely reuses an async HTTP client."""
    return run_coroutine_from_any_context(_fetch(url))

agent = Agent(
    name="Researcher",
    instructions="Fetch and summarise web pages",
    tools=[fetch_sync],
)
agent.start("Summarise https://example.com")

The user runs sync code that needs async I/O; the bridge executes coroutines without nested event loops.

From an async tool

Use run_sync_in_executor to call blocking code from an async tool without blocking the event loop:

from praisonaiagents import Agent
from praisonaiagents.utils.async_bridge import run_sync_in_executor
import time

def blocking_task(duration: int) -> str:
    time.sleep(duration)
    return f"Completed after {duration} seconds"

async def async_tool(duration: int) -> str:
    """Async tool that offloads blocking work."""
    return await run_sync_in_executor(blocking_task, duration)

agent = Agent(
    name="Worker",
    instructions="Handle blocking tasks efficiently",
    tools=[async_tool],
)

Detecting the context

Use is_async_context to create dual-mode helpers:

from praisonaiagents.utils.async_bridge import is_async_context, run_coroutine_from_any_context
import httpx

async def _async_fetch(url: str) -> str:
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).text

def smart_fetch(url: str) -> str:
    """Context-aware fetch that works in both sync and async."""
    if is_async_context():
        raise RuntimeError("Use await smart_fetch_async(url) in async context")
    return run_coroutine_from_any_context(_async_fetch(url))

async def smart_fetch_async(url: str) -> str:
    """Async version for use in async contexts."""
    return await _async_fetch(url)

How It Works

The bridge probes for a running event loop using asyncio.get_running_loop(). If no loop exists, it safely creates one with asyncio.run(). If a loop is already running, it raises RuntimeError to prevent deadlocks.

Configuration Options

Option	Type	Default	Description
`timeout`	`float`	`300`	Maximum seconds to wait for coroutine completion

Common Patterns

Reusing async SDKs from sync tools

from praisonaiagents import Agent
from praisonaiagents.utils.async_bridge import run_coroutine_from_any_context
import aiofiles

async def _read_file_async(path: str) -> str:
    async with aiofiles.open(path) as f:
        return await f.read()

def read_file(path: str) -> str:
    """Sync wrapper for async file operations."""
    return run_coroutine_from_any_context(_read_file_async(path))

agent = Agent(
    name="FileReader",
    instructions="Process files efficiently",
    tools=[read_file],
)

Offloading blocking calls from async tools

import subprocess
from praisonaiagents.utils.async_bridge import run_sync_in_executor

async def run_command(cmd: str) -> str:
    """Run shell command without blocking the event loop."""
    def _run():
        return subprocess.check_output(cmd, shell=True, text=True)
    
    return await run_sync_in_executor(_run)

Context-aware dual-mode helper

from praisonaiagents.utils.async_bridge import is_async_context, run_coroutine_from_any_context

def universal_helper(data):
    """Works in both sync and async contexts."""
    if is_async_context():
        raise RuntimeError("Use await universal_helper_async(data) in async context")
    
    async def _process():
        # async processing logic
        await asyncio.sleep(0.1)
        return f"Processed: {data}"
    
    return run_coroutine_from_any_context(_process())

Best Practices

Prefer await when you're already async

Calling run_coroutine_from_any_context inside an async def raises RuntimeError by design. If you’re in a coroutine, use await instead:

# Good
async def my_async_tool():
    result = await my_coroutine()
    
# Bad - will raise RuntimeError
async def my_async_tool():
    result = run_coroutine_from_any_context(my_coroutine())

Don't wrap everything

Only wrap at the true sync/async boundary. Avoid creating unnecessary bridge calls in the middle of your call stack:

# Good - bridge at the boundary
def sync_tool():
    return run_coroutine_from_any_context(async_logic())

# Bad - unnecessary nesting
def sync_tool():
    def inner():
        return run_coroutine_from_any_context(async_logic())
    return inner()

Set a sensible timeout

The default 300 seconds is large for most use cases. Tighten for latency-critical tools:

# Good for quick operations
result = run_coroutine_from_any_context(quick_api_call(), timeout=10)

# Good for long operations
result = run_coroutine_from_any_context(model_training(), timeout=3600)

Check is_async_context() for dual-mode helpers

When building utilities that work in both sync and async contexts, check the context first:

def smart_helper():
    if is_async_context():
        raise RuntimeError("Use await smart_helper_async() in async context")
    return run_coroutine_from_any_context(async_implementation())

async def smart_helper_async():
    return await async_implementation()

Used by

The following synchronous APIs route through run_sync() and therefore honour PRAISONAI_RUN_SYNC_TIMEOUT consistently:

praisonai.bots.WebhookApproval.request_approval_sync()
praisonai.bots.HTTPApproval.request_approval_sync()
praisonai.integrations.get_available_integrations()
praisonai._run_praisonai (added PR #1681) — boots the InteractiveRuntime on the persistent background loop. If you call PraisonAI.run() from inside a running event loop, you now get a clear RuntimeError instead of a silent deadlock.
All ~77 wrapper-side run_sync call sites (gateway, a2u, mcp_server, scheduler) — see PR #1583 for the full list.
praisonai.auto.BaseAutoGenerator._structured_completion / _run_coro_sync runner threads (added in the fix for #3340) — inherit the caller’s scoped_bridge() via contextvars.copy_context().
praisonai_code.cli.features.agent_tools._run_sync (added via PR #3361) — a separate, module-local bridge in the Tier-2 praisonai-code package that ACP/LSP agent-centric tools route through. Does not import praisonai._async_bridge (C7 gate), but reads the same PRAISONAI_RUN_SYNC_TIMEOUT env var and enforces the same 300 s default. Safe under a running loop (offloads to a ThreadPoolExecutor), and returns promptly on timeout without waiting for the abandoned worker.

These sync wrappers now raise RuntimeError("run_sync() cannot be called from a running event loop; await the coroutine directly instead.") when called from inside an active asyncio loop. Previously they would silently spawn a worker thread. If you call any of these from async code, switch to await request_approval(...) (or the equivalent async method) directly. This is a deliberate fail-fast change — the silent thread spawn was masking architectural bugs in multi-agent setups.PR #1692 — cancellation on timeout (May 2026). When a run_sync() call hits its timeout (default 300 s, or whatever PRAISONAI_RUN_SYNC_TIMEOUT is set to), the underlying coroutine is now actively cancelled on the background loop. The bridge waits up to 1 s for cancellation to propagate before re-raising TimeoutError. This means slow DB queries (SurrealDB, async MySQL), HTTP calls, and subprocess waits now release their connection / socket / pipe instead of leaking. Cancellation also fires on KeyboardInterrupt, SystemExit, and GeneratorExit.

The wrapper-layer bridge (praisonai._async_bridge) creates its background loop lazily on the first run_sync() call. Pure imports do not allocate a loop or thread. Calling the module-level shutdown() before any run_sync() is a safe no-op — it only affects the shared default bridge, not any AsyncBridge() instances you create yourself.

Troubleshooting

RuntimeError: run_coroutine_from_any_context() cannot be called from async context

You’re trying to use the bridge inside a coroutine. Use await instead:

# Bad
async def my_coroutine():
    return run_coroutine_from_any_context(other_coroutine())

# Good
async def my_coroutine():
    return await other_coroutine()

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

This error used to leak from SDK internals before the async bridge was implemented. If you see this on current versions, upgrade to the latest release.

Symptom	Cause	Resolution
`TimeoutError` raised but you also see your coroutine’s `finally:` block run after the exception	Expected: cancellation propagated, cleanup ran.	No action needed; this is the new PR #1692 behaviour.

Test reference: praisonai/tests/unit/test_async_bridge.py::TestBridgeIntegration::test_timeout_cancels_coroutine_and_runs_finally — quote this in the page so users can verify the behaviour locally.

TimeoutError: run_sync_or_offload() worker did not complete within s

The offloaded coroutine did not finish inside timeout + 1s. Check for blocking I/O or a missing await, or raise PRAISONAI_RUN_SYNC_TIMEOUT if the work is legitimately long-running.

PermissionError in approval system

The approval system now fails fast in async contexts. Configure a non-console backend:

from praisonaiagents.approval import get_approval_registry, WebhookBackend

# Configure for async compatibility
get_approval_registry().set_backend(WebhookBackend(url="http://localhost:8080/approve"))

Wrapper Bridge (`praisonai._async_bridge`)

The wrapper layer provides a module-level run_sync() for CLI scripts and single-tenant servers, plus a public AsyncBridge class when you need an isolated loop per tenant or service.

Module-level (default)
Per-instance AsyncBridge

from praisonai._async_bridge import run_sync, shutdown

async def async_helper(data: str) -> str:
    await asyncio.sleep(0.1)
    return f"Processed: {data}"

def sync_entry_point(data: str) -> str:
    return run_sync(async_helper(data), timeout=60)

import atexit
atexit.register(shutdown)  # shuts down ONLY the shared default bridge

from praisonai._async_bridge import AsyncBridge

bridge = AsyncBridge()  # per-tenant / per-service instance

async def fetch(url: str) -> str:
    ...

result = bridge.run_sync(fetch("https://example.com"), timeout=10)
bridge.shutdown()  # only this bridge — not the shared default

When to use a per-instance bridge

Situation	Module-level `run_sync()`	Your own `AsyncBridge()`
CLI script / one-shot job	✅	—
Single-tenant server	✅	—
Multi-tenant gateway (loop scoped to tenant lifecycle)	—	✅
Embedding PraisonAI inside another async framework	—	✅
Tests needing clean shutdown without affecting others	—	✅

API Reference:

Class / Function	Signature	Description
`AsyncBridge`	`class AsyncBridge`	Per-instance async runner; create one per tenant or service.
`AsyncBridge.run_sync`	`(self, coro, *, timeout=300) -> T`	Run a coroutine on this bridge’s background loop.
`AsyncBridge.shutdown`	`(self, timeout=5.0, *, permanent=False) -> None`	Cancel pending tasks and stop this bridge’s loop. With `permanent=True`, the bridge cannot be reused; later `run_sync`/`submit` raise `RuntimeError`. `scoped_bridge()` uses `permanent=True` for scope-owned bridges. PR #2122: releases the bridge lock before awaiting cancellation.
`AsyncBridge.submit`	`(self, coro) -> concurrent.futures.Future`	Submit without waiting; returns a `concurrent.futures.Future`.
`AsyncBridge.get`	`(self) -> asyncio.AbstractEventLoop`	Get (lazily spawn) the bridge’s event loop.
`run_sync` (module-level)	`(coro, *, timeout=300) -> T`	Routes through the shared default `AsyncBridge`. Behaviour unchanged.
`shutdown` (module-level)	`() -> None`	Shuts down only the shared default bridge. User-owned instances are untouched.

Environment:

PRAISONAI_RUN_SYNC_TIMEOUT: Default timeout in seconds (300). Read by both praisonai._async_bridge (this page) and praisonai_code.cli.features.agent_tools._run_sync (see Agent-Centric Tools → Timeouts & Cancellation).

Do not call run_sync from inside async def — use await instead. The function raises RuntimeError if called from within a running event loop to prevent deadlocks.The module-level shutdown() only stops the shared default bridge. Per-instance AsyncBridge objects must be shut down via bridge.shutdown() on each instance.

Used by:

CLI approval protocol (ACP/LSP tools)
Interactive runtime start/stop operations
Deployment scheduler
Gateway operations

See also: Approval Protocol and Gateway.

`run_sync_or_offload` — safe from any calling context

Use run_sync_or_offload() when your code path can be reached from either a normal script or from inside a running event loop (FastAPI, Jupyter, async tests) — one helper, no RuntimeError.

Import and call

from praisonai._async_bridge import run_sync_or_offload

async def fetch(): return 42

value = run_sync_or_offload(fetch())

Inside a FastAPI handler

Same helper, no code change — it just works:

@app.post("/run")
def run(prompt: str):
    return {"answer": run_sync_or_offload(agent.astart(prompt))}

Configuration Options

Option	Type	Default	Description
`coro`	`Awaitable[T]`	—	The coroutine to await.
`timeout`	`float \| None`	`PRAISONAI_RUN_SYNC_TIMEOUT` (300s)	Same env var as `run_sync`.
`thread_name`	`str`	`"praisonai-sync-offload"`	Name of the offload worker thread (shows up in stack traces).

Called from a plain sync caller, it dispatches to the active AsyncBridge via run_sync, sharing its background loop and connection pools. Called from inside a running loop, it copies the caller’s ContextVars onto a worker thread that hands the coroutine to the same bridge — never a fresh asyncio.new_event_loop() — so a caller-installed scoped_bridge() binding still wins and LiteLLM/HTTPX per-loop connection pools are preserved. Exceptions re-raise on the caller thread.

How it differs from `run_sync`

Helper	Plain sync caller	Inside a running loop
`run_sync`	✅ works	❌ `RuntimeError`
`run_sync_or_offload`	✅ works	✅ offloads onto a worker thread (blocks the caller)
`arun_sync_or_offload`	— (must be awaited)	✅ non-blocking await; the loop stays responsive

run_sync_or_offload() now bounds thread.join() at timeout + 1.0s. If the worker is still alive it raises TimeoutError("run_sync_or_offload() worker did not complete within {timeout}s"). The extra second covers thread hand-off/teardown so the join does not spuriously time out before the worker records its own error.

Best Practices

Prefer run_sync_or_offload for any 3-way surface (CLI + YAML + Python)

The caller’s context isn’t knowable at write time — the same code path can be reached from a plain script, a YAML run, or a Python call inside a running loop. run_sync_or_offload handles all three without a RuntimeError, while still reusing the shared bridge and its connection pools.

from praisonai._async_bridge import run_sync_or_offload

def surface_entry(prompt: str) -> str:
    return run_sync_or_offload(agent.astart(prompt))

Keep run_sync for sync-only paths

run_sync fails loudly inside a loop, which is what you want for a code path that must never be called from one — the RuntimeError surfaces an architectural bug instead of silently offloading.

from praisonai._async_bridge import run_sync

def cli_only(prompt: str) -> str:
    return run_sync(agent.astart(prompt))

This helper landed in PraisonAI #3492; its callers (praisonai.auto, persistence.orchestrator, api.agent_invoke) are migrated onto it.

run_sync

Module-level runner for sync-only paths.

scoped_bridge

Per-session bridge that run_sync_or_offload preserves across the offload hop.

Agent-Centric Tools

The Tier-2 _run_sync (PR #3361) mirrors the same design.

`arun_sync_or_offload` — await from an async context

Use arun_sync_or_offload() from async callers (FastAPI/Starlette handlers, Jupyter cells, async tests) — await it instead of parking the loop thread with the sync helper.

Import and await

from praisonai._async_bridge import arun_sync_or_offload

async def fetch(): return 42

async def handler():
    value = await arun_sync_or_offload(fetch())
    return value

Inside a FastAPI handler

await, don’t park — the loop stays responsive while the agent runs:

from fastapi import FastAPI
from praisonaiagents import Agent
from praisonai._async_bridge import arun_sync_or_offload

app = FastAPI()
agent = Agent(name="Analyst", instructions="Answer briefly")

@app.post("/ask")
async def ask(q: str):
    answer = await arun_sync_or_offload(agent.astart(q))
    return {"answer": answer}

The user calls an async endpoint; the agent’s coroutine runs on the shared background bridge while the request loop stays free to serve other traffic.

Configuration Options

Option	Type	Default	Description
`coro`	`Awaitable[T]`	—	The coroutine to await.
`timeout`	`float \| None`	`PRAISONAI_RUN_SYNC_TIMEOUT` (300 s)	Same env var as `run_sync`. On expiry the in-flight future is cancelled (`fut.cancel()`) and `asyncio.TimeoutError` re-raises.

Unlike run_sync_or_offload, this variant does not park the running loop thread. It submits coro to the active AsyncBridge and awaits the result, so the loop keeps serving other work. The coroutine runs on the shared background bridge loop — never a fresh asyncio.new_event_loop() — so a caller-installed scoped_bridge() binding still wins and per-loop LiteLLM/HTTPX connection pools are preserved. On timeout or cancellation, the in-flight future is cancelled before the exception re-raises.

Best Practices

Await it from async handlers, not run_sync_or_offload

Inside a running loop, run_sync_or_offload offloads onto a worker thread and blocks the caller until the coroutine finishes. arun_sync_or_offload awaits instead, so the loop keeps serving other requests.

# Good — loop stays responsive
answer = await arun_sync_or_offload(agent.astart(q))

# Blocks the caller until done
answer = run_sync_or_offload(agent.astart(q))

Keep the sync helper for sync-only paths

Plain scripts and CLI entry points can’t await. Use run_sync_or_offload (or run_sync) there and reserve arun_sync_or_offload for code already inside a coroutine.

run_sync_or_offload

Sync sibling for non-async callers.

scoped_bridge

Per-session bridge preserved across the await.

Per-Session Scoped Bridge

Servers and gateways that handle multiple concurrent sessions need each session to run on its own loop+thread binding. current_bridge() and scoped_bridge() provide ContextVar-backed per-session isolation so sessions never share a bridge accidentally.

When to use scoped bridges

Use scoped_bridge() inside any request handler that may run concurrently with other handlers — for example a FastAPI endpoint, a Starlette WebSocket handler, or a custom bot session dispatcher. The isolation now extends into sync-completion runner threads used by AutoGenerator.generate() — a scoped_bridge() set on the caller is honoured inside the worker thread via contextvars.copy_context().

from praisonai._async_bridge import AsyncBridge, scoped_bridge, current_bridge

async def handle_request(session_id: str, message: str) -> str:
    """Each concurrent request gets its own isolated bridge."""
    bridge = AsyncBridge()

    async with scoped_bridge(bridge):
        # All code in this block sees `current_bridge()` as `bridge`
        result = bridge.run_sync(process_message(session_id, message))
        return result

`scoped_bridge()` context manager

from praisonai._async_bridge import scoped_bridge, AsyncBridge

bridge = AsyncBridge()

async with scoped_bridge(bridge):
    # current_bridge() returns `bridge` inside this block
    ...

# Outside: current_bridge() returns None (or the enclosing scope's bridge)

The context manager uses a ContextVar so nested scopes work correctly in async tasks and threads — each concurrent task sees only its own bridge.

When scoped_bridge() creates the bridge for you (no argument), it shuts down with permanent=True on exit. If code tries to call run_sync or submit on that bridge afterward, you get:

RuntimeError: AsyncBridge has been shut down and cannot be reused; this usually means a context outlived its scoped_bridge() block

That guard stops an orphaned loop and thread from outliving the scope that owned them. The shared default bridge always shuts down with permanent=False.

Scope-owned bridge (preferred)

from praisonaiagents import Agent
from praisonai._async_bridge import scoped_bridge, run_sync

def handle_session(session_id: str, prompt: str) -> str:
    with scoped_bridge() as bridge:
        data = run_sync(fetch_session_data(session_id))
        agent = Agent(name="Assistant", instructions=f"Context: {data}")
        return agent.start(prompt)
    # bridge.shutdown(permanent=True) ran automatically — do not reuse `bridge` here

With no argument, scoped_bridge() creates a fresh bridge and tears it down with permanent=True when the with block ends. Internal code that calls module-level run_sync() inside the block uses the scoped bridge via contextvars — no import changes required.

`current_bridge()` for introspection

current_bridge() returns the bridge bound to the current async task, or None when no scope is active. Use it to inspect which bridge is in use without passing it explicitly through call stacks.

from praisonai._async_bridge import current_bridge

def my_util():
    bridge = current_bridge()
    if bridge is None:
        raise RuntimeError("No scoped bridge active — wrap with scoped_bridge()")
    return bridge.run_sync(some_coroutine())

Multi-session server example

import asyncio
from praisonaiagents import Agent
from praisonai._async_bridge import AsyncBridge, scoped_bridge

agent = Agent(name="server-agent", instructions="Respond to requests.")

async def handle_session(session_id: str, user_message: str) -> str:
    bridge = AsyncBridge()
    async with scoped_bridge(bridge):
        result = bridge.run_sync(agent.achat(user_message))
        bridge.shutdown()
    return result

async def main():
    sessions = [
        handle_session("user-1", "Hello from session 1"),
        handle_session("user-2", "Hello from session 2"),
    ]
    results = await asyncio.gather(*sessions)
    for r in results:
        print(r)

asyncio.run(main())

API Reference

Symbol	Signature	Description
`scoped_bridge`	`(bridge: Optional[AsyncBridge] = None) -> Iterator[AsyncBridge]`	Context manager. Binds a per-scope bridge via `ContextVar`. With `None`, creates and owns a fresh bridge (shut down with `permanent=True` on exit). With a caller-provided bridge, the caller retains lifecycle ownership.
`current_bridge`	`() -> AsyncBridge`	Returns the bridge bound by an enclosing `scoped_bridge()` block, else the shared default.

Async Agents Guide

Thread Safety & Concurrency

​Quick Start

​How It Works

​Configuration Options

​Common Patterns

​Reusing async SDKs from sync tools

​Offloading blocking calls from async tools

​Context-aware dual-mode helper

​Best Practices

​Used by

​Troubleshooting

​RuntimeError: run_coroutine_from_any_context() cannot be called from async context

​asyncio.run() cannot be called from a running event loop

​TimeoutError: run_sync_or_offload() worker did not complete within s

​PermissionError in approval system

​Wrapper Bridge (praisonai._async_bridge)

​When to use a per-instance bridge

​run_sync_or_offload — safe from any calling context

​Configuration Options

​How it differs from run_sync

​Best Practices

​Related

run_sync

scoped_bridge

Agent-Centric Tools

​arun_sync_or_offload — await from an async context

​Configuration Options

​Best Practices

​Related

run_sync_or_offload

scoped_bridge

​Per-Session Scoped Bridge

​When to use scoped bridges

​scoped_bridge() context manager

​Scope-owned bridge (preferred)

​current_bridge() for introspection

​Multi-session server example

​API Reference

​Related

Quick Start

How It Works

Configuration Options

Common Patterns

Reusing async SDKs from sync tools

Offloading blocking calls from async tools

Context-aware dual-mode helper

Best Practices

Used by

Troubleshooting

RuntimeError: run_coroutine_from_any_context() cannot be called from async context

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

TimeoutError: run_sync_or_offload() worker did not complete within s

PermissionError in approval system

Wrapper Bridge (`praisonai._async_bridge`)

When to use a per-instance bridge

`run_sync_or_offload` — safe from any calling context

Configuration Options

How it differs from `run_sync`

Best Practices

Related

`arun_sync_or_offload` — await from an async context

Configuration Options

Best Practices

Related

Per-Session Scoped Bridge

When to use scoped bridges

`scoped_bridge()` context manager

Scope-owned bridge (preferred)

`current_bridge()` for introspection

Multi-session server example

API Reference

Related