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

​
Quick Start

1

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")
2

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],
)
3

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

OptionTypeDefaultDescription
timeoutfloat300Maximum 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

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())
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()
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)
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.
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 shutdown() before any run_sync() is a safe no-op.

​
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.
SymptomCauseResolution
TimeoutError raised but you also see your coroutine’s finally: block run after the exceptionExpected: 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.

​
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 has its own async bridge for PraisonAI-level operations: 
from praisonai._async_bridge import run_sync, shutdown

# Example: sync entry point calling async helper
async def async_helper(data: str) -> str:
    await asyncio.sleep(0.1)
    return f"Processed: {data}"

def sync_entry_point(data: str) -> str:
    """Sync function that needs to call async code."""
    return run_sync(async_helper(data), timeout=60)

# For long-running servers
import atexit
atexit.register(shutdown)
API Reference:
FunctionSignatureDescription
run_sync(coro, *, timeout=300) -> TRun coroutine on background loop
shutdown() -> NoneClean shutdown of background loop
Environment:
  • PRAISONAI_RUN_SYNC_TIMEOUT: Default timeout in seconds (300)
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.
Used by:
  • CLI approval protocol (ACP/LSP tools)
  • Interactive runtime start/stop operations
  • Deployment scheduler
  • Gateway operations
See also: Approval Protocol and Gateway.

Async Agents Guide

Thread Safety & Concurrency