Concurrency

Limit parallel agent runs and bound tool execution time

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

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

2. Same, async

   Use async context for better resource utilization:

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

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

Retune the limit safely at any time

Call set_limit() at any time — even while agents already hold permits — and the new cap takes effect immediately without letting phantom permits leak.

from praisonaiagents.agent.concurrency import ConcurrencyRegistry

reg = ConcurrencyRegistry()
reg.set_limit("worker", 3)

# ... three "worker" runs are already in flight ...

reg.set_limit("worker", 5)  # widen mid-flight — the same limiter is resized in place
reg.set_limit("worker", 2)  # shrink again — in-flight holders stay on the same counter

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.

Dynamic reconfiguration safety

Since PR #5258, set_limit() and remove_limit() are safe to call while permits are still outstanding. Each acquire() / acquire_sync() now returns the exact semaphore it obtained, and the matching release() frees that same instance — so a runtime cap change never redirects a release onto a fresh semaphore and never silently inflates the effective cap.

Before the fix, calling set_limit("agent", N) or remove_limit("agent") while a permit was outstanding could permanently corrupt the cap: the outstanding release landed on a newer semaphore, which threading.Semaphore (unlike BoundedSemaphore) accepts without raising. From that point on the configured cap was meaningless and the drift was undetectable from the public API.

sequenceDiagram
    participant Caller
    participant Registry as ConcurrencyRegistry
    participant S1 as Semaphore (old)
    participant S2 as Semaphore (new)

    Caller->>Registry: acquire("agent")
    Registry->>S1: acquire permit
    Registry-->>Caller: return S1 handle
    Note over Registry,S2: set_limit("agent", N) mid-flight → S2 replaces S1
    Caller->>Registry: release("agent")
    Registry->>S1: release the acquired handle (S1), not S2
    Note over S2: new cap stays intact — no silent inflation

    classDef reg fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef old fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef new fill:#10B981,stroke:#7C90A0,color:#fff

    class Registry reg
    class S1 old
    class S2 new

Sync vs Async Rule

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

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=…).

graph TB
    Call[🛠️ Tool call] --> Which{framework = praisonai<br/>AND plain callable?}
    Which -->|Yes| Core[🧠 Core enforces<br/>Agent.tool_config.timeout]
    Which -->|No, framework tool object<br/>or non-native framework| Wrap[🛡️ Wrapper enforces<br/>per-run ThreadPoolExecutor]
    Core --> CoreTO{Within timeout?}
    Wrap --> WrapTO{Within timeout?}
    CoreTO -->|Yes| OK1[✅ result]
    CoreTO -->|No| Dict["{error: Tool timed out after Ns, timeout: true}"]
    WrapTO -->|Yes| OK2[✅ result]
    WrapTO -->|No| Err[raise ToolTimeoutError]

    classDef call fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef core fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef wrap fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef check fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef ok fill:#10B981,stroke:#7C90A0,color:#fff
    classDef error fill:#8B0000,stroke:#7C90A0,color:#fff

    class Call call
    class Which,CoreTO,WrapTO check
    class Core core
    class Wrap wrap
    class OK1,OK2 ok
    class Dict,Err error

Timeout Return Shape

On timeout, each layer surfaces the timeout differently:

Effective Timeout Precedence

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

1. 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.
2. 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.
3. 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).
4. 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.

The timeout pool size (not the per-tool budget above) is per-run configurable via tool_timeout_workers on AgentsGenerator (or cli_config["tool_timeout_workers"]), falling back to PRAISONAI_TOOL_TIMEOUT_WORKERS (default 32). See Tool Configuration.

Executor Details

• One executor per Agent instance (lazy creation)
• max_workers=5 threads per agent — matches the default ParallelToolCallExecutor width, so a parallel batch of up to 5 tool calls never queues
• tool_timeout starts when the tool body starts, not when it is submitted — time spent waiting for a worker never times out a healthy tool, but that wait is itself capped at tool_timeout so a pool stuck behind hung workers still fails
• 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

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

Calling registry.set_limit("chat_agent", N) again on a running server — a config reload or hot re-tune — is now correct even with requests in flight. Each in-flight request releases the exact semaphore it acquired, so the new cap takes effect for subsequent acquires without the old outstanding releases inflating it (PR #5258).

@app.post("/admin/retune")
async def retune(new_limit: int):
    registry.set_limit("chat_agent", new_limit)  # safe with permits outstanding
    return {"chat_agent_limit": new_limit}

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

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

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

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
)

# 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),
)

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

Shared JSON state under concurrent writes

BaseJSONStore‘s file lock now tolerates long critical sections without corruption. It judges a lock as stale by the lock file’s mtime age and refreshes that mtime with a background heartbeat while held, so a legitimately slow read-modify-write (large JSON write, slow disk) stays exclusive instead of being reclaimed by an impatient waiter and clobbered. See Thread Safety for details (PR #5136).

Related