Provider adapters

Each file in packages/ai/src/api/ is an adapter: on the way out it converts Pi's transcript into one vendor's request format, and on the way back it converts that vendor's streaming events into Pi's event protocol. Ten wire formats (KnownApi) cover more than forty providers, because most vendors speak a dialect of OpenAI Chat Completions, OpenAI Responses or Anthropic Messages. This chapter dissects the Anthropic adapter as a specimen, then studies the cross-cutting machinery every adapter shares: Server-Sent Events, cross-provider message transformation, prompt caching, retries and context-overflow detection.

Learning goals. Read any adapter and identify its request-building, streaming and finalization phases; parse SSE by hand; explain why replaying a conversation on a different model requires rewriting it; implement exponential backoff with jitter that respects Retry-After and cancellation.

adapter pipeline · request and response paths

The top row turns Pi's transcript into one vendor's request; the bottom row turns that vendor's byte stream back into Pi's event protocol.

4.1 Anatomy of an adapter: Anthropic Messages#

api/anthropic-messages.ts exports stream (and streamSimple). Its skeleton is the pattern all adapters follow:

1. Create the stream and return it synchronously. The work runs in an immediately-invoked async function, (async () => { … })(), whose promise is deliberately not awaited.
2. Prepare an output message with content: [], zeroed usage and stopReason: "pending". Every event will carry this object as partial.
3. Resolve the transcript for this API — resolveTranscript(context, compat.supportsMidConvoSystemMessages) collapses mid-conversation system messages when the API cannot take them — and compute the current tool set.
4. Build the request (buildParams), give the caller a chance to inspect or replace it (onPayload), and send it through the vendor SDK with retries (§4.6).
5. Push start, then translate each vendor event into Pi events, mutating output as it goes.
6. Finalize: check for abort, for a missing stop reason, for an error stop; push done; end().
7. On any exception, clean scratch fields out of the content blocks, set stopReason to aborted if the signal fired or error otherwise, push an error event and end().

Step 7 is why callers never see a thrown error once they hold a stream (Chapter 3). Note how it scrubs scratch state before the message can be persisted:

} catch (error) {
	for (const block of output.content) {
		delete (block as { index?: number }).index;
		// partialJson is only a streaming scratch buffer; never persist it.
		delete (block as { partialJson?: string }).partialJson;
	}
	output.stopReason = options?.signal?.aborted ? "aborted" : "error";
	output.errorMessage = error instanceof Error ? error.message : JSON.stringify(error);
	stream.push({ type: "error", reason: output.stopReason, error: output });
	stream.end();
}

The request loop contains a nice example of graceful degradation. Pi may ask Anthropic to enforce tool schemas strictly. If the API rejects the request with HTTP 400 because "The compiled grammar is too large" (or too many strict tools, or too many union types), the adapter sets disablePreferredStrictTools = true, records a diagnostic, rebuilds the request without strict schemas and tries again. The comment marks the safety condition: "Only downgrade a rejected request, never an already-started stream." Retrying is only safe while nothing has been emitted.

4.2 Server-Sent Events, by hand#

Streaming LLM APIs use Server-Sent Events (SSE), a text format over a long-lived HTTP response. Each event is a group of field: value lines terminated by a blank line:

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hel"}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"lo"}}

The Anthropic adapter parses this itself in iterateSseMessages() rather than relying on the SDK's parser. The algorithm is a framing problem: bytes arrive in arbitrary chunks that do not respect line or event boundaries.

buffer += decoder.decode(value, { stream: true });
let consumed = consumeLine(buffer);
while (consumed) {
	buffer = consumed.rest;
	const event = decodeSseLine(consumed.line, state);
	if (event) {
		yield event;
	}
	consumed = consumeLine(buffer);
}

Four details reward attention:

