Build with Direct Responses

This tutorial walks you through building on Direct Responses, POST /ai/v2/direct/responses, one step at a time. Want the concepts first? Read the Direct Responses guide. For every parameter and response field, see the Direct Responses API reference.

Why Direct Responses? It sends your request straight to the model you name, with one request and response shape across every provider, and it bills the model’s published rate. It takes the same request body as the guarded Responses endpoint. Choose Direct Responses when you want the provider’s model as-is and handle safety yourself, and guarded Responses when you want Gloo’s guardrails and values alignment built in. See Which endpoint should I use?

Prerequisites

Before starting, make sure you have:

A Gloo AI Studio account
Your API key from the API Keys page, exported as GLOO_API_KEY
Authentication setup: complete the Authentication Tutorial first
Python 3.9+ with requests (pip install requests), or Node.js 18+ (for built-in fetch)

export GLOO_API_KEY="your_api_key_here"

What You’ll Build

A small command-line assistant, built up in five steps:

First request: send a prompt and read the typed output[] array
Choose a model: list the catalog and pick a model ID
Stream: print the answer as it is generated
Multi-turn: keep a conversation going with instructions and history
Tool call: let the model call a function you define, then answer with its result

Then you’ll add error handling and put it all together in Complete Code.

Step 1: Your First Request

Send a model and an input. input can be a plain string or an array of {role, content} items.

On a reasoning model, max_output_tokens holds the model’s thinking as well as its answer, and Responses sends it as you set it. The examples here set 4,096 to leave room for both. See Set your output limit for reasoning models.

curl -X POST 'https://platform.ai.gloo.com/ai/v2/direct/responses' \
  -H "Authorization: Bearer $GLOO_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "gloo-anthropic-claude-sonnet-4.6",
    "input": "Suggest three warm, short welcome messages for first-time church visitors.",
    "max_output_tokens": 4096
  }'

import os
import requests

API_KEY = os.environ["GLOO_API_KEY"]
API_URL = "https://platform.ai.gloo.com/ai/v2/direct/responses"
MODEL = "gloo-anthropic-claude-sonnet-4.6"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}


def output_text(response: dict) -> str:
    """Join the text of every message item in output[]."""
    return "".join(
        part.get("text", "")
        for item in response.get("output", [])
        if item.get("type") == "message"
        for part in item.get("content", [])
        if part.get("type") == "output_text"
    )


r = requests.post(API_URL, headers=HEADERS, json={
    "model": MODEL,
    "input": "Suggest three warm, short welcome messages for first-time church visitors.",
    "max_output_tokens": 4096,
}, timeout=120)
r.raise_for_status()
result = r.json()

print(output_text(result))
print(result["usage"])

const API_KEY = process.env.GLOO_API_KEY!;
const API_URL = "https://platform.ai.gloo.com/ai/v2/direct/responses";
const MODEL = "gloo-anthropic-claude-sonnet-4.6";

/** Join the text of every message item in output[]. */
function outputText(response: any): string {
  return (response.output ?? [])
    .filter((item: any) => item.type === "message")
    .flatMap((item: any) => item.content ?? [])
    .filter((part: any) => part.type === "output_text")
    .map((part: any) => part.text ?? "")
    .join("");
}

const res = await fetch(API_URL, {
  method: "POST",
  headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
  body: JSON.stringify({
    model: MODEL,
    input: "Suggest three warm, short welcome messages for first-time church visitors.",
    max_output_tokens: 4096,
  }),
});
if (!res.ok) throw new Error(`Request failed: ${res.status} ${await res.text()}`);
const result = await res.json();

console.log(outputText(result));
console.log(result.usage);

Understanding the Response

The answer arrives as a typed output[] array. Plain text comes back as a message item whose content[] holds output_text parts:

{
  "id": "resp_...",
  "object": "response",
  "model": "gloo-anthropic-claude-sonnet-4.6",
  "output": [
    {
      "type": "message",
      "role": "assistant",
      "content": [
        { "type": "output_text", "text": "1. We're so glad you're here today..." }
      ]
    }
  ],
  "usage": { "input_tokens": 24, "output_tokens": 38, "total_tokens": 62 }
}

