Concurrency

from praisonaiagents import Agent

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

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

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

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

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

@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}

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

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
    )

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

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

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