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

# 流式工具调用

> 组装交错参数分片，验证完成后执行并关联工具结果。

当前 Chat ↔ Responses 函数工具转换没有复制 strict 字段；不要将原生声明、协议转换与上游默认值视为相同保证。工具参数仍需应用验证。详见 [工具选择与并行控制](/zh/guides/tool-controls.md)。

本页在[非流式工具调用](/zh/guides/tool-calling.md)基础上完成“接收参数分片 → 校验完整调用 → 执行本地乘法 → 回传工具结果 → 获取最终回答”。先确认模型和渠道支持对应协议的流式函数调用及严格参数 Schema。

## 两种组装方式

| 阶段 | Chat Completions | Responses |
| - | - | - |
| 关联分片 | `choices[0].delta.tool_calls[].index` | `output_index` 与 `item_id` |
| 参数增量 | `function.arguments` 字符串 | `response.function_call_arguments.delta` 的 `delta` |
| 调用身份 | `id` 与 `function.name` | output item 的 `id`、`call_id`、`name` |
| 结束验证 | `finish_reason: tool_calls`，且迭代无异常结束 | arguments.done、output\_item.done、response.completed 相互一致，且迭代无异常结束 |
| 回传结果 | `role: tool` 与 `tool_call_id` | `type: function_call_output` 与 `call_id` |

同一轮不同工具的分片可能交错，不能用一个全局参数字符串。逐片累加参数，不对半截 JSON 解析或执行。Chat 本例要求 ID/名称首次出现时完整，后续只能省略或重复同值，不拼接名称。SDK 负责 SSE / UTF-8 解码；本例不直接验证 SDK 已消费的 `[DONE]` 字符串。

## 运行准备

按 [TypeScript](/zh/guides/typescript.md)或 [Python](/zh/guides/python.md)安装已有 OpenAI SDK；无需新增依赖。预先设置网关令牌 `API_KEY`，然后设置含 `/v1` 的 Base URL：

```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.
```

在文档目录分别运行 `node examples/stream-tools.mts` 或 `.venv/bin/python examples/stream-responses-tools.py`。单独保存下方代码时可运行 `node stream-tools.mts` 或 `python stream-responses-tools.py`。例子请求计算 7 × 8；最终措辞由模型决定，不强制输出某个固定句子。

## 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.")
```

## 执行与失败策略

* 两个脚本都先等完整一轮结束，再验证工具白名单、关联 ID 和全部参数；仅实现无外部副作用的 `multiply`。严格 Schema 不能替代本地校验。
* 每轮最多 4 个调用、512 个事件；最多 3 次模型请求，第三次必须产生最终文本。参数长度上限为 16,384（TypeScript 按 UTF-16 code unit，Python 按字符）；Chat 文本缓冲最多 100,000 code unit。这些是示例限制，不是服务端限额，也不限制 SDK 解码前单个事件的字节大小。
* Responses 示例核对增量参数、done 参数和最终 output，重放完整输出项（含 reasoning / encrypted\_content / phase）及结果。`store: false` 不依赖响应 ID 续接；目标渠道仍需支持对应字段保留。
* 两个脚本缓冲最终文本，流结束并通过完成校验后才打印。HTTP 401/429、事件级错误、拒答、截断、缺少完成标记、未知工具、重复 ID、非法参数或上限触发时退出非零，不发送下一轮，也不自动重试。

这个执行屏障不等于业务事务。若换成支付、发送消息或数据库写入，需在应用侧设计授权、幂等键、执行日志和结果恢复；网络失败不能成为重新执行已完成工具的依据。当前示例只执行纯计算，无需外部凭据或服务。

## 本地验证与协议范围

`npm run check:examples` 用本地 HTTP fixture 执行这些文件，检查交错调用、UTF-8 跨网络块、完整历史与工具结果关联、第二轮请求，以及异常时没有后续请求。模拟响应不能证明真实模型一定选择工具或目标渠道保留所有事件；上线前须用获授权的测试模型验证。

[Claude 原生](/zh/guides/native-claude.md)与 [Gemini 原生](/zh/guides/native-gemini.md)使用不同内容块、事件及结果结构；本页脚本不适用于这两种原生协议。

参考：[OpenAI 函数调用](https://developers.openai.com/api/docs/guides/function-calling)、[Responses 流式事件](https://developers.openai.com/api/reference/resources/responses/streaming-events)。

原生协议的可运行 SSE 与工具闭环见 [Claude](/zh/guides/native-claude-streaming.md)和 [Gemini](/zh/guides/native-gemini-streaming.md)，包含完成校验与签名保留。

[Read in English](/en/guides/streaming-tools.md)


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