Other item types share the same array: a reasoning summary arrives as a reasoning item, a tool call as a function_call item (Step 5) and a generated image as an image_generation_call item. Walking output[] by type, as output_text does, keeps your code working whichever items come back.

Already using the OpenAI SDKs? Point the base URL at https://platform.ai.gloo.com/ai/v2/direct and call responses.create. This tutorial uses plain HTTP so every field is visible.

Step 2: Choose a Model

Direct Responses calls exactly the model you name, so the model ID is your main choice. The live catalog is at GET /platform/v2/models, which needs no authentication. Each entry has an id to use as model, plus capability flags such as supports_tools, supports_streaming and supports_reasoning, a context_window (the context window: how much input the model can take), list pricing, and is_deprecated. List the models that can stream and call tools, and aren’t being retired:

curl -s 'https://platform.ai.gloo.com/platform/v2/models' \
  | jq -r '.data[]
      | select(.supports_tools and .supports_streaming and (.is_deprecated | not))
      | "\(.id)\t\(.family)\t\(.context_window)"'

import requests

catalog = requests.get("https://platform.ai.gloo.com/platform/v2/models", timeout=30).json()

for m in catalog["data"]:
    if m.get("supports_tools") and m.get("supports_streaming") and not m.get("is_deprecated"):
        print(f'{m["id"]:<45} {m["family"]:<12} {m.get("context_window")}')

const catalog = await (await fetch("https://platform.ai.gloo.com/platform/v2/models")).json();

for (const m of catalog.data) {
  if (m.supports_tools && m.supports_streaming && !m.is_deprecated) {
    console.log(m.id.padEnd(45), m.family.padEnd(12), m.context_window);
  }
}

What to weigh when you pick:

Capabilities. Steps 3 and 5 need supports_streaming and supports_tools. Image input needs a vision-capable model (supports_vision).
Cost. You pay the model’s published rate per token, with nothing added on direct endpoints. See Pricing and token spend.
Reasoning. Models with supports_reasoning can think before answering; set reasoning.effort to control how much. See Reasoning and the Models That Think lesson.
Deprecations. A deprecated model lists a replacement_model. See Model Lifecycle.

This tutorial keeps gloo-anthropic-claude-sonnet-4.6 in one MODEL constant, so trying another model is a one-line change. The Supported Models page shows the same catalog as a table.

Step 3: Stream the Answer

Set "stream": true to stream the answer as Server-Sent Events. Each event is an event: line followed by a data: line with a JSON payload. Three event types carry what you need:

Event	What to do
`response.output_text.delta`	Print or buffer `delta`
`response.completed` or `response.incomplete`	Read `status` and `usage`; the stream is done. `response.incomplete` means the answer reached `max_output_tokens`.
`response.failed`	The stream failed after output started; read `response.error`

Other events (response.output_item.added, response.content_part.done and so on) mark structure and can be ignored for plain text. The final event’s shape depends on the model. For OpenAI models its fields sit under a response key; for other models they’re at the top level, without output[]. Read data.response when present and fall back to data, and build the answer from the deltas.

curl -N -X POST 'https://platform.ai.gloo.com/ai/v2/direct/responses' \
  -H "Authorization: Bearer $GLOO_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "gloo-anthropic-claude-sonnet-4.6",
    "input": "Write a two-sentence blessing for a volunteer team.",
    "max_output_tokens": 4096,
    "stream": true
  }'

import json


def stream_answer(payload: dict) -> str:
    """Stream a response, printing text as it arrives. Returns the full text."""
    with requests.post(API_URL, headers=HEADERS, json={**payload, "stream": True},
                       stream=True, timeout=300) as r:
        r.raise_for_status()  # a failure before any output is a normal HTTP error
        r.encoding = "utf-8"  # SSE is always UTF-8; don't guess from the header
        event, parts = None, []
        for line in r.iter_lines(decode_unicode=True):
            if not line:
                continue
            if line.startswith("event:"):
                event = line[len("event:"):].strip()
                continue
            if not line.startswith("data:"):
                continue
            data = json.loads(line[len("data:"):].strip())
            kind = event or data.get("type")

            if kind == "response.output_text.delta":
                print(data["delta"], end="", flush=True)
                parts.append(data["delta"])
            elif kind in ("response.completed", "response.incomplete"):
                final = data.get("response", data)
                print(f'\n[{final.get("status")}] {final.get("usage")}')
            elif kind == "response.failed":
                error = data.get("response", {}).get("error", {})
                raise RuntimeError(f'{error.get("code")}: {error.get("message")}')
        return "".join(parts)


