> ## Documentation Index > Fetch the complete documentation index at: https://praison.ai/docs/llms.txt > Use this file to discover all available pages before exploring further. # 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. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph LR subgraph "Concurrency Control Flow" User[👤 User] --> Registry[🛂 ConcurrencyRegistry] Registry --> Agent[🧠 Agent] Agent --> ToolPool[🔧 Tool Pool
2-thread executor] ToolPool --> Timeout{⏱️ Timeout?} Timeout -->|Yes| Error[❌ Timeout Result] Timeout -->|No| Success[✅ Tool Result] end classDef user fill:#8B0000,stroke:#7C90A0,color:#fff classDef registry fill:#6366F1,stroke:#7C90A0,color:#fff classDef pool fill:#F59E0B,stroke:#7C90A0,color:#fff classDef result fill:#10B981,stroke:#7C90A0,color:#fff classDef error fill:#8B0000,stroke:#7C90A0,color:#fff classDef agent fill:#8B0000,color:#fff classDef tool fill:#189AB4,color:#fff class User user class Registry registry class Agent agent class ToolPool pool class Success result class Error error ``` ## Quick Start Control how many instances of the same agent can run concurrently: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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") ``` Use async context for better resource utilization: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} await registry.acquire("researcher") try: await agent.astart("Research Mars exploration") finally: registry.release("researcher") ``` Prevent slow tools from blocking agent execution: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonaiagents.config.feature_configs 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 ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant User participant Registry as ConcurrencyRegistry participant Agent participant Executor as Tool Executor User->>Registry: acquire_sync("researcher") Registry->>Registry: Check semaphore limit Registry->>Agent: Grant execution slot Agent->>Executor: submit(tool, timeout=30s) alt Tool completes in time Executor->>Agent: Return result else Tool times out Executor->>Agent: {"error": "Tool timed out after 30s", "timeout": True} end Agent->>User: Response User->>Registry: release("researcher") ``` | Component | Purpose | Thread Safety | | --------------------- | -------------------------- | ---------------- | | `ConcurrencyRegistry` | Limits parallel agent runs | ✅ Thread-safe | | Tool Executor | Runs tools with timeout | ✅ Per-agent pool | | Plugin APIs | Enable/disable plugins | ✅ Lock-protected | *** ## Sync vs Async Rule The concurrency registry enforces strict separation between sync and async contexts: | Context | Method | What happens if you mix | | --------- | ------------------------------ | ----------------------- | | **Sync** | `registry.acquire_sync(name)` | ✅ Works correctly | | **Async** | `await registry.acquire(name)` | ✅ Works correctly | | **Mixed** | `acquire_sync()` in async | ❌ Raises `RuntimeError` | Calling `acquire_sync()` from an async context raises `RuntimeError("acquire_sync('') cannot be called with a running event loop; use async acquire() in async contexts.")`. Use `await acquire()` instead. **Example error:** ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} async def bad_example(): registry.acquire_sync("agent") # RuntimeError! async def good_example(): await registry.acquire("agent") # ✅ Correct ``` *** ## 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=…)`.* ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph TB subgraph "Two-Layer Tool Timeout" Call[Tool Call] --> WrapShim[Wrapper Shim
tool_timeout enforced] WrapShim --> Check{Has timeout?} Check -->|No| Direct[Direct execution] Check -->|Yes| Pool[Per-agent 2-thread pool] Pool --> Submit[submit(tool, timeout)] Submit --> Wait{Within timeout?} Wait -->|Yes| Result[Return result] WrapShim -.->|wrapper timeout| Dict1[raise ToolTimeoutError] Wait -->|No| Dict2[{"error": "Tool timed out after Ns", "timeout": true}] end classDef call fill:#8B0000,stroke:#7C90A0,color:#fff classDef safety fill:#189AB4,stroke:#7C90A0,color:#fff classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff classDef result fill:#10B981,stroke:#7C90A0,color:#fff classDef error fill:#8B0000,stroke:#7C90A0,color:#fff class Call call class WrapShim safety class Check,Pool,Submit process class Direct,Result result class Dict1,Dict2 error ``` ### 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](/docs/features/async-tool-safety#wrapper-level-timeout-yaml-framework-praisonai). | | 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](/docs/features/tool-call-executor-timeout). | ### 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`. As of [PR #4477](https://github.com/MervinPraison/PraisonAI/pull/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](https://github.com/MervinPraison/PraisonAI/pull/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-` — 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](https://github.com/MervinPraison/PraisonAI/pull/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:** ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} flowchart TD Start[Tool Type] --> IO{Network IO?} IO -->|Yes| Net[30-60s] IO -->|No| Local{Local computation?} Local -->|Yes| CPU[5-10s] Local -->|No| None[No timeout needed] classDef decision fill:#6366F1,stroke:#7C90A0,color:#fff classDef timeout fill:#189AB4,stroke:#7C90A0,color:#fff class IO,Local decision class Net,CPU,None timeout ``` *** ## Common Patterns ### Limit FastAPI Route Concurrency ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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 ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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 ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonaiagents.config.feature_configs 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 Prevents deadlocks when exceptions occur: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} registry.acquire_sync("agent") try: # Agent work here agent.start("task") finally: registry.release("agent") # Always runs ``` Keep acquisition method consistent with execution context: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # ✅ 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") try: await agent.astart("task") finally: registry.release("agent") ``` Any tool that does network IO should have a timeout: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent from praisonaiagents.config.feature_configs 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 logs by agent name using the thread prefix: ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # 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: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents import Agent, ExecutionConfig from praisonaiagents.config.feature_configs import ToolConfig from praisonaiagents.tools.retry import RetryPolicy # Enable retry with defaults agent = Agent( name="resilient_agent", tools=[web_search, api_tool], tool_retry_policy=RetryPolicy() ) # Using consolidated tool config (preferred) agent = Agent( name="modern_agent", tools=[flaky_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](/docs/features/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`. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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](https://github.com/MervinPraison/PraisonAI/pull/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. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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.") ``` *** ## Related Automatically retry failed tool calls with exponential backoff Tool timeout settings and performance tuning Safe sync↔async boundary crossing utilities Chat history and state protection mechanisms