> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Async Bridge
> Safely call async code from sync, and sync code from async, without deadlocking the event loop.
The async bridge lets your tools and callbacks move between sync and async without crashing the event loop.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Async Bridge"
A[๐ Sync Caller] --> B{๐ Loop Check}
B -->|No Loop| C[๐ Run Coroutine]
B -->|Loop Running| D[โ RuntimeError]
C --> E[โ
Result]
F[๐ Async Caller] --> G[๐งต Executor]
G --> H[โ
Result]
end
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
class A,F agent
class B,C,D,E,G,H tool
```
## Quick Start
Use `run_coroutine_from_any_context` to call async code from a sync tool:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
Use `run_sync_in_executor` to call blocking code from an async tool without blocking the event loop:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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],
)
```
Use `is_async_context` to create dual-mode helpers:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Caller
participant Bridge
participant EventLoop
participant Executor
Caller->>Bridge: run_coroutine_from_any_context(coro)
Bridge->>EventLoop: get_running_loop()
EventLoop-->>Bridge: RuntimeError (no loop)
Bridge->>EventLoop: asyncio.run(coro)
EventLoop-->>Bridge: Result
Bridge-->>Caller: Result
Note over Bridge,EventLoop: If loop exists, raises RuntimeError
Caller->>Bridge: await run_sync_in_executor(func, args)
Bridge->>Executor: submit(func, args)
Executor-->>Bridge: Result
Bridge-->>Caller: Result
```
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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 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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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.
### PermissionError in approval system
The approval system now fails fast in async contexts. Configure a non-console backend:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Per-Instance AsyncBridge"
TA[๐ค Tenant A] --> BA[๐ AsyncBridge A]
TB[๐ค Tenant B] --> BB[๐ AsyncBridge B]
BA --> LA[๐ Loop A]
BB --> LB[๐ Loop B]
end
subgraph "Shared default"
S[๐ run_sync module-level] --> SD[๐ _default_bridge]
SD --> SL[๐ Default Loop]
end
classDef tenant fill:#6366F1,stroke:#7C90A0,color:#fff
classDef bridge fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef loop fill:#10B981,stroke:#7C90A0,color:#fff
class TA,TB,S tenant
class BA,BB,SD bridge
class LA,LB,SL loop
```
### 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)
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](/docs/features/approval-protocol) and [Gateway](/docs/features/gateway).
***
## 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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Scoped Bridge Per Session"
S1[๐ค Session A] --> SB1[๐ scoped_bridge A]
S2[๐ค Session B] --> SB2[๐ scoped_bridge B]
SB1 --> L1[๐ Loop A]
SB2 --> L2[๐ Loop B]
end
classDef session fill:#6366F1,stroke:#7C90A0,color:#fff
classDef bridge fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef loop fill:#10B981,stroke:#7C90A0,color:#fff
class S1,S2 session
class SB1,SB2 bridge
class L1,L2 loop
```
### 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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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. |
***
## Related
Async Agents Guide
Thread Safety & Concurrency