Concurrency

Concurrency controls let you limit parallel agent execution and set timeouts for tool calls to prevent resource exhaustion.

from praisonaiagents import Agent

agent = Agent(
    name="worker",
    instructions="Respect concurrency limits across tool calls",
)

agent.start("Process these jobs without overloading APIs")

The user runs parallel work; concurrency controls cap simultaneous operations and tool timeouts.

Quick Start

Limit parallel runs of an agent

Control how many instances of the same agent can run concurrently:

from praisonaiagents import Agent
from praisonaiagents.agent.concurrency import ConcurrencyRegistry

registry = ConcurrencyRegistry()
registry.set_limit("researcher", 2)  # at most 2 concurrent runs

agent = Agent(name="researcher", instructions="Research topics")

# Sync context
registry.acquire_sync("researcher")
try:
    agent.start("Research Mars exploration")
finally:
    registry.release("researcher")

Same, async

Use async context for better resource utilization:

await registry.acquire("researcher")
try:
    await agent.astart("Research Mars exploration")
finally:
    registry.release("researcher")

Bound tool time with ToolConfig

Prevent slow tools from blocking agent execution:

from praisonaiagents import Agent
from praisonaiagents.config import ToolConfig

agent = Agent(
    name="Assistant",
    instructions="Use tools to help users",
    tools=["get_weather"],
    tool_config=ToolConfig(timeout=30),  # seconds; slow tools return a timeout dict
)
agent.start("What's the weather in Tokyo?")

How It Works

Component	Purpose	Thread Safety
`ConcurrencyRegistry`	Limits parallel agent runs — a true global cap across every event loop and thread	✅ Thread-safe
Tool Executor	Runs tools with timeout	✅ Per-agent pool
Plugin APIs	Enable/disable plugins	✅ Lock-protected
`AgentTeam` / `PraisonAIAgents` run guard	Refuses concurrent runs on one instance	✅ Per-instance re-entrancy lock

PraisonAI’s concurrency controls at a glance:

ConcurrencyRegistry caps how many agent runs happen in parallel.
Per-agent tool executor applies tool_timeout and recycles a hung worker.
AgentTeam / PraisonAIAgents instances are single-run — a second concurrent start() / astart() on the same object raises RuntimeError. Batched runs via start_for_each / astart_for_each are unaffected. See AgentTeam Batch Runs → Concurrent-run safety.

Global cap across loops and threads. Each agent’s limit is backed by a loop-neutral threading.Semaphore keyed by agent name, so the configured cap holds across every event loop and thread. Sync and async callers share the same per-agent permit pool. Multi-loop / multi-thread deployments no longer crash with “bound to a different event loop”. See PR #5050.

Sync vs Async Rule

acquire_sync() is safe from any context; in async code prefer await acquire() to avoid blocking the loop.

Context	Method	What happens
Sync	`registry.acquire_sync(name)`	✅ Works correctly
Async	`await registry.acquire(name)`	✅ Works correctly — yields to the loop
Mixed	`acquire_sync()` in async	⚠️ Works, but blocks the calling thread — prefer `await acquire()`

acquire_sync() is safe to call from any context (sync or async). Under the hood it acquires a loop-neutral threading.Semaphore, so it never raises “bound to a different event loop”. In an async context, however, it blocks the current thread until a permit is available — prefer await acquire() there so other tasks on the loop keep running. Sync and async callers share the same per-agent permit pool, so the configured limit is a true global cap across every event loop and thread.

Prefer await acquire() in async code:

async def works_but_blocks():
    registry.acquire_sync("agent")  # ⚠️ permitted, but blocks the event loop thread

async def recommended():
    await registry.acquire("agent")  # ✅ yields to the loop

Tool Timeout Behavior

When tool_config=ToolConfig(timeout=...) is set, tools run in a dedicated executor with these characteristics: In YAML the field name is still tool_timeout:; in Python use tool_config=ToolConfig(timeout=…).

Timeout Return Shape

On timeout, each layer surfaces the timeout differently:

Layer	Trigger	Behaviour
SDK Agent executor	`Agent(tool_config=ToolConfig(timeout=30))` directly in Python	`{"error": "Tool timed out after 30s", "timeout": True}` (seconds)
Wrapper boundary	YAML `tool_timeout: 30` or CLI `--tool-timeout 30` (`framework: praisonai`)	Raises `praisonai.agents_generator.ToolTimeoutError` (a `TimeoutError` subclass) with `tool_name`, `timeout_seconds`, and `background_work_may_continue` (seconds). Framework adapters catch it and translate it per framework. See Async Tool Safety → Wrapper-Level Timeout.
SDK tool-call executor	`Agent(llm={"tool_timeout_ms": 30000})`	Returns `ToolResult` with `error=praisonaiagents.tools.ToolTimeoutError` and `error_kind="timeout"` (milliseconds). See Tool Call Executor Timeout.

Effective Timeout Precedence

When tool_timeout values are declared, the wrapper resolves each agent’s budget independently:

