Streaming et appels d’outils avec l’API BetterToken

Activez le streaming avec stream: true et définissez les outils du client selon le schéma requis par votre API. La prise en charge du streaming et des appels d’outils varie selon le modèle ; testez une petite requête avec votre Model ID avant d’utiliser ces fonctions dans une application.

Prérequis

Commencez par réussir une requête de texte simple avec les exemples de SDK. Copiez depuis la model plaza le Model ID exact pour le protocole à tester. Le guide de sélection des modèles explique les points à vérifier. Installez openai et anthropic dans l’environnement virtuel Python de votre projet, puis remplacez les valeurs d’exemple dans le même terminal :

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

Chaque bloc Python ci-dessous s’exécute indépendamment. Le délai d’attente de 120 secondes est un exemple pour ces tests courts ; choisissez celui de votre application avec les conseils sur les délais d’attente. Vérifiez les limites de tokens de sortie du modèle avant de définir max_tokens ou max_output_tokens.

Recevoir des réponses de texte en streaming

Le streaming transmet des événements envoyés par le serveur (SSE) dans une seule réponse HTTP. Un événement SSE ne correspond pas nécessairement à une phrase complète ni à un objet JSON d’arguments complet.

API	Paramètre de requête	Événements ou fragments de texte	Fin de la réponse
Chat Completions	`stream: true`	`choices[].delta.content` ; les fragments d’appels d’outils utilisent `delta.tool_calls`	Chaque choix fournit `finish_reason` ; le flux brut se termine par `data: [DONE]`
Responses	`stream: true`	Événements typés, tels que `response.output_text.delta`	`response.completed` ; gérez aussi les événements d’échec ou de réponse incomplète
Messages	`stream: true`	`content_block_delta` avec `text_delta` ; les arguments d’outils utilisent `input_json_delta`	`message_delta` contient le motif d’arrêt, puis vient `message_stop`

Le SDK décode le format SSE pour vous. Ces exemples affichent les fragments de texte et les informations de fin ; les applications doivent aussi gérer les événements d’erreur.

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)

Définir un outil météo pour chaque protocole

Le modèle demande un appel de fonction ; votre application valide les arguments, exécute la fonction et renvoie le résultat. Les exemples utilisent une fonction get_weather avec un argument location et renvoient des données de démonstration fixes, pas la météo en temps réel. Aucun service météo ni aucune Key supplémentaire n’est nécessaire.

Champ de requête ou de réponse	Chat Completions	Responses	Messages
Prompt d’entrée	`messages`	`input`	`messages`
Déclaration d’outil dans `tools[]`	`type: "function"`, avec `name`, `description` et `parameters` dans `function`	`type: "function"`, avec `name`, `description` et `parameters` à la racine de l’outil	`name`, `description` et `input_schema`
Appel renvoyé	`choices[].message.tool_calls[]`	Élément de `output[]` avec `type: "function_call"`	Bloc de `content[]` avec `type: "tool_use"`
Arguments	Chaîne JSON dans `function.arguments`	Chaîne JSON dans `arguments`	Objet dans `input`
Renvoi du résultat	Message avec `role: "tool"`, `tool_call_id` et `content`	Élément d’entrée avec `type: "function_call_output"`, `call_id` et `output` sous forme de chaîne	Bloc de contenu utilisateur avec `type: "tool_result"`, `tool_use_id` et `content`

La prise en charge varie selon le modèle et le protocole. Testez cette petite requête d’outil avec votre Model ID même si le texte simple fonctionne déjà. Déclarer un outil n’active ni la recherche hébergée ni d’autres outils exécutés par le fournisseur.

Exécuter un appel d’outil et renvoyer son résultat

Ces exemples sans streaming montrent un cycle complet d’appel et de résultat pour vous permettre d’examiner les arguments complets. Chacun traite tous les appels d’outils de la première réponse et autorise une requête de suivi. Un workflow d’agent plus long nécessite une boucle bornée et une condition d’arrêt.

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)

Conservez la réponse complète de l’assistant avant d’ajouter les résultats des outils. Pour Responses, gardez tous les éléments de output, y compris ceux de raisonnement lorsqu’ils sont renvoyés. Pour Messages, préservez les blocs de contenu et leur ordre. Associez les résultats aux ID d’appel reçus, sans inventer d’ID.

Questions fréquentes

Puis-je exécuter les arguments au fur et à mesure de leur arrivée ?

Accumulez les fragments par index d’appel d’outil ou ID d’élément, puis analysez et validez les arguments complets. Un JSON partiel peut être illisible. Commencez par les exemples sans streaming avant de combiner streaming et outils.

Pourquoi le texte simple fonctionne-t-il alors que les outils échouent ?

La prise en charge varie selon le modèle et le protocole ; vérifiez une petite requête d’outil avec votre Model ID. Contrôlez le schéma, les arguments renvoyés, l’ID du résultat et l’ordre des messages. Pour valider du JSON avec des types, consultez la FAQ sur la sortie structurée.

Faut-il relancer un flux interrompu ?

Vérifiez l’enregistrement de la requête avant de la renvoyer. Une déconnexion ne permet pas de savoir si la requête s’est terminée. Une nouvelle tentative ne doit pas répéter une action d’outil déjà exécutée. Consultez les conseils sur les erreurs et les nouvelles tentatives.

Ressources associées

Références officielles

Ces références définissent les champs des protocoles en amont. Vérifiez la fonction avec votre Model ID BetterToken avant de l’utiliser.

Référence de l'API

Guides de développement

Streaming et appels d’outils avec l’API BetterToken

Prérequis

Recevoir des réponses de texte en streaming

Définir un outil météo pour chaque protocole

Exécuter un appel d’outil et renvoyer son résultat

Questions fréquentes

Puis-je exécuter les arguments au fur et à mesure de leur arrivée ?

Pourquoi le texte simple fonctionne-t-il alors que les outils échouent ?

Faut-il relancer un flux interrompu ?

Ressources associées

Références officielles

​Prérequis

​Recevoir des réponses de texte en streaming

​Définir un outil météo pour chaque protocole

​Exécuter un appel d’outil et renvoyer son résultat

​Questions fréquentes

​Puis-je exécuter les arguments au fur et à mesure de leur arrivée ?

​Pourquoi le texte simple fonctionne-t-il alors que les outils échouent ?

​Faut-il relancer un flux interrompu ?

​Ressources associées

​Références officielles

Prérequis

Recevoir des réponses de texte en streaming

Définir un outil météo pour chaque protocole

Exécuter un appel d’outil et renvoyer son résultat

Questions fréquentes

Puis-je exécuter les arguments au fur et à mesure de leur arrivée ?

Pourquoi le texte simple fonctionne-t-il alors que les outils échouent ?

Faut-il relancer un flux interrompu ?

Ressources associées

Références officielles