Provider streaming API

Custom Streaming API

For providers with non-standard APIs, implement streamSimple. Study the existing API implementations before writing your own: Reference implementations: Atomic uses provider implementations from its installed @bastani/pi-ai dependency. The streaming implementations behind the api field live under node_modules/@bastani/pi-ai/dist/api/, including:

anthropic-messages.d.ts / anthropic-messages.js - Anthropic Messages API
mistral-conversations.d.ts / mistral-conversations.js - Mistral Conversations/Chat streaming
openai-completions.d.ts / openai-completions.js - OpenAI Chat Completions
openai-responses.d.ts / openai-responses.js - OpenAI Responses API
google-generative-ai.d.ts / google-generative-ai.js - Google Generative AI
bedrock-converse-stream.d.ts / bedrock-converse-stream.js - Amazon Bedrock Converse API Per-vendor provider configurations (base URLs, auth, model catalogs) live under dist/providers/, for example anthropic.d.ts / anthropic.js and mistral.d.ts / mistral.js.

Stream Pattern

All providers follow the same pattern:

import {
  type AssistantMessage,
  type AssistantMessageEventStream,
  type Context,
  type Model,
  type SimpleStreamOptions,
  calculateCost,
  createAssistantMessageEventStream,
} from "@bastani/pi-ai";

function streamMyProvider(
  model: Model<any>,
  context: Context,
  options?: SimpleStreamOptions
): AssistantMessageEventStream {
  const stream = createAssistantMessageEventStream();

  (async () => {
    // Initialize output message
    const output: AssistantMessage = {
      role: "assistant",
      content: [],
      api: model.api,
      provider: model.provider,
      model: model.id,
      usage: {
        input: 0,
        output: 0,
        cacheRead: 0,
        cacheWrite: 0,
        totalTokens: 0,
        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
      },
      stopReason: "pending",
      timestamp: Date.now(),
    };

    try {
      // Push start event
      stream.push({ type: "start", partial: output });

      // Make API request and process response...
      // Push content events as they arrive, and set output.stopReason from the
      // terminal event. A reason your provider sends that you do not map must
      // become an error, not a silent "stop".
      if (output.stopReason === "pending") {
        throw new Error("Provider stream ended without a stop reason");
      }
      if (output.stopReason === "error" || output.stopReason === "aborted") {
        throw new Error(output.errorMessage || "An unknown error occurred");
      }

      // Push done event
      stream.push({
        type: "done",
        reason: output.stopReason,
        message: output
      });
      stream.end();
    } catch (error) {
      output.stopReason = options?.signal?.aborted ? "aborted" : "error";
      output.errorMessage = error instanceof Error ? error.message : String(error);
      stream.push({ type: "error", reason: output.stopReason, error: output });
      stream.end();
    }
  })();

  return stream;
}

Event Types

Push events via stream.push() in this order:

{ type: "start", partial: output } - Stream started
Content events (repeatable, track contentIndex for each block):
- { type: "text_start", contentIndex, partial } - Text block started
- { type: "text_delta", contentIndex, delta, partial } - Text chunk
- { type: "text_end", contentIndex, content, partial } - Text block ended
- { type: "thinking_start", contentIndex, partial } - Thinking started
- { type: "thinking_delta", contentIndex, delta, partial } - Thinking chunk
- { type: "thinking_end", contentIndex, content, partial } - Thinking ended
- { type: "toolcall_start", contentIndex, partial } - Tool call started
- { type: "toolcall_delta", contentIndex, delta, partial } - Tool call JSON chunk
- { type: "toolcall_end", contentIndex, toolCall, partial } - Tool call ended
{ type: "done", reason, message } or { type: "error", reason, error } - Stream ended

The partial field in each event contains the current AssistantMessage state. Update output.content as you receive data, then include output as the partial.

Stop Reasons

StopReason is "pending" | "stop" | "length" | "toolUse" | "error" | "aborted". Start the partial message at "pending". It is the reason every in-flight message carries, and it says the terminal event has not arrived yet — it is not a default standing in for "stop". Set the real reason when the provider says the turn ended, then push done with it. Two checks belong immediately before done:

a stream that reached the end while still "pending" never received a terminal event, so raise rather than report a stop that did not happen;
"error" and "aborted" are failures, so raise them with output.errorMessage and let the catch push an error event.

done accepts only "stop", "length", and "toolUse", which is exactly what those two checks leave, so the as "stop" | "length" | "toolUse" cast older implementations used is no longer needed. Map each raw reason your provider can send onto one of the five terminal values, and raise on one you do not recognise rather than falling back to "stop". This is what the built-in providers do: an unmapped reason becomes a provider error naming the raw value, so a new truncation or safety signal is visible instead of arriving as a turn that looks like it finished normally. The optional rawStopReason field on AssistantMessage is where the provider’s own string belongs when you want to keep it.

Content Blocks

Add content blocks to output.content as they arrive:

// Text block
output.content.push({ type: "text", text: "" });
stream.push({ type: "text_start", contentIndex: output.content.length - 1, partial: output });

// As text arrives
const block = output.content[contentIndex];
if (block.type === "text") {
  block.text += delta;
  stream.push({ type: "text_delta", contentIndex, delta, partial: output });
}

// When block completes
stream.push({ type: "text_end", contentIndex, content: block.text, partial: output });

Tool Calls

Tool calls require accumulating JSON and parsing:

// Start tool call
output.content.push({
  type: "toolCall",
  id: toolCallId,
  name: toolName,
  arguments: {}
});
stream.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });

// Accumulate JSON
let partialJson = "";
partialJson += jsonDelta;
try {
  block.arguments = JSON.parse(partialJson);
} catch {}
stream.push({ type: "toolcall_delta", contentIndex, delta: jsonDelta, partial: output });

// Complete
stream.push({
  type: "toolcall_end",
  contentIndex,
  toolCall: { type: "toolCall", id, name, arguments: block.arguments },
  partial: output
});

Usage and Cost

Update usage from API response and calculate cost:

output.usage.input = response.usage.input_tokens;
output.usage.output = response.usage.output_tokens;
output.usage.cacheRead = response.usage.cache_read_tokens ?? 0;
output.usage.cacheWrite = response.usage.cache_write_tokens ?? 0;
output.usage.totalTokens = output.usage.input + output.usage.output +
                           output.usage.cacheRead + output.usage.cacheWrite;
calculateCost(model, output.usage);

calculateCost() selects one rate set for the whole request. Aggregate input is usage.input + usage.cacheRead + usage.cacheWrite; a tier applies only when that sum is strictly greater than inputTokensAbove, and the matching tier with the highest threshold wins. Every tier must provide complete input, output, cacheRead, and cacheWrite rates. Extension-registered models preserve these tiers, and matching models.json modelOverrides use the same replacement rules described in Custom Models.

Registration

pi.registerProvider("my-provider", {
  baseUrl: "https://api.example.com",
  apiKey: "$MY_API_KEY",
  api: "my-custom-api",
  models: [...],
  streamSimple: streamMyProvider
});

Build with Atomic

Extend a session

Orchestrate work

Distribute

Models and providers

Programmatic use

Provider streaming API

Provider streaming API

Custom Streaming API

Stream Pattern

Event Types

Stop Reasons

Content Blocks

Tool Calls

Usage and Cost

Registration

​Provider streaming API

​Custom Streaming API

​Stream Pattern

​Event Types

​Stop Reasons

​Content Blocks

​Tool Calls

​Usage and Cost

​Registration

Provider streaming API

Custom Streaming API

Stream Pattern

Event Types

Stop Reasons

Content Blocks

Tool Calls

Usage and Cost

Registration