Streaming tool calls

Assemble interleaved arguments, validate completion, and return tool results.

Current Chat ↔ Responses function-tool conversion does not copy strict. Native declarations, conversion, and upstream defaults do not establish equivalent guarantees. Validate tool arguments in the application. See tool selection and parallel controls.

This page extends nonstreaming tool calls into a complete loop: collect argument fragments, validate completed calls, multiply locally, submit tool results, and obtain a final answer. Confirm that your channel and model support streamed function calls and strict parameter schemas for the chosen protocol.

Two assembly protocols

Stage	Chat Completions	Responses
Associate fragments	choices[0].delta.tool_calls[].index	output_index and item_id
Argument delta	function.arguments string	delta in response.function_call_arguments.delta
Call identity	id and function.name	Output item id, call_id, and name
Completion gate	finish_reason: tool_calls and clean iterator completion	Consistent arguments.done, output_item.done, response.completed, and clean iterator completion
Return results	role: tool and tool_call_id	type: function_call_output and call_id

Fragments from different calls may interleave. Accumulate arguments per call; never parse or execute partial JSON. The Chat example expects complete IDs/names when first supplied, then omission or repetition of the same value; it does not concatenate names. The SDK decodes SSE and UTF-8. The example does not independently verify the [DONE] string consumed by the SDK.

Setup

Install the existing OpenAI SDK following TypeScript or Python; no new dependencies are required. Set the gateway token in API_KEY, then configure a Base URL including /v1:

export API_BASE_URL="https://gateway.example.com/v1"
export API_MODEL="YOUR_TOOL_CAPABLE_MODEL_ID"
# API_KEY must already be set in the process environment.

From the documentation directory, run node examples/stream-tools.mts or .venv/bin/python examples/stream-responses-tools.py. If you save the code separately, run node stream-tools.mts or python stream-responses-tools.py. The prompt requests 7 × 8; the model chooses the wording of its final answer.

TypeScript: Chat Completions

stream-tools.mts

import OpenAI from 'openai';

const { API_BASE_URL, API_KEY, API_MODEL } = process.env;
if (!API_BASE_URL || !API_KEY || !API_MODEL) {
  throw new Error('Set API_BASE_URL, API_KEY, and API_MODEL first.');
}
const client = new OpenAI({ baseURL: API_BASE_URL.replace(/\/+$/, ''), apiKey: API_KEY,
  timeout: 60_000, maxRetries: 0 });
const messages: OpenAI.Chat.Completions.ChatCompletionMessageParam[] = [
  { role: 'user', content: 'What is 7 times 8? Use multiply.' },
];
const tools: OpenAI.Chat.Completions.ChatCompletionTool[] = [{
  type: 'function', function: { name: 'multiply', description: 'Multiply two finite numbers.', strict: true,
    parameters: { type: 'object', properties: { a: { type: 'number' }, b: { type: 'number' } },
      required: ['a', 'b'], additionalProperties: false } },
}];
const seenIds = new Set<string>();