CLI wins for every agent. An explicit --tool-timeout N on the command line (or cli_config={"tool_timeout": N} when embedding) is used verbatim for all agents.
Uniform declared values take the shared-wrap fast path. When every agent under roles: and agents: declares tool_timeout and every declared value is identical, a single guard wraps the shared tool dict. If any agent omits the field, the shared wrap is skipped and the per-agent resolver runs instead.
Heterogeneous per-agent values are honoured per agent. When agents declare different values, each agent’s tools carry its own budget (tool objects stay shared; only the guard closure differs). The tightest value no longer collapses onto every agent. Agents that omit tool_timeout get no wrap (they do not inherit another agent’s declared value).
Otherwise, no wrapping. If nothing declares a timeout, tools run without wrapper-layer enforcement (the SDK executor-layer enforcement still applies if tool_config=ToolConfig(timeout=…) is set in Python).

The uniform path is AgentsGenerator._resolve_uniform_tool_timeout(config); heterogeneous budgets use make_agent_tool_wrap_resolver(config) / resolve_agent_tool_timeout(agent_key, config) — see praisonai/agents_generator.py.

As of PR #4477, heterogeneous per-agent tool_timeout values are honoured per agent instead of collapsing to the tightest value. A slow agent that previously inherited a fast agent’s tighter budget (masking a real timeout) may now run longer. Uniform declarations are unaffected. (Earlier, PR #3176 had reversed the collapse from max() to min().) PR #4468 follows up by tightening uniform detection: a single declared tool_timeout on one agent no longer counts as uniform — if any agent omits the field, undeclared agents get no wrap instead of silently inheriting the lone declared budget.

YAML boolean values are ignored, not coerced. Because bool subclasses int in Python, tool_timeout: yes or tool_timeout: true used to silently become a 1-second cap on every tool. As of PR #2609 the resolver explicitly rejects bool values — such entries are treated as “not declared” and fall through to the next precedence level. Use an integer or float (e.g. tool_timeout: 30).

Executor Details

One executor per Agent instance (lazy creation)
max_workers=2 threads per agent
Thread name prefix: tool-<agent_name> — useful for log filtering
Reused across calls — no resource leak
Recycled on timeout — a tool that hangs past tool_timeout is not reclaimable, so the executor is shutdown(wait=False) and the next call gets a fresh worker

Self-healing after a hang. The pool has 2 workers by default; before PR #3960, two consecutive hangs would deadlock the pool because future.cancel() cannot stop a thread that has already started. Now the executor is recycled on timeout — the next call starts on a fresh worker — so a hung tool cannot progressively degrade throughput toward a deadlock. Recycling is bounded so repeated timeouts cannot leak an unbounded number of stuck threads.

Which timeout to choose:

Common Patterns

Limit FastAPI Route Concurrency

from praisonaiagents import Agent
from praisonaiagents.agent.concurrency import ConcurrencyRegistry

registry = ConcurrencyRegistry()
registry.set_limit("chat_agent", 5)

@app.post("/chat")
async def chat_endpoint(message: str):
    await registry.acquire("chat_agent")
    try:
        agent = Agent(name="chat_agent", instructions="Help users")
        response = await agent.astart(message)
        return {"response": response}
    finally:
        registry.release("chat_agent")

Async Context Manager Helper

from contextlib import asynccontextmanager

@asynccontextmanager
async def throttled_agent(name: str, max_concurrent: int = 3):
    registry = ConcurrencyRegistry()
    registry.set_limit(name, max_concurrent)
    await registry.acquire(name)
    try:
        yield
    finally:
        registry.release(name)

# Usage
async with throttled_agent("researcher", 2):
    agent = Agent(name="researcher", instructions="Research topics")
    result = await agent.astart("Study quantum computing")

Timeout Selection by Tool Type

from praisonaiagents import Agent
from praisonaiagents.config import ToolConfig

def get_agent_with_timeouts():
    return Agent(
        name="MultiTool Assistant",
        instructions="Help with various tasks",
        tools=[
            "web_search",      # Network IO
            "file_processor",  # Local computation  
            "simple_math"      # Fast operation
        ],
        tool_config=ToolConfig(timeout=45)  # Good balance for mixed workload
    )

Best Practices

Always release in finally blocks

Prevents deadlocks when exceptions occur:

registry.acquire_sync("agent")
try:
    # Agent work here
    agent.start("task")
finally:
    registry.release("agent")  # Always runs

Prefer async acquire in async code

