Streaming und Tool-Aufrufe mit der BetterToken-API

Aktiviere Streaming mit stream: true und definiere clientseitige Tools mit dem Schema deiner API. Die Unterstützung für Streaming und Tool-Aufrufe hängt vom Modell ab; teste eine kleine Anfrage mit deiner Model ID, bevor du diese Funktionen in einer Anwendung nutzt.

Voraussetzungen

Führe zuerst mit den SDK-Beispielen eine reine Textanfrage erfolgreich aus. Kopiere aus der model plaza die exakte Model ID für das zu testende Protokoll. Die Anleitung zur Modellauswahl erklärt, was du prüfen solltest. Installiere openai und anthropic in der virtuellen Python-Umgebung deines Projekts und ersetze die Platzhalter im selben Terminal:

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

Jeder folgende Python-Block lässt sich unabhängig ausführen. Der Timeout von 120 Sekunden ist ein Beispiel für diese kurzen Tests; wähle den Timeout deiner Anwendung anhand der Timeout-Hinweise. Prüfe die Limits für Ausgabetokens des Modells, bevor du max_tokens oder max_output_tokens festlegst.

Textantworten streamen

Streaming überträgt vom Server gesendete Ereignisse (SSE) innerhalb einer HTTP-Antwort. Ein SSE-Ereignis ist nicht unbedingt ein vollständiger Satz oder ein vollständiges JSON-Argumentobjekt.

API	Anfrageeinstellung	Textereignisse oder Fragmente	Ende der Antwort
Chat Completions	`stream: true`	`choices[].delta.content`; Tool-Fragmente verwenden `delta.tool_calls`	Eine Auswahl liefert `finish_reason`; der rohe Stream endet mit `data: [DONE]`
Responses	`stream: true`	Typisierte Ereignisse wie `response.output_text.delta`	`response.completed`; behandle auch Fehler und Ereignisse für unvollständige Antworten
Messages	`stream: true`	`content_block_delta` mit `text_delta`; Tool-Argumente verwenden `input_json_delta`	`message_delta` enthält den Abbruchgrund, danach folgt `message_stop`

Das SDK verarbeitet das SSE-Format für dich. Diese Beispiele geben Textfragmente und Abschlussinformationen aus; Anwendungen sollten auch Fehlerereignisse behandeln.

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)

Ein Wetter-Tool für jedes Protokoll definieren

Das Modell fordert einen Funktionsaufruf an; deine Anwendung prüft die Argumente, führt die Funktion aus und gibt das Ergebnis zurück. Die Beispiele verwenden eine Funktion get_weather mit einem Argument location und liefern feste Demodaten, keine aktuellen Wetterdaten. Du benötigst weder einen Wetterdienst noch einen zusätzlichen Key.

Anfrage- oder Antwortfeld	Chat Completions	Responses	Messages
Eingabe-Prompt	`messages`	`input`	`messages`
Tool-Deklaration in `tools[]`	`type: "function"`, mit `name`, `description` und `parameters` innerhalb von `function`	`type: "function"`, mit `name`, `description` und `parameters` auf der obersten Ebene des Tools	`name`, `description` und `input_schema`
Zurückgegebener Aufruf	`choices[].message.tool_calls[]`	Element in `output[]` mit `type: "function_call"`	Block in `content[]` mit `type: "tool_use"`
Argumente	JSON-Zeichenfolge in `function.arguments`	JSON-Zeichenfolge in `arguments`	Objekt in `input`
Ergebnis zurückgeben	Nachricht mit `role: "tool"`, `tool_call_id` und `content`	Eingabeelement mit `type: "function_call_output"`, `call_id` und `output` als Zeichenfolge	Inhaltsblock des Nutzers mit `type: "tool_result"`, `tool_use_id` und `content`

Die Unterstützung hängt von Modell und Protokoll ab. Teste diese kleine Tool-Anfrage mit deiner Model ID auch dann, wenn reiner Text bereits funktioniert. Eine Tool-Deklaration aktiviert keine gehostete Suche oder andere vom Anbieter ausgeführte Tools.