stream_answer({
    "model": MODEL,
    "input": "Write a two-sentence blessing for a volunteer team.",
    "max_output_tokens": 4096,
})

/** Stream a response, printing text as it arrives. Returns the full text. */
async function streamAnswer(payload: Record<string, unknown>): Promise<string> {
  const res = await fetch(API_URL, {
    method: "POST",
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({ ...payload, stream: true }),
  });
  // A failure before any output is a normal HTTP error.
  if (!res.ok) throw new Error(`Request failed: ${res.status} ${await res.text()}`);

  const reader = res.body!.getReader();
  const decoder = new TextDecoder();
  const parts: string[] = [];
  let buffer = "";
  let event: string | null = null;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() ?? ""; // keep the incomplete last line

    for (const raw of lines) {
      const line = raw.trim();
      if (!line) continue;
      if (line.startsWith("event:")) {
        event = line.slice("event:".length).trim();
        continue;
      }
      if (!line.startsWith("data:")) continue;
      const data = JSON.parse(line.slice("data:".length).trim());
      const kind = event ?? data.type;

      if (kind === "response.output_text.delta") {
        process.stdout.write(data.delta);
        parts.push(data.delta);
      } else if (kind === "response.completed" || kind === "response.incomplete") {
        const final = data.response ?? data;
        console.log(`\n[${final.status}]`, final.usage);
      } else if (kind === "response.failed") {
        const error = data.response?.error ?? {};
        throw new Error(`${error.code}: ${error.message}`);
      }
    }
  }
  return parts.join("");
}

await streamAnswer({
  model: MODEL,
  input: "Write a two-sentence blessing for a volunteer team.",
  max_output_tokens: 4096,
});

A stream can fail after text is already on screen. Keep the partial text rather than erasing it, and don’t silently start over. See Handling Streaming Failures for when to retry, when to offer Continue, and retry ceilings.

Step 4: Hold a Multi-Turn Conversation

Use instructions for the system prompt and send the whole conversation in input on every request. Conversation state lives in your application: previous_response_id and conversation (server-side chaining) are not supported.

curl -X POST 'https://platform.ai.gloo.com/ai/v2/direct/responses' \
  -H "Authorization: Bearer $GLOO_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "gloo-anthropic-claude-sonnet-4.6",
    "instructions": "You are a friendly assistant for a church front desk. Keep answers to two sentences.",
    "max_output_tokens": 4096,
    "input": [
      { "role": "user", "content": "We want to start a meal train for a new family." },
      { "role": "assistant", "content": "What a kind idea. A shared sign-up with dates and dietary needs keeps it simple." },
      { "role": "user", "content": "What should the sign-up ask for?" }
    ]
  }'

INSTRUCTIONS = "You are a friendly assistant for a church front desk. Keep answers to two sentences."
conversation = []


def ask(question: str) -> str:
    conversation.append({"role": "user", "content": question})
    r = requests.post(API_URL, headers=HEADERS, json={
        "model": MODEL,
        "instructions": INSTRUCTIONS,
        "input": conversation,
        "max_output_tokens": 4096,
    }, timeout=120)
    r.raise_for_status()
    answer = output_text(r.json())
    conversation.append({"role": "assistant", "content": answer})
    return answer


print(ask("We want to start a meal train for a new family."))
print(ask("What should the sign-up ask for?"))  # the model sees the first turn

const INSTRUCTIONS =
  "You are a friendly assistant for a church front desk. Keep answers to two sentences.";
const conversation: any[] = [];

