Async Tool Safety - PraisonAI

Async tool calls in agent.achat() now route through the same safety mechanisms as sync execution, including approval checks, circuit breakers, and timeouts.

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	✅	✅

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

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)

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)

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

Core Controls

Concurrency

​Quick Start

​How It Works

​Safety Mechanisms

​Approval Checks

​Circuit Breaker Protection

​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

Quick Start

How It Works

Safety Mechanisms

Approval Checks

Circuit Breaker Protection

Timeout Controls

Wrapper-Level Timeout (YAML / framework: praisonai)

Task-Scoped Tools

Trace Events

Migration Notes

Best Practices

Safe Defaults

Related