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

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

准备工作

先按 SDK 示例完成一次普通文本请求。从模型广场复制适用于目标协议的准确 Model ID。模型选择指南列出了需要检查的项目。在项目的 Python 虚拟环境中安装 openai 和 anthropic，然后在同一终端中替换以下占位值：

python -m pip install openai anthropic
export BETTERTOKEN_API_KEY="YOUR_API_KEY"
export BETTERTOKEN_MODEL_ID="YOUR_MODEL_ID"

下方每个 Python 代码块均可独立运行。120 秒超时仅用于这些简短测试；应用的超时设置请参阅超时配置。设置 max_tokens 或 max_output_tokens 前，请检查模型的输出 Token 上限。

流式接收文本

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

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

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

Chat Completions
Responses
Messages

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)

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)

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
Responses
Messages

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))

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)

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 进行类型校验时，请参阅结构化输出常见问题。

流中断后应该重试吗？

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

官方参考

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

接口文档

开发者指南

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

准备工作

流式接收文本

为三种协议定义天气工具

执行工具调用并回传结果

常见问题

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

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

流中断后应该重试吗？

相关文档

官方参考

​准备工作

​流式接收文本

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

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

​常见问题

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

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

​流中断后应该重试吗？

​相关文档

​官方参考

准备工作

流式接收文本

为三种协议定义天气工具

执行工具调用并回传结果

常见问题

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

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

流中断后应该重试吗？

相关文档

官方参考