Einen Tool-Aufruf ausführen und das Ergebnis zurückgeben

Diese Beispiele ohne Streaming zeigen einen vollständigen Ablauf von Aufruf und Ergebnis, damit du die vollständigen Argumente prüfen kannst. Jedes Beispiel verarbeitet alle Tool-Aufrufe der ersten Antwort und erlaubt eine Folgeanfrage. Ein längerer Agenten-Workflow benötigt eine eigene begrenzte Schleife und eine Abbruchbedingung.

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)

Bewahre die vollständige Assistentenantwort auf, bevor du Tool-Ergebnisse ergänzt. Behalte bei Responses alle Elemente aus output, einschließlich zurückgegebener Reasoning-Elemente. Bewahre bei Messages die Inhaltsblöcke und ihre Reihenfolge. Ordne Ergebnisse den zurückgegebenen Aufruf-IDs zu, statt eigene IDs zu erfinden.

Häufige Fragen

Kann ich Tool-Argumente sofort beim Eintreffen ausführen?

Sammle Fragmente anhand des Tool-Aufrufindex oder der Element-ID. Parse und validiere anschließend die vollständigen Argumente. Unvollständiges JSON lässt sich möglicherweise nicht parsen. Beginne mit den Beispielen ohne Streaming, bevor du Streaming und Tools kombinierst.

Warum funktioniert reiner Text, während Tools fehlschlagen?

Die Unterstützung hängt von Modell und Protokoll ab; prüfe eine kleine Tool-Anfrage mit deiner Model ID. Kontrolliere Tool-Schema, zurückgegebene Argumente, Ergebnis-ID und Nachrichtenreihenfolge. Für typisierte JSON-Validierung hilft die FAQ zu strukturierten Ausgaben.

Sollte ich einen unterbrochenen Stream erneut anfordern?

Prüfe vor dem erneuten Senden den Anfrageeintrag. Ein Verbindungsabbruch zeigt nicht, ob die Anfrage abgeschlossen wurde. Eine Wiederholung startet einen weiteren Versuch und darf keine bereits ausgeführte Tool-Aktion wiederholen. Siehe Hinweise zu Fehlern und Wiederholungen.

Weitere Informationen

Offizielle Referenzen

Diese Referenzen definieren die Protokollfelder der Anbieter. Prüfe die Funktion mit deiner BetterToken Model ID, bevor du sie verwendest.

API-Referenz

Entwicklerleitfäden

Streaming und Tool-Aufrufe mit der BetterToken-API

Voraussetzungen

Textantworten streamen

Ein Wetter-Tool für jedes Protokoll definieren

Einen Tool-Aufruf ausführen und das Ergebnis zurückgeben

Häufige Fragen

Kann ich Tool-Argumente sofort beim Eintreffen ausführen?

Warum funktioniert reiner Text, während Tools fehlschlagen?

Sollte ich einen unterbrochenen Stream erneut anfordern?

Weitere Informationen

Offizielle Referenzen

​Voraussetzungen

​Textantworten streamen

​Ein Wetter-Tool für jedes Protokoll definieren

​Einen Tool-Aufruf ausführen und das Ergebnis zurückgeben

​Häufige Fragen

​Kann ich Tool-Argumente sofort beim Eintreffen ausführen?

​Warum funktioniert reiner Text, während Tools fehlschlagen?

​Sollte ich einen unterbrochenen Stream erneut anfordern?

​Weitere Informationen

​Offizielle Referenzen

Voraussetzungen

Textantworten streamen

Ein Wetter-Tool für jedes Protokoll definieren

Einen Tool-Aufruf ausführen und das Ergebnis zurückgeben

Häufige Fragen

Kann ich Tool-Argumente sofort beim Eintreffen ausführen?

Warum funktioniert reiner Text, während Tools fehlschlagen?

Sollte ich einen unterbrochenen Stream erneut anfordern?

Weitere Informationen

Offizielle Referenzen