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 を指定する前に、モデルの出力トークン上限を確認します。

テキスト応答をストリーミングする

ストリーミングは、一つの 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`	`content_block_delta` 内の `text_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 でこの小さなツールリクエストを試してください。ツールを定義しても、ホスト型の検索やプロバイダー側で実行する他のツールが有効になるわけではありません。

ツールを呼び出し、結果を返す

以下の非ストリーミングの例では、引数全体を確認できるよう、ツール呼び出しを一往復だけ行います。最初の応答に含まれるすべてのツール呼び出しを処理し、追加のリクエストを一回送信します。より長いエージェントの処理には、上限付きのループと停止条件を別途設けてください。

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 の検証については、構造化出力の FAQを参照してください。

ストリームが切断されたら再試行すべきですか？

再送信する前にリクエスト記録を確認します。切断しただけでは、リクエストが完了したかどうかは判断できません。再試行は新たな試行になるため、実行済みのツール操作を繰り返さないようにしてください。エラーと再試行のガイドを参照してください。

公式リファレンス

以下の資料は、上流 API のプロトコルフィールドを定義しています。利用する前に、BetterToken の Model ID で機能を検証してください。

API リファレンス

開発者ガイド

BetterToken API のストリーミングとツール呼び出し

前提条件

テキスト応答をストリーミングする

プロトコルごとに天気ツールを定義する

ツールを呼び出し、結果を返す

よくある質問

ツールの引数を受信した順に実行してもよいですか？

通常のテキストは動くのに、ツールが失敗するのはなぜですか？

ストリームが切断されたら再試行すべきですか？

関連ガイド

公式リファレンス

​前提条件

​テキスト応答をストリーミングする

​プロトコルごとに天気ツールを定義する

​ツールを呼び出し、結果を返す

​よくある質問

​ツールの引数を受信した順に実行してもよいですか？

​通常のテキストは動くのに、ツールが失敗するのはなぜですか？

​ストリームが切断されたら再試行すべきですか？

​関連ガイド

​公式リファレンス

前提条件

テキスト応答をストリーミングする

プロトコルごとに天気ツールを定義する

ツールを呼び出し、結果を返す

よくある質問

ツールの引数を受信した順に実行してもよいですか？

通常のテキストは動くのに、ツールが失敗するのはなぜですか？

ストリームが切断されたら再試行すべきですか？

関連ガイド

公式リファレンス