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

# Streaming and tool calling with BetterToken APIs

> Add streaming and tool calling to BetterToken API requests. Compare Chat Completions, Responses, and Messages events, function schemas, and tool results.

Enable streaming with `stream: true` and define client tools with the schema required by your API. Streaming and tool calling support varies by model; test a small request with your Model ID before using either feature in an application.

## Prerequisites

First complete a plain text request using the [SDK examples](/en/guides/sdk-examples.md). Copy an exact Model ID from the [model plaza](https://bettertoken.ai/pricing) for the protocol you will test. The [model selection guide](/en/faq/model-calling/model-selection-guide.md) explains what to check.

Install `openai` and `anthropic` in your project's Python virtual environment, then replace the placeholders in the same terminal:

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

Each Python block below runs independently. The 120-second timeout is an example for these short tests; choose your application's timeout using the [timeout guidance](/en/best-practices.md#set-a-generous-timeout). Check the model's [output token limits](/en/faq/model-calling/max-tokens.md) before setting `max_tokens` or `max_output_tokens`.

## Stream text responses

Streaming sends server-sent events (SSE) over one HTTP response. An SSE event is not necessarily a complete sentence or JSON argument object.

| API | Request setting | Text events or chunks | End of response |
| - | - | - | - |
| [Chat Completions](/en/api-reference/chat-completions.md) | `stream: true` | `choices[].delta.content`; tool fragments use `delta.tool_calls` | A choice supplies `finish_reason`; the raw stream ends with `data: [DONE]` |
| [Responses](/en/api-reference/responses.md) | `stream: true` | Typed events such as `response.output_text.delta` | `response.completed`; also handle failure or incomplete-response events |
| [Messages](/en/api-reference/messages.md) | `stream: true` | `content_block_delta` with `text_delta`; tool arguments use `input_json_delta` | `message_delta` carries the stop reason, followed by `message_stop` |

The SDK parses SSE framing for you. These examples print text deltas and completion information; applications should also handle error events.

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

## Define a weather tool for each protocol

The model requests a function call; your application validates the arguments, runs the function, and returns the result. The examples use one `get_weather` function with a `location` argument and return **fixed demonstration data, not live weather**. No weather service or additional Key is required.

| Request or response field | Chat Completions | Responses | Messages |
| - | - | - | - |
| Prompt | `messages` | `input` | `messages` |
| Tool declaration in `tools[]` | `type: "function"`, with `name`, `description`, and `parameters` inside `function` | `type: "function"`, with `name`, `description`, and `parameters` at the tool root | `name`, `description`, and `input_schema` |
| Returned call | `choices[].message.tool_calls[]` | `output[]` item of `type: "function_call"` | `content[]` block of `type: "tool_use"` |
| Arguments | JSON string in `function.arguments` | JSON string in `arguments` | Object in `input` |
| Return result | Message with `role: "tool"`, `tool_call_id`, and `content` | Input item with `type: "function_call_output"`, `call_id`, and string `output` | User content block with `type: "tool_result"`, `tool_use_id`, and `content` |

Support varies by model and protocol. Test this small tool request with your Model ID even if plain text already works. A tool declaration does not enable hosted search or other provider-run tools.

## Run a tool call and return its result

These non-streaming examples show one tool round trip so you can inspect complete arguments. Each handles every tool call in the first response and allows one follow-up request. A longer agent workflow needs its own bounded loop and stop condition.

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

Keep the complete assistant response before adding tool results. For Responses, retain all `output` items, including reasoning items when returned. For Messages, preserve content blocks and their order. Match results to the returned call IDs rather than inventing IDs.

## Common questions

### Can I execute tool arguments as they arrive?

Accumulate fragments by tool-call index or item ID, then parse and validate the completed arguments. Partial JSON may not parse. Start with the non-streaming examples above before combining streaming and tools.

### Why does plain text work while tools fail?

Support varies by model and protocol; verify a small tool request with your Model ID. Check the tool schema, returned arguments, result ID, and message order. For typed JSON validation, use the [structured output FAQ](/en/faq/model-calling/structured-json-pydantic.md).

### Should I retry an interrupted stream?

Check the request record before resending. A disconnect does not establish whether the request finished. A retry starts another attempt and must not repeat a tool action that already ran. See [errors and retry guidance](/en/api-reference/errors.md).

## Related

* [SDK setup and authentication](/en/guides/sdk-examples.md)
* [API reference overview](/en/api-reference/introduction.md)
* [Output token limits](/en/faq/model-calling/max-tokens.md)
* [Structured JSON and Pydantic](/en/faq/model-calling/structured-json-pydantic.md)
* [Timeout settings](/en/best-practices.md#set-a-generous-timeout)
* [Errors and limits](/en/api-reference/errors.md)

## Official references

These references define upstream protocol fields. Verify the feature with your BetterToken Model ID before using it.

* [OpenAI function calling](https://developers.openai.com/api/docs/guides/function-calling)
* [OpenAI streaming responses](https://developers.openai.com/api/docs/guides/streaming-responses)
* [Anthropic tool definitions](https://platform.claude.com/docs/en/agents-and-tools/tool-use/define-tools)
* [Anthropic tool results](https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls)
* [Anthropic streaming 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.