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.

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 whether or not an event loop is already running.

An async guardrail on a plain Agent, reached from a FastAPI route, runs on a worker thread with the caller’s contextvars — the timeout is enforced inside that worker’s loop.

from fastapi import FastAPI
from praisonaiagents import Agent, TaskOutput

async def content_check(output: TaskOutput):
    # async because it hits a moderation API
    ok = await moderation_service.is_safe(output.raw)
    return (ok, output if ok else "flagged by moderation")

agent = Agent(
    name="writer",
    instructions="Write a short reply.",
    guardrails=content_check,   # async guardrail
)

app = FastAPI()

@app.post("/reply")
async def reply(prompt: str):
    # Before PR #5377 this raised "Agent guardrail validation error".
    # Now the async guardrail runs on a worker thread with the caller's
    # contextvars and returns its real result.
    return {"answer": agent.start(prompt)}

The user posts a prompt from a FastAPI route; the agent’s async guardrail runs on a worker thread with the caller’s contextvars, and the timeout is enforced inside that worker’s 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")

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

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. If no loop exists, it drives the coroutine directly on the caller’s thread via asyncio.run (wrapped in asyncio.wait_for when a timeout is set). If a loop is already running (FastAPI, Jupyter, a bot handler calling agent.chat() from inside async def), it offloads onto a dedicated worker thread with its own loop — never a deadlock, never a RuntimeError. The caller’s contextvars are copied into the worker thread with contextvars.copy_context(), so trace, session, and approval context set by the caller stays visible inside the coroutine.

Which helper do I call?

Pick between await, run_coroutine_from_any_context, and the wrapper-layer run_sync:

Async guardrails, callbacks, and approvals inside a running loop (PR #5377)

Prior to PR #5377, run_coroutine_from_any_context raised RuntimeError when reached from inside a running event loop (FastAPI async route, Jupyter cell, Discord/Slack bot handler, any async def calling agent.start(), task.execute_callback_sync(), or an approval callback). That surfaced as a spurious “Agent guardrail validation error” and burned LLM retries on a valid async guardrail. It also masked real summariser errors as an “event loop already running” fallback to naive truncation. From that PR onwards:

• Async guardrails on Agent(...) and Task(...) return their real result inside a running loop.
• Async task callbacks reached via Task.execute_callback_sync run to completion instead of raising.
• Async ApprovalCallback(approve_fn=<async fn>) works from FastAPI / Textual / Jupyter without the caller having to opt in.
• ContextVars set by the caller (trace ids, session ids, approval context) are visible inside the coroutine.
• The timeout is enforced. A coroutine that ignores cancellation (e.g. a blocking C call) can no longer pin the caller past timeout — the worker is abandoned (shutdown(wait=False)).
• LLM_SUMMARIZE context compaction runs the real async summariser under a running loop instead of silently falling back to naive truncation, and provider errors from the summariser now surface instead of being masked.

The five previously-broken sync→async call sites (Agent._process_guardrail, Task._process_guardrail, Task.execute_callback_sync, planning/approval.py, bus/bus.py) all keep their existing code and now work correctly. The sync tool-calling helper run_async_in_sync_context is a thin delegate to run_coroutine_from_any_context(coro, timeout=None).

Internal sync surfaces now routed through the shared bridge (PR #5589)

PR #5589 routes ~11 internal sync→async call sites that used to hand-roll their own asyncio.run + ThreadPoolExecutor dance through the shared run_coroutine_from_any_context helper. The hand-rolled copies had two defects the shared helper does not:

• Their try/except RuntimeError wrapped the worker’s future.result(), so a provider RuntimeError was replaced by a misleading "asyncio.run() cannot be called from a running event loop". Fixed.
• They never copied contextvars.copy_context(), so trace / session / approval context set by the caller was invisible inside the coroutine. Fixed — the shared helper copies the context.

If you call agent.start(...) from a FastAPI route and your async guardrail, backend, aggregator, or tool used to lose trace context or report an opaque loop error, PR #5589 fixes that. The converted user-reachable surfaces:

PR #6239 (Issue #6238) removed three more hand-rolled sync-over-async copies in favour of the shared bridge.

Three sites are intentionally not converted because they have distinct semantics: hooks/runner.py (raises on an in-loop call), tools/edit_tools.py (defers to a fallback), and tools/messaging_tools.py (routes via run_coroutine_threadsafe to a registered gateway loop).

Configuration Options

run_coroutine_from_any_context (the praisonaiagents.utils.async_bridge helper) takes two arguments:

Timeout resolution: omitted vs None vs number

The timeout argument on the wrapper-layer helpers — run_sync, run_sync_or_offload, arun_sync_or_offload, and AsyncBridge.run_sync — uses a private _UNSET sentinel as its default, so the bridge can tell “argument omitted” apart from an explicit timeout=None.

The sync-scheduler bridges (praisonai/integration/bridges/schedules_runner.py, praisonai/cli/commands/schedule.py) pass timeout=None so a long-running claimed job is never cancelled after 300 s.

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

# Good — no bridge overhead, work stays on the caller's loop
async def my_async_tool():
    result = await my_coroutine()

# Works, but takes a worker thread — reserve for sync callers
async def my_async_tool():
    result = run_coroutine_from_any_context(my_coroutine())

# 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()

# 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)

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.
• Ten CLI subcommand leaves route through the public run_cli_coro() helper (PR #5322) — background {list, status, cancel, clear}, managed {list-check, stop}, standardise, sandbox {execute, shell}, chat REPL /tasks {list, cancel, status}, and the direct_prompt external-agent pass-through — so per-loop connection pools survive across invocations instead of a fresh asyncio.run(...) per call.

Troubleshooting

RuntimeError: run_coroutine_from_any_context() cannot be called from async context

This specific RuntimeError no longer exists. PR #5377 removed it — previously-broken async guardrails, callbacks, and approvals now succeed inside a running loop by offloading to a worker thread. If you still see this message, you are on an older release; upgrade to the version that ships PR #5377.

My async guardrail / callback / approval now works but I’d rather await it directly

You can. Inside async def code, await the coroutine yourself instead of routing through the bridge. The bridge is there for the sync-only surfaces (Agent.start, Task.execute_callback_sync, sync tool functions) that cannot await.

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.

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 {N}s

This timeout branch is only reachable when the caller is inside a running loop and has opted into PRAISONAI_ALLOW_LOOP_BLOCKING=true (the offload path). A plain-sync caller still hits the same TimeoutError via run_sync. 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. Prefer migrating to await arun_sync_or_offload(...) / await praisonai.arun(...) so the loop is never blocked.

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.

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

API Reference:

Environment:

• PRAISONAI_RUN_SYNC_TIMEOUT: Default timeout in seconds (300). Resolved per call (not at import) via _default_timeout(); a malformed value falls back to 300.0, and late-set env vars (dotenv loaded after import, per-request reconfig) are honoured. An explicit timeout=None opts into an unbounded wait (used by the sync-scheduler bridges for long-lived jobs). Read by both praisonai._async_bridge (this page) and praisonai_code.cli.features.agent_tools._run_sync (see Agent-Centric Tools → Timeouts & Cancellation).

Used by:

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

See also: Approval Protocol and Gateway.

run_cli_coro() — the CLI-leaf entry point

run_cli_coro() is the public helper a synchronous CLI leaf should call instead of a bare asyncio.run(coro). It routes the coroutine onto the shared background loop so per-loop connection pools survive across invocations, any embedder-installed scoped_bridge() binding is honoured, and no fresh loop is spawned on every call (PR #5322).

from praisonai._async_bridge import run_cli_coro

def handle_list() -> None:
    tasks = run_cli_coro(handler.list_tasks())
    for t in tasks:
        print(t)

Migration from asyncio.run()

Replace the bare asyncio.run(...) at a CLI leaf with run_cli_coro(...) — the coroutine now shares the background loop instead of tearing down a fresh one on every invocation.

# Before
import asyncio
asyncio.run(handler.list_tasks())

# After
from praisonai._async_bridge import run_cli_coro
run_cli_coro(handler.list_tasks())

sequenceDiagram
    participant CLI as praisonai CLI leaf
    participant RCC as run_cli_coro()
    participant Bridge as shared AsyncBridge
    participant Loop as background loop

    CLI->>RCC: run_cli_coro(handler.list_tasks())
    RCC->>RCC: get_running_loop()? → RuntimeError → sync
    RCC->>Bridge: run_sync(coro, timeout=None)
    Bridge->>Loop: schedule on shared loop
    Loop-->>Bridge: result
    Bridge-->>RCC: result
    RCC-->>CLI: result
    Note over Loop: pool + httpx.AsyncClient stay alive<br/>for the next invocation

Migrated CLI sites

Ten CLI subcommand leaves now dispatch through run_cli_coro() so repeated invocations reuse the background loop:

dispatch_maybe_awaitable — one helper for sync-hook dispatch

dispatch_maybe_awaitable() is the single owner of the “a sync call may return a coroutine; run it correctly” policy. It picks one of four branches from the running-loop state and an explicit DispatchKind.

DispatchKind makes read-vs-write intent explicit at the call site instead of depending on a name-string allow-list. The four branches:

This is the shared helper the PraisonAIDB._call_store, _merge_and_set, and on_agent_end paths route through — the split policy is no longer hand-rolled per call site (PR #5038). See Async DB Hooks → Shared dispatch for the db-adapter view.

The contract tests in praisonai/tests/unit/test_async_bridge.py::TestDispatchMaybeAwaitable pin the four branches:

• test_non_awaitable_passthrough — a plain value is returned unchanged.
• test_no_loop_read_blocks_and_returns_value — no running loop + READ blocks and returns the value.
• test_no_loop_write_blocks_and_returns_value — no running loop + WRITE runs to completion (fire-and-forget is a running-loop concern).
• test_running_loop_write_is_tracked_fire_and_forget — running loop + WRITE submits, tracks, and returns None while the write still completes.
• test_running_loop_failed_write_logs_and_does_not_raise — a failed deferred WRITE is swallowed (logged) via the done-callback, never surfacing to the sync caller.

run_sync_or_offload — strict inside a running loop

Use run_sync_or_offload() on a code path that can be reached from a plain script or from inside a running event loop (FastAPI, Jupyter, async tests). On a plain-sync caller it dispatches to run_sync. Inside a running loop it now raises RuntimeError by default (PR #4261) and steers you to the awaitable siblings — it never silently pins the loop.

1. Plain sync caller — works

   from praisonai._async_bridge import run_sync_or_offload
   
   async def fetch(): return 42
   
   value = run_sync_or_offload(fetch())  # no running loop → run_sync

2. Inside a FastAPI handler — await instead

   A sync call inside a running loop now raises RuntimeError. Make the handler async and await the coroutine directly:

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

   For an existing sync entry point you cannot convert to async def, await the awaitable sibling:

   from praisonai._async_bridge import arun_sync_or_offload
   
   @app.post("/run")
   async def run(prompt: str):
       return {"answer": await arun_sync_or_offload(agent.astart(prompt))}
       # or: await praisonai.arun(...)

Configuration Options

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 with PRAISONAI_ALLOW_LOOP_BLOCKING=true, 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.

The strict RuntimeError

In the default (strict) mode, a call from inside a running loop raises with this message — grep for it in a traceback:

run_sync_or_offload() would block the running event loop for up to 300s. From an async context, await one of:
  - praisonai.arun(...)
  - adapter.arun(...)
  - praisonai._async_bridge.arun_sync_or_offload(coro)
Set PRAISONAI_ALLOW_LOOP_BLOCKING=true to opt into the legacy blocking behaviour.

How it differs from run_sync

Migration

Any caller that today relies on offload-inside-a-loop must either (a) migrate to the awaitable sibling (await praisonai.arun(...), await adapter.arun(...), or await arun_sync_or_offload(...)), or (b) set PRAISONAI_ALLOW_LOOP_BLOCKING=true as an interim measure. See PR #4261.

Best Practices

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

# Raises RuntimeError inside a running loop (strict default)
answer = run_sync_or_offload(agent.astart(prompt))

from praisonai._async_bridge import run_sync_or_offload

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

This helper landed in PraisonAI #3492; the strict-inside-a-loop default landed in PR #4261. Its callers (praisonai.auto, persistence.orchestrator, api.agent_invoke) are migrated onto it.

Related

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.

1. 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

2. 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

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

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

# Raises RuntimeError inside a running loop (strict default)
answer = run_sync_or_offload(agent.astart(q))

Related

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().

• Multi-tenant sync embedders no longer need a manual scoped_bridge() around praisonai workflow run (or a process: workflow YAML dispatched through generate_crew_and_kickoff()) — _run_yaml_workflow wraps workflow.start(...) in its own scoped_bridge().

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.

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

async_scoped_bridge() — async-safe context manager

async_scoped_bridge() is the async-safe sibling of scoped_bridge() for async def callers — same binding semantics, but the scope-owned bridge is torn down off-loop so scope exit never parks the caller’s event loop.

Wrap your async handler in one line — a stuck coroutine in one tenant’s request can no longer freeze the loop for every other tenant:

from praisonai._async_bridge import async_scoped_bridge

async def handle(...):
    async with async_scoped_bridge():
        ...

When to use it

Pick the context manager that matches your caller.

• scoped_bridge() — sync callers (CLI, generate_crew_and_kickoff, sync request handlers).
• async_scoped_bridge() — inside async def (FastAPI handlers, agenerate_crew_and_kickoff, custom async gateways). Preferred over the sync sibling for any async def, because scope exit on the sync context manager parks the loop thread for up to ~10s while shutdown() waits for cancellation and joins the background thread.

FastAPI multi-tenant handler

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

app = FastAPI()

@app.post("/tenants/{tenant_id}/ask")
async def ask(tenant_id: str, prompt: str):
    # Per-request isolated bridge — a stuck coroutine here cannot
    # park the loop for other tenants.
    async with async_scoped_bridge():
        agent = Agent(name=f"agent-{tenant_id}", instructions="Answer briefly")
        return {"answer": await agent.astart(prompt)}

Teardown internals

A scope-owned bridge (no argument) is torn down via asyncio.to_thread(bridge.shutdown, permanent=True) under asyncio.shield, so the blocking cancel-and-join runs on a worker thread instead of the event loop. asyncio.shield keeps that teardown running to completion even when the surrounding scope is being cancelled, so the loop+thread is neither leaked nor blocked. A caller-provided bridge is left untouched — the caller owns its lifecycle.

Behavioural contract

API Reference

Per-run auto-scoping (sequential, hierarchical, workflow)

Both generate_crew_and_kickoff() (sync) and agenerate_crew_and_kickoff() (async) now auto-wrap each run in a per-run scoped bridge — the sync path uses scoped_bridge(), the async path uses async_scoped_bridge(). Every run isolates its run_sync-driven work onto its own loop+thread, so a stuck coroutine in one agent/tenant can no longer park the shared default loop for the rest.

Each entry point has two arms — a sequential/hierarchical arm and a YAML-workflow arm — so there are four corners in total. All four are now symmetric.

The same per-run scoping now covers process: workflow on the sync path. _run_yaml_workflow (invoked by praisonai workflow run ... and AgentsGenerator.generate_crew_and_kickoff() when the YAML’s process: is workflow) wraps workflow.start(...) in scoped_bridge(). The async workflow path (_arun_yaml_workflow) already runs on the caller’s loop and remains unchanged.

PR #5360 closes the last asymmetry: the sync YAML workflow arm now wraps its _run_yaml_workflow(config) call in scoped_bridge(), matching the other three corners so a stuck coroutine in one tenant’s YAML workflow can no longer park the shared default loop for the rest.

Callers doing multi-tenant runs no longer need to add their own scoped_bridge() around either entry point — with this PR that is now true for all four corners — a stuck coroutine in one run cannot affect another. On the async path this closes a live multi-tenancy regression: praisonai serve, the gateway, and FastAPI-embed deployments no longer let a stuck coroutine in one tenant park the shared default loop+thread for the others. The earlier reasoning that async callers await arun directly did not hold in practice — adapter internals, user tools, the sync DeliveryRouter._finalize_delivery, and BlueprintAgent.start() (via run_sync(...)) all reach back into the sync bridge, which previously resolved to the process-default bridge under the async path.

See Wrapper → Lifecycle / cleanup for the embedder view.

CLI leaves

Every CLI leaf in praisonai that owns a top-level coroutine routes it through run_cli_coro() instead of a bare asyncio.run(...). This keeps the AsyncBridge contract (per-loop pools, embedder scoped_bridge()) honored end to end.

Adding a new CLI entry point that owns a top-level coroutine? Use from praisonai._async_bridge import run_cli_coro and avoid asyncio.run at the leaf — a grep -rn "asyncio.run(" src/praisonai/praisonai/ outside _async_bridge.py must stay at zero.

Related