How to Implement Parallel Tool Execution in Agents

How to Implement Parallel Tool Execution in Agents

Modern AI agents increasingly rely on external tools to accomplish complex tasks — fetching data, querying databases, calling APIs, and running computations. When an agent needs to invoke multiple independent tools to answer a single user query, executing them sequentially creates unnecessary latency. Parallel tool execution solves this by allowing the agent to dispatch several tool calls at once, dramatically reducing response times and improving throughput. In this tutorial, we'll explore what parallel tool execution is, why it matters, how to implement it from scratch, and the best practices you should follow.

What Is Parallel Tool Execution?

Parallel tool execution is the ability of an AI agent to issue multiple tool calls in a single reasoning step and execute them concurrently rather than one after another. Instead of waiting for Tool A to finish before calling Tool B, the agent identifies that both tools are needed and dispatches them simultaneously. Once all results return, the agent aggregates them and continues its reasoning.

Many modern LLMs — including OpenAI's GPT-4 Turbo and later, Anthropic's Claude 3.5, and Gemini 1.5 — natively support parallel function calling. The model returns multiple tool-call objects in a single response, and the orchestration layer is responsible for executing them concurrently and feeding the results back.

Why It Matters

• Lower latency: If each tool call takes 500ms and you have four independent calls, sequential execution takes 2 seconds. Parallel execution can bring that down to roughly 500ms.
• Better user experience: Faster responses feel more conversational and keep users engaged.
• Cost efficiency: Fewer round-trips to the LLM means fewer tokens spent on intermediate reasoning and less compute overhead.
• Scalability: Agents that fan out to many data sources can handle higher load when calls are parallelized.
• Natural reasoning: Some questions inherently require multiple independent lookups. Forcing sequential execution adds artificial constraints to the agent's logic.

How to Use It: A Practical Implementation

Let's build a minimal agent loop that supports parallel tool execution. We'll use Python with asyncio for concurrency, and we'll simulate an OpenAI-style API that returns multiple tool calls in one response.

First, let's define a few simple tools that our agent can call:

import asyncio
import time
from typing import Any, Callable

# Registry of available tools
TOOLS: dict[str, Callable] = {}

def tool(name: str):
    def decorator(fn):
        TOOLS[name] = fn
        return fn
    return decorator

@tool("get_weather")
async def get_weather(city: str) -> dict[str, Any]:
    await asyncio.sleep(0.5)  # simulate network latency
    return {"city": city, "temp_c": 22, "condition": "sunny"}

@tool("get_stock_price")
async def get_stock_price(symbol: str) -> dict[str, Any]:
    await asyncio.sleep(0.5)
    return {"symbol": symbol, "price": 184.32, "currency": "USD"}

@tool("get_news")
async def get_news(topic: str) -> dict[str, Any]:
    await asyncio.sleep(0.5)
    return {"topic": topic, "headlines": ["Market hits record high", "AI stocks surge"]}

Each tool is an async function that simulates some I/O latency. In a real application, these would be HTTP calls, database queries, or file reads.

Next, we need a function that executes a list of tool calls in parallel:

async def execute_tool_calls(tool_calls: list[dict]) -> list[dict]:
    """Execute multiple tool calls concurrently and return results."""
    async def run_single(call: dict) -> dict:
        tool_name = call["name"]
        arguments = call.get("arguments", {})
        if tool_name not in TOOLS:
            return {
                "tool_call_id": call["id"],
                "error": f"Unknown tool: {tool_name}"
            }
        try:
            result = await TOOLS[tool_name](**arguments)
            return {"tool_call_id": call["id"], "result": result}
        except Exception as e:
            return {"tool_call_id": call["id"], "error": str(e)}

    # Fan out all calls concurrently
    results = await asyncio.gather(
        *[run_single(c) for c in tool_calls],
        return_exceptions=False
    )
    return list(results)

The key here is asyncio.gather, which schedules all tool executions on the event loop simultaneously. They all start at once, and gather waits for all of them to complete before returning.

Now let's build the main agent loop. We'll simulate the LLM's response with a mock function, but the structure mirrors what you'd build with a real API client:

async def mock_llm_response(messages: list[dict]) -> dict:
    """Simulate an LLM that returns parallel tool calls."""
    user_msg = messages[-1]["content"].lower()

    if "weather" in user_msg and "stock" in user_msg:
        return {
            "role": "assistant",
            "tool_calls": [
                {"id": "call_1", "name": "get_weather", "arguments": {"city": "Tokyo"}},
                {"id": "call_2", "name": "get_stock_price", "arguments": {"symbol": "AAPL"}},
            ]
        }
    elif "weather" in user_msg:
        return {
            "role": "assistant",
            "tool_calls": [
                {"id": "call_1", "name": "get_weather", "arguments": {"city": "Paris"}}
            ]
        }
    else:
        return {
            "role": "assistant",
            "content": "I can help with weather, stock prices, and news. What would you like to know?"
        }


async def run_agent(user_query: str) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful assistant with access to tools."},
        {"role": "user", "content": user_query},
    ]

    max_iterations = 5
    for _ in range(max_iterations):
        response = await mock_llm_response(messages)
        messages.append(response)

        tool_calls = response.get("tool_calls")
        if not tool_calls:
            return response.get("content", "")

        # Execute all tool calls in parallel
        print(f"Dispatching {len(tool_calls)} tool call(s) in parallel...")
        start = time.perf_counter()
        results = await execute_tool_calls(tool_calls)
        elapsed = time.perf_counter() - start
        print(f"All tool calls completed in {elapsed:.3f}s")

        # Feed results back to the model
        for r in results:
            messages.append({
                "role": "tool",
                "tool_call_id": r["tool_call_id"],
                "content": str(r.get("result") or r.get("error"))
            })

    return "Max iterations reached."

Let's run the agent with a query that requires two independent tool calls:

async def main():
    answer = await run_agent("What's the weather in Tokyo and the stock price of AAPL?")
    print("Final answer:", answer)

asyncio.run(main())

Output:

Dispatching 2 tool call(s) in parallel...
All tool calls completed in 0.501s
Final answer: I can help with weather, stock prices, and news. What would you like to know?

Notice that both tool calls — each taking 500ms individually — completed in just over 500ms total rather than 1 second. That's the power of parallel execution.

Using the OpenAI API with Parallel Tool Calls

If you're using the OpenAI Python SDK, parallel tool calling is supported out of the box. The model may return multiple entries in response.choices[0].message.tool_calls. Here's how to handle that:

from openai import AsyncOpenAI

client = AsyncOpenAI()

tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "Get current stock price",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string"}
                },
                "required": ["symbol"]
            }
        }
    }
]

async def run_openai_agent(user_query: str):
    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": user_query},
    ]

    while True:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools_schema,
        )

        msg = response.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content

        # Execute all tool calls in parallel
        tasks = []
        for tc in msg.tool_calls:
            fn_name = tc.function.name
            import json
            args = json.loads(tc.function.arguments)
            tasks.append(TOOLS[fn_name](**args))

        results = await asyncio.gather(*tasks)

        for tc, result in zip(msg.tool_calls, results):
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": str(result),
            })

Best Practices

• Identify independence: Only run tools in parallel when they have no data dependencies. If Tool B needs the output of Tool A, they must run sequentially. The LLM usually handles this correctly, but validate in your orchestration layer if needed.
• Set timeouts: A single slow tool shouldn't block the entire batch. Use asyncio.wait_for or similar to enforce per-call timeouts and return partial results or errors for calls that exceed the limit.
• Handle errors gracefully: One tool failing shouldn't crash the whole batch. Catch exceptions per-tool and return structured error objects so the LLM can reason about what went wrong and retry or adapt.
• Limit concurrency: If an agent fans out to dozens of tools, use a semaphore to cap concurrent executions. This prevents overwhelming downstream APIs or exhausting connection pools.
• Cache idempotent calls: If the same tool is called with the same arguments repeatedly, cache the result to avoid redundant work.
• Log and trace: Parallel execution makes debugging harder. Log each tool call with a unique ID, start time, duration, and status so you can reconstruct what happened.
• Test with realistic latency: Parallel execution benefits are most visible under real network conditions. Test with actual API latencies, not just instant local mocks.
• Consider rate limits: Parallel calls can trigger rate limits faster than sequential ones. Implement backoff and retry logic, and respect Retry-After headers from upstream APIs.

Parallel tool execution is one of the highest-impact optimizations you can make when building production AI agents. By dispatching independent calls concurrently, you cut latency, reduce cost, and let agents reason more naturally about multi-faceted questions. The implementation is straightforward with async primitives like asyncio.gather, and most modern LLM APIs already support returning multiple tool calls in a single turn. Pair the technique with solid error handling, timeouts, and concurrency limits, and your agents will be both fast and resilient.