Agents/Providers — Adapter Pattern

Multi-Provider Adapter Pattern Schema Sanitization IAM Auth

Contents

Overview

flowchart TD
  subgraph Registry["Model Registry"]
    MODEL["model.ts (24 KB)\nModel registry & resolution"]
    NORM["model.provider-normalization.ts\nProvider ID normalization"]
    EXTRA["extra-params.ts\nProvider-specific params"]
  end
  subgraph Selection["Provider Selection"]
    RESOLVE["resolveProvider(modelId)"]
    NORM --> RESOLVE
    MODEL --> RESOLVE
    EXTRA --> RESOLVE
  end
  subgraph Wrappers["Stream Wrappers"]
    OAI["openai-stream-wrappers.ts"]
    GOO["google-stream-wrappers.ts"]
    ANT["anthropic-cache-retention.ts"]
    BED["bedrock-stream-wrappers.ts"]
    MOON["moonshot-stream-wrappers.ts"]
    MINI["minimax-stream-wrappers.ts"]
    PROXY["proxy-stream-wrappers.ts"]
    ZAI["zai-stream-wrappers.ts"]
  end
  subgraph Output["Normalized Output"]
    CHUNKS["Uniform chunk protocol"]
    USAGE["UsageAccumulator"]
  end
  RESOLVE --> OAI
  RESOLVE --> GOO
  RESOLVE --> ANT
  RESOLVE --> BED
  RESOLVE --> MOON
  RESOLVE --> MINI
  RESOLVE --> PROXY
  RESOLVE --> ZAI
  OAI --> CHUNKS
  GOO --> CHUNKS
  ANT --> CHUNKS
  BED --> CHUNKS
  MOON --> CHUNKS
  MINI --> CHUNKS
  PROXY --> CHUNKS
  ZAI --> CHUNKS
  CHUNKS --> USAGE

The flow begins in model.ts where the model registry resolves a model identifier to a concrete provider. Provider ID normalization in model.provider-normalization.ts handles special cases such as Google and XAI model prefixes. Once the provider is identified, the corresponding stream wrapper is instantiated with any extra parameters from extra-params.ts.

Decision Points

How It Works

The adapter pattern in OpenClaw follows a three-phase approach: resolve the target provider, wrap the provider-specific HTTP stream, and normalize the output into uniform chunks. Each wrapper file implements the same contract while encapsulating all provider-specific logic.

Phase 1: Model Resolution

The model registry in model.ts (24 KB) maps model identifiers to provider metadata. When a model string arrives, model.provider-normalization.ts normalizes provider prefixes — for example, Google models may arrive as google/gemini-pro or gemini-pro, and XAI models may use vendor-specific prefixes that need canonicalization.

// model.provider-normalization.ts — simplified provider resolution
export function normalizeProviderId(rawModel: string): ProviderInfo {
  const lower = rawModel.toLowerCase();

  // Google special cases: "google/gemini-*" or bare "gemini-*"
  if (lower.startsWith("google/") || lower.startsWith("gemini-")) {
    return { provider: "google", modelId: lower.replace("google/", "") };
  }

  // XAI special cases: "xai/grok-*" or "grok-*"
  if (lower.startsWith("xai/") || lower.startsWith("grok-")) {
    return { provider: "xai", modelId: lower.replace("xai/", "") };
  }

  // Default: split on first "/" — "anthropic/claude-3" → provider="anthropic"
  const slashIdx = lower.indexOf("/");
  if (slashIdx > 0) {
    return { provider: lower.slice(0, slashIdx), modelId: lower.slice(slashIdx + 1) };
  }

  // Bare model name: look up in registry
  return resolveFromRegistry(rawModel);
}

Phase 2: Stream Wrapping

Once the provider is resolved, the runtime instantiates the corresponding stream wrapper. Each wrapper implements the same streaming contract but encapsulates provider-specific concerns:

Phase 3: Output Normalization

Every wrapper transforms raw provider events into the same chunk protocol. The normalization ensures that downstream code in attempt.ts and payloads.ts never needs provider-conditional logic.

// Simplified normalized chunk interface
interface NormalizedChunk {
  type: "text" | "tool_call" | "thinking" | "usage" | "done";
  content?: string;
  toolCall?: {
    id: string;
    name: string;
    arguments: string; // always JSON string
  };
  thinking?: string;
  usage?: { inputTokens: number; outputTokens: number };
}

// Each wrapper implements this contract:
type StreamWrapper = (
  response: Response,
  onChunk: (chunk: NormalizedChunk) => void,
  signal: AbortSignal,
) => Promise<void>;

Provider Wrapper Class Diagram

