Streaming tool calls

export API_BASE_URL="https://gateway.example.com/v1"
export API_MODEL="YOUR_TOOL_CAPABLE_MODEL_ID"
# API_KEY must already be set in the process environment.

import OpenAI from 'openai';

const { API_BASE_URL, API_KEY, API_MODEL } = process.env;
if (!API_BASE_URL || !API_KEY || !API_MODEL) {
  throw new Error('Set API_BASE_URL, API_KEY, and API_MODEL first.');
}
const client = new OpenAI({ baseURL: API_BASE_URL.replace(/\/+$/, ''), apiKey: API_KEY,
  timeout: 60_000, maxRetries: 0 });
const messages: OpenAI.Chat.Completions.ChatCompletionMessageParam[] = [
  { role: 'user', content: 'What is 7 times 8? Use multiply.' },
];
const tools: OpenAI.Chat.Completions.ChatCompletionTool[] = [{
  type: 'function', function: { name: 'multiply', description: 'Multiply two finite numbers.', strict: true,
    parameters: { type: 'object', properties: { a: { type: 'number' }, b: { type: 'number' } },
      required: ['a', 'b'], additionalProperties: false } },
}];
const seenIds = new Set<string>();

try {
  for (let round = 0; round < 3; round++) {
    const calls = new Map<number, { id: string; type: 'function'; function: { name: string; arguments: string } }>();
    let finish: string | null = null;
    let text = '';
    let events = 0;
    const stream = await client.chat.completions.create({ model: API_MODEL, messages, tools, stream: true });
    for await (const chunk of stream) {
      if (++events > 512) throw new Error('Stream event limit reached.');
      for (const choice of chunk.choices) {
        if (choice.index !== 0 || finish !== null) throw new Error('Unexpected choice or event after completion.');
        if (choice.delta.refusal || choice.delta.function_call) throw new Error('Refusal or unsupported legacy call.');
        text += choice.delta.content ?? '';
        if (text.length > 100_000) throw new Error('Text limit reached.');
        for (const delta of choice.delta.tool_calls ?? []) {
          if (!Number.isInteger(delta.index) || delta.index < 0 || delta.index >= 4) throw new Error('Invalid call index.');
          let call = calls.get(delta.index);
          if (!call) {
            call = { id: '', type: 'function', function: { name: '', arguments: '' } };
            calls.set(delta.index, call);
          }
          if (delta.type && delta.type !== 'function') throw new Error('Unsupported tool type.');
          if (delta.id) {
            if (call.id && call.id !== delta.id) throw new Error('Call ID changed.');
            call.id = delta.id;
          }
          if (delta.function?.name) {
            if (call.function.name && call.function.name !== delta.function.name) throw new Error('Tool name changed.');
            call.function.name = delta.function.name;
          }
          call.function.arguments += delta.function?.arguments ?? '';
          if (call.function.arguments.length > 16_384) throw new Error('Arguments limit reached.');
        }
        if (choice.finish_reason !== null) finish = choice.finish_reason;
      }
    }
    // No function executes until the whole stream has ended with a valid finish reason.
    if (calls.size === 0) {
      if (finish !== 'stop' || !text.trim()) throw new Error('No complete answer.');
      console.log(text);
      break;
    }
    if (finish !== 'tool_calls' || round === 2) throw new Error('Incomplete calls or round limit reached.');
    const ordered = [...calls.entries()].sort(([a], [b]) => a - b).map(([, call]) => call);
    const outputs: OpenAI.Chat.Completions.ChatCompletionToolMessageParam[] = [];
    for (const call of ordered) {
      if (!call.id.trim() || seenIds.has(call.id) || call.function.name !== 'multiply') throw new Error('Invalid tool call.');
      const args = JSON.parse(call.function.arguments);
      if (!args || Array.isArray(args) || Object.keys(args).length !== 2
        || !Number.isFinite(args.a) || !Number.isFinite(args.b)) throw new Error('Invalid arguments.');
      const value = args.a * args.b;
      if (!Number.isFinite(value)) throw new Error('Invalid result.');
      seenIds.add(call.id);
      outputs.push({ role: 'tool', tool_call_id: call.id, content: JSON.stringify({ value }) });
    }
    messages.push({ role: 'assistant', content: text || null, tool_calls: ordered }, ...outputs);
  }
} catch (error) {
  console.error(error instanceof OpenAI.APIError && error.status
    ? `Request failed (HTTP ${error.status}). No automatic retry.`
    : 'Streaming tool workflow failed; no incomplete answer published.');
  process.exitCode = 1;
}

import json
import math
import os
from openai import OpenAI, APIConnectionError, APIStatusError, APIError

