import { Effect, Schema } from "effect" import { Tool } from "@opencode-ai/schema/tool" import { Route } from "../route/client" import { Auth } from "../route/auth" import { Endpoint } from "../route/endpoint" import { Framing } from "../route/framing" import { Protocol } from "../route/protocol" import { AIError, LLMEvent, mergeJsonRecords, Usage, type CacheHint, type FinishReasonDetails, type FinishReason, type JsonSchema, type LLMRequest, type MediaPart, type ProviderOptions, type ProviderMetadata, type ToolCallPart, type ToolDefinition, type ToolResultPart, } from "../schema" import { JsonObject, optionalArray, optionalNull, ProviderShared } from "./shared" import { classifyProviderFailure } from "../provider-error" import * as Cache from "./utils/cache" import { Lifecycle } from "./utils/lifecycle" import { ToolSchemaProjection } from "./utils/tool-schema" import { ToolStream } from "./utils/tool-stream" const ADAPTER = "anthropic-messages" const MEDIA_MIMES = new Set([...ProviderShared.IMAGE_MIMES, ...ProviderShared.PDF_MIMES]) export const DEFAULT_BASE_URL = "https://api.anthropic.com/v1" export const PATH = "/messages" export type ThinkingInput = | { readonly type: "adaptive" readonly display?: "summarized" | "omitted" } | { readonly type: "disabled" } | ({ readonly type: "enabled" } & ( | { readonly budgetTokens: number; readonly budget_tokens?: number } | { readonly budgetTokens?: number; readonly budget_tokens: number } )) export interface OptionsInput { readonly [key: string]: unknown readonly thinking?: ThinkingInput readonly effort?: string } export type ProviderOptionsInput = ProviderOptions & { readonly anthropic?: OptionsInput } // ============================================================================= // Request Body Schema // ============================================================================= const AnthropicCacheControl = Schema.Struct({ type: Schema.tag("ephemeral"), ttl: Schema.optional(Schema.Literals(["5m", "1h"])), }) const AnthropicTextBlock = Schema.Struct({ type: Schema.tag("text"), text: Schema.String, cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicTextBlock = Schema.Schema.Type const AnthropicImageBlock = Schema.Struct({ type: Schema.tag("image"), source: Schema.Struct({ type: Schema.tag("base64"), media_type: Schema.String, data: Schema.String, }), cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicImageBlock = Schema.Schema.Type const AnthropicDocumentBlock = Schema.Struct({ type: Schema.tag("document"), source: Schema.Struct({ type: Schema.tag("base64"), media_type: Schema.Literal("application/pdf"), data: Schema.String, }), cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicDocumentBlock = Schema.Schema.Type const AnthropicThinkingBlock = Schema.Struct({ type: Schema.tag("thinking"), thinking: Schema.String, signature: Schema.optional(Schema.String), cache_control: Schema.optional(AnthropicCacheControl), }) // Safety-filtered thinking arrives as an opaque encrypted `data` payload with // no visible text. It must round-trip verbatim so multi-turn thinking + tool // use conversations keep their reasoning continuity. const AnthropicRedactedThinkingBlock = Schema.Struct({ type: Schema.tag("redacted_thinking"), data: Schema.String, cache_control: Schema.optional(AnthropicCacheControl), }) const AnthropicToolUseBlock = Schema.Struct({ type: Schema.tag("tool_use"), id: Schema.String, name: Schema.String, input: Schema.Unknown, cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicToolUseBlock = Schema.Schema.Type const AnthropicServerToolUseBlock = Schema.Struct({ type: Schema.tag("server_tool_use"), id: Schema.String, name: Schema.String, input: Schema.Unknown, cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicServerToolUseBlock = Schema.Schema.Type // Server tool result blocks: web_search_tool_result, code_execution_tool_result, // and web_fetch_tool_result. The provider executes the tool and inlines the // structured result into the assistant turn — there is no client tool_result // round-trip. We round-trip the structured `content` payload as opaque JSON so // the next request can echo it back when continuing the conversation. const AnthropicServerToolResultType = Schema.Literals([ "web_search_tool_result", "code_execution_tool_result", "web_fetch_tool_result", ]) type AnthropicServerToolResultType = Schema.Schema.Type const AnthropicServerToolResultBlock = Schema.Struct({ type: AnthropicServerToolResultType, tool_use_id: Schema.String, content: Schema.Unknown, cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicServerToolResultBlock = Schema.Schema.Type // Anthropic accepts either a plain string or an ordered array of text, image, and // document blocks inside `tool_result.content`. The array form keeps media as native // model input instead of JSON-stringifying base64 into prompt text. const AnthropicToolResultContent = Schema.Union([AnthropicTextBlock, AnthropicImageBlock, AnthropicDocumentBlock]) const AnthropicToolResultBlock = Schema.Struct({ type: Schema.tag("tool_result"), tool_use_id: Schema.String, content: Schema.Union([Schema.String, Schema.Array(AnthropicToolResultContent)]), is_error: Schema.optional(Schema.Boolean), cache_control: Schema.optional(AnthropicCacheControl), }) const AnthropicUserBlock = Schema.Union([ AnthropicTextBlock, AnthropicImageBlock, AnthropicDocumentBlock, AnthropicToolResultBlock, ]) type AnthropicUserBlock = Schema.Schema.Type const AnthropicAssistantBlock = Schema.Union([ AnthropicTextBlock, AnthropicThinkingBlock, AnthropicRedactedThinkingBlock, AnthropicToolUseBlock, AnthropicServerToolUseBlock, AnthropicServerToolResultBlock, ]) type AnthropicAssistantBlock = Schema.Schema.Type type AnthropicToolResultBlock = Schema.Schema.Type const AnthropicMessage = Schema.Union([ Schema.Struct({ role: Schema.Literal("user"), content: Schema.Array(AnthropicUserBlock) }), Schema.Struct({ role: Schema.Literal("assistant"), content: Schema.Array(AnthropicAssistantBlock) }), Schema.Struct({ role: Schema.Literal("system"), content: Schema.Array(AnthropicTextBlock) }), ]).pipe(Schema.toTaggedUnion("role")) type AnthropicMessage = Schema.Schema.Type const AnthropicTool = Schema.Struct({ name: Schema.String, description: Schema.String, input_schema: JsonObject, cache_control: Schema.optional(AnthropicCacheControl), }) type AnthropicTool = Schema.Schema.Type const AnthropicToolChoice = Schema.Union([ Schema.Struct({ type: Schema.Literals(["auto", "any", "none"]) }), Schema.Struct({ type: Schema.tag("tool"), name: Schema.String }), ]) const AnthropicThinking = Schema.Union([ Schema.Struct({ type: Schema.tag("enabled"), budget_tokens: Schema.Number, }), Schema.Struct({ type: Schema.tag("adaptive"), display: Schema.optional(Schema.Literals(["summarized", "omitted"])), }), Schema.Struct({ type: Schema.tag("disabled"), }), ]) const AnthropicOutputConfig = Schema.Struct({ effort: Schema.optional(Schema.String), }) const AnthropicBodyFields = { model: Schema.String, system: optionalArray(AnthropicTextBlock), messages: Schema.Array(AnthropicMessage), tools: optionalArray(AnthropicTool), tool_choice: Schema.optional(AnthropicToolChoice), stream: Schema.Literal(true), max_tokens: Schema.Number, temperature: Schema.optional(Schema.Number), top_p: Schema.optional(Schema.Number), top_k: Schema.optional(Schema.Number), stop_sequences: optionalArray(Schema.String), thinking: Schema.optional(AnthropicThinking), output_config: Schema.optional(AnthropicOutputConfig), } export const AnthropicMessagesBody = Schema.Struct(AnthropicBodyFields) export type AnthropicMessagesBody = Schema.Schema.Type const AnthropicUsage = Schema.StructWithRest( Schema.Struct({ input_tokens: Schema.optional(Schema.Number), output_tokens: Schema.optional(Schema.Number), cache_creation_input_tokens: optionalNull(Schema.Number), cache_read_input_tokens: optionalNull(Schema.Number), server_tool_use: optionalNull( Schema.StructWithRest(Schema.Struct({ web_search_requests: Schema.optional(Schema.Number) }), [ Schema.Record(Schema.String, Schema.Unknown), ]), ), output_tokens_details: optionalNull( Schema.StructWithRest(Schema.Struct({ thinking_tokens: Schema.optional(Schema.Number) }), [ Schema.Record(Schema.String, Schema.Unknown), ]), ), }), [Schema.Record(Schema.String, Schema.Unknown)], ) type AnthropicUsage = Schema.Schema.Type const AnthropicStreamBlock = Schema.Struct({ type: Schema.String, id: Schema.optional(Schema.String), name: Schema.optional(Schema.String), text: Schema.optional(Schema.String), thinking: Schema.optional(Schema.String), signature: Schema.optional(Schema.String), // redacted_thinking blocks arrive whole in content_block_start with the // encrypted payload in `data`; there is no streaming delta sequence. data: Schema.optional(Schema.String), input: Schema.optional(Schema.Unknown), // *_tool_result blocks arrive whole as content_block_start (no streaming // delta) with the structured payload in `content` and the originating // server_tool_use id in `tool_use_id`. tool_use_id: Schema.optional(Schema.String), content: Schema.optional(Schema.Unknown), }) const AnthropicStreamDelta = Schema.Struct({ type: Schema.optional(Schema.String), text: Schema.optional(Schema.String), thinking: Schema.optional(Schema.String), partial_json: Schema.optional(Schema.String), signature: Schema.optional(Schema.String), stop_reason: optionalNull(Schema.String), stop_sequence: optionalNull(Schema.String), }) const AnthropicEvent = Schema.Struct({ type: Schema.String, index: Schema.optional(Schema.Number), message: Schema.optional(Schema.Struct({ usage: Schema.optional(AnthropicUsage) })), content_block: Schema.optional(AnthropicStreamBlock), delta: Schema.optional(AnthropicStreamDelta), usage: Schema.optional(AnthropicUsage), // `type` and `message` are both required per Anthropic's spec, but // OpenAI-compatible proxies and gateway translations occasionally drop one // or the other; mark them optional so a partial payload still parses and // the parser can fall back to whichever field is populated. error: Schema.optional( Schema.Struct({ type: Schema.optional(Schema.String), message: Schema.optional(Schema.String) }), ), }) type AnthropicEvent = Schema.Schema.Type interface ParserState { readonly tools: ToolStream.State readonly reasoningSignatures: Readonly> readonly usage?: Usage readonly pendingFinish?: { readonly reason: FinishReasonDetails readonly providerMetadata?: ProviderMetadata } readonly lifecycle: Lifecycle.State } const invalid = ProviderShared.invalidRequest // ============================================================================= // Request Lowering // ============================================================================= // Anthropic accepts at most 4 explicit cache_control breakpoints per request, // across `tools`, `system`, and `messages`. Beyond the cap the API returns a // 400 — so the lowering layer counts emitted markers and silently drops any // that exceed it. const ANTHROPIC_BREAKPOINT_CAP = 4 const EPHEMERAL_5M = { type: "ephemeral" as const } const EPHEMERAL_1H = { type: "ephemeral" as const, ttl: "1h" as const } const cacheControl = (breakpoints: Cache.Breakpoints, cache: CacheHint | undefined) => { if (cache?.type !== "ephemeral" && cache?.type !== "persistent") return undefined if (breakpoints.remaining <= 0) { breakpoints.dropped += 1 return undefined } breakpoints.remaining -= 1 return Cache.ttlBucket(cache.ttlSeconds) === "1h" ? EPHEMERAL_1H : EPHEMERAL_5M } const anthropicMetadata = (metadata: Record): ProviderMetadata => ({ anthropic: metadata }) const signatureFromMetadata = (metadata: ProviderMetadata | undefined): string | undefined => { const anthropic = metadata?.anthropic if (!ProviderShared.isRecord(anthropic)) return undefined return typeof anthropic.signature === "string" ? anthropic.signature : undefined } const redactedDataFromMetadata = (metadata: ProviderMetadata | undefined): string | undefined => { const anthropic = metadata?.anthropic if (!ProviderShared.isRecord(anthropic)) return undefined return typeof anthropic.redactedData === "string" ? anthropic.redactedData : undefined } const lowerTool = (breakpoints: Cache.Breakpoints, tool: ToolDefinition, inputSchema: JsonSchema): AnthropicTool => ({ name: tool.name, description: tool.description, input_schema: inputSchema, cache_control: cacheControl(breakpoints, tool.cache), }) const lowerToolChoice = (toolChoice: NonNullable) => ProviderShared.matchToolChoice("Anthropic Messages", toolChoice, { auto: () => ({ type: "auto" as const }), none: () => ({ type: "none" as const }), required: () => ({ type: "any" as const }), tool: (name) => ({ type: "tool" as const, name }), }) const lowerToolCall = (part: ToolCallPart): AnthropicToolUseBlock => ({ type: "tool_use", id: part.id, name: part.name, input: part.input, }) const lowerServerToolCall = (part: ToolCallPart): AnthropicServerToolUseBlock => ({ type: "server_tool_use", id: part.id, name: part.name, input: part.input, }) // Server tool result blocks are typed by name. Anthropic ships three today; // extend this list when new server tools land. The block content is the // structured payload returned by the provider, which we round-trip as-is. const serverToolResultType = (name: string): AnthropicServerToolResultType | undefined => { if (name === "web_search") return "web_search_tool_result" if (name === "code_execution") return "code_execution_tool_result" if (name === "web_fetch") return "web_fetch_tool_result" return undefined } const lowerServerToolResult = Effect.fn("AnthropicMessages.lowerServerToolResult")(function* (part: ToolResultPart) { const wireType = serverToolResultType(part.name) if (!wireType) return yield* invalid(`Anthropic Messages does not know how to round-trip server tool result for ${part.name}`) // Prefer the provider-owned replay payload; fall back to the result value for // histories constructed directly from provider events. const payload = part.providerMetadata?.anthropic?.["result"] ?? part.result.value return { type: wireType, tool_use_id: part.id, content: payload } satisfies AnthropicServerToolResultBlock }) const lowerMedia = Effect.fn("AnthropicMessages.lowerMedia")(function* (part: MediaPart) { const media = yield* ProviderShared.validateMedia("Anthropic Messages", part, MEDIA_MIMES) if (media.mime === "application/pdf") return { type: "document" as const, source: { type: "base64" as const, media_type: "application/pdf" as const, data: media.base64, }, } satisfies AnthropicDocumentBlock return { type: "image" as const, source: { type: "base64" as const, media_type: media.mime, data: media.base64, }, } satisfies AnthropicImageBlock }) // Tool results may carry structured text, images, and documents. Keep media as provider-native // content instead of JSON-stringifying base64 into a prompt string. const lowerToolResultContentItem = Effect.fn("AnthropicMessages.lowerToolResultContentItem")(function* ( item: Tool.Content, ) { if (item.type === "text") return { type: "text" as const, text: item.text } satisfies AnthropicTextBlock return yield* lowerMedia({ type: "media", mediaType: item.mime, data: item.uri, filename: item.name }) }) const lowerToolResultContent = Effect.fn("AnthropicMessages.lowerToolResultContent")(function* (part: ToolResultPart) { // Text / json / error results stay as a string for backward compatibility // with existing cassettes and provider expectations. if (part.result.type !== "content") return ProviderShared.toolResultText(part) // Preserve the narrowed array element type when compiled through a consumer package. const content: ReadonlyArray = part.result.value return yield* Effect.forEach(content, lowerToolResultContentItem) }) // Mid-conversation system messages are a native Claude API feature only for // Opus 4.8. Other Anthropic models intentionally use the same visible wrapped- // user fallback as non-Anthropic routes rather than sending a role they reject. const supportsNativeSystemUpdates = (request: LLMRequest) => String(request.model.id) === "claude-opus-4-8" const endsInServerToolUse = (message: LLMRequest["messages"][number]) => { const last = message.content.at(-1) return message.role === "assistant" && last?.type === "tool-call" && last.providerExecuted === true } const canUseNativeSystemUpdate = (messages: LLMRequest["messages"], index: number) => { const previous = messages[index - 1] const next = messages[index + 1] return ( previous !== undefined && previous.role !== "system" && (previous.role === "user" || previous.role === "tool" || endsInServerToolUse(previous)) && next?.role !== "system" && (next === undefined || next.role === "assistant") ) } const splitsLocalToolResults = (messages: LLMRequest["messages"], index: number) => { const pending = new Set() for (const message of messages.slice(0, index)) { for (const part of message.content) { if (message.role === "assistant" && part.type === "tool-call" && part.providerExecuted !== true) pending.add(part.id) if (message.role === "tool" && part.type === "tool-result") pending.delete(part.id) } } return pending.size > 0 } const lowerNativeSystemUpdate = Effect.fn("AnthropicMessages.lowerNativeSystemUpdate")(function* ( message: LLMRequest["messages"][number], breakpoints: Cache.Breakpoints, ) { const content = yield* ProviderShared.systemUpdateText("Anthropic Messages", message) return { role: "system" as const, content: content.map((part) => ({ type: "text" as const, text: part.text, cache_control: cacheControl(breakpoints, part.cache), })), } }) const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* ( request: LLMRequest, breakpoints: Cache.Breakpoints, ) { const messages: AnthropicMessage[] = [] for (const [index, message] of request.messages.entries()) { if (message.role === "system") { if (splitsLocalToolResults(request.messages, index)) return yield* invalid("Anthropic Messages system updates cannot split a local tool call from its tool result") if (supportsNativeSystemUpdates(request) && canUseNativeSystemUpdate(request.messages, index)) { messages.push(yield* lowerNativeSystemUpdate(message, breakpoints)) continue } const part = yield* ProviderShared.wrappedSystemUpdate("Anthropic Messages", message) const block = { type: "text" as const, text: part.text, cache_control: cacheControl(breakpoints, part.cache) } const previous = messages.at(-1) if (previous?.role === "user") messages[messages.length - 1] = { role: "user", content: [...previous.content, block] } else messages.push({ role: "user", content: [block] }) continue } if (message.role === "user") { const content: AnthropicUserBlock[] = [] for (const part of message.content) { if (part.type === "text") { content.push({ type: "text", text: part.text, cache_control: cacheControl(breakpoints, part.cache) }) continue } if (part.type === "media") { content.push(yield* lowerMedia(part)) continue } return yield* ProviderShared.unsupportedContent("Anthropic Messages", "user", ["text", "media"]) } messages.push({ role: "user", content }) continue } if (message.role === "assistant") { const content: AnthropicAssistantBlock[] = [] for (const part of message.content) { if (part.type === "text") { content.push({ type: "text", text: part.text, cache_control: cacheControl(breakpoints, part.cache) }) continue } if (part.type === "reasoning") { // Mirrors Vercel's @ai-sdk/anthropic: a signature marks visible // thinking; only signature-less parts carrying redactedData // round-trip as opaque redacted_thinking blocks. const signature = part.encrypted ?? signatureFromMetadata(part.providerMetadata) const redactedData = redactedDataFromMetadata(part.providerMetadata) if (signature === undefined && redactedData !== undefined) { content.push({ type: "redacted_thinking", data: redactedData }) continue } content.push({ type: "thinking", thinking: part.text, signature }) continue } if (part.type === "tool-call") { content.push(part.providerExecuted ? lowerServerToolCall(part) : lowerToolCall(part)) continue } if (part.type === "tool-result" && part.providerExecuted) { content.push(yield* lowerServerToolResult(part)) continue } return yield* invalid( `Anthropic Messages assistant messages only support text, reasoning, and tool-call content for now`, ) } messages.push({ role: "assistant", content }) continue } const content: AnthropicToolResultBlock[] = [] for (const part of message.content) { if (!ProviderShared.supportsContent(part, ["tool-result"])) return yield* ProviderShared.unsupportedContent("Anthropic Messages", "tool", ["tool-result"]) content.push({ type: "tool_result", tool_use_id: part.id, content: yield* lowerToolResultContent(part), is_error: part.result.type === "error" ? true : undefined, cache_control: cacheControl(breakpoints, part.cache), }) } messages.push({ role: "user", content }) } return messages }) const resolveOptions = Effect.fn("AnthropicMessages.resolveOptions")(function* (request: LLMRequest) { const input = request.providerOptions?.anthropic return { thinking: yield* resolveThinking(input?.thinking), effort: typeof input?.effort === "string" ? input.effort : undefined, } }) const resolveThinking = Effect.fn("AnthropicMessages.resolveThinking")(function* (input: unknown) { if (!ProviderShared.isRecord(input)) return undefined if (input.type === "adaptive") { const display = input.display === "summarized" ? ("summarized" as const) : input.display === "omitted" ? ("omitted" as const) : undefined return { type: "adaptive" as const, ...(display === undefined ? {} : { display }) } } if (input.type === "disabled") return { type: "disabled" as const } if (input.type !== "enabled") return undefined const budget = typeof input.budgetTokens === "number" ? input.budgetTokens : typeof input.budget_tokens === "number" ? input.budget_tokens : undefined if (budget === undefined) return yield* ProviderShared.invalidRequest("Anthropic thinking provider option requires budgetTokens") return { type: "enabled" as const, budget_tokens: budget } }) const fromRequest = Effect.fn("AnthropicMessages.fromRequest")(function* (request: LLMRequest) { const generation = request.generation const toolSchemaCompatibility = request.model.compatibility?.toolSchema const outputLimit = request.model.defaults?.limits?.output ?? request.model.route.defaults.limits?.output ?? 4096 // Allocate the 4-breakpoint budget in invalidation order: tools → system → // messages. Tools live highest in the cache hierarchy, so when callers // over-mark we keep their tool hints and shed the message-tail ones first. const breakpoints = Cache.newBreakpoints(ANTHROPIC_BREAKPOINT_CAP) const tools = request.tools.length === 0 ? undefined : request.tools.map((tool) => lowerTool( breakpoints, tool, ToolSchemaProjection.modelCompatibility(tool.inputSchema, toolSchemaCompatibility), ), ) // Anthropic rejects tool_choice when tools are absent; "none" is only meaningful with tools present. const toolChoice = tools === undefined || !request.toolChoice ? undefined : yield* lowerToolChoice(request.toolChoice) const system = request.system.length === 0 ? undefined : request.system.map((part) => ({ type: "text" as const, text: part.text, cache_control: cacheControl(breakpoints, part.cache), })) const messages = yield* lowerMessages(request, breakpoints) if (breakpoints.dropped > 0) { yield* Effect.logWarning( `Anthropic Messages: dropped ${breakpoints.dropped} cache breakpoint(s); the API allows at most ${ANTHROPIC_BREAKPOINT_CAP} per request.`, ) } const options = yield* resolveOptions(request) return { model: request.model.id, system, messages, tools, tool_choice: toolChoice, stream: true as const, max_tokens: generation?.maxTokens ?? outputLimit, temperature: generation?.temperature, top_p: generation?.topP, top_k: generation?.topK, stop_sequences: generation?.stop, thinking: options.thinking, output_config: options.effort === undefined ? undefined : { effort: options.effort }, } }) // ============================================================================= // Stream Parsing // ============================================================================= const mapFinishReason = (reason: string | null | undefined): FinishReason => { if (reason === "end_turn" || reason === "stop_sequence" || reason === "pause_turn") return "stop" if (reason === "max_tokens" || reason === "model_context_window_exceeded") return "length" if (reason === "tool_use") return "tool-calls" if (reason === "refusal") return "content-filter" return "unknown" } // Anthropic reports the non-overlapping breakdown natively — its // `input_tokens` is the *non-cached* count per the Messages API docs, with // cache reads and writes as separate fields. We sum them to derive the // inclusive `inputTokens` the rest of the contract expects. Extended // thinking tokens are included in `output_tokens`; newer responses also // expose that subset through `output_tokens_details.thinking_tokens`. const mapUsage = (usage: AnthropicUsage | undefined): Usage | undefined => { if (!usage) return undefined const nonCached = usage.input_tokens const cacheRead = usage.cache_read_input_tokens ?? undefined const cacheWrite = usage.cache_creation_input_tokens ?? undefined const inputTokens = ProviderShared.sumTokens(nonCached, cacheRead, cacheWrite) return new Usage({ inputTokens, outputTokens: usage.output_tokens, nonCachedInputTokens: nonCached, cacheReadInputTokens: cacheRead, cacheWriteInputTokens: cacheWrite, reasoningTokens: usage.output_tokens_details?.thinking_tokens, totalTokens: ProviderShared.totalTokens(inputTokens, usage.output_tokens, undefined), providerMetadata: { anthropic: usage }, }) } // Anthropic emits usage on `message_start` and again on `message_delta` — the // final delta carries the authoritative totals. Right-biased merge: each // field prefers `right` when defined, falls back to `left`. `inputTokens` is // recomputed from the merged breakdown so the inclusive total stays // consistent with `nonCached + cacheRead + cacheWrite`. const mergeUsage = (left: Usage | undefined, right: Usage | undefined) => { if (!left) return right if (!right) return left const nonCachedInputTokens = right.nonCachedInputTokens ?? left.nonCachedInputTokens const cacheReadInputTokens = right.cacheReadInputTokens ?? left.cacheReadInputTokens const cacheWriteInputTokens = right.cacheWriteInputTokens ?? left.cacheWriteInputTokens const inputTokens = ProviderShared.sumTokens(nonCachedInputTokens, cacheReadInputTokens, cacheWriteInputTokens) const outputTokens = right.outputTokens ?? left.outputTokens const reasoningTokens = right.reasoningTokens ?? left.reasoningTokens return new Usage({ inputTokens, outputTokens, nonCachedInputTokens, cacheReadInputTokens, cacheWriteInputTokens, reasoningTokens, totalTokens: ProviderShared.totalTokens(inputTokens, outputTokens, undefined), providerMetadata: { anthropic: mergeJsonRecords(left.providerMetadata?.["anthropic"], right.providerMetadata?.["anthropic"]) ?? {}, }, }) } // Server tool result blocks come whole in `content_block_start` (no streaming // delta sequence). We convert the payload to a `tool-result` event with // `providerExecuted: true`. The runtime appends it to the assistant message // for round-trip; downstream consumers can inspect `result.value` for the // structured payload. const SERVER_TOOL_RESULT_NAMES: Record = { web_search_tool_result: "web_search", code_execution_tool_result: "code_execution", web_fetch_tool_result: "web_fetch", } const isServerToolResultType = (type: string): type is AnthropicServerToolResultType => type in SERVER_TOOL_RESULT_NAMES const serverToolResultEvent = (block: NonNullable): LLMEvent | undefined => { if (!block.type || !isServerToolResultType(block.type)) return undefined const errorPayload = typeof block.content === "object" && block.content !== null && "type" in block.content ? String((block.content as Record).type) : "" const isError = errorPayload.endsWith("_tool_result_error") return LLMEvent.toolResult({ id: block.tool_use_id ?? "", name: SERVER_TOOL_RESULT_NAMES[block.type], result: isError ? { type: "error", value: block.content } : { type: "json", value: block.content }, providerExecuted: true, // The complete payload is irreducible provider replay state: subsequent // stateless requests must round-trip the typed result block verbatim. providerMetadata: anthropicMetadata({ blockType: block.type, result: block.content }), }) } type StepResult = readonly [ParserState, ReadonlyArray] const NO_EVENTS: StepResult["1"] = [] const onMessageStart = (state: ParserState, event: AnthropicEvent): StepResult => { const usage = mapUsage(event.message?.usage) return [usage ? { ...state, usage: mergeUsage(state.usage, usage) } : state, NO_EVENTS] } const onContentBlockStart = (state: ParserState, event: AnthropicEvent): StepResult => { const block = event.content_block if (!block) return [state, NO_EVENTS] if ((block.type === "tool_use" || block.type === "server_tool_use") && event.index !== undefined) { const events: LLMEvent[] = [] const lifecycle = Lifecycle.stepStart(state.lifecycle, events) return [ { ...state, lifecycle, tools: ToolStream.start(state.tools, event.index, { id: block.id ?? String(event.index), name: block.name ?? "", input: block.input !== undefined && (!ProviderShared.isRecord(block.input) || Object.keys(block.input).length > 0) ? ProviderShared.encodeJson(block.input) : undefined, providerExecuted: block.type === "server_tool_use", }), }, [ ...events, LLMEvent.toolInputStart({ id: block.id ?? String(event.index), name: block.name ?? "", providerExecuted: block.type === "server_tool_use" ? true : undefined, }), ], ] } if (block.type === "text" && block.text !== undefined) { const events: LLMEvent[] = [] const id = `text-${event.index ?? 0}` const lifecycle = Lifecycle.textStart(state.lifecycle, events, id) return [ { ...state, lifecycle: block.text ? Lifecycle.textDelta(lifecycle, events, id, block.text) : lifecycle }, events, ] } if (block.type === "thinking" && block.thinking !== undefined) { const events: LLMEvent[] = [] const id = `reasoning-${event.index ?? 0}` const providerMetadata = block.signature === undefined ? undefined : anthropicMetadata({ signature: block.signature }) const lifecycle = Lifecycle.reasoningStart(state.lifecycle, events, id, providerMetadata) return [ { ...state, lifecycle: block.thinking ? Lifecycle.reasoningDelta(lifecycle, events, id, block.thinking, providerMetadata) : lifecycle, reasoningSignatures: event.index === undefined || block.signature === undefined ? state.reasoningSignatures : { ...state.reasoningSignatures, [event.index]: block.signature }, }, events, ] } // Redacted thinking surfaces as an empty reasoning part carrying the opaque // payload as `redactedData` metadata (same model as Vercel's // @ai-sdk/anthropic). The existing content_block_stop closes the part. if (block.type === "redacted_thinking" && block.data !== undefined) { const events: LLMEvent[] = [] return [ { ...state, lifecycle: Lifecycle.reasoningStart( state.lifecycle, events, `reasoning-${event.index ?? 0}`, anthropicMetadata({ redactedData: block.data }), ), }, events, ] } const result = serverToolResultEvent(block) if (!result) return [state, NO_EVENTS] const events: LLMEvent[] = [] return [{ ...state, lifecycle: Lifecycle.stepStart(state.lifecycle, events) }, [...events, result]] } const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(function* ( state: ParserState, event: AnthropicEvent, ) { const delta = event.delta if (delta?.type === "text_delta" && delta.text) { const events: LLMEvent[] = [] return [ { ...state, lifecycle: Lifecycle.textDelta(state.lifecycle, events, `text-${event.index ?? 0}`, delta.text) }, events, ] satisfies StepResult } if (delta?.type === "thinking_delta" && delta.thinking) { const events: LLMEvent[] = [] return [ { ...state, lifecycle: Lifecycle.reasoningDelta(state.lifecycle, events, `reasoning-${event.index ?? 0}`, delta.thinking), }, events, ] satisfies StepResult } if (delta?.type === "signature_delta" && delta.signature) { const index = event.index ?? 0 return [ { ...state, reasoningSignatures: { ...state.reasoningSignatures, [index]: delta.signature }, }, NO_EVENTS, ] satisfies StepResult } if (delta?.type === "input_json_delta" && event.index !== undefined) { if (!delta.partial_json) return [state, NO_EVENTS] satisfies StepResult const result = ToolStream.appendExisting( ADAPTER, state.tools, event.index, delta.partial_json, "Anthropic Messages tool argument delta is missing its tool call", ) if (ToolStream.isError(result)) return yield* result const events: LLMEvent[] = [] const lifecycle = result.events.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle events.push(...result.events) return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult } return [state, NO_EVENTS] satisfies StepResult }) const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(function* ( state: ParserState, event: AnthropicEvent, ) { if (event.index === undefined) return [state, NO_EVENTS] satisfies StepResult const result = yield* ToolStream.finish(ADAPTER, state.tools, event.index) const events: LLMEvent[] = [] const resultEvents = result.events ?? [] const signature = state.reasoningSignatures[event.index] const lifecycle = resultEvents.length ? Lifecycle.stepStart(state.lifecycle, events) : Lifecycle.reasoningEnd( Lifecycle.textEnd(state.lifecycle, events, `text-${event.index}`), events, `reasoning-${event.index}`, signature === undefined ? undefined : anthropicMetadata({ signature }), ) events.push(...resultEvents) const reasoningSignatures = { ...state.reasoningSignatures } delete reasoningSignatures[event.index] return [{ ...state, lifecycle, tools: result.tools, reasoningSignatures }, events] satisfies StepResult }) const onMessageDelta = (state: ParserState, event: AnthropicEvent): StepResult => { const usage = mergeUsage(state.usage, mapUsage(event.usage)) return [ { ...state, usage, pendingFinish: { reason: { normalized: mapFinishReason(event.delta?.stop_reason), raw: event.delta?.stop_reason ?? undefined, }, providerMetadata: event.delta?.stop_sequence === null || event.delta?.stop_sequence === undefined ? undefined : anthropicMetadata({ stopSequence: event.delta.stop_sequence }), }, }, NO_EVENTS, ] } const onMessageStop = (state: ParserState): StepResult => { const events: LLMEvent[] = [] const lifecycle = Lifecycle.finish(state.lifecycle, events, { reason: state.pendingFinish?.reason ?? { normalized: "unknown", raw: undefined, }, usage: state.usage, providerMetadata: state.pendingFinish?.providerMetadata, }) return [{ ...state, lifecycle }, events] } // Prefix `error.type` so overloads, rate limits, and quota errors are visible // even when the provider message is generic or empty. const providerErrorMessage = (event: AnthropicEvent): string => { const type = event.error?.type const message = event.error?.message if (type && message) return `${type}: ${message}` return message || type || "Anthropic Messages stream error" } const onError = (event: AnthropicEvent) => new AIError({ module: ADAPTER, method: "stream", reason: classifyProviderFailure({ message: providerErrorMessage(event), code: event.error?.type }), }) const step = (state: ParserState, event: AnthropicEvent) => { if (event.type === "message_start") return Effect.succeed(onMessageStart(state, event)) if (event.type === "content_block_start") return Effect.succeed(onContentBlockStart(state, event)) if (event.type === "content_block_delta") return onContentBlockDelta(state, event) if (event.type === "content_block_stop") return onContentBlockStop(state, event) if (event.type === "message_delta") return Effect.succeed(onMessageDelta(state, event)) if (event.type === "message_stop") return Effect.succeed(onMessageStop(state)) if (event.type === "error") return onError(event) return Effect.succeed([state, NO_EVENTS]) } // ============================================================================= // Protocol And Anthropic Route // ============================================================================= /** * The Anthropic Messages protocol — request body construction, body schema, * and the streaming-event state machine. Used by native Anthropic Cloud and * (once registered) Vertex Anthropic / Bedrock-hosted Anthropic passthrough. */ export const protocol = Protocol.make({ id: ADAPTER, body: { schema: AnthropicMessagesBody, from: fromRequest, }, stream: { event: Protocol.jsonEvent(AnthropicEvent), initial: () => ({ tools: ToolStream.empty(), reasoningSignatures: {}, lifecycle: Lifecycle.initial(), }), step, }, }) export const route = Route.make({ id: ADAPTER, provider: "anthropic", providerMetadataKey: "anthropic", protocol, endpoint: Endpoint.path(PATH, { baseURL: DEFAULT_BASE_URL }), auth: Auth.none, framing: Framing.sse, headers: () => ({ "anthropic-version": "2023-06-01" }), }) export * as AnthropicMessages from "./anthropic-messages"