Parallel Tool Calls and Streaming

Capsule overview

You already know how to create tools with @tool, connect them to models with bind_tools(), and run the full tool execution loop. But so far, every interaction involved a single tool call: the user asks something, the model calls a tool, you run it, and the model produces the final answer.

In practice, users ask questions that need several data sources at once. "What's the weather in Madrid and Paris?" needs two calls to the same tool. "What's the price of Bitcoin and the temperature in Tokyo?" needs two different tools. Modern models solve this with parallel tool calls — instead of calling one tool at a time, the model returns multiple tool_calls in a single AIMessage.

On top of that, when your application has a user interface, you need the tool calls to arrive progressively — not to wait for the model to finish "thinking" before you know which tools it wants to call. That's where tool call streaming comes in: receiving the tools' arguments as partial chunks while the model generates them.

Parallel tool calls: multiple calls in one response

When a model with tools gets a question that needs data from several sources, it can return several tool_calls in a single AIMessage.

A basic example: weather in two cities

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {
        "Madrid": "Sunny, 22°C",
        "Paris": "Cloudy, 15°C",
        "Tokyo": "Rainy, 18°C",
    }
    return weathers.get(city, f"Weather unavailable for {city}")

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

response = model_with_tools.invoke("What's the weather in Madrid and Paris?")

print(f"Number of tool_calls: {len(response.tool_calls)}")
for tc in response.tool_calls:
    print(f"  {tc['name']}({tc['args']})")
# Expected output:
# Number of tool_calls: 2
#   get_weather({'city': 'Madrid'})
#   get_weather({'city': 'Paris'})

Every tool_call has its own id, name, and args. The id is critical: when you return the result as a ToolMessage, you have to include the matching tool_call_id so the model knows which result belongs to which call.

Processing parallel tool calls

Processing multiple tool calls follows the same pattern as a single one, but you iterate over response.tool_calls and create a ToolMessage for each.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C", "Tokyo": "Rainy, 18°C"}
    return weathers.get(city, f"Weather unavailable for {city}")

tools = [get_weather]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

messages = [HumanMessage(content="What's the weather in Madrid, Paris and Tokyo?")]

# Step 1: the model generates tool_calls
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)

print(f"The model wants to call {len(ai_message.tool_calls)} tools")

# Step 2: run EACH tool call and create a ToolMessage for each one
for tc in ai_message.tool_calls:
    tool_fn = tools_by_name[tc["name"]]
    result = tool_fn.invoke(tc["args"])
    messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    print(f"  {tc['args']['city']}: {result}")

# Step 3: the model produces the final answer with ALL the results
final_response = model_with_tools.invoke(messages)
print(f"\n{final_response.content}")
# Expected output:
# The model wants to call 3 tools
#   Madrid: Sunny, 22°C
#   Paris: Cloudy, 15°C
#   Tokyo: Rainy, 18°C
#
# Here's the current weather in all three cities:
# - **Madrid**: Sunny, 22°C
# - **Paris**: Cloudy, 15°C
# - **Tokyo**: Rainy, 18°C

The fundamental rule: one ToolMessage per tool_call

Every tool_call must have its matching ToolMessage. If the model generates 3 tool calls and you only return 2 ToolMessages, the response will be wrong or it'll throw an error.

When do models use parallel calls?

The model can call not only the same tool several times — it can also call different tools in parallel (for example, get_weather + get_population for the same city). The processing pattern is identical.

Models that support parallel tool calls

• ✅ OpenAI GPT-4.1, GPT-4.1 Mini, GPT-4o
• ✅ Anthropic Claude 4 Sonnet, Claude 4 Opus
• ✅ Google Gemini 2.5 Pro, Gemini 2.5 Flash
• ⚠️ Older or local models may not support it

If a model doesn't support parallel calls, it'll return the tool calls one at a time across separate turns. Your code should handle both cases.

Real parallel execution with asyncio

The model returning multiple tool calls doesn't mean you're running them in parallel automatically. The for tc in tool_calls loop runs them sequentially. To actually run them in parallel, use asyncio.gather:

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C", "Tokyo": "Rainy, 18°C"}
    return weathers.get(city, f"Weather unavailable for {city}")

tools = [get_weather]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

messages = [HumanMessage(content="Weather in Madrid, Paris and Tokyo?")]
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)

async def execute_tool(tc):
    tool_fn = tools_by_name[tc["name"]]
    result = await tool_fn.ainvoke(tc["args"])
    return ToolMessage(content=result, tool_call_id=tc["id"])

async def main():
    tool_messages = await asyncio.gather(
        *[execute_tool(tc) for tc in ai_message.tool_calls]
    )
    for tm in tool_messages:
        messages.append(tm)

    final = model_with_tools.invoke(messages)
    print(final.content)

