Async Tool Safety

Safety mechanisms now apply to async tool execution paths

Async tool calls in agent.achat() now route through the same safety mechanisms as sync execution, including approval checks, timeouts, and Loop Guard’s doom-loop / no-progress protection (PR #4005). Each async tool invocation now also runs through CircuitBreaker.acall(...) — event-loop safe, using the same per-agent breaker name (tool_{id(self)}_{function_name}) and weakref.finalize cleanup as the sync path (PR #4469).

Two more sync/async parity gaps closed in PR #4634:

• MCP tool resolution — MCP instances passed via tools= are now resolved on achat(), and MCP transport / timeout failures ("Error: MCP tool call timed out after Ns", "Error: MCP initialization timed out after Ns") are normalized to the same {"error": …, "timeout": True} dict shape used on the sync path.
• Tool-name self-repair — a case / separator hallucination (e.g. WebSearch for web_search) is quietly re-matched on achat() too. A genuine miss now returns the same corrective error dict with available_tools and a closest-match suggestion.

A third parity gap closed in PR #4858:

• Doom-loop approval override — a critical doom-loop verdict on achat() / astart() now routes through the same approval pipeline as chat() / start(), so a PermissionManager rule (doom_loop: allow), YAML permission, or interactive backend continue can override the stop. The async path uses approve_async(...) natively, so event-loop-bound approval backends work.

from praisonaiagents import Agent

agent = Agent(
    name="assistant",
    instructions="Call tools safely from async runs.",
)

The user runs tools via achat(); approvals, circuit breakers, and timeouts apply the same way as synchronous execution.

Quick Start

1. Async Tool with Approval

   from praisonaiagents import Agent
   
   def risky_tool():
       """Tool that requires approval"""
       return "Executed risky operation"
   
   agent = Agent(
       name="Safety Agent",
       instructions="Execute tools safely",
       tools=[risky_tool],
       require_approval=True  # Now works in async mode too
   )
   
   # Approval prompt will appear during async execution
   response = await agent.achat("Use the risky tool")

2. Task-Scoped Tools

   from praisonaiagents import Agent, Task
   
   def special_tool():
       return "Special operation completed"
   
   agent = Agent(name="Tool Agent")
   
   # Task-specific tools override agent tools in async mode
   task = Task(
       description="Use special tool",
       agent=agent,
       tools=[special_tool]  # Available only for this task
   )
   
   # special_tool is available during async chat execution

How It Works

[^parity]: The async column was broken before PR #4141 — a raised exception was retried zero times regardless of RetryPolicy. See Tool Retry Policy → Sync / async parity.

[^cb-async]: The async column was aspirational before PR #4469 — async tool calls bypassed the breaker entirely. The async path now wraps each invocation in CircuitBreaker.acall(...), so five consecutive failures open the breaker just like sync. See Circuit Breaker Protection.

[^mcp-async]: Before PR #4634 the async path did not consult MCP at all — MCP instances in tools= hard-failed with "Function ... not found in tools" on achat(), and MCP transport / timeout failures were handed to the model as bare "Error: …" strings that looked like successful results. Both are now at sync parity via the shared _resolve_mcp_tool_result / _normalize_mcp_result helpers.

[^selfrepair-async]: Before PR #4634 the async path hard-failed on a slightly wrong tool name and returned only "Function ... not found in tools". It now runs the same case / separator-insensitive re-match as the sync path and, on a genuine miss, returns the same corrective error dict ({"error": "...", "available_tools": [...]}) so the model can retry with a valid name.

[^doomloop-approve-async]: Before PR #4858 the async path (achat() / astart()) hard-stopped a critical doom-loop verdict with a bare return, never consulting the unified approval pipeline. A PermissionManager rule like doom_loop: allow, a YAML permissions: { doom_loop: allow }, PRAISONAI_AUTO_APPROVE, or an interactive backend continue — all honoured on chat() / start() since PR #3776 — were silently ignored on the async path. The async path now routes the critical verdict through _doom_loop_approved_async(...), which calls the registry’s native approve_async(...) so an async-only / event-loop-bound approval backend runs on the caller’s loop. Fail-closed semantics are preserved: deny, timeout, no backend, or any error still blocks with loop_blocked: True.

[^tool-approve-async-4878]: On the tool-approval async path (not the doom-loop gate above), PRAISONAI_AUTO_APPROVE — along with YAML approve: and [s]/[a] session grants — is now honoured on attached-backend agents as well since PR #4878. Before #4878 the backend branch of _resolve_approval_decision went straight to the prompt and ignored these standing grants (measured: PRAISONAI_AUTO_APPROVE=true prompted 3 of 3 identical calls; 0 of 3 after the fix). Fail-closed: no standing grant, a denied backend response, or a bookkeeping exception → the historical behaviour is preserved. See Standing grants and attached backends.

Architecture: The unified async dispatcher now calls execute_tool_async for tool invocations, so long-running tools no longer block the event loop.

Safety Mechanisms

Approval Checks

User approval prompts now appear during async tool execution.[^tool-approve-async-4878]

from praisonaiagents import Agent

agent = Agent(
    name="Approved Agent",
    tools=[dangerous_operation],
    require_approval=True
)

# Approval workflow works in async mode
async def safe_execution():
    response = await agent.achat("Delete all files")
    # User sees: "Agent wants to use dangerous_operation. Approve? (y/n)"
    return response

Circuit Breaker Protection

Every async tool call runs through CircuitBreaker.acall(...) — the same per-agent / per-tool breaker as sync (tool_{id(self)}_{function_name}), with the same failure_threshold=5, recovery_timeout=60.0, and graceful_degradation=True defaults (PR #4469).

from praisonaiagents import Agent

agent = Agent(
    name="Circuit Breaker Agent", 
    tools=[unreliable_api],
    # Circuit breaker automatically protects async calls
)

# Failed async calls contribute to circuit breaker state
await agent.achat("Call unreliable API")  # May be blocked if failure rate is high

Two kinds of failures count toward the breaker on the async path:

1. A raised exception inside the tool, except ValueError, TypeError, AttributeError — these are caller/programming errors, raised through breaker.acall() wrapped in a private _NeutralOutcome sentinel, then unwrapped and re-raised to the caller without changing the breaker’s counters (PR #5861).
2. A tool result dict carrying error — surfaced to the breaker through a _ToolFailure sentinel so the dict is preserved — except when the dict is tagged _praison_retryable: False (schema mismatch / caller error). Tagged dicts are neutral and returned to the caller as-is (PR #5861).

Approval / permission / policy / guardrail denials never count as failures either — the async path excludes the same keys the sync _ToolFailure wrapper does: approval_denied, permission_denied, approval_error, policy_denied, guardrail_denied. A tool that raises RuntimeError or returns a plain {"error": …} dict still counts as a failure, so repeated outages still open the breaker.

When the breaker is open, CircuitBreaker.acall(...) raises CircuitBreakerException and the async path returns:

{
    "error": "Tool 'X' circuit breaker open - too many recent failures",
    "circuit_open": True,
    "agent_name": "...",
    "session_id": "...",
    "remediation": "Wait for recovery_timeout (60s) or investigate recent tool failures.",
}

The circuit_open: True flag activates the async retry-skip, so retries stop immediately instead of hammering the open breaker. Loop Guard also records the rejection, so open-circuit rejections still count toward its BLOCK/HALT thresholds.

A repeatedly-failing async tool opens the breaker after 5 failures; the 6th call short-circuits:

import asyncio
from praisonaiagents import Agent, tool

@tool
async def flaky_upstream(query: str) -> str:
    """Async tool that keeps failing on a bad upstream."""
    raise RuntimeError("upstream 503")

agent = Agent(name="Researcher", tools=[flaky_upstream])

async def main():
    # First 5 calls raise and count as failures; the breaker opens.
    # The 6th call short-circuits with {"circuit_open": True}, so the
    # async retry loop terminates immediately instead of retrying.
    result = await agent.achat("Query the upstream repeatedly")
    print(result)

asyncio.run(main())

MCP + Name Self-Repair Parity

MCP tools and hallucinated tool names now resolve identically on achat() and chat() (PR #4634).

from praisonaiagents import Agent
from praisonaiagents.mcp import MCP

agent = Agent(
    name="MCP Agent",
    instructions="Use the MCP server tools to answer.",
    tools=MCP("npx -y @modelcontextprotocol/server-filesystem /tmp"),
)

# Before PR #4634: hard-failed with "Function read_file not found in tools".
# Now: resolves through the MCP runner and returns the file contents.
result = await agent.achat("What is in /tmp/notes.txt?")

# Model emits 'WebSearch' but the tool is named 'web_search'.
# Before PR #4634: async path hard-failed; sync path quietly re-matched.
# Now: async path re-matches too and calls web_search(...) as intended.
await agent.achat("Search the web for the latest fusion news.")

Loop Guard on the async path

Every async tool call in agent.achat() now runs through Loop Guard the same way sync calls always did (PR #4005). Repeated identical calls, repeated timeouts, and repeated exceptions all accumulate toward WARN → BLOCK → HALT on the current turn.

from praisonaiagents import Agent

agent = Agent(
    name="Async Poller",
    instructions="Poll a status endpoint until it succeeds.",
)

# Identical failing calls escalate through Loop Guard on the async path.
# A HALT propagates as ToolExecutionError(is_retryable=False) — retry
# policies also honour this and will not re-run a HALT'd call.
result = await agent.achat("Watch job-42 until it is done.")

Async-specific details:

• Async tool timeouts (asyncio.wait_for firing after tool_timeout seconds) are recorded as failures and count toward the BLOCK/HALT thresholds. Previously they early-returned and bypassed the streak. Middleware-wrapped async tools (see Middleware) also honour ToolConfig.timeout as of PR #5080 — before that fix, the middleware path blocked on future.result() with no bound and timeouts on the middleware path were silently ignored (they were not recorded as failures and did not count toward BLOCK/HALT thresholds).
• Async tool exceptions are also recorded as failures — a tool that keeps raising escalates the same way a tool that keeps returning {"error": …} does.
• A Loop Guard HALT propagates as ToolExecutionError(is_retryable=False). The async retry wrapper honours both loop_blocked (in the returned dict) and the exception, so no custom retry policy can re-run a terminally blocked call.

See Loop Guard for the full threshold table, tool classification, and configuration options.

Retry parity on the async path

Raised exceptions on agent.achat(...) / agent.execute_tool_async(...) are now retried under ToolConfig(retry_policy=RetryPolicy(...)), matching the sync path (PR #4141) — caller errors (ValueError, TypeError, AttributeError, LookupError, and file-path errors such as FileNotFoundError / PermissionError) and ToolExecutionError(is_retryable=False) (including Loop Guard HALT) stay terminal. Once retries are exhausted, the raised exception surfaces as ToolExecutionError (not an error dict), matching the sync path (PR #6109). See Tool Retry Policy for the full policy reference.

The reverse direction — not retrying when the tool asked not to be retried — is also at parity as of PR #4299. @tool(restart_safe=False) / .idempotent = False runs exactly once on agent.achat(...) / agent.execute_tool_async(...), matching the sync path. See Tool Retry & Backoff → Retrying Non-Idempotent Tools for the full veto rules.

Timeout Controls

Async tool execution respects timeout settings:

from praisonaiagents import Agent
from praisonaiagents.config import ToolConfig

def slow_tool():
    import time
    time.sleep(10)  # Long operation
    return "Done"

agent = Agent(
    name="Timeout Agent",
    tools=[slow_tool],
    tool_config=ToolConfig(timeout=5)  # 5 second limit applies to async calls
)

# Timeout enforced in async mode
await agent.achat("Use slow tool")  # Will timeout after 5 seconds

Timed sync tools don’t starve asyncio.to_thread

Since PR #6073, when achat() runs a sync tool under ToolConfig(timeout=…), it uses the same per-agent worker pool as the sync path — not asyncio’s shared default executor. A tool hung past its timeout only ties up an agent-scoped worker, so unrelated asyncio.to_thread(...) calls elsewhere in your app keep running.

If every pool worker is busy, a new timed call starts at once on a spill thread (up to 8 per agent) instead of queueing behind a hung worker. Past that cap, the call is refused fast:

{
    "error": "Tool worker pool exhausted: ... Refusing to queue 'slow_tool'.",
    "tool_pool_exhausted": True,
    "_praison_retryable": False,
}

This is the same signal described in Tool-Call Executor Timeout — fix the hung tool rather than raising the timeout.

graph TB
    Call[🔄 Timed sync tool<br/>via achat] --> Free{Pool worker<br/>free?}
    Free -->|Yes| Pool[👷 Per-agent pool]
    Free -->|No| Cap{8 spill threads<br/>running?}
    Cap -->|No| Spill[🧵 Spill thread]
    Cap -->|Yes| Refuse[🚫 tool_pool_exhausted]
    Pool --> Ok[✅ Result]
    Spill --> Ok

    classDef call fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef q fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef pool fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef ok fill:#10B981,stroke:#7C90A0,color:#fff
    classDef bad fill:#8B0000,stroke:#7C90A0,color:#fff

    class Call call
    class Free,Cap q
    class Pool,Spill pool
    class Ok ok
    class Refuse bad

With no tool_timeout set, nothing changes — async sync-tool calls still use the default executor and are not throttled.

Wrapper-Level Timeout (YAML / framework: praisonai)

When running through YAML or the CLI, the wrapper wraps every tool with a timeout-enforcing shim before handing the agent to the SDK. This provides defense-in-depth timeout enforcement even for pathological tools.

On timeout the wrapper raises ToolTimeoutError (a TimeoutError subclass) instead of returning a JSON dict. This preserves each tool’s declared return-type contract — a typed return value is never silently downgraded to a string. Framework adapters catch it and translate it per framework.

The wrapper handles sync and async tools differently:

• Sync tools run in an instance-owned ThreadPoolExecutor (see _get_tool_timeout_executor in agents_generator.py); on timeout the future is best-effort cancelled. A call that already started cannot be interrupted, so background_work_may_continue is True.
• Async tools are wrapped with asyncio.wait_for(...), which cancels the underlying task cleanly, so background_work_may_continue is False.

In Python, configure with tool_config=ToolConfig(timeout=…).

Exception raised on wrapper-level timeout:

from praisonaiagents import Agent
from praisonai.agents_generator import ToolTimeoutError

def slow_lookup(query: str) -> str:
    ...  # a tool that may hang

agent = Agent(
    name="Researcher",
    instructions="Look things up.",
    tools=[slow_lookup],
    tool_config={"timeout": 5},  # ToolConfig(timeout=5) also works
)

try:
    agent.start("Look up X")
except ToolTimeoutError as e:
    print(e.tool_name, e.timeout_seconds, e.background_work_may_continue)

To catch the executor-level version instead, use from praisonaiagents.tools import ToolTimeoutError — see Tool Call Executor Timeout.

ToolTimeoutError carries three attributes:

See Tool Configuration for full details and Concurrency for shape comparison.

Task-Scoped Tools

The tools_override parameter allows tasks to provide their own tool set for async execution:

from praisonaiagents.agent.execution_mixin import execute_tool_async

# Internal usage (SDK implementation detail)
result = await execute_tool_async(
    agent=agent,
    tool_call=tool_call,
    tools_override=task.tools  # Task tools take precedence
)

For users, this manifests as task-specific tools being available during async chat:

from praisonaiagents import Agent, Task

def task_specific_tool():
    return "Task-specific result"

agent = Agent(name="Base Agent", tools=[])

task = Task(
    description="Use task tool",
    agent=agent,
    tools=[task_specific_tool]
)

# task_specific_tool is available during task execution
# even though agent has no tools

Trace Events

Async tool execution now emits the same trace events as sync execution:

from praisonaiagents import Agent

# Trace events work the same for async calls
agent = Agent(
    name="Traced Agent",
    tools=[my_tool],
    # Observability hooks capture async tool calls
)

# Events emitted during async execution
await agent.achat("Use my tool")

Migration Notes

No code changes required – async tool safety is automatically enabled:

Before: Async calls bypassed safety mechanisms

# Previously: approval, circuit breaker, etc. were skipped
await agent.achat("Use dangerous tool")  # No approval prompt
# Async tool calls bypassed the circuit breaker even after 5 consecutive failures

After: Async calls use full safety pipeline

# Now: full safety pipeline including approval
await agent.achat("Use dangerous tool")  # Approval prompt appears
# Async tool calls open the breaker after 5 failures and short-circuit, same as sync (PR #4469)

Best Practices

import asyncio
from praisonaiagents import Agent

agent = Agent(require_approval=True, tools=[my_tool])

async def safe_async_execution():
    # Approval prompts work in async context
    result = await agent.achat("Use tool")
    return result

# Run with proper event loop
asyncio.run(safe_async_execution())

from praisonaiagents import Agent
from praisonaiagents.config import ToolConfig

agent = Agent(
    name="Async Agent",
    tools=[async_api_call],
    tool_config=ToolConfig(timeout=30)  # 30 second limit for async tools
)

from praisonaiagents import Agent

async def monitored_workflow():
    for i in range(10):
        try:
            result = await agent.achat(f"Process item {i}")
        except ToolExecutionError as e:
            if "circuit breaker" in str(e).lower():
                # Circuit breaker is open, wait before retrying
                await asyncio.sleep(60)

from praisonaiagents import Agent, Task

# Agent tools
def general_tool():
    return "General purpose"

# Task-specific tools (will override agent tools)  
def specialized_tool():
    return "Task-specific operation"

agent = Agent(tools=[general_tool])

task = Task(
    description="Specialized work",
    agent=agent,
    tools=[specialized_tool]  # Only this tool available during task
)

Safe Defaults

On a fresh interactive session, the runtime now routes dangerous tools through the CLI approval backend automatically — no approval= kwarg needed. Off-TTY (pipes, CI) keeps deny-by-default. See Tool Approval → Default Behaviour for the precedence ladder and bypass flags, and Approval Backends → Which class runs the prompt? for how the Python (ConsoleBackend) and CLI (InteractiveCLIApprovalBackend) paths differ.

Related