async function ask(question: string): Promise<string> {
  conversation.push({ role: "user", content: question });
  const res = await fetch(API_URL, {
    method: "POST",
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({
      model: MODEL,
      instructions: INSTRUCTIONS,
      input: conversation,
      max_output_tokens: 4096,
    }),
  });
  if (!res.ok) throw new Error(`Request failed: ${res.status} ${await res.text()}`);
  const answer = outputText(await res.json());
  conversation.push({ role: "assistant", content: answer });
  return answer;
}

console.log(await ask("We want to start a meal train for a new family."));
console.log(await ask("What should the sign-up ask for?")); // the model sees the first turn

The history grows with every turn and is billed as input tokens each time, so long conversations cost more and eventually approach the model’s context window. Trim or summarize older turns once they stop being useful.

You can also send output[] items back exactly as you received them, which is how tool calls work in the next step. On a reasoning model, a reasoning item you send back is not carried into the next request: each request is reasoned afresh from the conversation in input. See Reasoning across turns.

Step 5: Let the Model Call a Function

Tool use lets the model ask your code for information it doesn’t have. The round trip has three parts:

Send tools with your request. The model answers with a function_call item naming the function and its JSON arguments.
Run the function yourself.
Send the conversation again with the model’s function_call and a function_call_output holding your result under the same call_id. The model then answers as a normal message.

Define tools with the nested function schema, {"type": "function", "function": { ... }}, the same schema Completions V2 uses. See Tool Use.

# 1. Ask, with a tool the model can call
curl -X POST 'https://platform.ai.gloo.com/ai/v2/direct/responses' \
  -H "Authorization: Bearer $GLOO_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "gloo-anthropic-claude-sonnet-4.6",
    "input": [{ "role": "user", "content": "When are services at the Riverside campus?" }],
    "tools": [{
      "type": "function",
      "function": {
        "name": "get_service_times",
        "description": "Look up weekend service times for a church campus.",
        "parameters": {
          "type": "object",
          "properties": { "campus": { "type": "string" } },
          "required": ["campus"]
        }
      }
    }]
  }'

# 3. Send back the call and your result (use the call_id from step 1's function_call)
curl -X POST 'https://platform.ai.gloo.com/ai/v2/direct/responses' \
  -H "Authorization: Bearer $GLOO_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "gloo-anthropic-claude-sonnet-4.6",
    "input": [
      { "role": "user", "content": "When are services at the Riverside campus?" },
      { "type": "function_call", "call_id": "call_abc123", "name": "get_service_times",
        "arguments": "{\"campus\": \"Riverside\"}" },
      { "type": "function_call_output", "call_id": "call_abc123",
        "output": "{\"campus\": \"Riverside\", \"times\": [\"Sat 5:00 PM\", \"Sun 9:00 AM\", \"Sun 11:00 AM\"]}" }
    ]
  }'

TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_service_times",
        "description": "Look up weekend service times for a church campus.",
        "parameters": {
            "type": "object",
            "properties": {"campus": {"type": "string"}},
            "required": ["campus"],
        },
    },
}]


def get_service_times(campus: str) -> dict:
    # Your real lookup goes here.
    return {"campus": campus, "times": ["Sat 5:00 PM", "Sun 9:00 AM", "Sun 11:00 AM"]}


def run_tool(call: dict) -> str:
    """Run one function call. Bad arguments go back to the model as an error result."""
    try:
        args = json.loads(call.get("arguments") or "{}")
        campus = args["campus"]
        if call.get("name") != "get_service_times" or not isinstance(campus, str):
            raise ValueError("unexpected call")
    except (ValueError, KeyError, TypeError):
        return json.dumps({"error": "Call get_service_times with a string 'campus'."})
    return json.dumps(get_service_times(campus))


def create(payload: dict) -> dict:
    r = requests.post(API_URL, headers=HEADERS, json=payload, timeout=120)
    r.raise_for_status()
    return r.json()


items = [{"role": "user", "content": "When are services at the Riverside campus?"}]
first = create({"model": MODEL, "input": items, "tools": TOOLS})

calls = [item for item in first["output"] if item.get("type") == "function_call"]
if not calls:
    print(output_text(first))  # the model answered without the tool
