BetterToken API: streaming y llamadas a herramientas

Activa el streaming con stream: true y define las herramientas del cliente con el esquema que requiere tu API. La compatibilidad con streaming y llamadas a herramientas varía según el modelo; prueba una solicitud pequeña con tu Model ID antes de usar estas funciones en una aplicación.

Requisitos previos

Primero completa una solicitud de texto simple con los ejemplos de SDK. Copia del model plaza el Model ID exacto para el protocolo que vas a probar. La guía de selección de modelos explica qué debes comprobar. Instala openai y anthropic en el entorno virtual de Python de tu proyecto y sustituye los valores de ejemplo en el mismo terminal:

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

Cada bloque de Python se ejecuta de forma independiente. El tiempo de espera de 120 segundos es un ejemplo para estas pruebas breves; elige el de tu aplicación con la guía de tiempos de espera. Comprueba los límites de tokens de salida del modelo antes de configurar max_tokens o max_output_tokens.

Recibe respuestas de texto en streaming

El streaming envía eventos del servidor (SSE) a través de una única respuesta HTTP. Un evento SSE no equivale necesariamente a una frase completa ni a un objeto JSON de argumentos completo.

API	Ajuste de la solicitud	Eventos o fragmentos de texto	Fin de la respuesta
Chat Completions	`stream: true`	`choices[].delta.content`; los fragmentos de herramientas usan `delta.tool_calls`	Cada opción indica `finish_reason`; el flujo sin procesar termina con `data: [DONE]`
Responses	`stream: true`	Eventos tipados como `response.output_text.delta`	`response.completed`; también debes gestionar eventos de error o respuesta incompleta
Messages	`stream: true`	`content_block_delta` con `text_delta`; los argumentos de herramientas usan `input_json_delta`	`message_delta` contiene el motivo de finalización y le sigue `message_stop`

El SDK interpreta el formato SSE. Estos ejemplos imprimen los fragmentos de texto y la información de finalización; las aplicaciones también deben gestionar los eventos de error.

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 una herramienta meteorológica para cada protocolo

El modelo solicita una llamada a una función; tu aplicación valida los argumentos, ejecuta la función y devuelve el resultado. Los ejemplos usan una función get_weather con un argumento location y devuelven datos fijos de demostración, no información meteorológica en tiempo real. No necesitas un servicio meteorológico ni otra Key.

Campo de solicitud o respuesta	Chat Completions	Responses	Messages
Mensaje de entrada	`messages`	`input`	`messages`
Declaración de herramienta en `tools[]`	`type: "function"`, con `name`, `description` y `parameters` dentro de `function`	`type: "function"`, con `name`, `description` y `parameters` en la raíz de la herramienta	`name`, `description` e `input_schema`
Llamada recibida	`choices[].message.tool_calls[]`	Elemento de `output[]` con `type: "function_call"`	Bloque de `content[]` con `type: "tool_use"`
Argumentos	Cadena JSON en `function.arguments`	Cadena JSON en `arguments`	Objeto en `input`
Devolución del resultado	Mensaje con `role: "tool"`, `tool_call_id` y `content`	Elemento de entrada con `type: "function_call_output"`, `call_id` y `output` como cadena	Bloque de contenido del usuario con `type: "tool_result"`, `tool_use_id` y `content`

La compatibilidad varía según el modelo y el protocolo. Prueba esta solicitud pequeña con tu Model ID aunque el texto simple ya funcione. Declarar una herramienta no activa la búsqueda alojada ni otras herramientas ejecutadas por el proveedor.

Ejecuta la llamada y devuelve el resultado

Estos ejemplos sin streaming muestran un ciclo completo de llamada y resultado para que puedas revisar los argumentos completos. Cada uno procesa todas las llamadas a herramientas de la primera respuesta y permite una solicitud de seguimiento. Un flujo de agente más largo necesita un bucle con límites y una condición de parada.

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)

Conserva la respuesta completa del asistente antes de añadir los resultados de las herramientas. En Responses, conserva todos los elementos de output, incluidos los de razonamiento si se devuelven. En Messages, conserva los bloques de contenido y su orden. Asocia los resultados a los ID de llamada recibidos; no inventes ID.

Preguntas frecuentes

¿Puedo ejecutar los argumentos a medida que llegan?

Acumula los fragmentos por índice de llamada o ID de elemento y después analiza y valida los argumentos completos. Un JSON parcial puede no ser válido. Empieza por los ejemplos sin streaming antes de combinar streaming y herramientas.

¿Por qué funciona el texto simple pero fallan las herramientas?

La compatibilidad varía según el modelo y el protocolo; verifica una solicitud pequeña con tu Model ID. Revisa el esquema, los argumentos recibidos, el ID del resultado y el orden de los mensajes. Para validar JSON con tipos, consulta las preguntas sobre salida estructurada.

¿Debo reintentar un flujo interrumpido?

Revisa el registro de la solicitud antes de reenviarla. Una desconexión no permite saber si la solicitud terminó. Un reintento inicia otro intento y no debe repetir una acción de herramienta ya ejecutada. Consulta la guía de errores y reintentos.

Recursos relacionados

Referencias oficiales

Estas referencias definen los campos de los protocolos de origen. Verifica la función con tu Model ID de BetterToken antes de usarla.

Referencia de la API

Guías para desarrolladores

BetterToken API: streaming y llamadas a herramientas

Requisitos previos

Recibe respuestas de texto en streaming

Define una herramienta meteorológica para cada protocolo

Ejecuta la llamada y devuelve el resultado

Preguntas frecuentes

¿Puedo ejecutar los argumentos a medida que llegan?

¿Por qué funciona el texto simple pero fallan las herramientas?

¿Debo reintentar un flujo interrumpido?

Recursos relacionados

Referencias oficiales

​Requisitos previos

​Recibe respuestas de texto en streaming

​Define una herramienta meteorológica para cada protocolo

​Ejecuta la llamada y devuelve el resultado

​Preguntas frecuentes

​¿Puedo ejecutar los argumentos a medida que llegan?

​¿Por qué funciona el texto simple pero fallan las herramientas?

​¿Debo reintentar un flujo interrumpido?

​Recursos relacionados

​Referencias oficiales

Requisitos previos

Recibe respuestas de texto en streaming

Define una herramienta meteorológica para cada protocolo

Ejecuta la llamada y devuelve el resultado

Preguntas frecuentes

¿Puedo ejecutar los argumentos a medida que llegan?

¿Por qué funciona el texto simple pero fallan las herramientas?

¿Debo reintentar un flujo interrumpido?

Recursos relacionados

Referencias oficiales