• TextDecoder with { stream: true }. A UTF-8 character is one to four bytes. If a chunk boundary splits a character (an emoji, a CJK glyph), decoding chunks independently would produce replacement characters. Streaming mode keeps the incomplete bytes and finishes them with the next chunk. A final decoder.decode() flushes anything left.
• Three line terminators. consumeLine() accepts \n, \r and \r\n, taking care that a \r at the end of the buffer followed by \n in the next chunk is still one break where possible.
• The field grammar. In decodeSseLine(), a blank line flushes the accumulated event; a line beginning with : is a comment (servers send these as keep-alives); otherwise the text before the first : is the field name and one leading space is stripped from the value. Multiple data: lines join with \n.
• Cleanup in finally. reader.releaseLock() runs whether iteration finishes, throws, or the consumer stops early (a generator's finally block runs when its consumer breaks out of for await).

On top of SSE, iterateAnthropicEvents() checks protocol-level invariants: an event: error becomes an exception; unknown event types are skipped; and a stream that saw message_start but never message_stop throws "Anthropic stream ended before message_stop". That last check turns a silently truncated connection into a visible, retryable error.

4.3 Translating vendor events#

Anthropic's stream has its own vocabulary: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop. The mapping is direct but not trivial:

Two indexing schemes coexist. Anthropic numbers blocks with event.index; Pi's contentIndex is the position in output.content. The adapter stores the vendor index on each block temporarily and looks blocks up with blocks.findIndex((b) => b.index === event.index), deleting the field at content_block_stop. This is the adapter pattern at small scale: a translation table between two numbering systems, kept private to the adapter.

Thinking signatures. Reasoning models return an opaque, cryptographically signed form of their hidden reasoning, which must be sent back unchanged on the next request for multi-turn continuity. Pi stores it in thinkingSignature. A redacted_thinking block carries only encrypted data; Pi records it as a thinking block with text "[Reasoning redacted]", redacted: true, and the payload in the signature.

Stop reasons are mapped by mapStopReason(): end_turn → stop, max_tokens → length, tool_use → toolUse, refusal → error with the model's explanation, pause_turn → stop ("Stop is good enough -> resubmit"), and anything unknown throws "Unhandled stop reason" rather than guessing. Failing loudly on an unknown enum value is a defensive choice: vendors add values, and silently treating a new one as success could hide a refusal.

Usage arrives twice: message_start carries input and cache counts, message_delta carries output counts (and, for some proxies, partial data). The adapter updates only fields that are present ("Preserves input_tokens from message_start when proxies omit it in message_delta"), computes totalTokens because "Anthropic doesn't provide total_tokens", and calls calculateCost after each update so that even an aborted response has an accurate bill.

One vendor-specific branch deserves mention because it shows how much of an adapter is policy, not protocol. When the credential is a Claude subscription OAuth token rather than an API key, the adapter renames tools to the casing used by Claude Code (toClaudeCodeName, under a source comment "Stealth mode: Mimic Claude Code's tool naming exactly") and maps names back on the way in (fromClaudeCodeName).

4.4 The OpenAI Chat Completions dialect#

api/openai-completions.ts serves OpenAI itself and every "OpenAI-compatible" server: OpenRouter, Groq, Cerebras, DeepSeek, Together, vLLM, llama.cpp, Ollama, LM Studio and others. The streaming format is a sequence of JSON chunks whose choices[0].delta may contain content, a reasoning field, or tool_calls fragments.

Tool-call fragments are correlated by position, not by event type. A fragment carries an index and, usually only in its first appearance, an id and function.name. ensureToolCallBlock() looks the block up first by stream index, then by id, and creates it otherwise; subsequent fragments append function.arguments text to partialArgs.

Compatibility is the hard part. Reasoning text appears as reasoning_content (llama.cpp), reasoning, or reasoning_text depending on the server; the adapter uses "the first non-empty reasoning field to avoid duplication (e.g., chutes.ai returns both reasoning_content and reasoning with same content)". Some servers never send finish_reason, in which case the stop reason is inferred: toolUse if any tool call was produced, otherwise stop. Which quirks apply is decided by detectCompat(), which inspects model.provider and substrings of model.baseUrl (api.z.ai, openrouter.ai, api.x.ai, api.cloudflare.com, …), and can be overridden per model through model.compat. The thinkingFormat compat field alone has variants for zai, qwen, qwen-chat-template, chat-template, baseten, deepseek and openrouter — each a different way of asking for reasoning.

This is the reality of an interoperability layer: the abstraction is clean at the top and a catalogue of special cases underneath. Pi keeps those cases data-driven (compat flags on the model) so a user can describe a new server in models.json without changing code.

Other adapters follow the same shape with different vocabularies. Google's Gemini attaches a thoughtSignature that "can appear on ANY part type (text, functionCall, etc.)" and must be replayed in place, which is why ToolCall has an optional thoughtSignature field. The OpenAI Responses API produces long item ids and encrypted reasoning items. Bedrock speaks the Converse stream protocol through the AWS SDK. pi-messages is Pi's own protocol: "a single POST of { model, context, options } to <baseUrl>/messages, the response is an SSE stream of serialized assistant-message events" — the identity adapter, used by the Radius gateway.

4.5 Changing models mid-conversation#

Pi lets you switch models in the middle of a session. The history then contains assistant messages produced by a different vendor, with that vendor's signatures, ids and quirks. transformMessages() in api/transform-messages.ts rewrites the history for the target model in two passes.

Pass 1: per-message rewriting. An assistant message is "same model" when provider, api and model id all match. For other assistant messages:

• Redacted thinking is dropped ("opaque encrypted content, only valid for the same model").
• Readable thinking becomes plain text, so the new model can still see the reasoning.
• Google's thoughtSignature is removed from tool calls.
• Tool-call ids are normalized when the target needs it: "OpenAI Responses API generates IDs that are 450+ chars with special characters like |. Anthropic APIs require IDs matching ^[a-zA-Z0-9_-]+$ (max 64 chars)." A Map from old id to new id is filled as calls are seen and applied to the matching tool results.

Images are downgraded to placeholder text for models whose input does not include "image".

Pass 2: restoring protocol invariants. Every provider requires that each tool call is followed by exactly one result before the next assistant or user message. A history may violate this: the user aborted mid-tool, the process crashed, or an error interrupted the turn. The second pass walks the transcript with a small state machine (pendingToolCalls, existingToolResultIds) and, at each point where the call/result group must be closed (a new assistant message, a user message, or the end), inserts a synthetic result for every unanswered call:

result.push({
	role: "toolResult",
	toolCallId: tc.id,
	toolName: tc.name,
	content: [{ type: "text", text: "No result provided" }],
	isError: true,
	timestamp: Date.now(),
} as ToolResultMessage);

Assistant messages that ended in error or aborted are skipped entirely because they "may have partial content (reasoning without message, incomplete tool calls)" and "the model should retry from the last valid state". System messages that fall between a call and its results are held back and emitted after the results so they cannot split the group. The general lesson: when you persist a log written by an unreliable process, the reader must re-establish the invariants the consumer requires.

4.6 Prompt caching#

Providers can cache the processed prefix of a prompt. If the next request begins with the same tokens, the cached part is billed at a fraction of the price (cacheRead) and processed faster. Agent sessions are ideal: each request is the previous one plus a few messages.

Anthropic requires explicit cache breakpoints. The adapter puts cache_control: { type: "ephemeral" } (with ttl: "1h" for long retention where supported) on the last tool definition and on the last block of the last user or system message: "Add cache_control to the last user or system message to cache conversation history". Retention comes from the cacheRetention option (none, short or long, default short, also settable through PI_CACHE_RETENTION). OpenAI caches automatically; the completions adapter passes a cache key derived from the session id and asks for 24-hour retention when long retention is supported. This is also why Pi prefers appending system messages to rewriting the system prompt: any change to an early token invalidates everything after it.

4.7 Retries at two layers#

Transient failures (overload, rate limits, dropped connections) are common. Pi retries at two layers with different knowledge.

HTTP layer: retryProviderRequest() in utils/provider-retry.ts. The vendor SDKs have their own retry loops, but "Their built-in retry timers ignore the request AbortSignal, so callers must invoke the SDK with maxRetries: 0 and wrap the request with this helper." The policy mirrors the SDKs: retry if the server says x-should-retry: true, if there was no status (a network failure), or on 408, 409, 429 or any 5xx. The delay honours server hints first:

const retryAfterMs = error.headers?.get("retry-after-ms");
…
const retryAfter = error.headers?.get("retry-after");
if (retryAfter) {
	const seconds = Number.parseFloat(retryAfter);
	const delayMs = Number.isNaN(seconds) ? Date.parse(retryAfter) - Date.now() : seconds * 1000;
	…
}
const exponentialDelay = Math.min(0.5 * 2 ** retryIndex, 8) * 1000;
return exponentialDelay * (1 - Math.random() * 0.25);

Retry-After may be a number of seconds or an HTTP date, hence the two parses. Without a hint, the delay is 0.5 s, 1 s, 2 s, 4 s, capped at 8 s, reduced by up to 25 % at random. That randomness is jitter: if a server restart causes a thousand clients to fail at once, identical backoff schedules would make them all retry at the same instants (the thundering herd); jitter spreads them out. If the server asks for more than maxRetryDelayMs (default 60 s), the helper fails immediately with a message naming the delay, so a higher layer can show it to the user instead of freezing silently. The sleep is abortable: it listens for the signal's abort event and rejects at once.

Agent layer: retryAssistantCall() in utils/retry.ts. Once a stream has started, HTTP retry is no longer possible; the failure surfaces as an assistant message with stopReason: "error". This layer re-runs the whole call with exponential backoff (baseDelayMs * 2^(attempt-1), capped by maxAgentDelayMs), reports progress through callbacks so the UI can show "retrying in 4 s", never retries an abort, and classifies errors by message text. Two regular expressions do the classification: a long list of transient signatures ("overloaded", "rate.?limit", "502", "socket hang up", "stream ended before message_stop", …, each annotated with the issue that motivated it) and a list of non-retryable limits ("insufficient_quota", "Monthly usage limit reached", "billing", …). Retrying a quota error would only waste time; "deterministic errors fail fast".

4.8 Detecting context overflow#

When a request exceeds the context window, the right reaction is not a retry but compaction (Chapter 9). Each vendor phrases the error differently, so utils/overflow.ts keeps a catalogue of regular expressions with example messages: Anthropic's "prompt is too long: 213462 tokens > 200000 maximum", Google's "The input token count (1196265) exceeds the maximum number of tokens allowed (1048575)", llama.cpp's "the request exceeds the available context size", and about twenty more. A separate list of non-overflow patterns excludes rate-limit messages that would otherwise match.

Some servers do not report overflow at all. isContextOverflow() therefore also checks two silent cases using usage numbers: a successful response whose input exceeds the context window (z.ai truncates silently), and a length stop with zero output tokens and input filling 99 % of the window (Xiaomi MiMo truncates the input to fit and leaves no room to answer). Inferring a failure from metrics when the error channel is unreliable is a technique worth remembering.

4.9 Concepts in this chapter#

• Adapter pattern / anti-corruption layer. Translate between an external model and your own at a boundary, so vendor concepts never leak inward.
• Server-Sent Events and message framing. Delimiting messages in a byte stream is the first job of any protocol parser; SSE uses blank lines.
• Streaming text decoding. Multi-byte encodings require carrying partial characters across chunk boundaries.
• Generators and resource cleanup. finally in a generator runs on normal end, error, or early consumer exit.
• Invariant restoration. Repair data so it satisfies a downstream contract (every tool call answered) rather than trusting the writer.
• Retry safety and idempotency. Retry only operations that have had no visible effect; a request rejected before streaming is safe, a half-streamed response is not.
• Exponential backoff with jitter. Increase waits geometrically and randomize them to avoid synchronized retries.
• Error classification. Separate transient, permanent and "needs a different action" (overflow) failures; each deserves a different response.
• Graceful degradation. Fall back to a less capable but working request (non-strict tools) when the preferred one is refused.
• Caching by prefix. Stable prefixes are cheap; any early change invalidates the rest.

Exercises#

1. Feed iterateSseMessages a stream where a chunk boundary falls inside the four bytes of an emoji, and another where it falls between \r and \n. Predict the output and check it.
2. Trace transformMessages on: assistant (OpenAI Responses, two tool calls with long ids) → one tool result → user message, targeting an Anthropic model. List every inserted, removed and renamed item.
3. Why does the adapter delete partialJson in both the success and error paths? What would go wrong in a persisted session otherwise?
4. Compute the retry schedule retryProviderRequest would follow for three consecutive 503 errors without Retry-After, giving the minimum and maximum of each delay.
5. A provider returns HTTP 429 with retry-after: 300. What happens with the default options, and why is that better than sleeping for five minutes?
6. Write a new overflow pattern for a made-up server message and the test that proves it does not match a rate-limit message.
7. Sketch the minimal adapter for a fictional API whose stream sends whole tool calls in one event. Which Pi events must you still emit, and in what order?