else:
    for call in calls:
        items.append({"type": "function_call", "call_id": call["call_id"],
                      "name": call["name"], "arguments": call["arguments"]})
        items.append({"type": "function_call_output", "call_id": call["call_id"],
                      "output": run_tool(call)})
    final = create({"model": MODEL, "input": items})
    print(output_text(final))

const TOOLS = [{
  type: "function",
  function: {
    name: "get_service_times",
    description: "Look up weekend service times for a church campus.",
    parameters: {
      type: "object",
      properties: { campus: { type: "string" } },
      required: ["campus"],
    },
  },
}];

function getServiceTimes(campus: string) {
  // Your real lookup goes here.
  return { campus, times: ["Sat 5:00 PM", "Sun 9:00 AM", "Sun 11:00 AM"] };
}

/** Run one function call. Bad arguments go back to the model as an error result. */
function runTool(call: any): string {
  let args: any;
  try {
    args = JSON.parse(call.arguments || "{}");
  } catch {
    args = undefined;
  }
  if (call.name !== "get_service_times" || typeof args?.campus !== "string") {
    return JSON.stringify({ error: "Call get_service_times with a string 'campus'." });
  }
  return JSON.stringify(getServiceTimes(args.campus));
}

async function create(payload: Record<string, unknown>): Promise<any> {
  const res = await fetch(API_URL, {
    method: "POST",
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify(payload),
  });
  if (!res.ok) throw new Error(`Request failed: ${res.status} ${await res.text()}`);
  return res.json();
}

const items: any[] = [{ role: "user", content: "When are services at the Riverside campus?" }];
const first = await create({ model: MODEL, input: items, tools: TOOLS });

const calls = first.output.filter((item: any) => item.type === "function_call");
if (calls.length === 0) {
  console.log(outputText(first)); // the model answered without the tool
} else {
  for (const call of calls) {
    items.push({ type: "function_call", call_id: call.call_id, name: call.name, arguments: call.arguments });
    items.push({ type: "function_call_output", call_id: call.call_id, output: runTool(call) });
  }
  const final = await create({ model: MODEL, input: items });
  console.log(outputText(final));
}

The model’s first answer contains a function_call item like this:

{
  "type": "function_call",
  "id": "fc_abc123",
  "call_id": "call_abc123",
  "name": "get_service_times",
  "arguments": "{\"campus\": \"Riverside\"}"
}

arguments is a JSON string written by the model, so treat it like any other untrusted input: run_tool parses it, picks out only the campus field it expects, and sends an error result back to the model instead of crashing when the arguments don’t fit. To make the model call a tool rather than answer directly, send "tool_choice": "required".

Step 6: Handle Errors

Direct Responses errors follow the direct endpoint error contract. Every error carries:

a short message that is safe to show your users as is
a developer_message and retryable flag for you
an X-Gloo-Error-Code header with a stable lowercase code to branch on, and an X-Request-Id to include when you contact support
a Retry-After header on errors where Gloo knows how long to wait

The fields sit under error for provider and platform failures and under detail for refusals such as authentication or content policy. Read whichever is present, and never parse the wording of message: it can change, the codes don’t.

import random
import time
from typing import Optional

# Someone is waiting at the terminal: at most 2 retries, each wait 2 seconds at most.
MAX_RETRIES = 2
MAX_WAIT_S = 2.0


def _seconds(value: Optional[str]) -> Optional[float]:
    """Retry-After as seconds, or None when it's missing or not a number."""
    try:
        return max(0.0, float(value)) if value else None
    except ValueError:
        return None  # for example an HTTP date: fall back to backoff


def _backoff(attempt: int) -> float:
    """Full jitter, capped at MAX_WAIT_S."""
    return random.uniform(0, min(MAX_WAIT_S, 0.5 * 2 ** attempt))


class GlooError(Exception):
    def __init__(self, r: requests.Response):
        try:
            body = r.json()
        except ValueError:
            body = {}
        fields = next((body[k] for k in ("error", "detail") if isinstance(body.get(k), dict)), {})
        self.status = r.status_code
        self.code = r.headers.get("X-Gloo-Error-Code")
        self.message = fields.get("message", "Something went wrong.")  # show this to users
        self.developer_message = fields.get("developer_message") or body.get("developer_message")
        self.retryable = fields.get("retryable") is True
        self.retry_after = _seconds(r.headers.get("Retry-After"))
        self.request_id = r.headers.get("X-Request-Id")
        super().__init__(f"{self.status} {self.code}: {self.developer_message or self.message}")


