Streaming and Tools

Streaming with Function Calling

Streaming lets you receive the model’s response incrementally, token by token. This is essential for responsive user interfaces. But how does streaming interact with function calling?

The Fundamental Rule

Function calls are returned in a single chunk, not streamed progressively. This is a major difference from normal text, which arrives token by token.

In practice:

• Response text is streamed progressively
• tool_calls arrive all at once in a single event

This constraint exists for a good reason: a tool call must be complete (name + valid JSON arguments) for you to execute it. Partial JSON would be unusable.

Implementing Streaming with Tools

stream = client.chat.completions.create(
    model="grok-4.5",
    messages=messages,
    tools=tools,
    stream=True
)

collected_tool_calls = []
collected_content = ""

for chunk in stream:
    delta = chunk.choices[0].delta

    # Streamed text
    if delta.content:
        collected_content += delta.content
        print(delta.content, end="", flush=True)

    # Tool calls (arrive in a single block)
    if delta.tool_calls:
        for tc in delta.tool_calls:
            collected_tool_calls.append(tc)

    # Check for end of stream
    if chunk.choices[0].finish_reason == "tool_calls":
        # Model expects tool results
        break
    elif chunk.choices[0].finish_reason == "stop":
        # Response finished, no tool called
        break

Reconstructing Tool Calls in Streaming

In some cases, tool call arguments may arrive in multiple pieces within the stream. You need to reconstruct them:

tool_calls_buffer = {}

for chunk in stream:
    delta = chunk.choices[0].delta

    if delta.tool_calls:
        for tc_chunk in delta.tool_calls:
            idx = tc_chunk.index

            if idx not in tool_calls_buffer:
                tool_calls_buffer[idx] = {
                    "id": tc_chunk.id,
                    "function": {
                        "name": tc_chunk.function.name or "",
                        "arguments": ""
                    }
                }

            if tc_chunk.function.name:
                tool_calls_buffer[idx]["function"]["name"] = tc_chunk.function.name
            if tc_chunk.function.arguments:
                tool_calls_buffer[idx]["function"]["arguments"] += tc_chunk.function.arguments

This buffering approach ensures you reconstruct each tool call correctly, even if the data arrives fragmented.

Architecture for Streaming with Tools

In a real application, you need to handle two simultaneous streams:

1. Real-time display: show text as it comes in
2. Tool call accumulation: collect calls for execution

async def handle_streaming_response(stream):
    content_parts = []
    tool_calls = []
    current_tool = None

    async for chunk in stream:
        delta = chunk.choices[0].delta

        if delta.content:
            content_parts.append(delta.content)
            yield {"type": "text", "data": delta.content}

        if delta.tool_calls:
            for tc in delta.tool_calls:
                tool_calls.append(tc)

        if chunk.choices[0].finish_reason == "tool_calls":
            # Execute tools then restart
            results = await execute_tools(tool_calls)
            yield {"type": "tool_results", "data": results}

Latency and User Experience

Streaming with tools involves natural pauses in the flow:

1. The user asks their question
2. The model starts thinking (no visible tokens)
3. The model decides to call a tool (you receive the tool_call)
4. Pause — you execute the function
5. You send back the result
6. The model generates its response (streamed progressively)

For the user experience, display an indicator during step 4:

🔍 Searching for weather in Paris...

This prevents the user from thinking the application is frozen.

Server-Sent Events (SSE) for Web Applications

If you are building a web API, use SSE to relay the stream:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
async def chat(message: str):
    async def event_stream():
        stream = client.chat.completions.create(
            model="grok-4.5",
            messages=[{"role": "user", "content": message}],
            tools=tools,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {chunk.choices[0].delta.content}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

Key Takeaways

• Tool calls arrive in a single chunk, not streamed progressively
• Normal text continues to be streamed token by token
• Reconstruct fragmented tool calls with a buffer indexed by index
• Handle two streams in parallel: text display + tool call collection
• Display an indicator during tool execution for UX
• Use SSE for web applications