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

​
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

Safety LayerSync ModeAsync 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, 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. The wrapper handles sync and async tools differently:
  • Sync tools run in a daemon thread (threading.Thread(daemon=True)); on timeout the wrapper returns a JSON error dict and abandons the thread (daemon threads do not block process exit).
  • Async tools are wrapped with asyncio.wait_for(...); on timeout the wrapper returns the same JSON error dict.
In Python, configure with tool_config=ToolConfig(timeout=…). JSON return shape on wrapper-level timeout: 
{
  "error": "tool_timeout",
  "tool": "<function name>",
  "timeout_seconds": 30
}
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:
EventWhenData
TOOL_CALL_STARTBefore executiontool_name, arguments
TOOL_CALL_RESULTAfter executionresult, 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

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())
Set appropriate timeouts for async tool execution:
from praisonaiagents import Agent, ToolConfig

agent = Agent(
    name="Async Agent",
    tools=[async_api_call],
    tool_config=ToolConfig(timeout=30)  # 30 second limit for async tools
)
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)
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
)

Approval

Tool approval configuration

Tool Circuit Breaker

Tool failure protection