try {
  for (let round = 0; round < 3; round++) {
    const calls = new Map<number, { id: string; type: 'function'; function: { name: string; arguments: string } }>();
    let finish: string | null = null;
    let text = '';
    let events = 0;
    const stream = await client.chat.completions.create({ model: API_MODEL, messages, tools, stream: true });
    for await (const chunk of stream) {
      if (++events > 512) throw new Error('Stream event limit reached.');
      for (const choice of chunk.choices) {
        if (choice.index !== 0 || finish !== null) throw new Error('Unexpected choice or event after completion.');
        if (choice.delta.refusal || choice.delta.function_call) throw new Error('Refusal or unsupported legacy call.');
        text += choice.delta.content ?? '';
        if (text.length > 100_000) throw new Error('Text limit reached.');
        for (const delta of choice.delta.tool_calls ?? []) {
          if (!Number.isInteger(delta.index) || delta.index < 0 || delta.index >= 4) throw new Error('Invalid call index.');
          let call = calls.get(delta.index);
          if (!call) {
            call = { id: '', type: 'function', function: { name: '', arguments: '' } };
            calls.set(delta.index, call);
          }
          if (delta.type && delta.type !== 'function') throw new Error('Unsupported tool type.');
          if (delta.id) {
            if (call.id && call.id !== delta.id) throw new Error('Call ID changed.');
            call.id = delta.id;
          }
          if (delta.function?.name) {
            if (call.function.name && call.function.name !== delta.function.name) throw new Error('Tool name changed.');
            call.function.name = delta.function.name;
          }
          call.function.arguments += delta.function?.arguments ?? '';
          if (call.function.arguments.length > 16_384) throw new Error('Arguments limit reached.');
        }
        if (choice.finish_reason !== null) finish = choice.finish_reason;
      }
    }
    // No function executes until the whole stream has ended with a valid finish reason.
    if (calls.size === 0) {
      if (finish !== 'stop' || !text.trim()) throw new Error('No complete answer.');
      console.log(text);
      break;
    }
    if (finish !== 'tool_calls' || round === 2) throw new Error('Incomplete calls or round limit reached.');
    const ordered = [...calls.entries()].sort(([a], [b]) => a - b).map(([, call]) => call);
    const outputs: OpenAI.Chat.Completions.ChatCompletionToolMessageParam[] = [];
    for (const call of ordered) {
      if (!call.id.trim() || seenIds.has(call.id) || call.function.name !== 'multiply') throw new Error('Invalid tool call.');
      const args = JSON.parse(call.function.arguments);
      if (!args || Array.isArray(args) || Object.keys(args).length !== 2
        || !Number.isFinite(args.a) || !Number.isFinite(args.b)) throw new Error('Invalid arguments.');
      const value = args.a * args.b;
      if (!Number.isFinite(value)) throw new Error('Invalid result.');
      seenIds.add(call.id);
      outputs.push({ role: 'tool', tool_call_id: call.id, content: JSON.stringify({ value }) });
    }
    messages.push({ role: 'assistant', content: text || null, tool_calls: ordered }, ...outputs);
  }
} catch (error) {
  console.error(error instanceof OpenAI.APIError && error.status
    ? `Request failed (HTTP ${error.status}). No automatic retry.`
    : 'Streaming tool workflow failed; no incomplete answer published.');
  process.exitCode = 1;
}

Python: Responses

stream-responses-tools.py

import json
import math
import os
from openai import OpenAI, APIConnectionError, APIStatusError, APIError

