Asynchronous and Parallel Tools
On this page
Asynchronous and Parallel Tools
In production, your tools call external APIs, query databases, or perform I/O operations. These operations take time, and that time adds up: an agent chaining five one-second calls takes five seconds to respond. Making tools asynchronous and leveraging the SDK’s native parallelism is what separates a demo from a usable service.
Asynchronous Tools
The @function_tool decorator natively supports async functions. No configuration required: declare async def, use an asynchronous HTTP client, and the SDK handles the rest.
import httpx
from agents import function_tool
@function_tool
async def search_company(name: str) -> str:
"""Searches for company information via the Societe.com API."""
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.societe.com/v1/search",
params={"q": name},
timeout=10,
)
if response.status_code == 200:
data = response.json()
return f"Company: {data['name']}, SIREN: {data['siren']}, Revenue: {data['revenue']}"
return f"No results for '{name}'"
@function_tool
async def verify_solvency(siren: str) -> str:
"""Verifies a company's solvency by its SIREN number."""
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.credit-scoring.com/v1/{siren}",
timeout=10,
)
data = response.json()
return f"Solvency score: {data['score']}/100, risk: {data['risk_level']}"
When the Runner executes an async tool, it doesn’t block the event loop. In a FastAPI server, this means other requests continue processing while your agent waits for the Societe.com API response.
Automatic Parallelism
Two independent asynchronous tools can run simultaneously. The model decides this, and your instructions can explicitly encourage it.
from agents import Agent, Runner
agent = Agent(
name="B2B Analyst",
instructions="""You analyze companies for business decisions.
When asked to analyze a company, retrieve its information
AND verify its solvency at the same time.""",
tools=[search_company, verify_solvency],
model="gpt-5.6-terra",
)
# The model will launch both tools in parallel
result = await Runner.run(
agent,
"Analyze the company Acme Corp (SIREN 123456789)"
)
Since the SIREN is provided in the query, the solvency check doesn’t wait for the company search: the SDK executes both calls simultaneously. The total time equals the duration of the slowest tool, not the sum of both.
Controlling Parallelism
Parallelism isn’t always desirable. If your tools write to the same record or if order matters for data consistency, force sequential execution.
from agents import Agent
agent = Agent(
name="Cautious Agent",
instructions="Execute tools one by one to ensure data consistency.",
tools=[search_company, verify_solvency],
model="gpt-5.6-terra",
parallel_tool_calls=False, # Forces sequential execution
)
The parallel_tool_calls=False parameter forces the model to call only one tool at a time. You trade latency for predictability.
Pattern: Aggregating Parallel Results
The most common practical case involves gathering data from multiple sources for the same entity. Here, a city’s weather and cultural agenda have nothing to do with each other, but the response needs both.
import json
from agents import Agent, Runner, function_tool
@function_tool
async def get_weather(city: str) -> str:
"""Gets current weather for a city."""
# Simulation - in production: weather API call
weather_data = {
"Paris": {"temp": 22, "conditions": "sunny"},
"Lyon": {"temp": 19, "conditions": "cloudy"},
"Marseille": {"temp": 26, "conditions": "sunny"},
}
data = weather_data.get(city, {"temp": 0, "conditions": "unknown"})
return json.dumps({"city": city, **data}, ensure_ascii=False)
@function_tool
async def get_events(city: str) -> str:
"""Gets upcoming events in a city."""
events = {
"Paris": ["Concert at Stade de France", "Exhibition at the Louvre"],
"Lyon": ["Festival of Lights", "Gastronomy Fair"],
"Marseille": ["Mediterranean Regatta"],
}
return json.dumps({
"city": city,
"events": events.get(city, [])
}, ensure_ascii=False)
agent = Agent(
name="Travel Agent",
instructions="""You are a travel agent. When asked
for recommendations for a city, retrieve weather AND events
simultaneously to provide complete advice.""",
tools=[get_weather, get_events],
model="gpt-5.6-terra",
)
# The model will call both tools in parallel for each city
result = await Runner.run(
agent,
"I want to visit Paris this weekend. What do you recommend?"
)
print(result.final_output)
Both tools return JSON, making it easier for the model to recombine the data. Note the instruction phrasing: the “AND … simultaneously” triggers the parallel call rather than a lazy chain.
Timeouts and Resilience
An API that never responds blocks your agent just as surely as a synchronous tool. Bound every slow call with a timeout and turn failures into intelligible messages.
import asyncio
from agents import function_tool
@function_tool
async def slow_api_call(query: str) -> str:
"""Calls an API that might be slow."""
try:
async with asyncio.timeout(5): # 5-second timeout
# Your API call here
await asyncio.sleep(0.1) # Simulation
return f"Result for '{query}'"
except asyncio.TimeoutError:
return "The API did not respond in time. Please try again."
Performance: Impact of Parallelism
Let’s put the scale in perspective. Three 500 ms tools executed sequentially cost 1500 ms; the same three tools in parallel cost about 500 ms—the duration of the slowest one. For an agent using many tools, the difference is immediately noticeable in practice.
Key Takeaways
- Use
asyncfor tools that make network or I/O calls - The SDK automatically executes independent tools in parallel
parallel_tool_calls=Falseforces sequential execution when needed- Add timeouts with
asyncio.timeout()for resilience - Parallelism can divide latency by the number of tools