Your agent is slow for a boring reason: it asks one question at a time.
Ask it "what's the weather in Jakarta, Singapore, and Tokyo?" and watch what happens. The model returns a single tool call. Your loop runs it, appends the result, and asks again. Three cities, three round trips. Each round trip costs a network hop plus a full model generation, so the answer takes roughly three times as long as it should.
The APIs you already use support several tool calls in one response. OpenAI returns a tool_calls array. Anthropic returns multiple tool_use blocks in a single assistant turn. Current models do this by default when the question benefits from it. Anthropic's docs say Claude 4 and later make parallel tool calls by default. OpenAI's default lets the model call several functions in one turn, with an opt-out switch. Most agent loops never take advantage, because most tutorials show the one-call-at-a-time version.
The fix is small: run the calls concurrently, and format the results the way the API expects. Copy-paste Python for both APIs below, plus the cases where sequential is the right call.
How a batched response looks
OpenAI returns tool_calls as an array on the message. This is a trimmed example from the official docs:
[
{
"id": "call_12345xyz",
"type": "function",
"function": {
"name": "get_weather",
"arguments": "{\"location\": \"Paris, France\"}"
}
},
{
"id": "call_67890abc",
"type": "function",
"function": {
"name": "get_weather",
"arguments": "{\"location\": \"Bogotá, Colombia\"}"
}
}
]
Two details matter. arguments is a JSON string, not an object, so parse it with json.loads. And the same tool can appear twice with different ids, so match results by id, not by name.
Anthropic says the same thing with different words. The response has stop_reason: "tool_use" and content holds several blocks of type tool_use, each with id, name, and input. You reply with one tool_result per block, matched by tool_use_id.
Both APIs assume you can handle several calls. OpenAI's docs put it plainly: "it is best practice to assume there are several."
Prerequisites
- Python 3.10+
pip install openai anthropic- An API key for whichever provider you test first. The code reads
OPENAI_API_KEYorANTHROPIC_API_KEYfrom the environment.
Step 1: prove your model batches
Before touching your loop, confirm the model returns multiple calls for a multi-part question. This script counts the tool_use blocks in one response:
from anthropic import Anthropic
client = Anthropic()
tools = [
{
"name": "get_weather",
"description": "Get the current weather in a given location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country, e.g. Jakarta, Indonesia",
}
},
"required": ["location"],
},
},
{
"name": "get_time",
"description": "Get the current time in a given timezone",
"input_schema": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "IANA timezone, e.g. Asia/Jakarta",
}
},
"required": ["timezone"],
},
},
]
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
tools=tools,
messages=[
{
"role": "user",
"content": "What's the weather in Jakarta and Tokyo, and what time is it in both cities?",
}
],
)
tool_uses = [block for block in response.content if block.type == "tool_use"]
print(f"{len(tool_uses)} tool calls in one response")
for tool in tool_uses:
print(f"- {tool.name}: {tool.input}")
You will likely see four calls. If you see one, the model chose to serialize. The troubleshooting section at the end explains why.
Step 2: the Anthropic loop
This is the full pattern. The tools here are read-only, so all calls run concurrently with asyncio.gather:
import asyncio
from anthropic import Anthropic
client = Anthropic() # reads ANTHROPIC_API_KEY
TOOLS = [
{
"name": "get_weather",
"description": "Get the current weather in a given location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country, e.g. Jakarta, Indonesia",
}
},
"required": ["location"],
},
},
{
"name": "get_time",
"description": "Get the current time in a given timezone",
"input_schema": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "IANA timezone, e.g. Asia/Jakarta",
}
},
"required": ["timezone"],
},
},
]
def run_tool(name: str, tool_input: dict) -> str:
"""Replace this with a real API call, DB query, or whatever your tool does."""
if name == "get_weather":
return f"Sunny, 31C in {tool_input['location']}"
if name == "get_time":
return f"14:30 WIB in {tool_input['timezone']}"
raise ValueError(f"Unknown tool: {name}")
async def execute(tool_use) -> dict:
await asyncio.sleep(1) # simulate network latency so tools actually overlap
try:
result = run_tool(tool_use.name, tool_use.input)
return {"type": "tool_result", "tool_use_id": tool_use.id, "content": result}
except Exception as exc:
return {
"type": "tool_result",
"tool_use_id": tool_use.id,
"is_error": True,
"content": str(exc),
}
async def main():
messages = [
{
"role": "user",
"content": "What's the weather in Jakarta and Tokyo, and what time is it in both cities?",
}
]
for _ in range(5): # hard cap on turns so a buggy loop can't run forever
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
tools=TOOLS,
messages=messages,
)
# The assistant turn is kept verbatim, tool_use blocks included.
messages.append({"role": "assistant", "content": response.content})
tool_uses = [block for block in response.content if block.type == "tool_use"]
if not tool_uses:
break
# Run all calls concurrently. Fine for read-only tools.
results = await asyncio.gather(*[execute(t) for t in tool_uses])
# All results go in ONE user message, matched by tool_use_id.
# tool_result blocks must come before any text in this message.
messages.append({"role": "user", "content": results})
final_text = next(
block.text for block in response.content if block.type == "text"
)
print(final_text)
if __name__ == "__main__":
asyncio.run(main())
The formatting rules are the part people get wrong, so repeat them until they stick:
- Keep the assistant turn verbatim. Its
contentarray holds thetool_useblocks and you append it as-is. - Return every result in a single user message. One user message per batch, not one per tool.
- Match each result with
tool_use_id, and put alltool_resultblocks before any text in that message. - If you skip a call, return
is_error: truewith a short explanation instead of dropping it.
The loop above, run against mock responses with one second of simulated latency per call, finished four calls in about one second instead of four. That is the whole point.
Step 3: the OpenAI loop
Same idea. tool_calls is an array on the message, and each call's arguments is a JSON string you parse with json.loads:
import asyncio
import json
from openai import OpenAI
client = OpenAI() # reads OPENAI_API_KEY
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country, e.g. Jakarta, Indonesia",
}
},
"required": ["location"],
},
},
},
{
"type": "function",
"function": {
"name": "get_time",
"description": "Get the current time in a given timezone",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "IANA timezone, e.g. Asia/Jakarta",
}
},
"required": ["timezone"],
},
},
},
]
def run_tool(name: str, args: dict) -> str:
"""Replace this with a real API call, DB query, or whatever your tool does."""
if name == "get_weather":
return f"Sunny, 31C in {args['location']}"
if name == "get_time":
return f"14:30 WIB in {args['timezone']}"
raise ValueError(f"Unknown tool: {name}")
async def execute(tool_call) -> dict:
await asyncio.sleep(1) # simulate network latency so tools actually overlap
try:
result = run_tool(
tool_call.function.name, json.loads(tool_call.function.arguments)
)
return {"role": "tool", "tool_call_id": tool_call.id, "content": result}
except Exception as exc:
return {
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"Error: {exc}",
}
async def main():
messages = [
{"role": "user", "content": "What's the weather in Jakarta and Tokyo?"},
]
for _ in range(5): # hard cap on turns so a buggy loop can't run forever
response = client.chat.completions.create(
model="gpt-5",
tools=TOOLS,
messages=messages,
)
message = response.choices[0].message
if not message.tool_calls:
print(message.content)
break
# Keep the assistant turn verbatim, tool_calls included.
messages.append(message)
# Run all calls concurrently. Fine for read-only tools.
results = await asyncio.gather(*[execute(tc) for tc in message.tool_calls])
# One "tool" message per call, matched by tool_call_id.
messages.extend(results)
if __name__ == "__main__":
asyncio.run(main())
The assistant message is appended as-is, tool_calls included. Then each result comes back as a tool message carrying the matching tool_call_id. That is the whole OpenAI contract.
Step 4: when parallel is the wrong call
The docs are explicit. Independent, read-only operations are safe to run in parallel for lower latency. Tools with side effects, shared state, or ordering requirements are better run sequentially.
Side effects are the classic case: send_email, write_file, debit_account. Firing those concurrently risks double sends and race conditions. If the model batches them anyway, run the batch sequentially and stop on the first failure. Return is_error: true for the calls you skipped, with a short note such as "Not executed: the preceding write_file call failed." The model reissues them on the next turn.
Dependent calls work the same way. Tool B needs tool A's output, and they arrived in one batch. Run them in order, stop on failure. To reduce how often dependent calls batch together, add this to the system prompt: "Only batch tool calls that are independent of each other."
Step 5: the control knobs
OpenAI:
| Parameter | Effect |
|---|---|
parallel_tool_calls: false |
zero or one tool per turn |
tool_choice: "required" |
one or more tools |
tool_choice: {"type": "function", "name": "get_weather"} |
exactly that one tool |
tool_choice: "none" |
no tools at all |
Anthropic:
| Parameter | Effect |
|---|---|
tool_choice: {"type": "auto", "disable_parallel_tool_use": true} |
at most one tool per turn, plain text answers still allowed |
{"type": "any"} or {"type": "tool", ...} plus disable_parallel_tool_use: true |
exactly one tool |
One gotcha: disable_parallel_tool_use goes inside the tool_choice object, not at the top level of the request.
When would you disable it? Rate-limited APIs. Tools that mutate shared state. Or when you need a step-by-step trace for auditing and every call must be attributable in order.
Pitfalls that will bite you
Result formatting kills parallelism. This is the number one cause. Send each tool result as its own user message and the model learns to serialize. All results go in a single user message, results before any text. OpenAI is the same idea: one tool message per tool_call_id, and return every id, including failed calls. Put the error text in content.
Rate limits. asyncio.gather fires everything at once. Ten tools means ten concurrent HTTP calls. Wrap execution in asyncio.Semaphore(3) when your tools hit third-party APIs.
Schemas cost tokens on every request. OpenAI injects function definitions into the system message and bills them as input tokens. Keep descriptions short, and do not ship twenty tools when five will do. If you need to restrict which tools are callable without dropping schemas, allowed_tools does that, and it also keeps prompt caching intact.
Tool output is untrusted. Results can carry injected instructions, a pattern called indirect prompt injection. Keep them inside tool_result blocks and never echo them into the system prompt.
Streaming changes the shape. tool_calls arrive as deltas keyed by an index. Aggregate by index until the stream ends, then run the same gather loop.
Measure it
Track the average number of tool calls per assistant turn. Above 1.0 means batching is working. If it sits at 1.0, formatting or prompting is the problem:
tool_call_messages = [
msg for msg in messages
if any(block.type == "tool_use" for block in msg.content)
]
total_calls = sum(
len([b for b in msg.content if b.type == "tool_use"])
for msg in tool_call_messages
)
print(f"avg tools per message: {total_calls / len(tool_call_messages):.2f}")
Next steps
- OpenAI's
strict: trueforces every call to match its schema. - The Anthropic SDK's Tool Runner handles this whole loop for you, error wrapping included, if you would rather not maintain it.
- If you build on MCP, the same loop applies. Hosts receive several tool calls and execute them the same way.
My take: start with the manual loop so you actually understand the format rules, measure it, then decide if an abstraction is worth it.
The whole trick is small: assume several calls per turn, run the independent ones concurrently, and keep results formatted the way the API expects. That alone removes most of the latency from a naive agent loop.