def create(payload: dict) -> dict:
    for attempt in range(MAX_RETRIES + 1):
        try:
            r = requests.post(API_URL, headers=HEADERS, json=payload, timeout=120)
        except (requests.ConnectionError, requests.Timeout):
            if attempt == MAX_RETRIES:
                raise
            time.sleep(_backoff(attempt))  # no response at all: plausibly transient
            continue
        if r.ok:
            return r.json()
        error = GlooError(r)
        if not error.retryable or attempt == MAX_RETRIES:
            raise error
        if error.retry_after is None:
            time.sleep(_backoff(attempt))
        elif error.retry_after > MAX_WAIT_S:
            raise error  # longer than someone should wait: fail fast with error.message
        else:
            time.sleep(error.retry_after)

// Someone is waiting at the terminal: at most 2 retries, each wait 2 seconds at most.
const MAX_RETRIES = 2;
const MAX_WAIT_S = 2;

/** Retry-After as seconds, or undefined when it's missing or not a number. */
function seconds(value: string | null): number | undefined {
  const n = value === null || value.trim() === "" ? NaN : Number(value);
  return Number.isFinite(n) ? Math.max(0, n) : undefined; // e.g. an HTTP date: use backoff
}

/** Full jitter, capped at MAX_WAIT_S. */
const backoffS = (attempt: number) => Math.random() * Math.min(MAX_WAIT_S, 0.5 * 2 ** attempt);
const sleepS = (s: number) => new Promise((resolve) => setTimeout(resolve, s * 1000));

class GlooError extends Error {
  constructor(public userMessage: string, detail: string, public retryable: boolean,
              public retryAfter: number | undefined) {
    super(detail);
  }

  static async from(res: Response): Promise<GlooError> {
    const body: any = await res.json().catch(() => ({}));
    const f = [body.error, body.detail].find((v) => v && typeof v === "object") ?? {};
    const code = res.headers.get("X-Gloo-Error-Code");
    const requestId = res.headers.get("X-Request-Id"); // include when contacting support
    const detail = f.developer_message ?? body.developer_message ?? f.message;
    return new GlooError(
      f.message ?? "Something went wrong.", // show this to users
      `${res.status} ${code}: ${detail} (request ${requestId})`,
      f.retryable === true,
      seconds(res.headers.get("Retry-After")),
    );
  }
}

async function create(payload: Record<string, unknown>): Promise<any> {
  for (let attempt = 0; ; attempt++) {
    let res: Response;
    try {
      res = await fetch(API_URL, {
        method: "POST",
        headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
        body: JSON.stringify(payload),
        signal: AbortSignal.timeout(120_000),
      });
    } catch (err) {
      if (attempt >= MAX_RETRIES) throw err;
      await sleepS(backoffS(attempt)); // no response at all: plausibly transient
      continue;
    }
    if (res.ok) return res.json();
    const error = await GlooError.from(res);
    if (!error.retryable || attempt >= MAX_RETRIES) throw error;
    if (error.retryAfter === undefined) await sleepS(backoffS(attempt));
    else if (error.retryAfter > MAX_WAIT_S) throw error; // too long to wait: fail fast
    else await sleepS(error.retryAfter);
  }
}

create follows the retry rule for a request someone is waiting on: it retries only errors marked retryable and network failures with no response, at most twice, with waits of 2 seconds or less. When Retry-After asks for longer, such as the 30 seconds a 503 can ask for, it raises at once so you can show message instead of leaving the terminal silent. A Retry-After that isn’t a number of seconds falls back to backoff. For a batch job nobody is watching, raise the budget to 5 requests in total and the wait cap to 30 seconds. The same rules apply to a stream that fails before any output: it returns a normal HTTP status and body. A stream that fails after output has started ends with response.failed, whose response.error carries the string code, message, retryable and developer_message. For retry ceilings and partial-output handling, see Handling Streaming Failures.