history = [{"role": "user", "content": "What is 7 times 8? Use multiply."}]
tools = [{
    "type": "function", "name": "multiply", "description": "Multiply two finite numbers.",
    "strict": True,
    "parameters": {
        "type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
        "required": ["a", "b"], "additionalProperties": False,
    },
}]
seen_call_ids = set()
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
            api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
    try:
        for round_index in range(3):
            result = None
            pending = {}
            with client.responses.create(
                model=os.environ["API_MODEL"], input=history, tools=tools,
                instructions="Use multiply for arithmetic, then answer briefly.",
                store=False, include=["reasoning.encrypted_content"], stream=True,
            ) as stream:
                for event_count, event in enumerate(stream, start=1):
                    if event_count > 512 or result is not None:
                        raise ValueError("Event limit or event after completion.")
                    if event.type in ("error", "response.failed", "response.incomplete",
                                      "response.refusal.delta", "response.refusal.done"):
                        raise ValueError("Failed or refused response.")
                    if event.type == "response.output_item.added" and event.item.type == "function_call":
                        index = event.output_index
                        if type(index) is not int or index < 0 or index in pending or len(pending) >= 4:
                            raise ValueError("Invalid function item index.")
                        if (not event.item.id or not event.item.call_id
                                or not isinstance(event.item.arguments, str)):
                            raise ValueError("Missing function identity or arguments.")
                        pending[index] = {"id": event.item.id, "call_id": event.item.call_id,
                                          "name": event.item.name, "arguments": event.item.arguments,
                                          "arguments_done": False, "item_done": False}
                        if len(event.item.arguments) > 16_384:
                            raise ValueError("Arguments limit reached.")
                    elif event.type in ("response.function_call_arguments.delta", "response.function_call_arguments.done"):
                        call = pending[event.output_index]
                        if call["id"] != event.item_id or call["arguments_done"]:
                            raise ValueError("Unknown, changed, or finished function item.")
                        if event.type.endswith(".delta"):
                            call["arguments"] += event.delta
                            if len(call["arguments"]) > 16_384:
                                raise ValueError("Arguments limit reached.")
                        else:
                            if event.arguments != call["arguments"]:
                                raise ValueError("Final arguments differ from streamed arguments.")
                            call["arguments_done"] = True
                    elif event.type == "response.output_item.done" and event.item.type == "function_call":
                        call = pending[event.output_index]
                        if (not call["arguments_done"] or call["item_done"]
                                or any(getattr(event.item, key) != call[key]
                                       for key in ("id", "call_id", "name", "arguments"))):
                            raise ValueError("Incomplete or inconsistent function item.")
                        call["item_done"] = True
                    elif event.type == "response.completed":
                        result = event.response
            if result is None:
                raise ValueError("Stream ended without response.completed.")
            final_indices = set()
            for index, item in enumerate(result.output):
                if item.type != "function_call":
                    continue
                call = pending[index]
                if (not call["item_done"] or any(getattr(item, key) != call[key]
                                                for key in ("id", "call_id", "name", "arguments"))):
                    raise ValueError("Completed response differs from streamed calls.")
                final_indices.add(index)
            if final_indices != set(pending):
                raise ValueError("Completed response lost a streamed call.")
            # Execute only after the completed response and the entire stream validate.
            if result.status != "completed":
                raise ValueError("Response did not complete.")
            calls = []
            for item in result.output:
                if item.type == "function_call":
                    if getattr(item, "status", None) not in (None, "completed"):
                        raise ValueError("Function call did not complete.")
                    calls.append(item)
                elif item.type == "message":
                    if item.status != "completed" or any(part.type != "output_text" for part in item.content):
                        raise ValueError("Refusal or incomplete message.")
                elif item.type != "reasoning":
                    raise ValueError("Unsupported output item.")
            if not calls:
                if not result.output_text.strip():
                    raise ValueError("No final text.")
                print(result.output_text)
                break
            if round_index == 2 or len(calls) > 4:
                raise ValueError("Tool round or call limit reached.")
            outputs = []
            for call in calls:
                if (call.name != "multiply" or not isinstance(call.call_id, str)
                        or not call.call_id.strip() or call.call_id in seen_call_ids):
                    raise ValueError("Unknown tool or invalid call ID.")
                args = json.loads(call.arguments)
                if not isinstance(args, dict) or set(args) != {"a", "b"}:
                    raise ValueError("Invalid tool arguments.")
                if any(type(value) not in (int, float) or not math.isfinite(value) for value in args.values()):
                    raise ValueError("Expected finite numbers.")
                value = args["a"] * args["b"]
                if not math.isfinite(value):
                    raise ValueError("Invalid tool result.")
                seen_call_ids.add(call.call_id)
                outputs.append({
                    "type": "function_call_output", "call_id": call.call_id,
                    "output": json.dumps({"value": value}, allow_nan=False),
                })
            history.extend(item.model_dump(exclude_none=True) for item in result.output)
            history.extend(outputs)
    except APIStatusError as error:
        raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
    except APIConnectionError:
        raise SystemExit("Connection failed. No automatic retry.")
    except (ValueError, TypeError, AttributeError, KeyError, OverflowError, APIError):
        raise SystemExit("Streaming tool workflow failed; no incomplete answer published.")