Mixing is no longer an error — acquire_sync() in an async context simply blocks its thread. Sync and async callers contend for the same per-agent permit pool (see the test_sync_and_async_share_one_pool test in PR #5050). Still prefer await acquire() in async code so the event loop keeps running:

# ✅ Good - sync context, sync acquire
def sync_handler():
    registry.acquire_sync("agent")
    try:
        agent.start("task")
    finally:
        registry.release("agent")

# ✅ Good - async context, async acquire  
async def async_handler():
    await registry.acquire("agent")  # yields to the loop
    try:
        await agent.astart("task")
    finally:
        registry.release("agent")

Set tool_timeout for network tools

Any tool that does network IO should have a timeout:

from praisonaiagents import Agent
from praisonaiagents.config import ToolConfig

# Tools that need timeouts
network_tools = ["web_search", "api_call", "download_file"]
local_tools = ["calculate", "format_text", "parse_json"]

agent = Agent(
    name="Assistant",
    tools=network_tools + local_tools,
    tool_config=ToolConfig(timeout=30)  # Protects against slow network
)

Use thread names for debugging

Filter logs by agent name using the thread prefix:

# Filter tool execution logs by agent
grep "tool-researcher" app.log

# Or in Python logging
import logging
logging.basicConfig(format='%(threadName)s: %(message)s')

Retries

Tool failures can be automatically retried using the retry policy feature. This works alongside timeouts to handle transient errors:

from praisonaiagents import Agent, ExecutionConfig
from praisonaiagents.config import ToolConfig
from praisonaiagents.tools.retry import RetryPolicy

agent = Agent(
    name="resilient_agent",
    tools=[web_search, api_tool],
    tool_config=ToolConfig(
        timeout=30,
        retry_policy=RetryPolicy(
            max_attempts=3,
            retry_on={"timeout", "rate_limit", "connection_error"},
            backoff_factor=2.0,
            jitter=True
        )
    ),
    execution=ExecutionConfig(parallel_tool_calls=True),
)

ToolConfig.parallel is a deprecated alias for ExecutionConfig.parallel_tool_calls. Enable parallel tool calls with execution=ExecutionConfig(parallel_tool_calls=True) alongside ToolConfig for timeout and retry. Do not set both spellings to conflicting values \u2014 that raises TypeError.

For complete retry configuration and error handling strategies, see Tool Retry Policy.

Parallel tool calls inside one async turn

A single astart(...) turn can itself dispatch multiple independent tool calls concurrently when parallel_tool_calls=True.

import asyncio
from praisonaiagents import Agent, ExecutionConfig, ToolConfig

agent = Agent(
    instructions="Fetch multiple independent sources and summarise.",
    tools=[fetch_weather, fetch_news, fetch_stocks],
    tool_config=ToolConfig(timeout=15),
    execution=ExecutionConfig(parallel_tool_calls=True),
)

# Before PR #4634 the three tools ran sequentially on astart() even though
# `parallel_tool_calls=True` was set (silently absorbed). Now they run
# concurrently via asyncio.gather inside the single async turn.
asyncio.run(agent.astart("Summarise conditions in New York."))

Two situations users conflate compose cleanly: asyncio.gather(agent.astart(a), agent.astart(b)) runs two agents in parallel, while parallel_tool_calls=True runs several tools inside one agent turn in parallel.

Write-conflict guard for shell-like tools

When parallel_tool_calls=True batches two or more tool calls in one turn, PraisonAI runs them sequentially instead of concurrently if any pair could touch the same file. As of PraisonAI PR #4907, the guard also catches shell-like tools whose write target lives in a command string rather than a path/file_path argument — execute_command, acp_execute_command, and execute_code. Two calls to any of these in the same batch, or one of them alongside any other write, force sequential fallback. Path-only reads (read_file, list tools, etc.) still run concurrently.

from praisonaiagents import Agent, ExecutionConfig
from praisonaiagents.tools.shell_tools import execute_command

agent = Agent(
    instructions="Write two lines to the same report file.",
    tools=[execute_command],
    execution=ExecutionConfig(parallel_tool_calls=True),
)

# Two concurrent `echo … > /tmp/report.txt` calls now run one after the
# other, protecting the file from a write-conflict race.
agent.start("Append 'A' then 'B' to /tmp/report.txt using execute_command.")

Tool Retry Policy

Automatically retry failed tool calls with exponential backoff

Tool Configuration

Tool timeout settings and performance tuning

Async Bridge

Safe sync↔async boundary crossing utilities

Thread Safety

Chat history and state protection mechanisms

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

Sync vs Async Rule

Tool Timeout Behavior

Timeout Return Shape

Effective Timeout Precedence

Executor Details

Common Patterns

Limit FastAPI Route Concurrency

Async Context Manager Helper

Timeout Selection by Tool Type

Best Practices

Retries

Parallel tool calls inside one async turn

Write-conflict guard for shell-like tools

Tool Retry Policy

Tool Configuration

Async Bridge

Thread Safety

​Quick Start

​How It Works

​Sync vs Async Rule

​Tool Timeout Behavior

​Timeout Return Shape

​Effective Timeout Precedence

​Executor Details

​Common Patterns

​Limit FastAPI Route Concurrency

​Async Context Manager Helper

​Timeout Selection by Tool Type

​Best Practices

​Retries

​Parallel tool calls inside one async turn

​Write-conflict guard for shell-like tools

​Related

Tool Retry Policy

Tool Configuration

Async Bridge

Thread Safety

Quick Start

How It Works

Sync vs Async Rule

Tool Timeout Behavior

Timeout Return Shape

Effective Timeout Precedence

Executor Details

Common Patterns

Limit FastAPI Route Concurrency

Async Context Manager Helper

Timeout Selection by Tool Type

Best Practices

Retries

Parallel tool calls inside one async turn

Write-conflict guard for shell-like tools

Related