> ## Documentation Index
> Fetch the complete documentation index at: https://docs.bastani.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Provider streaming API

> Implement a custom streaming API: events, content blocks, tool calls, stop reasons, and usage.

# 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:

```typescript theme={null}
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:

1. `{ type: "start", partial: output }` - Stream started

2. 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

3. `{ 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:

```typescript theme={null}
// 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:

```typescript theme={null}
// 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:

```typescript theme={null}
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](/models/reference#request-wide-cost-tiers).

### Registration

Register your stream function:

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