Parallel tool calling in LangChain agents

Parallel tool calling lets agents invoke multiple functions in a single turn, cutting latency dramatically for workloads that fetch data from independent sources. This guide walks through enabling langchain parallel tool calling, configuring concurrency limits, handling partial failures, and verifying the behavior in a test harness you can run today.

Step 1: Install the right versions

LangChain’s parallel tool calling support landed in langchain-core>=0.2.0 and requires a model that exposes the parallel_tool_calls parameter (OpenAI gpt-4o, gpt-4o-mini, gpt-4-turbo, and Anthropic claude-3-5-sonnet-20241022 or newer). Pin these versions to avoid surprises:

pip install "langchain-core>=0.2.0" "langchain-openai>=0.1.0" "langchain-anthropic>=0.1.0"

Verify the imports work:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

print("Imports OK")

Step 2: Define tools that benefit from parallelism

Parallel execution only helps when tools are independent — no shared state, no ordering constraints. Good candidates: multiple API lookups, independent database queries, or separate search calls. Bad candidates: a tool that writes a file and another that reads it.

from langchain_core.tools import tool
import httpx
import asyncio

@tool
async def fetch_weather(city: str) -> dict:
    """Get current weather for a city."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": 0, "longitude": 0, "current_weather": "true"},
        )
        resp.raise_for_status()
        return {"city": city, "data": resp.json()}

@tool
async def fetch_timezone(city: str) -> dict:
    """Get timezone info for a city."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(
            "https://worldtimeapi.org/api/timezone",
            params={},
        )
        resp.raise_for_status()
        zones = [z for z in resp.json() if city.lower() in z.lower()]
        return {"city": city, "zones": zones[:3]}

