Streaming function calls with the Python OpenAI SDK

When you build agents that need to show progress while the model decides which tool to invoke, streaming function calls python openai sdk patterns are the only way to avoid blocking on a full response. The OpenAI Python SDK emits tool calls inside streaming chunks, but those chunks carry fragmented argument strings that you must reassemble before execution. This guide walks through a complete, runnable pipeline: stream a completion with tools, accumulate deltas, call your Python functions, and feed results back into a second stream.

Step 1: Install and configure the client

Install the official SDK:

pip install openai>=1.0.0

Create a client. If you route through an OpenAI-compatible gateway such as n4n.ai, set base_url to its single endpoint and the same streaming function calls python openai sdk code works across 240+ models with automatic fallback when a provider is degraded.

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-key",
    # base_url="https://your-gateway/v1"  # e.g. n4n.ai OpenAI-compatible endpoint
)

Keep the key in environment variables in real deployments. The client is thread-safe; reuse one instance.

Step 2: Define tools and a local dispatcher

The model needs a JSON Schema description of each function. Define a registry that maps names to callables so you can execute after streaming finishes.

import json
from typing import Callable

def get_weather(location: str) -> str:
    # stub: replace with real API
    return f"Sunny in {location}, 22C"

def search_docs(query: str) -> str:
    return f"Top result for '{query}': doc #42"

registry: dict[str, Callable] = {
    "get_weather": get_weather,
    "search_docs": search_docs,
}

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location",
            "parameters": {
                "type": "object",
                "properties": {"location": {"type": "string"}},
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_docs",
            "description": "Search internal documentation",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
]

Step 3: Send a streaming request with tools

Initialize the conversation and call create with stream=True. The first user message triggers the model to emit tool calls.

messages = [
    {"role": "user", "content": "What's the weather in Berlin and do we have docs on refunds?"}
]

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    tools=tools,
    stream=True,
)

The stream iterator yields ChatCompletionChunk objects. Deltas may contain tool_calls; they never contain a finished function call in one shot.

Step 4: Reassemble tool calls from deltas

Each delta.tool_calls entry has an index, an optional id, and a function object with name and arguments fragments. Accumulate by index:

tool_calls: dict[int, dict] = {}

for chunk in stream:
    delta = chunk.choices[0].delta
    if not delta.tool_calls:
        continue
    for tc in delta.tool_calls:
        idx = tc.index
        if idx not in tool_calls:
            tool_calls[idx] = {"id": "", "name": "", "args": ""}
        if tc.id:
            tool_calls[idx]["id"] = tc.id
        if tc.function:
            if tc.function.name:
                tool_calls[idx]["name"] += tc.function.name
            if tc.function.arguments:
                tool_calls[idx]["args"] += tc.function.arguments

# tool_calls is now {0: {"id": "call_abc", "name": "get_weather", "args": '{"location": "Berlin"}'}, ...}

This accumulation step is the core of streaming function calls python openai sdk work: the API will not hand you a parsed object, only string pieces.

Step 5: Validate and execute functions

Before calling, parse the argument string. It should be complete JSON, but a truncated stream can leave it invalid—wrap in try/except.

assistant_tool_calls = []
for tc in tool_calls.values():
    try:
        args = json.loads(tc["args"])
    except json.JSONDecodeError:
        args = {}  # or skip / log
    assistant_tool_calls.append({
        "id": tc["id"],
        "type": "function",
        "function": {"name": tc["name"], "arguments": tc["args"]},
    })
    fn = registry.get(tc["name"])
    if fn:
        result = fn(**args)
        messages.append({"role": "tool", "tool_call_id": tc["id"], "content": str(result)})

# Record the assistant message that produced the calls
messages.append({
    "role": "assistant",
    "content": None,
    "tool_calls": assistant_tool_calls,
})

Note the order: append the tool results after the assistant message containing tool_calls. The OpenAI API requires this sequence.

Step 6: Stream the follow-up response

With tool results in context, make a second streaming call. You can keep tools attached if you expect chained calls, or drop them for a final answer.

final_stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    tools=tools,
    stream=True,
)

print("Assistant: ", end="")
for chunk in final_stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="")
print()

If the model emits more tool calls, loop back to Step 4. A production agent typically wraps steps 3–6 in a while loop bounded by max iterations.

Step 7: Handle errors and partial streams

Network failures mid-stream leave tool_calls with incomplete JSON. Detect this by attempting json.loads and falling back to a re-request without streaming if the id is missing. When using a gateway with automatic fallback, a provider rate-limit raises a standard openai.RateLimitError; catch it and retry with exponential backoff.

from openai import RateLimitError, APIConnectionError

try:
    stream = client.chat.completions.create(...)
except (RateLimitError, APIConnectionError) as e:
    # gateway like n4n.ai may already have failed over; otherwise back off
    time.sleep(2)

Also honor tool_choice: if you set "tool_choice": {"type": "function", "function": {"name": "get_weather"}}, the model is forced to call that function, simplifying accumulation logic in tests.

Step 8: Verify success

Run the script and confirm three things:

1. The first stream prints no final text but populates tool_calls with correct name and parseable args.
2. The messages list ends with role: "tool" entries whose tool_call_id matches the assistant id.
3. The second stream prints a natural-language answer that references both the weather stub and the doc search stub.

A minimal assertion block:

assert any(tc["name"] == "get_weather" for tc in tool_calls.values())
assert messages[-2]["role"] == "assistant"
assert messages[-1]["role"] == "tool"
print("Streaming function call cycle verified.")

If you swap the model name to another behind the same OpenAI-compatible endpoint, the assertions should still pass—the streaming function calls python openai sdk contract is stable across compliant providers.

Practical notes

• Always key accumulation by tc.index. Parallel tool calls arrive interleaved; index keeps them separate.
• The arguments fragments are not guaranteed to be valid JSON until the stream ends. Don’t parse mid-stream.
• Set stream_options={"include_usage": True} if you need per-token metering; the final chunk then carries usage. Gateways that meter per token will report it here.
• If you forward provider cache-control hints (e.g., cache_control on system messages), they pass through unchanged to compliant backends.

The pattern above is the backbone of any tool-using agent that can’t afford to wait for a full response before showing intent. Get the delta accumulation right and the rest is standard message bookkeeping.