Complete Code

To run every step from one file, start with the imports and constants below, then paste in the functions you already wrote. Copy only the definitions, not each step’s example calls, and use Step 6’s retrying create in place of Step 5’s simpler one. Finish with main(), export GLOO_API_KEY, and run it.

"""direct_responses.py: run with `python direct_responses.py`."""
import json
import os
import random
import time
from typing import Optional

import requests

API_KEY = os.environ["GLOO_API_KEY"]
API_URL = "https://platform.ai.gloo.com/ai/v2/direct/responses"
MODEL = "gloo-anthropic-claude-sonnet-4.6"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

# Paste these definitions from the steps above:
#   output_text()                    Step 1
#   stream_answer()                  Step 3
#   TOOLS, get_service_times(),      Step 5
#   run_tool()
#   GlooError, create()              Step 6 (the retrying create)


def main() -> None:
    print("== Step 1: first request ==")
    first = create({"model": MODEL, "max_output_tokens": 4096,
                    "input": "Suggest three warm, short welcome messages for first-time church visitors."})
    print(output_text(first))

    print("\n== Step 3: streaming ==")
    stream_answer({"model": MODEL, "max_output_tokens": 4096,
                   "input": "Write a two-sentence blessing for a volunteer team."})

    print("\n== Step 4: multi-turn ==")
    conversation = []
    for question in ("We want to start a meal train for a new family.",
                     "What should the sign-up ask for?"):
        conversation.append({"role": "user", "content": question})
        answer = output_text(create({
            "model": MODEL, "max_output_tokens": 4096, "input": conversation,
            "instructions": "You are a friendly assistant for a church front desk. "
                            "Keep answers to two sentences.",
        }))
        conversation.append({"role": "assistant", "content": answer})
        print(f"> {question}\n{answer}\n")

    print("== Step 5: tool call ==")
    items = [{"role": "user", "content": "When are services at the Riverside campus?"}]
    response = create({"model": MODEL, "input": items, "tools": TOOLS})
    calls = [i for i in response["output"] if i.get("type") == "function_call"]
    for call in calls:
        items.append({"type": "function_call", "call_id": call["call_id"],
                      "name": call["name"], "arguments": call["arguments"]})
        items.append({"type": "function_call_output", "call_id": call["call_id"],
                      "output": run_tool(call)})
    if calls:
        response = create({"model": MODEL, "input": items})
    print(output_text(response))


if __name__ == "__main__":
    try:
        main()
    except GlooError as e:
        print(f"\nError for users: {e.message}\nFor you: {e}")

// direct-responses.ts: run with `npx tsx direct-responses.ts`
const API_KEY = process.env.GLOO_API_KEY;
if (!API_KEY) throw new Error("Set GLOO_API_KEY first");
const API_URL = "https://platform.ai.gloo.com/ai/v2/direct/responses";
const MODEL = "gloo-anthropic-claude-sonnet-4.6";

// Paste these definitions from the steps above:
//   outputText()                   Step 1
//   streamAnswer()                 Step 3
//   TOOLS, getServiceTimes(),      Step 5
//   runTool()
//   GlooError, create()            Step 6 (the retrying create)

async function main() {
  console.log("== Step 1: first request ==");
  const first = await create({ model: MODEL, max_output_tokens: 4096,
    input: "Suggest three warm, short welcome messages for first-time church visitors." });
  console.log(outputText(first));

  console.log("\n== Step 3: streaming ==");
  await streamAnswer({ model: MODEL, max_output_tokens: 4096,
    input: "Write a two-sentence blessing for a volunteer team." });

  console.log("\n== Step 4: multi-turn ==");
  const conversation: any[] = [];
  for (const question of ["We want to start a meal train for a new family.",
                          "What should the sign-up ask for?"]) {
    conversation.push({ role: "user", content: question });
    const answer = outputText(await create({
      model: MODEL, max_output_tokens: 4096, input: conversation,
      instructions: "You are a friendly assistant for a church front desk. Keep answers to two sentences.",
    }));
    conversation.push({ role: "assistant", content: answer });
    console.log(`> ${question}\n${answer}\n`);
  }

  console.log("== Step 5: tool call ==");
  const items: any[] = [{ role: "user", content: "When are services at the Riverside campus?" }];
  let response = await create({ model: MODEL, input: items, tools: TOOLS });
  const calls = response.output.filter((i: any) => i.type === "function_call");
  for (const call of calls) {
    items.push({ type: "function_call", call_id: call.call_id, name: call.name, arguments: call.arguments });
    items.push({ type: "function_call_output", call_id: call.call_id, output: runTool(call) });
  }
  if (calls.length) response = await create({ model: MODEL, input: items });
  console.log(outputText(response));
}