history = [{"role": "user", "content": "What is 7 times 8? Use multiply."}]
tools = [{
    "type": "function", "name": "multiply", "description": "Multiply two finite numbers.",
    "strict": True,
    "parameters": {
        "type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
        "required": ["a", "b"], "additionalProperties": False,
    },
}]
seen_call_ids = set()
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
            api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
    try:
        for round_index in range(3):
            result = None
            pending = {}
            with client.responses.create(
                model=os.environ["API_MODEL"], input=history, tools=tools,
                instructions="Use multiply for arithmetic, then answer briefly.",
                store=False, include=["reasoning.encrypted_content"], stream=True,
            ) as stream:
                for event_count, event in enumerate(stream, start=1):
                    if event_count > 512 or result is not None:
                        raise ValueError("Event limit or event after completion.")
                    if event.type in ("error", "response.failed", "response.incomplete",
                                      "response.refusal.delta", "response.refusal.done"):
                        raise ValueError("Failed or refused response.")
                    if event.type == "response.output_item.added" and event.item.type == "function_call":
                        index = event.output_index
                        if type(index) is not int or index < 0 or index in pending or len(pending) >= 4:
                            raise ValueError("Invalid function item index.")
                        if (not event.item.id or not event.item.call_id
                                or not isinstance(event.item.arguments, str)):
                            raise ValueError("Missing function identity or arguments.")
                        pending[index] = {"id": event.item.id, "call_id": event.item.call_id,
                                          "name": event.item.name, "arguments": event.item.arguments,
                                          "arguments_done": False, "item_done": False}
                        if len(event.item.arguments) > 16_384:
                            raise ValueError("Arguments limit reached.")
                    elif event.type in ("response.function_call_arguments.delta", "response.function_call_arguments.done"):
                        call = pending[event.output_index]
                        if call["id"] != event.item_id or call["arguments_done"]:
                            raise ValueError("Unknown, changed, or finished function item.")
                        if event.type.endswith(".delta"):
                            call["arguments"] += event.delta
                            if len(call["arguments"]) > 16_384:
                                raise ValueError("Arguments limit reached.")
                        else:
                            if event.arguments != call["arguments"]:
                                raise ValueError("Final arguments differ from streamed arguments.")
                            call["arguments_done"] = True
                    elif event.type == "response.output_item.done" and event.item.type == "function_call":
                        call = pending[event.output_index]
                        if (not call["arguments_done"] or call["item_done"]
                                or any(getattr(event.item, key) != call[key]
                                       for key in ("id", "call_id", "name", "arguments"))):
                            raise ValueError("Incomplete or inconsistent function item.")
                        call["item_done"] = True
                    elif event.type == "response.completed":
                        result = event.response
            if result is None:
                raise ValueError("Stream ended without response.completed.")
            final_indices = set()
            for index, item in enumerate(result.output):
                if item.type != "function_call":
                    continue
                call = pending[index]
                if (not call["item_done"] or any(getattr(item, key) != call[key]
                                                for key in ("id", "call_id", "name", "arguments"))):
                    raise ValueError("Completed response differs from streamed calls.")
                final_indices.add(index)
            if final_indices != set(pending):
                raise ValueError("Completed response lost a streamed call.")
            # Execute only after the completed response and the entire stream validate.
            if result.status != "completed":
                raise ValueError("Response did not complete.")
            calls = []
            for item in result.output:
                if item.type == "function_call":
                    if getattr(item, "status", None) not in (None, "completed"):
                        raise ValueError("Function call did not complete.")
                    calls.append(item)
                elif item.type == "message":
                    if item.status != "completed" or any(part.type != "output_text" for part in item.content):
                        raise ValueError("Refusal or incomplete message.")
                elif item.type != "reasoning":
                    raise ValueError("Unsupported output item.")
            if not calls:
                if not result.output_text.strip():
                    raise ValueError("No final text.")
                print(result.output_text)
                break
            if round_index == 2 or len(calls) > 4:
                raise ValueError("Tool round or call limit reached.")
            outputs = []
            for call in calls:
                if (call.name != "multiply" or not isinstance(call.call_id, str)
                        or not call.call_id.strip() or call.call_id in seen_call_ids):
                    raise ValueError("Unknown tool or invalid call ID.")
                args = json.loads(call.arguments)
                if not isinstance(args, dict) or set(args) != {"a", "b"}:
                    raise ValueError("Invalid tool arguments.")
                if any(type(value) not in (int, float) or not math.isfinite(value) for value in args.values()):
                    raise ValueError("Expected finite numbers.")
                value = args["a"] * args["b"]
                if not math.isfinite(value):
                    raise ValueError("Invalid tool result.")
                seen_call_ids.add(call.call_id)
                outputs.append({
                    "type": "function_call_output", "call_id": call.call_id,
                    "output": json.dumps({"value": value}, allow_nan=False),
                })
            history.extend(item.model_dump(exclude_none=True) for item in result.output)
            history.extend(outputs)
    except APIStatusError as error:
        raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
    except APIConnectionError:
        raise SystemExit("Connection failed. No automatic retry.")
    except (ValueError, TypeError, AttributeError, KeyError, OverflowError, APIError):
        raise SystemExit("Streaming tool workflow failed; no incomplete answer published.")

Execution and failure policy

• Both scripts wait for the complete turn, validate allowed tool names, IDs, and all arguments, and implement only a local multiply operation without external effects. Strict schemas do not replace local validation.
• Limits are four calls and 512 events per turn, with at most three model requests; the third must return final text. Arguments are limited to 16,384 UTF-16 code units in TypeScript or characters in Python. Chat text is limited to 100,000 code units. These are example safeguards, not server limits, and do not bound individual event bytes before SDK decoding.
• Responses compares argument deltas, done arguments, and final output, then replays complete output items, including reasoning / encrypted_content / phase, and tool results. store: false avoids response-ID continuation; field preservation still depends on the channel.
• Final text is buffered until stream completion validates. HTTP 401/429, event errors, refusals, truncation, missing completion markers, unknown tools, repeated IDs, invalid arguments, and limit violations exit nonzero without another request or automatic retry.

This completion gate is not a business transaction. Replacing multiplication with payments, messaging, or database writes requires application authorization, idempotency keys, execution records, and result recovery. A network failure must not trigger re-execution of an already completed tool. These examples perform only pure computation, requiring no external service credentials.

Local verification and protocol scope

npm run check:examples executes these files against a local HTTP fixture. It checks interleaved calls, UTF-8 split across network chunks, complete history and result association, second requests, and absence of further requests after failures. Simulated responses cannot prove that a real model selects tools or that a channel preserves every event. Validate with an authorized test model before deployment.

Native Claude and native Gemini use different blocks, events, and result structures; these scripts do not implement those native protocols.

References: OpenAI function calling, Responses streaming events.

Runnable native SSE and tool loops for Claude and Gemini include completion validation and signature preservation.

阅读中文