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

# BetterToken API 流式输出与工具调用指南

> 为 BetterToken API 添加流式输出与工具调用，对比三种文本接口的 SSE 事件、函数定义和结果回传格式，并用 Python 示例验证模型支持情况。

通过 `stream: true` 开启流式输出，并按所选 API 的格式定义由客户端执行的工具。流式输出和工具调用的支持情况因模型而异；接入应用前，先用你的 Model ID 发送小请求验证。

## 准备工作

先按 [SDK 示例](/zh/guides/sdk-examples.md)完成一次普通文本请求。从[模型广场](https://bettertoken.ai/pricing)复制适用于目标协议的准确 Model ID。[模型选择指南](/zh/faq/model-calling/model-selection-guide.md)列出了需要检查的项目。

在项目的 Python 虚拟环境中安装 `openai` 和 `anthropic`，然后在同一终端中替换以下占位值：

```bash theme={null}
python -m pip install openai anthropic
export BETTERTOKEN_API_KEY="YOUR_API_KEY"
export BETTERTOKEN_MODEL_ID="YOUR_MODEL_ID"
```

下方每个 Python 代码块均可独立运行。120 秒超时仅用于这些简短测试；应用的超时设置请参阅[超时配置](/zh/best-practices.md#超时配置)。设置 `max_tokens` 或 `max_output_tokens` 前，请检查模型的[输出 Token 上限](/zh/faq/model-calling/max-tokens.md)。

## 流式接收文本

流式输出在一个 HTTP 响应中通过服务器发送事件（SSE）传输内容。单个 SSE 事件不一定包含完整句子或完整的 JSON 参数对象。

| API | 请求设置 | 文本事件或片段 | 响应结束标志 |
| - | - | - | - |
| [Chat Completions](/zh/api-reference/chat-completions.md) | `stream: true` | `choices[].delta.content`；工具片段使用 `delta.tool_calls` | 某个候选项返回 `finish_reason`；原始流以 `data: [DONE]` 结束 |
| [Responses](/zh/api-reference/responses.md) | `stream: true` | `response.output_text.delta` 等带类型的事件 | `response.completed`；还需处理失败或响应不完整的事件 |
| [Messages](/zh/api-reference/messages.md) | `stream: true` | 含 `text_delta` 的 `content_block_delta`；工具参数使用 `input_json_delta` | `message_delta` 携带停止原因，随后是 `message_stop` |

SDK 会解析 SSE 格式。以下示例打印文本增量和结束信息；应用还应处理错误事件。

**Chat Completions**

```python theme={null}
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://bettertoken.ai/v1",
    timeout=120.0,
    max_retries=0,
)
with client.chat.completions.create(
    model=os.environ["BETTERTOKEN_MODEL_ID"],
    messages=[{"role": "user", "content": "Say hello in one sentence."}],
    stream=True,
) as stream:
    for chunk in stream:
        for choice in chunk.choices:
            if choice.delta.content:
                print(choice.delta.content, end="", flush=True)
            if choice.finish_reason:
                print("\nFinish reason:", choice.finish_reason)
```

**Responses**

```python theme={null}
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://bettertoken.ai/v1",
    timeout=120.0,
    max_retries=0,
)
with client.responses.create(
    model=os.environ["BETTERTOKEN_MODEL_ID"],
    input="Say hello in one sentence.",
    stream=True,
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
        elif event.type in (
            "response.completed", "response.failed", "response.incomplete", "error"
        ):
            print("\n", event)
```

**Messages**

```python theme={null}
import os
from anthropic import Anthropic

client = Anthropic(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://bettertoken.ai",
    timeout=120.0,
    max_retries=0,
)
with client.messages.create(
    model=os.environ["BETTERTOKEN_MODEL_ID"],
    max_tokens=256,
    messages=[{"role": "user", "content": "Say hello in one sentence."}],
    stream=True,
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            print("\nStop reason:", event.delta.stop_reason)
```

## 为三种协议定义天气工具

模型发出函数调用请求；你的应用负责校验参数、执行函数并回传结果。示例中的 `get_weather` 接收 `location` 参数，返回**固定的演示数据，并非实时天气**，无需接入天气服务或申请额外的 Key。

| 请求或响应字段 | Chat Completions | Responses | Messages |
| - | - | - | - |
| 提示词 | `messages` | `input` | `messages` |
| `tools[]` 中的工具声明 | `type: "function"`，`name`、`description`、`parameters` 放在 `function` 内 | `type: "function"`，`name`、`description`、`parameters` 放在工具对象顶层 | `name`、`description`、`input_schema` |
| 返回的调用 | `choices[].message.tool_calls[]` | `output[]` 中 `type: "function_call"` 的项 | `content[]` 中 `type: "tool_use"` 的块 |
| 参数 | `function.arguments` 中的 JSON 字符串 | `arguments` 中的 JSON 字符串 | `input` 中的对象 |
| 回传结果 | 含 `role: "tool"`、`tool_call_id`、`content` 的消息 | 含 `type: "function_call_output"`、`call_id` 和字符串 `output` 的输入项 | 含 `type: "tool_result"`、`tool_use_id`、`content` 的用户消息内容块 |

支持情况取决于模型和协议。即使普通文本请求已成功，也请用你的 Model ID 测试这个小型工具请求。声明工具不会启用托管搜索或其他由服务商执行的工具。

## 执行工具调用并回传结果

以下非流式示例演示一轮完整的工具调用，方便你检查完整参数。每个示例处理首个响应中的所有工具调用，并允许一次后续请求。更长的 Agent 工作流需要自行设置有限循环和停止条件。

**Chat Completions**

```python theme={null}
import json
import os
from openai import OpenAI

def get_weather(location):
    if not isinstance(location, str) or not location.strip():
        raise ValueError("location must be a non-empty string")
    return {"location": location, "temperature_c": 20, "source": "demo"}

client = OpenAI(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://bettertoken.ai/v1", timeout=120.0, max_retries=0,
)
model = os.environ["BETTERTOKEN_MODEL_ID"]
schema = {
    "type": "object",
    "properties": {"location": {"type": "string"}},
    "required": ["location"], "additionalProperties": False,
}
tools = [{"type": "function", "function": {
    "name": "get_weather", "description": "Get demonstration weather for a location.",
    "parameters": schema,
}}]
messages = [{"role": "user", "content": "Call get_weather for Paris and report the demo result."}]
first = client.chat.completions.create(model=model, messages=messages, tools=tools)
message = first.choices[0].message
calls = message.tool_calls or []
if not calls:
    print("No tool call returned:", message.content)
else:
    messages.append(message.model_dump(exclude_none=True))
    for call in calls:
        if call.type != "function" or call.function.name != "get_weather":
            raise ValueError("Unexpected tool")
        args = json.loads(call.function.arguments)
        result = get_weather(**args)
        messages.append({
            "role": "tool", "tool_call_id": call.id,
            "content": json.dumps(result),
        })
    final = client.chat.completions.create(model=model, messages=messages, tools=tools)
    print(final.choices[0].message.model_dump(exclude_none=True))
```

**Responses**

```python theme={null}
import json
import os
from openai import OpenAI

def get_weather(location):
    if not isinstance(location, str) or not location.strip():
        raise ValueError("location must be a non-empty string")
    return {"location": location, "temperature_c": 20, "source": "demo"}

client = OpenAI(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://bettertoken.ai/v1", timeout=120.0, max_retries=0,
)
model = os.environ["BETTERTOKEN_MODEL_ID"]
schema = {
    "type": "object",
    "properties": {"location": {"type": "string"}},
    "required": ["location"], "additionalProperties": False,
}
tools = [{
    "type": "function", "name": "get_weather",
    "description": "Get demonstration weather for a location.", "parameters": schema,
}]
history = [{"role": "user", "content": "Call get_weather for Paris and report the demo result."}]
first = client.responses.create(model=model, input=history, tools=tools)
calls = [item for item in first.output if item.type == "function_call"]
if not calls:
    print("No tool call returned:", first.output)
else:
    history.extend(first.output)
    for call in calls:
        if call.name != "get_weather":
            raise ValueError("Unexpected tool")
        args = json.loads(call.arguments)
        result = get_weather(**args)
        history.append({
            "type": "function_call_output", "call_id": call.call_id,
            "output": json.dumps(result),
        })
    final = client.responses.create(model=model, input=history, tools=tools)
    print(final.output)
```

**Messages**

```python theme={null}
import json
import os
from anthropic import Anthropic

def get_weather(location):
    if not isinstance(location, str) or not location.strip():
        raise ValueError("location must be a non-empty string")
    return {"location": location, "temperature_c": 20, "source": "demo"}

client = Anthropic(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://bettertoken.ai", timeout=120.0, max_retries=0,
)
model = os.environ["BETTERTOKEN_MODEL_ID"]
schema = {
    "type": "object",
    "properties": {"location": {"type": "string"}},
    "required": ["location"], "additionalProperties": False,
}
tools = [{
    "name": "get_weather", "description": "Get demonstration weather for a location.",
    "input_schema": schema,
}]
messages = [{"role": "user", "content": "Call get_weather for Paris and report the demo result."}]
first = client.messages.create(model=model, max_tokens=1024, messages=messages, tools=tools)
calls = [block for block in first.content if block.type == "tool_use"]
if first.stop_reason == "max_tokens":
    raise RuntimeError("Response truncated; inspect it before executing tools")
if not calls:
    print("No tool call returned:", first.content)
else:
    messages.append({"role": "assistant", "content": first.content})
    results = []
    for call in calls:
        if call.name != "get_weather":
            raise ValueError("Unexpected tool")
        result = get_weather(**call.input)
        results.append({
            "type": "tool_result", "tool_use_id": call.id,
            "content": json.dumps(result),
        })
    messages.append({"role": "user", "content": results})
    final = client.messages.create(model=model, max_tokens=1024, messages=messages, tools=tools)
    print(final.content)
```

添加工具结果前，保留完整的助手响应。Responses 需要保留所有 `output` 项，包括返回的推理项；Messages 需要保留内容块及其顺序。使用返回的调用 ID 关联结果，不要自行编造 ID。

## 常见问题

### 参数还在流式传输时，可以执行工具吗？

先按工具调用索引或项目 ID 累积片段，等参数完整后再解析和校验。未完整接收的 JSON 可能无法解析。组合流式输出与工具调用前，先跑通上方非流式示例。

### 为什么普通文本成功，工具调用却失败？

支持情况取决于模型和协议。先用你的 Model ID 验证小型工具请求，再检查工具定义、返回参数、结果 ID 和消息顺序。需要对 JSON 进行类型校验时，请参阅[结构化输出常见问题](/zh/faq/model-calling/structured-json-pydantic.md)。

### 流中断后应该重试吗？

重新发送前，先检查请求记录。断开连接不能确定请求是否已经完成。重试会发起新的一次请求，且不能重复执行已经完成的工具操作。参阅[错误与重试说明](/zh/api-reference/errors.md)。

## 相关文档

* [SDK 配置与认证](/zh/guides/sdk-examples.md)
* [API 参考概览](/zh/api-reference/introduction.md)
* [输出 Token 上限](/zh/faq/model-calling/max-tokens.md)
* [结构化 JSON 与 Pydantic](/zh/faq/model-calling/structured-json-pydantic.md)
* [超时配置](/zh/best-practices.md#超时配置)
* [错误与限制](/zh/api-reference/errors.md)

## 官方参考

以下资料定义了上游协议字段。使用相关功能前，请先用你的 BetterToken Model ID 验证。

* [OpenAI 函数调用](https://developers.openai.com/api/docs/guides/function-calling)
* [OpenAI 流式响应](https://developers.openai.com/api/docs/guides/streaming-responses)
* [Anthropic 工具定义](https://platform.claude.com/docs/en/agents-and-tools/tool-use/define-tools)
* [Anthropic 工具结果处理](https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls)
* [Anthropic Messages 流式响应](https://platform.claude.com/docs/en/build-with-claude/streaming)


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