main().catch((e) => {
  if (e instanceof GlooError) console.error(`\nError for users: ${e.userMessage}\nFor you: ${e.message}`);
  else console.error(e);
  process.exit(1);
});

Troubleshooting

401 with X-Gloo-Error-Code: authentication_error : The API key is missing, invalid or expired. Check that GLOO_API_KEY is exported in the shell that runs the script, and that the header reads Authorization: Bearer <key>. 400 naming an unknown model : The model ID isn’t in the catalog. Copy an id from GET /platform/v2/models (Step 2). The answer stops early, or status is "incomplete" : The response reached max_output_tokens. On a reasoning model the thinking counts toward that limit too. Raise max_output_tokens or lower reasoning.effort; see Set your output limit for reasoning models. The streaming loop prints nothing : Match on the event: line (or the payload’s type), and read text only from response.output_text.delta. Don’t wait for output[] on the final event: for non-OpenAI models, response.completed has no output[]. The model never returns a function_call : Check that the tool uses the nested {"type": "function", "function": {...}} schema and that the model has supports_tools in the catalog. To require a call, send "tool_choice": "required". 400 mentioning function calling or tools : The model takes no tools, for example a Gemini image model. Remove tools, or pick a model with supports_tools. The second tool request fails or ignores the result : The function_call and function_call_output items need the same call_id, the call_id from the model’s function_call item. output must be a string, so JSON.stringify / json.dumps your result. A parameter seems to have no effect : Some models don’t take every sampling parameter (for example temperature on some reasoning models). The request still succeeds, and the parameters that weren’t applied are listed in the response’s ignored_params.

Next Steps

Direct Responses guide

Reasoning summaries, vision, image generation and pricing

Direct Responses API reference

Every request parameter and response field

Handling Streaming Failures

Retry ceilings, continuation and partial-output UX

Grounded Responses tutorial

Answer from your own content with the same request shape

Guarded Responses tutorial

The same body with Gloo’s guardrails and values-alignment

Tool Use

Tool schemas, tool results and tool use on Completions V2

Get Started

Responses & Completions

Production

Content

Search & Recommendations

Embeddings

Legacy

Build with Direct Responses

Prerequisites

What You’ll Build

Step 1: Your First Request

Understanding the Response

Step 2: Choose a Model

Step 3: Stream the Answer

Step 4: Hold a Multi-Turn Conversation

Step 5: Let the Model Call a Function

Step 6: Handle Errors

Complete Code

Troubleshooting

Next Steps

Direct Responses guide

Direct Responses API reference

Handling Streaming Failures

Grounded Responses tutorial

Guarded Responses tutorial

Tool Use

​Prerequisites

​What You’ll Build

​Step 1: Your First Request

​Understanding the Response

​Step 2: Choose a Model

​Step 3: Stream the Answer

​Step 4: Hold a Multi-Turn Conversation

​Step 5: Let the Model Call a Function

​Step 6: Handle Errors

​Complete Code

​Troubleshooting

​Next Steps

Direct Responses guide

Direct Responses API reference

Handling Streaming Failures

Grounded Responses tutorial

Guarded Responses tutorial

Tool Use

Prerequisites

What You’ll Build

Step 1: Your First Request

Understanding the Response

Step 2: Choose a Model

Step 3: Stream the Answer

Step 4: Hold a Multi-Turn Conversation

Step 5: Let the Model Call a Function

Step 6: Handle Errors

Complete Code

Troubleshooting

Next Steps