asyncio.run(main())
# Expected output:
# Here's the current weather:
# - **Madrid**: Sunny, 22°C
# - **Paris**: Cloudy, 15°C
# - **Tokyo**: Rainy, 18°C

With asyncio.gather, the three tool calls run at the same time. If each one takes 1 second, the total is ~1 second instead of ~3. That's especially valuable when the tools make HTTP calls to external APIs.

Streaming tool calls

With stream(), you can receive tool calls progressively as partial chunks instead of waiting for the model to finish.

Why stream tool calls?

• ✅ Show which tools the model is about to call while it generates them
• ✅ Give the user visual feedback about progress
• ✅ Start prep work before the arguments are complete

Example: streaming with tool calls

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C"}
    return weathers.get(city, f"Weather unavailable for {city}")

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

chunks = []
for chunk in model_with_tools.stream("Weather in Madrid?"):
    chunks.append(chunk)
    if chunk.tool_call_chunks:
        print(f"Tool chunk: {chunk.tool_call_chunks}")
# Expected output (varies by model):
# Tool chunk: [{'name': 'get_weather', 'args': '', 'id': 'call_abc123', 'index': 0, ...}]
# Tool chunk: [{'name': None, 'args': '{"ci', 'id': None, 'index': 0, ...}]
# Tool chunk: [{'name': None, 'args': 'ty":', 'id': None, 'index': 0, ...}]
# Tool chunk: [{'name': None, 'args': ' "Mad', 'id': None, 'index': 0, ...}]
# Tool chunk: [{'name': None, 'args': 'rid"}', 'id': None, 'index': 0, ...}]

The first chunk usually carries the name and the id. The following ones bring fragments of the args as a partial JSON string.

AIMessageChunk and tool_call_chunks

Accumulating chunks to get complete tool calls

The tool_call_chunks are partial fragments — you need to accumulate them with the + operator to get the complete tool calls:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C"}
    return weathers.get(city, f"Weather unavailable for {city}")

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

chunks = []
for chunk in model_with_tools.stream("Weather in Madrid and Paris?"):
    chunks.append(chunk)

full_message = chunks[0]
for chunk in chunks[1:]:
    full_message = full_message + chunk

print(f"Complete tool calls: {len(full_message.tool_calls)}")
for tc in full_message.tool_calls:
    print(f"  {tc['name']}({tc['args']}) id={tc['id']}")
# Expected output:
# Complete tool calls: 2
#   get_weather({'city': 'Madrid'}) id=call_abc123
#   get_weather({'city': 'Paris'}) id=call_def456

The full loop: stream + execute + stream the answer

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C"}
    return weathers.get(city, f"Weather unavailable for {city}")

tools = [get_weather]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)
messages = [HumanMessage(content="Weather in Madrid and Paris?")]

# Phase 1: stream the tool calls
print("Generating tool calls...")
chunks = []
for chunk in model_with_tools.stream(messages):
    chunks.append(chunk)
    if chunk.tool_call_chunks:
        for tcc in chunk.tool_call_chunks:
            if tcc["name"]:
                print(f"  → Tool detected: {tcc['name']}")

full_message = chunks[0]
for c in chunks[1:]:
    full_message = full_message + c
messages.append(full_message)

# Phase 2: run the tools
print("\nRunning tools...")
for tc in full_message.tool_calls:
    result = tools_by_name[tc["name"]].invoke(tc["args"])
    messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    print(f"  {tc['name']}({tc['args']}) → {result}")

# Phase 3: stream the final answer
print("\nAnswer:")
for chunk in model_with_tools.stream(messages):
    if chunk.content:
        print(chunk.content, end="", flush=True)
print()
# Expected output:
# Generating tool calls...
#   → Tool detected: get_weather
#   → Tool detected: get_weather
#
# Running tools...
#   get_weather({'city': 'Madrid'}) → Sunny, 22°C
#   get_weather({'city': 'Paris'}) → Cloudy, 15°C
#
# Answer:
# The current weather is:
# - **Madrid**: Sunny, 22°C
# - **Paris**: Cloudy, 15°C

This is the complete pattern for a chat UI with tools: stream the tool calls to show progress, run the tools, and stream the final answer.

Comparison: parallel calls vs sequential calls

With 5 tools, the sequential version needs 6 calls to the model; the parallel one only needs 2. The difference in latency and cost is significant.

Connection to the project

In the Assistant with External Tools (Capsule 08), you'll implement parallel tool calls for scenarios like "What's the weather in the 3 capitals on my itinerary?" and streaming will be essential for the UX: while the model generates tool calls you'll show progress indicators, and the final answer will arrive as streamed tokens.

Troubleshooting

Problem 1: the model returns sequential tool calls instead of parallel ones

Cause: The model or provider doesn't support parallel tool calls, or the question isn't clear enough. Fix: Use a model that supports parallel calls (GPT-4.1, Claude 4 Sonnet). Rephrase the question to be explicit:

# More likely to trigger parallel calls
response = model_with_tools.invoke("Give me the weather in Madrid AND Paris")

Problem 2: a missing ToolMessage causes an error

Cause: You're not creating a ToolMessage for every tool_call. Fix: Always iterate over all the tool_calls:

for tc in ai_message.tool_calls:
    result = tools_by_name[tc["name"]].invoke(tc["args"])
    messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

Problem 3: tool_call_chunks is empty during streaming

Cause: Not every chunk carries tool_call_chunks. Fix: Check before accessing it:

for chunk in model_with_tools.stream("Weather in Madrid?"):
    if chunk.tool_call_chunks:
        for tcc in chunk.tool_call_chunks:
            print(tcc)

Problem 4: tool_calls is empty after accumulating chunks

Cause: The accumulation wasn't done correctly. Fix: Use the + operator between AIMessageChunk objects:

full = chunks[0]
for chunk in chunks[1:]:
    full = full + chunk
print(full.tool_calls)

Exercises

Exercise 1: basic parallel calls (Easy)

Create two tools: get_weather(city) and get_time(city). Send a question that asks for a city's weather and time. Check that the model returns 2 tool calls in parallel, run them, and get the final answer.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Tokyo": "Rainy, 18°C"}
    return weathers.get(city, f"Weather unavailable for {city}")

@tool
def get_time(city: str) -> str:
    """Get the current time in a city."""
    times = {"Madrid": "14:30 CET", "Tokyo": "22:30 JST"}
    return times.get(city, f"Time unavailable for {city}")

tools = [get_weather, get_time]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

messages = [HumanMessage(content="What's the weather and time in Madrid?")]
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)

print(f"Tool calls: {len(ai_message.tool_calls)}")
for tc in ai_message.tool_calls:
    result = tools_by_name[tc["name"]].invoke(tc["args"])
    messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    print(f"  {tc['name']}({tc['args']}) → {result}")

final = model_with_tools.invoke(messages)
print(f"\n{final.content}")
# Expected output:
# Tool calls: 2
#   get_weather({'city': 'Madrid'}) → Sunny, 22°C
#   get_time({'city': 'Madrid'}) → 14:30 CET
#
# In Madrid: sunny weather at 22°C, and it's 14:30 CET.

Exercise 2: counting tool calls per question (Easy)

Create a get_weather tool and send three questions: one city, two cities, and a question that needs no tools. Print how many tool calls each one generates.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return "Sunny, 22°C"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

questions = [
    "What's the weather in Madrid?",
    "What's the weather in Madrid and Paris?",
    "What's 2 + 2?",
]

for q in questions:
    response = model_with_tools.invoke(q)
    n = len(response.tool_calls)
    print(f"'{q}' → {n} tool call(s)")
    if n == 0:
        print(f"  Direct answer: {response.content[:60]}...")
# Expected output:
# 'What's the weather in Madrid?' → 1 tool call(s)
# 'What's the weather in Madrid and Paris?' → 2 tool call(s)
# 'What's 2 + 2?' → 0 tool call(s)
#   Direct answer: 2 + 2 equals 4...

Exercise 3: streaming with tool feedback (Medium)

Stream a question that generates tool calls. Print a message every time you detect a new tool. At the end, accumulate the chunks and show the complete tool calls.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return "Sunny, 22°C"

@tool
def get_population(city: str) -> str:
    """Get a city's population."""
    return "3.2 million"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather, get_population])

chunks = []
detected = []
for chunk in model_with_tools.stream("Give me the weather and population of Madrid"):
    chunks.append(chunk)
    if chunk.tool_call_chunks:
        for tcc in chunk.tool_call_chunks:
            if tcc.get("name"):
                detected.append(tcc["name"])
                print(f"  [Detected] → {tcc['name']}")

full = chunks[0]
for c in chunks[1:]:
    full = full + c

print(f"\nDetected during streaming: {detected}")
print(f"Complete tool calls: {len(full.tool_calls)}")
for tc in full.tool_calls:
    print(f"  {tc['name']}({tc['args']})")
# Expected output:
#   [Detected] → get_weather
#   [Detected] → get_population
#
# Detected during streaming: ['get_weather', 'get_population']
# Complete tool calls: 2
#   get_weather({'city': 'Madrid'})
#   get_population({'city': 'Madrid'})

Exercise 4: the full loop with streaming in 3 phases (Medium)

Implement the professional flow: stream the tool calls → run the tools → stream the final answer. Use get_weather and get_time with a question that triggers both.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C"}
    return weathers.get(city, f"Unavailable for {city}")

@tool
def get_time(city: str) -> str:
    """Get the current time in a city."""
    times = {"Madrid": "14:30 CET", "Paris": "14:30 CET"}
    return times.get(city, f"Unavailable for {city}")

