BetterToken API 스트리밍과 도구 호출 구현

stream: true로 스트리밍을 켜고 API에 맞는 스키마로 클라이언트 도구를 정의하세요. 스트리밍과 도구 호출 지원은 모델마다 다릅니다. 앱에 적용하기 전에 사용할 Model ID로 작은 요청을 테스트하세요.

준비 사항

먼저 SDK 예제로 일반 텍스트 요청을 성공시키세요. 모델 광장에서 테스트할 프로토콜에 맞는 정확한 Model ID를 복사하세요. 확인할 항목은 모델 선택 가이드를 참고하세요. 프로젝트의 Python 가상 환경에 openai와 anthropic을 설치하고 같은 터미널에서 자리표시자를 바꾸세요.

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

아래 Python 코드 블록은 각각 독립적으로 실행할 수 있습니다. 120초 타임아웃은 짧은 테스트를 위한 예시입니다. 앱의 타임아웃은 타임아웃 설정 안내에 따라 정하세요. max_tokens 또는 max_output_tokens를 설정하기 전에 모델의 출력 토큰 한도를 확인하세요.

텍스트 응답 스트리밍하기

스트리밍은 하나의 HTTP 응답으로 서버 전송 이벤트(SSE)를 보냅니다. SSE 이벤트 하나가 완전한 문장이나 JSON 인수 객체에 해당하지는 않습니다.

API	요청 설정	텍스트 이벤트 또는 청크	응답 종료
Chat Completions	`stream: true`	`choices[].delta.content`, 도구 조각은 `delta.tool_calls`	각 선택지가 `finish_reason`을 제공하며 원시 스트림은 `data: [DONE]`으로 종료
Responses	`stream: true`	`response.output_text.delta` 같은 유형별 이벤트	`response.completed`, 실패 및 불완전 응답 이벤트도 처리
Messages	`stream: true`	`content_block_delta` 안의 `text_delta`, 도구 인수는 `input_json_delta`	`message_delta`가 종료 이유를 전달한 뒤 `message_stop` 수신

SDK가 SSE 형식을 파싱합니다. 예제는 텍스트의 증분과 완료 정보를 출력합니다. 앱에서는 오류 이벤트도 처리하세요.

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)

프로토콜별 날씨 도구 정의하기

모델이 함수 호출을 요청하면 앱이 인수를 검증하고 함수를 실행한 뒤 결과를 반환합니다. 예제는 get_weather 함수 하나에 location 인수를 전달하며, 실시간 날씨가 아닌 고정된 데모 데이터를 반환합니다. 날씨 서비스나 추가 Key는 필요하지 않습니다.

요청 또는 응답 필드	Chat Completions	Responses	Messages
프롬프트	`messages`	`input`	`messages`
`tools[]`의 도구 선언	`type: "function"`, `function` 안에 `name`, `description`, `parameters` 지정	`type: "function"`, 도구 최상위에 `name`, `description`, `parameters` 지정	`name`, `description`, `input_schema`
반환된 호출	`choices[].message.tool_calls[]`	`output[]`의 `type: "function_call"` 항목	`content[]`의 `type: "tool_use"` 블록
인수	`function.arguments`의 JSON 문자열	`arguments`의 JSON 문자열	`input`의 객체
결과 반환	`role: "tool"`, `tool_call_id`, `content`를 포함한 메시지	`type: "function_call_output"`, `call_id`, 문자열 `output`을 포함한 입력 항목	`type: "tool_result"`, `tool_use_id`, `content`를 포함한 사용자 콘텐츠 블록

지원 여부는 모델과 프로토콜에 따라 다릅니다. 일반 텍스트가 작동하더라도 사용할 Model ID로 이 작은 도구 요청을 테스트하세요. 도구를 선언한다고 호스팅된 검색이나 제공자 측에서 실행하는 다른 도구가 활성화되지는 않습니다.

도구를 호출하고 결과 반환하기

아래 비스트리밍 예제는 완성된 인수를 확인할 수 있도록 도구 호출과 결과 반환을 한 차례 수행합니다. 첫 응답의 모든 도구 호출을 처리하고 후속 요청을 한 번 보냅니다. 더 긴 에이전트 워크플로에는 별도의 횟수 제한과 종료 조건을 갖춘 루프가 필요합니다.

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)

도구 결과를 추가하기 전에 어시스턴트의 전체 응답을 보존하세요. Responses는 추론 항목이 반환되면 이를 포함한 모든 output 항목을 유지하세요. Messages는 콘텐츠 블록과 순서를 보존하세요. ID를 새로 만들지 말고 반환된 호출 ID에 결과를 연결하세요.

자주 묻는 질문

도구 인수를 받는 즉시 실행해도 되나요?

도구 호출 인덱스나 항목 ID별로 조각을 모은 뒤 완성된 인수를 파싱하고 검증하세요. 일부만 받은 JSON은 파싱되지 않을 수 있습니다. 스트리밍과 도구를 결합하기 전에 위의 비스트리밍 예제로 시작하세요.

일반 텍스트는 작동하는데 도구는 왜 실패하나요?

지원 여부는 모델과 프로토콜에 따라 다릅니다. 사용할 Model ID로 작은 도구 요청을 검증하세요. 도구 스키마, 반환된 인수, 결과 ID, 메시지 순서를 확인하세요. 타입을 이용한 JSON 검증은 구조화된 출력 FAQ를 참고하세요.

스트림이 끊기면 재시도해야 하나요?

다시 보내기 전에 요청 기록을 확인하세요. 연결이 끊겼다는 사실만으로 요청 완료 여부를 알 수는 없습니다. 재시도는 별도의 시도이므로 이미 실행한 도구 작업을 반복하지 않아야 합니다. 오류 및 재시도 안내를 참고하세요.

공식 참고 자료

아래 자료는 업스트림 프로토콜 필드를 정의합니다. 사용하기 전에 BetterToken Model ID로 해당 기능을 검증하세요.

API 레퍼런스

개발자 가이드

BetterToken API 스트리밍과 도구 호출 구현

준비 사항

텍스트 응답 스트리밍하기

프로토콜별 날씨 도구 정의하기

도구를 호출하고 결과 반환하기

자주 묻는 질문

도구 인수를 받는 즉시 실행해도 되나요?

일반 텍스트는 작동하는데 도구는 왜 실패하나요?

스트림이 끊기면 재시도해야 하나요?

관련 가이드

공식 참고 자료

​준비 사항

​텍스트 응답 스트리밍하기

​프로토콜별 날씨 도구 정의하기

​도구를 호출하고 결과 반환하기

​자주 묻는 질문

​도구 인수를 받는 즉시 실행해도 되나요?

​일반 텍스트는 작동하는데 도구는 왜 실패하나요?

​스트림이 끊기면 재시도해야 하나요?

​관련 가이드

​공식 참고 자료

준비 사항

텍스트 응답 스트리밍하기

프로토콜별 날씨 도구 정의하기

도구를 호출하고 결과 반환하기

자주 묻는 질문

도구 인수를 받는 즉시 실행해도 되나요?

일반 텍스트는 작동하는데 도구는 왜 실패하나요?

스트림이 끊기면 재시도해야 하나요?

관련 가이드

공식 참고 자료