Streaming and tool calling with BetterToken APIs

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. Copy an exact Model ID from the model plaza for the protocol you will test. The model selection guide explains what to check. Install openai and anthropic in your project’s Python virtual environment, then replace the placeholders in the same terminal:

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. Check the model’s output token limits 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	`stream: true`	`choices[].delta.content`; tool fragments use `delta.tool_calls`	A choice supplies `finish_reason`; the raw stream ends with `data: [DONE]`
Responses	`stream: true`	Typed events such as `response.output_text.delta`	`response.completed`; also handle failure or incomplete-response events
Messages	`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
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)

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

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.

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.

Official references

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

API Reference

Developer guides

Streaming and tool calling with BetterToken APIs

Prerequisites

Stream text responses

Define a weather tool for each protocol

Run a tool call and return its result

Common questions

Can I execute tool arguments as they arrive?

Why does plain text work while tools fail?

Should I retry an interrupted stream?

Official references

​Prerequisites

​Stream text responses

​Define a weather tool for each protocol

​Run a tool call and return its result

​Common questions

​Can I execute tool arguments as they arrive?

​Why does plain text work while tools fail?

​Should I retry an interrupted stream?

​Related

​Official references

Prerequisites

Stream text responses

Define a weather tool for each protocol

Run a tool call and return its result

Common questions

Can I execute tool arguments as they arrive?

Why does plain text work while tools fail?

Should I retry an interrupted stream?

Related

Official references