@tool
async def fetch_currency(country: str) -> dict:
    """Get currency code for a country."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(f"https://restcountries.com/v3.1/name/{country}")
        resp.raise_for_status()
        data = resp.json()[0]
        currencies = list(data.get("currencies", {}).keys())
        return {"country": country, "currencies": currencies}

Each tool is async, uses its own HTTP client, and has no side effects — ideal for parallel execution.

Step 3: Enable parallel tool calling on the model

Pass parallel_tool_calls=True when constructing the chat model. This tells the provider to return multiple tool_calls in a single response message.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    parallel_tool_calls=True,  # the key flag
).bind_tools([fetch_weather, fetch_timezone, fetch_currency])

For Anthropic, the parameter name differs:

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
).bind_tools([fetch_weather, fetch_timezone, fetch_currency])
# Anthropic enables parallel calls by default when multiple tools are bound

Step 4: Build the agent executor with a concurrency limiter

LangChain’s create_tool_calling_agent and AgentExecutor handle the loop, but they invoke tools sequentially by default. Wrap the tool list with a semaphore to bound concurrency and prevent thundering-herd problems against downstream APIs.

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
import asyncio

# Semaphore limits concurrent tool invocations (tune per downstream API)
MAX_CONCURRENT_TOOLS = 3
_tool_semaphore = asyncio.Semaphore(MAX_CONCURRENT_TOOLS)

async def _limited_tool_coroutine(tool, args, config=None):
    async with _tool_semaphore:
        return await tool.ainvoke(args, config=config)

# Patch each tool's ainvoke to respect the semaphore
for t in [fetch_weather, fetch_timezone, fetch_currency]:
    original_ainvoke = t.ainvoke
    async def wrapped_ainvoke(args, config=None, _orig=original_ainvoke):
        return await _limited_tool_coroutine(t, args, config)
    t.ainvoke = wrapped_ainvoke

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools. Use them in parallel when possible."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [fetch_weather, fetch_timezone, fetch_currency], prompt)
executor = AgentExecutor(agent=agent, tools=[fetch_weather, fetch_timezone, fetch_currency], verbose=True)

The semaphore ensures at most three tools run simultaneously, protecting both your rate limits and the provider’s.

Step 5: Invoke the agent and inspect the tool call batch

Run a query that naturally triggers multiple independent lookups. The model should emit a single AIMessage with multiple tool_calls.

import asyncio

async def main():
    result = await executor.ainvoke({
        "input": "What's the weather in Tokyo, the timezone in London, and the currency used in Brazil?"
    })
    print("Final answer:", result["output"])
    return result

if __name__ == "__main__":
    asyncio.run(main())

Expected output structure (truncated):

> Entering new AgentExecutor chain...
Invoking: fetch_weather with {'city': 'Tokyo'}
Invoking: fetch_timezone with {'city': 'London'}
Invoking: fetch_currency with {'country': 'Brazil'}
[parallel execution — all three start before any finishes]
...
> Finished chain.
Final answer: Tokyo weather shows 18°C... London timezone is Europe/London... Brazil uses BRL.

The verbose=True flag prints each tool invocation. You should see all three Invoking: lines appear before any Finished: lines — that’s your visual confirmation of parallelism.

Step 6: Verify parallelism programmatically

Don’t rely on logs alone. Add a test that measures wall-clock time and asserts it’s less than the sum of sequential latencies.

import time
import pytest

@pytest.mark.asyncio
async def test_parallel_tool_calling_is_faster_than_sequential():
    # Warm up (cold starts skew timing)
    await executor.ainvoke({"input": "warmup"})

    # Measure parallel execution
    start = time.perf_counter()
    await executor.ainvoke({
        "input": "Weather in Paris, timezone in New York, currency in Japan"
    })
    parallel_elapsed = time.perf_counter() - start

    # Measure sequential baseline (same tools, forced serial)
    async def sequential():
        await fetch_weather.ainvoke({"city": "Paris"})
        await fetch_timezone.ainvoke({"city": "New York"})
        await fetch_currency.ainvoke({"country": "Japan"})

    start = time.perf_counter()
    await sequential()
    sequential_elapsed = time.perf_counter() - start

    # Parallel should be at least 30% faster (allows for overhead)
    assert parallel_elapsed < sequential_elapsed * 0.7, (
        f"Parallel {parallel_elapsed:.2f}s not sufficiently faster than sequential {sequential_elapsed:.2f}s"
    )
    print(f"Parallel: {parallel_elapsed:.2f}s, Sequential: {sequential_elapsed:.2f}s")

Run with pytest -xvs test_parallel.py. A passing test proves the runtime actually parallelized the calls.

Step 7: Handle partial failures gracefully

When one tool fails, the agent should still receive results from the others. LangChain’s AgentExecutor captures exceptions per tool call and surfaces them as ToolMessage with status="error". Customize the error payload so the model can reason about what succeeded.

from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig

async def _safe_tool_coroutine(tool, args, config=None):
    async with _tool_semaphore:
        try:
            return await tool.ainvoke(args, config=config)
        except Exception as e:
            # Return a structured error the model can read
            return ToolMessage(
                content=f"Error: {type(e).__name__}: {str(e)}",
                tool_call_id=args.get("tool_call_id", "unknown"),
                status="error",
            )

# Re-patch with error handling
for t in [fetch_weather, fetch_timezone, fetch_currency]:
    original_ainvoke = t.ainvoke
    async def wrapped_ainvoke(args, config=None, _orig=original_ainvoke):
        return await _safe_tool_coroutine(t, args, config)
    t.ainvoke = wrapped_ainvoke

Now if fetch_timezone times out, the agent still gets weather and currency data and can answer partially.

Step 8: Configure model-level timeouts and retries

Provider SDKs respect timeout and max_retries on the chat model. Set these to avoid hanging the entire agent on a single slow provider.

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    parallel_tool_calls=True,
    timeout=30.0,          # total request timeout
    max_retries=2,         # retry on transient errors
).bind_tools([fetch_weather, fetch_timezone, fetch_currency])

For streaming responses with parallel tools, the first chunk arrives after all tools complete. If you need incremental results, consider a custom callback handler that yields partial tool outputs — but that’s a separate pattern.

Step 9: Observe token usage and routing in production

Parallel tool calling increases prompt tokens (more tool schemas in the system message) and completion tokens (multiple tool_calls blocks). Meter per-request usage to catch regressions.

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict

class TokenMeter(BaseCallbackHandler):
    def __init__(self):
        self.usage = {"prompt": 0, "completion": 0, "total": 0}

    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
        if hasattr(response, "llm_output") and response.llm_output:
            usage = response.llm_output.get("token_usage", {})
            self.usage["prompt"] += usage.get("prompt_tokens", 0)
            self.usage["completion"] += usage.get("completion_tokens", 0)
            self.usage["total"] += usage.get("total_tokens", 0)

meter = TokenMeter()
result = await executor.ainvoke(
    {"input": "Weather in Sydney, timezone in Tokyo, currency in Canada"},
    config={"callbacks": [meter]}
)
print("Token usage:", meter.usage)

If you route requests across multiple providers (e.g., OpenAI primary, Anthropic fallback), ensure each provider receives the same parallel_tool_calls configuration. Some gateways forward provider-specific parameters automatically; others require explicit mapping.

Step 10: Tune the concurrency limit per workload

The optimal MAX_CONCURRENT_TOOLS depends on downstream API rate limits, not CPU. Start with 3–5, then load-test.

async def load_test(concurrency: int, requests: int = 20):
    global MAX_CONCURRENT_TOOLS, _tool_semaphore
    MAX_CONCURRENT_TOOLS = concurrency
    _tool_semaphore = asyncio.Semaphore(concurrency)

    tasks = [
        executor.ainvoke({"input": f"Weather in city {i}, timezone in city {i}, currency in country {i}"})
        for i in range(requests)
    ]
    start = time.perf_counter()
    await asyncio.gather(*tasks)
    elapsed = time.perf_counter() - start
    return elapsed

for c in [1, 2, 3, 5, 8]:
    t = await load_test(c, requests=15)
    print(f"Concurrency {c}: {t:.2f}s total ({15/t:.1f} req/s)")

Plot the results. The sweet spot is where throughput plateaus — higher concurrency just adds queueing delay.

Step 11: Debug when the model refuses to parallelize

Sometimes the model emits sequential calls despite parallel_tool_calls=True. Common causes:

1. Prompt implies ordering — “First get weather, then get timezone” forces serial. Fix: “Get weather, timezone, and currency for these locations.”
2. Tool descriptions suggest dependence — “Use the weather result to determine the timezone.” Fix: keep descriptions independent.
3. Model doesn’t support the parameter — Older models ignore parallel_tool_calls. Check the provider’s model card.

Force the behavior with a system reminder:

prompt = ChatPromptTemplate.from_messages([
    ("system", "You have access to multiple tools. ALWAYS invoke all required tools in a single response when they are independent. Do not wait for one tool to finish before calling the next."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

Step 12: Verify end-to-end with a realistic scenario

Combine everything into a runnable script that exercises the full path: parallel calls, error injection, token metering, and latency assertion.

# test_full_flow.py
import asyncio
import time
from langchain_core.messages import ToolMessage

async def test_full_parallel_flow():
    # 1. Successful parallel call
    start = time.perf_counter()
    result = await executor.ainvoke({
        "input": "Weather in Berlin, timezone in Paris, currency in Germany"
    })
    elapsed = time.perf_counter() - start
    assert "Berlin" in result["output"]
    assert "Paris" in result["output"]
    assert "Germany" in result["output"] or "EUR" in result["output"]
    print(f"✓ Success case: {elapsed:.2f}s")

    # 2. Partial failure (inject a bad tool)
    @tool
    async def failing_tool(city: str) -> dict:
        raise ValueError("Simulated downstream outage")

    # Temporarily add failing tool
    original_tools = executor.tools
    executor.tools = [*executor.tools, failing_tool]
    executor.agent = create_tool_calling_agent(llm, executor.tools, prompt)

    result = await executor.ainvoke({
        "input": "Weather in Rome and call failing_tool for Rome"
    })
    assert "Rome" in result["output"]
    assert "Error" in result["output"] or "outage" in result["output"].lower()
    print("✓ Partial failure handled")

    # Restore
    executor.tools = original_tools
    executor.agent = create_tool_calling_agent(llm, executor.tools, prompt)

    # 3. Latency check
    assert elapsed < 8.0, f"Parallel call too slow: {elapsed:.2f}s"
    print("✓ Latency within budget")

if __name__ == "__main__":
    asyncio.run(test_full_parallel_flow())

Run it: python test_full_flow.py. Three checkmarks mean your langchain parallel tool calling pipeline is production-ready.

Parallel tool calling is a force multiplier for agent latency, but it only works when the model, the tools, and the executor all align. Enable the flag, bound concurrency, handle partial failures, and measure. That’s the loop.