> ## Documentation Index
> Fetch the complete documentation index at: https://docs.playapi.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 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](/en/guides/tool-controls).

This page extends [nonstreaming tool calls](/en/guides/tool-calling) 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](/en/guides/typescript) or [Python](/en/guides/python); no new dependencies are required. Set the gateway token in `API_KEY`, then configure a Base URL including `/v1`:

```sh theme={null}
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`

```typescript theme={null}
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`

```python theme={null}
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](/en/guides/native-claude) and [native Gemini](/en/guides/native-gemini) use different blocks, events, and result structures; these scripts do not implement those native protocols.

References: [OpenAI function calling](https://developers.openai.com/api/docs/guides/function-calling), [Responses streaming events](https://developers.openai.com/api/reference/resources/responses/streaming-events).

Runnable native SSE and tool loops for [Claude](/en/guides/native-claude-streaming) and [Gemini](/en/guides/native-gemini-streaming) include completion validation and signature preservation.

[阅读中文](/zh/guides/streaming-tools)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.