classDiagram
  class StreamWrapper {
    <<interface>>
    +wrap(response, onChunk, signal) Promise~void~
  }
  class OpenAIWrapper {
    -parseSSELine(line) NormalizedChunk
    -assembleToolCallDelta(delta) void
    +wrap(response, onChunk, signal) Promise~void~
  }
  class GoogleWrapper {
    -cleanForGemini(schema) object
    -GEMINI_UNSUPPORTED_SCHEMA_KEYWORDS string[]
    -parseGeminiChunk(data) NormalizedChunk
    +wrap(response, onChunk, signal) Promise~void~
  }
  class AnthropicWrapper {
    -injectCacheRetention(headers) Headers
    -parseCacheStatus(event) CacheInfo
    +wrap(response, onChunk, signal) Promise~void~
  }
  class BedrockWrapper {
    -signWithSigV4(request) SignedRequest
    -decodeEventStream(buffer) Event[]
    +wrap(response, onChunk, signal) Promise~void~
  }
  class MoonshotWrapper {
    -splitThinkingStream(chunk) NormalizedChunk[]
    +wrap(response, onChunk, signal) Promise~void~
  }
  class MinimaxWrapper {
    +wrap(response, onChunk, signal) Promise~void~
  }
  class ProxyWrapper {
    +wrap(response, onChunk, signal) Promise~void~
  }
  class ZAIWrapper {
    -mapEventName(raw) string
    +wrap(response, onChunk, signal) Promise~void~
  }
  StreamWrapper <|.. OpenAIWrapper
  StreamWrapper <|.. GoogleWrapper
  StreamWrapper <|.. AnthropicWrapper
  StreamWrapper <|.. BedrockWrapper
  StreamWrapper <|.. MoonshotWrapper
  StreamWrapper <|.. MinimaxWrapper
  StreamWrapper <|.. ProxyWrapper
  StreamWrapper <|.. ZAIWrapper

All eight wrapper classes implement the same StreamWrapper interface. The class diagram above shows the private helper methods unique to each provider and the shared public wrap() contract that the runtime depends on.

Annotated Code Walkthrough

Walk through how an OpenAI-compatible stream wrapper normalizes SSE events into the common chunk protocol.

// openai-stream-wrappers.ts — entry point
export async function wrapOpenAIStream(
  response: Response,
  onChunk: (chunk: NormalizedChunk) => void,
  signal: AbortSignal,
): Promise<void> {
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (!signal.aborted) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() ?? "";
    for (const line of lines) {
      processSSELine(line, onChunk);
    }
  }
  onChunk({ type: "done" });
}

function processSSELine(
  line: string,
  onChunk: (chunk: NormalizedChunk) => void,
): void {
  if (!line.startsWith("data: ")) return;
  const payload = line.slice(6);
  if (payload === "[DONE]") return;

  const event = JSON.parse(payload);
  const delta = event.choices?.[0]?.delta;
  if (!delta) return;

  // Text content
  if (delta.content) {
    onChunk({ type: "text", content: delta.content });
  }

  // Tool call accumulation
  if (delta.tool_calls) {
    for (const tc of delta.tool_calls) {
      onChunk({
        type: "tool_call",
        toolCall: {
          id: tc.id ?? currentToolCallId,
          name: tc.function?.name ?? currentToolCallName,
          arguments: tc.function?.arguments ?? "",
        },
      });
    }
  }

  // Usage metadata (final chunk)
  if (event.usage) {
    onChunk({
      type: "usage",
      usage: {
        inputTokens: event.usage.prompt_tokens,
        outputTokens: event.usage.completion_tokens,
      },
    });
  }
}

// anthropic-cache-retention.ts — header injection
function buildAnthropicHeaders(baseHeaders: Headers, cacheConfig: CacheConfig): Headers {
  const headers = new Headers(baseHeaders);

  if (cacheConfig.enabled) {
    headers.set("anthropic-beta", "prompt-caching-2024-07-31");
    headers.set("anthropic-cache-control", JSON.stringify({
      type: "ephemeral",
      ttl: cacheConfig.ttlSeconds ?? 300,
    }));
  }

  return headers;
}

Anthropic Tool Call Compatibility

The anthropic-family-tool-payload-compat.ts module bridges the gap between OpenAI-style and Anthropic-style tool call formats. Anthropic expects tool inputs as a top-level input object rather than a function.arguments JSON string:

// anthropic-family-tool-payload-compat.ts — format bridge
export function toAnthropicToolUse(openaiToolCall: OpenAIToolCall): AnthropicToolUse {
  return {
    type: "tool_use",
    id: openaiToolCall.id,
    name: openaiToolCall.function.name,
    input: JSON.parse(openaiToolCall.function.arguments),
  };
}

export function fromAnthropicToolResult(result: AnthropicToolResult): OpenAIToolMessage {
  return {
    role: "tool",
    tool_call_id: result.tool_use_id,
    content: typeof result.content === "string"
      ? result.content
      : JSON.stringify(result.content),
  };
}

Edge Cases & Error Handling

Configuration

Related Topics

See also:

• Providers overview — the parent sub-module page
• Token Management — API key rotation and auth profile health
• Model Selection — model resolution, fallback chains, and context window sizing
• Streaming pipeline — how normalized chunks flow through the agent runtime

Directory Map

pi-embedded-runner/
├── model.ts                                    ← Model registry & resolution (24 KB)
├── model.provider-normalization.ts             ← Provider ID normalization
├── extra-params.ts                             ← Per-provider extra request params
├── openai-stream-wrappers.ts                   ← OpenAI SSE adapter
├── google-stream-wrappers.ts                   ← Google/Gemini adapter + cleanForGemini
├── anthropic-cache-retention.ts                ← Anthropic cache-retention headers
├── anthropic-family-tool-payload-compat.ts     ← Anthropic tool call format bridge
├── bedrock-stream-wrappers.ts                  ← AWS Bedrock + IAM SigV4 adapter
├── moonshot-stream-wrappers.ts                 ← Moonshot adapter
├── minimax-stream-wrappers.ts                  ← Minimax adapter
├── proxy-stream-wrappers.ts                    ← Generic proxy pass-through
└── zai-stream-wrappers.ts                      ← Z AI adapter