> ## 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 のストリーミングとツール呼び出しを実装。Chat Completions・Responses・Messages のイベントと関数スキーマ、結果の返し方を Python 例で確認します。

`stream: true` でストリーミングを有効にし、API に合ったスキーマでクライアント側のツールを定義します。ストリーミングとツール呼び出しへの対応はモデルによって異なります。アプリに組み込む前に、使用する Model ID で小さなリクエストを試してください。

## 前提条件

まず [SDK のサンプル](/ja/guides/sdk-examples.md)で通常のテキストリクエストを成功させます。[モデル広場](https://bettertoken.ai/pricing)から、テストするプロトコルに対応する正確な Model ID をコピーしてください。確認項目は[モデル選択ガイド](/ja/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 秒のタイムアウトは、この短いテスト用の設定例です。アプリのタイムアウトは[タイムアウトの設定ガイド](/ja/best-practices.md#set-a-generous-timeout)に従って決めてください。`max_tokens` や `max_output_tokens` を指定する前に、モデルの[出力トークン上限](/ja/faq/model-calling/max-tokens.md)を確認します。

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

ストリーミングは、一つの HTTP 応答でサーバー送信イベント（SSE）を送ります。SSE の各イベントが、完全な文や JSON の引数オブジェクトに対応するとは限りません。

| API | リクエスト設定 | テキストイベントやチャンク | 応答の終了 |
| - | - | - | - |
| [Chat Completions](/ja/api-reference/chat-completions.md) | `stream: true` | `choices[].delta.content`。ツールの断片は `delta.tool_calls` | 各選択肢が `finish_reason` を返し、生のストリームは `data: [DONE]` で終了 |
| [Responses](/ja/api-reference/responses.md) | `stream: true` | `response.output_text.delta` などの型付きイベント | `response.completed`。失敗や未完了の応答イベントも処理する |
| [Messages](/ja/api-reference/messages.md) | `stream: true` | `content_block_delta` 内の `text_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 でこの小さなツールリクエストを試してください。ツールを定義しても、ホスト型の検索やプロバイダー側で実行する他のツールが有効になるわけではありません。

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

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

**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 の検証については、[構造化出力の FAQ](/ja/faq/model-calling/structured-json-pydantic.md)を参照してください。

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

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

## 関連ガイド

* [SDK の設定と認証](/ja/guides/sdk-examples.md)
* [API リファレンスの概要](/ja/api-reference/introduction.md)
* [出力トークン上限](/ja/faq/model-calling/max-tokens.md)
* [構造化 JSON と Pydantic](/ja/faq/model-calling/structured-json-pydantic.md)
* [タイムアウトの設定](/ja/best-practices.md#set-a-generous-timeout)
* [エラーと制限](/ja/api-reference/errors.md)

## 公式リファレンス

以下の資料は、上流 API のプロトコルフィールドを定義しています。利用する前に、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.