tools = [get_weather, get_time]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)
messages = [HumanMessage(content="Weather and time in Madrid and Paris?")]

# Phase 1: stream the tool calls
print("=== Phase 1: Detecting tools ===")
chunks = []
for chunk in model_with_tools.stream(messages):
    chunks.append(chunk)
    if chunk.tool_call_chunks:
        for tcc in chunk.tool_call_chunks:
            if tcc.get("name"):
                print(f"  → {tcc['name']}")

full = chunks[0]
for c in chunks[1:]:
    full = full + c
messages.append(full)

# Phase 2: run the tools
print(f"\n=== Phase 2: Running {len(full.tool_calls)} tools ===")
for tc in full.tool_calls:
    result = tools_by_name[tc["name"]].invoke(tc["args"])
    messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    print(f"  {tc['name']}({tc['args']}) → {result}")

# Phase 3: stream the final answer
print("\n=== Phase 3: Answer ===")
for chunk in model_with_tools.stream(messages):
    if chunk.content:
        print(chunk.content, end="", flush=True)
print()
# Expected output:
# === Phase 1: Detecting tools ===
#   → get_weather
#   → get_time
#   → get_weather
#   → get_time
#
# === Phase 2: Running 4 tools ===
#   get_weather({'city': 'Madrid'}) → Sunny, 22°C
#   get_time({'city': 'Madrid'}) → 14:30 CET
#   get_weather({'city': 'Paris'}) → Cloudy, 15°C
#   get_time({'city': 'Paris'}) → 14:30 CET
#
# === Phase 3: Answer ===
# - **Madrid**: Sunny, 22°C — 14:30 CET
# - **Paris**: Cloudy, 15°C — 14:30 CET

Exercise 5: a generic tool execution function with streaming (Hard)

Write stream_tool_loop(model_with_tools, tools, question) that wraps up: stream tool calls → accumulate → run → check whether there are more tool calls → stream the final answer. It must handle multiple rounds.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

def stream_tool_loop(model_with_tools, tools: list, question: str, max_rounds: int = 5):
    """Full tool loop with streaming and multi-round support."""
    tools_by_name = {t.name: t for t in tools}
    messages = [HumanMessage(content=question)]

    for round_num in range(max_rounds):
        chunks = []
        for chunk in model_with_tools.stream(messages):
            chunks.append(chunk)
            if chunk.tool_call_chunks:
                for tcc in chunk.tool_call_chunks:
                    if tcc.get("name"):
                        print(f"  [R{round_num + 1}] → {tcc['name']}")
            if chunk.content:
                print(chunk.content, end="", flush=True)

        full = chunks[0]
        for c in chunks[1:]:
            full = full + c
        messages.append(full)

        if not full.tool_calls:
            print()
            return full

        for tc in full.tool_calls:
            result = tools_by_name[tc["name"]].invoke(tc["args"])
            messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
            print(f"  [Result] {tc['name']} → {result}")

    return None

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 22°C", "Paris": "Cloudy, 15°C"}
    return weathers.get(city, f"Unavailable for {city}")

tools = [get_weather]
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

stream_tool_loop(model_with_tools, tools, "Weather in Madrid and Paris?")
# Expected output:
#   [R1] → get_weather
#   [R1] → get_weather
#   [Result] get_weather → Sunny, 22°C
#   [Result] get_weather → Cloudy, 15°C
# Madrid: Sunny, 22°C. Paris: Cloudy, 15°C.

Summary

In this capsule you learned:

• Parallel tool calls let the model call multiple tools in a single turn — cutting down calls to the model and total latency
• Every tool_call has a name, args, and id — you must create a ToolMessage with the matching tool_call_id for each one
• Models trigger parallel calls when the question needs independent data from multiple sources
• For real parallel execution, use asyncio.gather with the tools' async versions
• Tool call streaming arrives as tool_call_chunks — partial fragments you accumulate with +
• The professional pattern has 3 phases: stream the tool calls → run the tools → stream the final answer
• Not every model supports parallel calls — your code should handle both scenarios

Next capsule: Structured Output with Tools — how to use tool calling not just to run actions, but to extract structured data from text.

Further reading

1. Tool Calling — LangChain Docs — Official conceptual guide to tool calling
2. How to call tools in parallel — Tutorial on parallel tool calls
3. How to stream tool calls — Streaming tool call chunks
4. AIMessageChunk API Reference — AIMessageChunk reference
5. ToolMessage API Reference — ToolMessage reference
6. OpenAI Parallel Function Calling — OpenAI's implementation
7. Anthropic Tool Use — Tool calling at Anthropic
8. AsyncIO Documentation — For real parallel execution of tools

Module 2 — LangChain & LangGraph: From Chains to Agents