Async Tool Safety

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). The sync-only execute_tool_with_circuit_breaker wrapper is not reused on the async path (it uses time.sleep); circuit-breaker state tracked elsewhere still applies.

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

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

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

Safety Layer	Sync Mode	Async Mode
Approval checks	✅	✅
Argument type casting	✅	✅
Circuit breaker	✅	✅
Tool timeout	✅	✅
Doom-loop tracking	✅	✅
Trace events	✅	✅
Output truncation	✅	✅
Error wrapping	✅	✅
Tool retry policy for raised exceptions	✅	✅ ¹

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:

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

Tool failure rates are tracked across async calls:

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

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.
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) — programming errors (ValueError, TypeError, AttributeError) and ToolExecutionError(is_retryable=False) (including Loop Guard HALT) stay terminal. See Tool Retry Policy for the full policy reference.

Timeout Controls

Async tool execution respects timeout settings:

from praisonaiagents import Agent
from praisonaiagents.config.feature_configs 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

Wrapper-Level Timeout (YAML / framework: praisonai)

Two ToolTimeoutError classes exist in PraisonAI. This section covers praisonai.agents_generator.ToolTimeoutError — the wrapper-level version raised on YAML/CLI tool_timeout (seconds). The SDK tool-call executor has its own praisonaiagents.tools.ToolTimeoutError (milliseconds, llm={"tool_timeout_ms": N}), surfaced as a ToolResult.error with error_kind="timeout". See Tool Call Executor Timeout.

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:

Attribute	Type	Description
`tool_name`	`str`	Name of the tool that timed out
`timeout_seconds`	`float`	The per-call limit that was exceeded
`background_work_may_continue`	`bool`	`True` only when a started sync worker could not be cancelled; async tools cancel cleanly (`False`)

Once half the pool’s workers are permanently leaked to stuck sync tools, the pool is automatically recycled: leaked threads continue until their syscall returns, but new tool calls get a fresh pool instead of queueing behind them.

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:

Event	When	Data
`TOOL_CALL_START`	Before execution	tool_name, arguments
`TOOL_CALL_RESULT`	After execution	result, duration, errors

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

After: Async calls use full safety pipeline

# Now: full safety pipeline including approval
await agent.achat("Use dangerous tool")  # Approval prompt appears

Best Practices

Handle Approval in Async Context

When using approval in async environments, ensure your event loop can handle user input:

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

Configure Timeouts for Async Tools

Set appropriate timeouts for async tool execution:

from praisonaiagents import Agent
from praisonaiagents.config.feature_configs import ToolConfig

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

Monitor Circuit Breaker in Async Workflows

Circuit breaker state affects all calls - monitor in async workflows:

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)

Task Tools Override Agent Tools

Design task tools to be self-contained since they override agent tools:

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

Approval

Tool approval configuration

Tool Circuit Breaker

Tool failure protection

Tool Approval

Safe-by-default behaviour, bypass flags, and risk levels

Loop Guard

Always-on per-turn tool-call guardrails, now on both sync and async paths

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

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Quick Start

How It Works

Safety Mechanisms

Approval Checks

Circuit Breaker Protection

Loop Guard on the async path

Retry parity on the async path

Timeout Controls

Wrapper-Level Timeout (YAML / framework: praisonai)

Task-Scoped Tools

Trace Events

Migration Notes

Best Practices

Safe Defaults

Approval

Tool Circuit Breaker

Tool Approval

Loop Guard

​Quick Start

​How It Works

​Safety Mechanisms

​Approval Checks

​Circuit Breaker Protection

​Loop Guard on the async path

​Retry parity on the async path

​Timeout Controls

​Wrapper-Level Timeout (YAML / framework: praisonai)

​Task-Scoped Tools

​Trace Events

​Migration Notes

​Best Practices

​Safe Defaults

​Related

Approval

Tool Circuit Breaker

Tool Approval

Loop Guard

Footnotes

Quick Start

How It Works

Safety Mechanisms

Approval Checks

Circuit Breaker Protection

Loop Guard on the async path

Retry parity on the async path

Timeout Controls

Wrapper-Level Timeout (YAML / framework: praisonai)

Task-Scoped Tools

Trace Events

Migration Notes

Best Practices

Safe Defaults

Related