源码
packages/ai/src/api/anthropic-messages.ts
// Adapted from Pi cd34e17ff039502f3664d8363b3c0a23f93a2ca3.
// Copyright (c) 2025 Mario Zechner. MIT: licenses/third-party-notices.txt.
import Anthropic from "@anthropic-ai/sdk";
import type {
BetaStopReason,
BetaThinkingDroppedInputTransformation,
BetaTool,
BetaCacheControlEphemeral as CacheControlEphemeral,
BetaContentBlockParam as ContentBlockParam,
MessageCreateParamsStreaming,
BetaMessageParam as MessageParam,
BetaRawMessageStreamEvent as RawMessageStreamEvent,
BetaRefusalStopDetails as RefusalStopDetails,
} from "@anthropic-ai/sdk/resources/beta/messages/messages.js";
import type {
Api,
AssistantMessage,
CacheRetention,
ImageContent,
JsonObject,
Message,
Model,
ProviderEnv,
ProviderHeaders,
SimpleStreamOptions,
StopReason,
StreamFunction,
StreamOptions,
TextContent,
ThinkingContent,
Tool,
ToolCall,
ToolResultMessage,
TranscriptContext,
} from "../types.ts";
import { measureContextBreakdown } from "../utils/context-breakdown.ts";
import { calculateCost } from "../utils/cost.ts";
import { appendAssistantMessageDiagnostic } from "../utils/diagnostics.ts";
import { formatProviderError, normalizeProviderError } from "../utils/error-body.ts";
import type { AssistantMessageEventStream } from "../utils/event-stream.ts";
import { headersToRecord } from "../utils/headers.ts";
import { parseJsonWithRepair } from "../utils/json-parse.ts";
import { projectContextForModelMediaPolicy } from "../utils/media-policy.ts";
import { MeasuredAssistantStream } from "../utils/model-performance.ts";
import { modelFailure } from "../utils/model-retry.ts";
import { assertProviderCacheControlLimit, finalizeProviderCacheControl } from "../utils/provider-cache.ts";
import { getProviderEnvValue } from "../utils/provider-env.ts";
import { retryProviderRequest } from "../utils/provider-retry.ts";
import { reasoningParameters } from "../utils/reasoning.ts";
import { redactProviderError } from "../utils/redact-error.ts";
import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts";
import { renderSystemMessageUpdate } from "../utils/text.ts";
import {
contentText,
getCurrentTools,
getDeclaredTools,
getInitialSystemMessage,
hasToolRedefinitions,
isJsonObject,
} from "../utils/transcript.ts";
import { getJsonSchemaToolParameters, resolveJsonSchemaStrictSampling } from "./constrained-sampling.ts";
import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./github-copilot-headers.ts";
import { adjustMaxTokensForThinking, buildBaseOptions, clampMaxTokensToContext } from "./simple-options.ts";
import { transformMessages } from "./transform-messages.ts";
/**
* Resolve cache retention preference.
* Defaults to "short" and uses PI_CACHE_RETENTION for backward compatibility.
*/
function resolveCacheRetention(cacheRetention?: CacheRetention, env?: ProviderEnv): CacheRetention {
if (cacheRetention) {
return cacheRetention;
}
if (getProviderEnvValue("PI_CACHE_RETENTION", env) === "long") {
return "long";
}
return "short";
}
function getCacheControl(
model: Model<"anthropic-messages">,
cacheRetention?: CacheRetention,
env?: ProviderEnv,
): { retention: CacheRetention; cacheControl?: CacheControlEphemeral } {
const retention = resolveCacheRetention(cacheRetention, env);
if (retention === "none") {
return { retention };
}
const ttl = retention === "long" && getAnthropicCompat(model).supportsLongCacheRetention ? "1h" : undefined;
return {
retention,
cacheControl: { type: "ephemeral", ...(ttl && { ttl }) },
};
}
// Stealth mode: Mimic Claude Code's tool naming exactly
const claudeCodeVersion = "2.1.251";
// Claude Code 2.x tool names (canonical casing)
// Source: https://cchistory.mariozechner.at/data/prompts-2.1.11.md
// To update: https://github.com/badlogic/cchistory
const claudeCodeTools = [
"Read",
"Write",
"Edit",
"Bash",
"Grep",
"Glob",
"AskUserQuestion",
"EnterPlanMode",
"ExitPlanMode",
"KillShell",
"NotebookEdit",
"Skill",
"Task",
"TaskOutput",
"TodoWrite",
"WebFetch",
"WebSearch",
];
const ccToolLookup = new Map(claudeCodeTools.map((t) => [t.toLowerCase(), t]));
// Convert tool name to CC canonical casing if it matches (case-insensitive)
const toClaudeCodeName = (name: string) => ccToolLookup.get(name.toLowerCase()) ?? name;
const fromClaudeCodeName = (name: string, tools?: Tool[]) => {
if (tools && tools.length > 0) {
const lowerName = name.toLowerCase();
const matchedTool = tools.find((tool) => tool.name.toLowerCase() === lowerName);
if (matchedTool) return matchedTool.name;
}
return name;
};
/**
* Convert content blocks to Anthropic API format
*/
function convertContentBlocks(content: (TextContent | ImageContent)[]):
| string
| Array<
| { type: "text"; text: string }
| {
type: "image";
source: {
type: "base64";
media_type: "image/jpeg" | "image/png" | "image/gif" | "image/webp";
data: string;
};
}
> {
// If only text blocks, return as concatenated string for simplicity
const hasImages = content.some((c) => c.type === "image");
if (!hasImages) {
return sanitizeSurrogates(content.map((c) => (c as TextContent).text).join("\n"));
}
// If we have images, convert to content block array
const blocks = content.map((block) => {
if (block.type === "text") {
return {
type: "text" as const,
text: sanitizeSurrogates(block.text),
};
}
return {
type: "image" as const,
source: {
type: "base64" as const,
media_type: block.mimeType as "image/jpeg" | "image/png" | "image/gif" | "image/webp",
data: block.data,
},
};
});
// If only images (no text), add placeholder text block
const hasText = blocks.some((b) => b.type === "text");
if (!hasText) {
blocks.unshift({
type: "text" as const,
text: "(see attached image)",
});
}
return blocks;
}
export type AnthropicEffort = "low" | "medium" | "high" | "xhigh" | "max";
export type AnthropicThinkingDisplay = "summarized" | "omitted";
const FINE_GRAINED_TOOL_STREAMING_BETA = "fine-grained-tool-streaming-2025-05-14";
const INTERLEAVED_THINKING_BETA = "interleaved-thinking-2025-05-14";
const SERVER_SIDE_FALLBACK_BETA = "server-side-fallback-2026-07-01";
const MID_CONVERSATION_OUTPUT_CONFIG_BETA = "mid-conversation-output-config-2026-07-01";
const THINKING_BINDING_CONTROLS_BETA = "thinking-binding-controls-2026-08-01";
const MID_CONVERSATION_TOOL_CHANGES_BETA = "mid-conversation-tool-changes-2026-07-01";
/**
* Stable deferred tool declared whenever native tool changes are in use. Anthropic adds
* hidden prompt scaffolding as soon as any tool has `defer_loading`; declaring this
* placeholder from the first request keeps that scaffolding in the cached prefix, so the
* first real late tool does not invalidate the cache (measured: full miss without it).
* It is never activated and the model cannot see it.
*/
const DEFERRED_TOOL_PLACEHOLDER: BetaTool = {
name: "__pi_deferred_placeholder__",
description: "Reserved placeholder. Never available. Never call this.",
input_schema: { type: "object", properties: {}, required: [] },
defer_loading: true,
};
function shouldUseServerSideFallbackBeta(model: Model<"anthropic-messages">): boolean {
return (model.compat?.allowedFallbackModels?.length ?? 0) > 0;
}
function getAnthropicCompat(model: Model<"anthropic-messages">) {
const isOpenRouter = model.provider === "openrouter" || model.baseUrl.includes("openrouter.ai");
return {
supportsEagerToolInputStreaming: model.compat?.supportsEagerToolInputStreaming ?? true,
supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true,
sendSessionAffinityHeaders: model.compat?.sendSessionAffinityHeaders ?? isOpenRouter,
sessionAffinityFormat: model.compat?.sessionAffinityFormat ?? (isOpenRouter ? "openrouter" : undefined),
supportsCacheControlOnTools: model.compat?.supportsCacheControlOnTools ?? true,
supportsTemperature: model.compat?.supportsTemperature ?? true,
allowEmptySignature: model.compat?.allowEmptySignature ?? false,
supportsStrictTools: model.compat?.supportsStrictTools ?? false,
supportsMidConvoSystemMessages: model.compat?.supportsMidConvoSystemMessages ?? false,
supportsMidConvoToolChanges: model.compat?.supportsMidConvoToolChanges ?? false,
};
}
export interface AnthropicOptions extends StreamOptions {
/**
* Enable extended thinking.
* For adaptive thinking models: the model decides when/how much to think.
* For older models: uses budget-based thinking with thinkingBudgetTokens.
* Default: undefined (thinking is omitted unless `streamSimple()` maps
* a simple reasoning level to this option, or callers set it explicitly).
*/
thinkingEnabled?: boolean;
/**
* Token budget for extended thinking (older models only).
* Ignored for adaptive thinking models.
* Default: 1024 when `thinkingEnabled` is true and no budget is provided.
*/
thinkingBudgetTokens?: number;
/**
* Effort level for adaptive thinking models.
* Controls how much thinking Claude allocates:
* - "max": Always thinks with no constraints (Opus 4.6 only)
* - "xhigh": Highest reasoning level (Opus 4.7+, Fable 5)
* - "high": Always thinks, deep reasoning
* - "medium": Moderate thinking, may skip for simple queries
* - "low": Minimal thinking, skips for simple tasks
* Ignored for older models.
* Default: omitted unless `streamSimple()` maps a simple reasoning
* level to this option.
*/
effort?: AnthropicEffort;
/**
* Controls how thinking content is returned in API responses.
* - "summarized": Thinking blocks contain summarized thinking text.
* - "omitted": Thinking blocks return an empty thinking field; the encrypted
* signature still travels back for multi-turn continuity. Use for faster
* time-to-first-text-token when your UI does not surface thinking.
*
* Note: Anthropic's API default for Claude Opus 4.7 and Claude Mythos Preview
* is "omitted". We default to "summarized" here to keep behavior consistent
* with older Claude 4 models. Set this explicitly to "omitted" to opt in.
* Default: "summarized" when thinking is enabled.
*/
thinkingDisplay?: AnthropicThinkingDisplay;
/**
* Whether to request the interleaved thinking beta header for non-adaptive
* thinking models. Adaptive thinking models have interleaved thinking built in,
* so the header is skipped for them regardless of this setting.
* Default: true.
*/
interleavedThinking?: boolean;
/**
* Anthropic tool choice behavior. String values map to Anthropic's built-in
* choices; `{ type: "tool", name }` forces a specific tool.
* Default: omitted (Anthropic default behavior, currently equivalent to auto).
*/
toolChoice?: "auto" | "any" | "none" | { type: "tool"; name: string };
/**
* Pre-built Anthropic client instance. When provided, skips internal client
* construction entirely. Use this to inject alternative SDK clients such as
* `AnthropicVertex` that shares the same messaging API.
*/
client?: Anthropic;
}
function mergeHeaders(...headerSources: (ProviderHeaders | undefined)[]): ProviderHeaders {
const merged: ProviderHeaders = {};
for (const headers of headerSources) {
if (headers) {
Object.assign(merged, headers);
}
}
return merged;
}
function mergeClientHeaders(...headerSources: (ProviderHeaders | undefined)[]): ProviderHeaders {
return mergeHeaders({ "User-Agent": "ZPI" }, ...headerSources);
}
function hasHeader(headers: ProviderHeaders | undefined, name: string): boolean {
if (!headers) return false;
const expected = name.toLowerCase();
for (const [key, value] of Object.entries(headers)) {
if (key.toLowerCase() === expected && value !== null && value.trim().length > 0) return true;
}
return false;
}
function assertRequestAuth(
provider: string,
apiKey: string | undefined,
headers: ProviderHeaders | undefined,
): void {
if (apiKey) return;
if (
hasHeader(headers, "authorization") ||
hasHeader(headers, "x-api-key") ||
hasHeader(headers, "cf-aig-authorization")
) {
return;
}
throw new Error(`No API key for provider: ${provider}`);
}
interface ServerSentEvent {
event: string | null;
data: string;
raw: string[];
}
interface SseDecoderState {
event: string | null;
data: string[];
raw: string[];
}
const ANTHROPIC_MESSAGE_EVENTS: ReadonlySet<string> = new Set([
"message_start",
"message_delta",
"message_stop",
"content_block_start",
"content_block_delta",
"content_block_stop",
]);
function flushSseEvent(state: SseDecoderState): ServerSentEvent | null {
if (state.data.length === 0) {
state.event = null;
state.raw = [];
return null;
}
const event: ServerSentEvent = {
event: state.event,
data: state.data.join("\n"),
raw: [...state.raw],
};
state.event = null;
state.data = [];
state.raw = [];
return event;
}
function decodeSseLine(line: string, state: SseDecoderState): ServerSentEvent | null {
if (line === "") {
return flushSseEvent(state);
}
state.raw.push(line);
if (line.startsWith(":")) {
return null;
}
const delimiterIndex = line.indexOf(":");
const fieldName = delimiterIndex === -1 ? line : line.slice(0, delimiterIndex);
let value = delimiterIndex === -1 ? "" : line.slice(delimiterIndex + 1);
if (value.startsWith(" ")) {
value = value.slice(1);
}
if (fieldName === "event") {
state.event = value;
} else if (fieldName === "data") {
state.data.push(value);
}
return null;
}
function nextLineBreakIndex(text: string): number {
const carriageReturnIndex = text.indexOf("\r");
const newlineIndex = text.indexOf("\n");
if (carriageReturnIndex === -1) {
return newlineIndex;
}
if (newlineIndex === -1) {
return carriageReturnIndex;
}
return Math.min(carriageReturnIndex, newlineIndex);
}
function consumeLine(text: string, final = false): { line: string; rest: string } | null {
const lineBreakIndex = nextLineBreakIndex(text);
if (lineBreakIndex === -1) {
return null;
}
if (!final && text[lineBreakIndex] === "\r" && lineBreakIndex === text.length - 1) return null;
let nextIndex = lineBreakIndex + 1;
if (text[lineBreakIndex] === "\r" && text[nextIndex] === "\n") {
nextIndex += 1;
}
return {
line: text.slice(0, lineBreakIndex),
rest: text.slice(nextIndex),
};
}
async function* iterateSseMessages(
body: ReadableStream<Uint8Array>,
signal?: AbortSignal,
): AsyncGenerator<ServerSentEvent> {
const reader = body.getReader();
const decoder = new TextDecoder();
const state: SseDecoderState = { event: null, data: [], raw: [] };
let buffer = "";
let finished = false;
const cancel = () => reader.cancel().catch(() => {});
signal?.addEventListener("abort", cancel, { once: true });
try {
while (true) {
if (signal?.aborted) {
throw new Error("Request was aborted");
}
const { value, done } = await reader.read();
if (done) {
finished = true;
break;
}
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);
}
}
buffer += decoder.decode();
let consumed = consumeLine(buffer, true);
while (consumed) {
buffer = consumed.rest;
const event = decodeSseLine(consumed.line, state);
if (event) {
yield event;
}
consumed = consumeLine(buffer, true);
}
if (buffer.length > 0) {
const event = decodeSseLine(buffer, state);
if (event) {
yield event;
}
}
const trailingEvent = flushSseEvent(state);
if (trailingEvent) {
yield trailingEvent;
}
} finally {
signal?.removeEventListener("abort", cancel);
if (!finished) await cancel();
reader.releaseLock();
}
}
async function* iterateAnthropicEvents(
response: Response,
signal?: AbortSignal,
): AsyncGenerator<RawMessageStreamEvent> {
if (!response.body) {
throw new Error("Attempted to iterate over an Anthropic response with no body");
}
let sawMessageStart = false;
let sawMessageEnd = false;
for await (const sse of iterateSseMessages(response.body, signal)) {
if (sse.event === "error") {
throw new Error(sse.data);
}
if (!ANTHROPIC_MESSAGE_EVENTS.has(sse.event ?? "")) {
continue;
}
try {
const event = parseJsonWithRepair<RawMessageStreamEvent>(sse.data);
if (event.type === "message_start") {
sawMessageStart = true;
} else if (event.type === "message_stop") {
sawMessageEnd = true;
}
yield event;
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
throw new Error(
`Could not parse Anthropic SSE event ${sse.event}: ${message}; data=${sse.data}; raw=${sse.raw.join("\\n")}`,
);
}
}
if (sawMessageStart && !sawMessageEnd) {
throw new Error("Anthropic stream ended before message_stop");
}
}
export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
model: Model<"anthropic-messages">,
context: TranscriptContext,
options?: AnthropicOptions,
): AssistantMessageEventStream => {
const stream = new MeasuredAssistantStream(options);
let normalizedContext = context;
const currentTools = getCurrentTools(normalizedContext.messages);
(async () => {
const providerThinkingLevel = model.compat?.supportsMidConvoEffort
? (options?.effort ?? "high")
: undefined;
const output: AssistantMessage = {
role: "assistant",
content: [],
api: model.api as Api,
provider: model.provider,
model: model.id,
...(providerThinkingLevel === undefined ? {} : { providerThinkingLevel }),
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
usageAvailable: false,
timestamp: Date.now(),
};
let receivedStopReason = false;
const completeCalls = new Set<ToolCall>();
let invalidToolInput: unknown;
try {
normalizedContext = projectContextForModelMediaPolicy(normalizedContext, model, {
onProjection: options?.onMediaProjection,
});
let client: Anthropic;
let isOAuth: boolean;
let usageModel = model;
let inputTransformations: BetaThinkingDroppedInputTransformation[] | undefined;
if (options?.client) {
client = options.client;
isOAuth = false;
} else {
const apiKey = options?.apiKey;
assertRequestAuth(model.provider, apiKey, mergeHeaders(model.headers, options?.headers));
let copilotDynamicHeaders: Record<string, string> | undefined;
if (model.provider === "github-copilot") {
const hasImages = hasCopilotVisionInput(normalizedContext.messages);
copilotDynamicHeaders = buildCopilotDynamicHeaders({
messages: normalizedContext.messages,
hasImages,
});
}
const cacheRetention = resolveCacheRetention(options?.cacheRetention, options?.env);
const cacheSessionId = cacheRetention === "none" ? undefined : options?.sessionId;
const created = createClient(
model,
apiKey,
options?.headers,
options?.fetch,
copilotDynamicHeaders,
cacheSessionId,
);
client = created.client;
isOAuth = created.isOAuthToken;
}
let params = buildParams(model, normalizedContext, isOAuth, options);
const nextParams = await options?.onPayload?.(params, model);
if (nextParams !== undefined) {
params = { ...(nextParams as MessageCreateParamsStreaming), stream: true };
}
output.contextBreakdown = measureContextBreakdown(params as unknown as JsonObject);
const requestOptions = {
...(options?.signal ? { signal: options.signal } : {}),
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
maxRetries: 0,
};
const response = await retryProviderRequest(
() => {
stream.startRequest();
return client.beta.messages.create(params, requestOptions).asResponse();
},
{
maxRetries: options?.maxRetries ?? 2,
maxRetryDelayMs: options?.maxRetryDelayMs,
signal: options?.signal,
onRetry: options?.onRetry,
},
);
await options?.onResponse?.(
{ status: response.status, headers: headersToRecord(response.headers) },
model,
);
stream.push({ type: "start", partial: output });
type Block = (ThinkingContent | TextContent | (ToolCall & { partialJson: string })) & { index: number };
const blocks = output.content as Block[];
for await (const event of iterateAnthropicEvents(response, options?.signal)) {
stream.received();
if (event.type === "message_start") {
output.responseId = event.message.id;
output.usageAvailable = true;
output.cacheUsageAvailable = true;
const transformations = event.message.input_transformations;
if (Array.isArray(transformations)) inputTransformations = transformations;
const responseModel = event.message.model;
if (responseModel !== model.id) output.responseModel = responseModel;
const fallbackCost =
responseModel === model.id
? undefined
: model.compat?.allowedFallbackModels?.find(
(fallback) => fallback.provider === model.provider && fallback.model === responseModel,
)?.cost;
usageModel = fallbackCost ? { ...model, id: responseModel, cost: fallbackCost } : model;
// Capture initial token usage from message_start event
// This ensures we have input token counts even if the stream is aborted early
output.usage.input = event.message.usage.input_tokens || 0;
output.usage.output = event.message.usage.output_tokens || 0;
output.usage.cacheRead = event.message.usage.cache_read_input_tokens || 0;
output.usage.cacheWrite = event.message.usage.cache_creation_input_tokens || 0;
output.usage.cacheWrite1h = event.message.usage.cache_creation?.ephemeral_1h_input_tokens || 0;
// Anthropic doesn't provide total_tokens, compute from components
output.usage.totalTokens =
output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
calculateCost(usageModel, output.usage);
} else if (event.type === "content_block_start") {
if (event.content_block.type === "fallback") {
if (output.content.length > 0) {
throw new Error("Anthropic performed an unsupported mid-output model fallback");
}
continue;
}
if (event.content_block.type === "text") {
const block: Block = {
type: "text",
text: event.content_block.text ?? "",
index: event.index,
};
output.content.push(block);
stream.push({ type: "text_start", contentIndex: output.content.length - 1, partial: output });
} else if (event.content_block.type === "thinking") {
const block: Block = {
type: "thinking",
thinking: event.content_block.thinking ?? "",
thinkingSignature: event.content_block.signature ?? "",
index: event.index,
};
output.content.push(block);
stream.push({ type: "thinking_start", contentIndex: output.content.length - 1, partial: output });
} else if (event.content_block.type === "redacted_thinking") {
const block: Block = {
type: "thinking",
thinking: "[Reasoning redacted]",
thinkingSignature: event.content_block.data,
redacted: true,
index: event.index,
};
output.content.push(block);
stream.push({ type: "thinking_start", contentIndex: output.content.length - 1, partial: output });
} else if (event.content_block.type === "tool_use") {
const block: Block = {
type: "toolCall",
id: event.content_block.id,
name: isOAuth
? fromClaudeCodeName(event.content_block.name, currentTools)
: event.content_block.name,
arguments: (event.content_block.input as JsonObject) ?? {},
partialJson: "",
index: event.index,
};
output.content.push(block);
stream.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });
}
} else if (event.type === "content_block_delta") {
if (event.delta.type === "text_delta") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block && block.type === "text") {
block.text += event.delta.text;
stream.push({
type: "text_delta",
contentIndex: index,
delta: event.delta.text,
partial: output,
});
}
} else if (event.delta.type === "thinking_delta") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block && block.type === "thinking") {
block.thinking += event.delta.thinking;
stream.push({
type: "thinking_delta",
contentIndex: index,
delta: event.delta.thinking,
partial: output,
});
}
} else if (event.delta.type === "input_json_delta") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block && block.type === "toolCall") {
block.partialJson += event.delta.partial_json;
// Previews consume deltas. Re-parsing the growing input on every token
// makes large write/edit calls quadratic; only closed calls need arguments.
stream.push({
type: "toolcall_delta",
contentIndex: index,
delta: event.delta.partial_json,
partial: output,
});
}
} else if (event.delta.type === "signature_delta") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block && block.type === "thinking") {
block.thinkingSignature = block.thinkingSignature || "";
block.thinkingSignature += event.delta.signature;
}
}
} else if (event.type === "content_block_stop") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block) {
delete (block as { index?: number }).index;
if (block.type === "text") {
stream.push({
type: "text_end",
contentIndex: index,
content: block.text,
partial: output,
});
} else if (block.type === "thinking") {
stream.push({
type: "thinking_end",
contentIndex: index,
content: block.thinking,
partial: output,
});
} else if (block.type === "toolCall") {
try {
const args: unknown = block.partialJson
? parseJsonWithRepair(block.partialJson)
: block.arguments;
if (!isJsonObject(args)) throw new Error("Incomplete tool call");
block.arguments = args;
} catch (error) {
// Never repair a cutoff into executable input. The later stop reason distinguishes length.
invalidToolInput ??= error;
continue;
} finally {
delete (block as { partialJson?: string }).partialJson;
}
completeCalls.add(block);
stream.push({
type: "toolcall_end",
contentIndex: index,
toolCall: block,
partial: output,
});
}
}
} else if (event.type === "message_delta") {
const transformations = event.input_transformations;
if (Array.isArray(transformations)) inputTransformations = transformations;
if (event.delta.stop_reason) {
receivedStopReason = true;
output.rawStopReason = event.delta.stop_reason;
const stopReasonResult = mapStopReason(event.delta.stop_reason, event.delta.stop_details);
output.stopReason = stopReasonResult.stopReason;
if (stopReasonResult.errorMessage) {
output.errorMessage = stopReasonResult.errorMessage;
}
}
// Only update usage fields if present (not null).
// Preserves input_tokens from message_start when proxies omit it in message_delta.
if (event.usage) {
if (event.usage.input_tokens != null) {
output.usage.input = event.usage.input_tokens;
}
if (event.usage.output_tokens != null) {
output.usage.output = event.usage.output_tokens;
stream.recordUsage(event.usage.output_tokens);
}
if (event.usage.cache_read_input_tokens != null) {
output.usage.cacheRead = event.usage.cache_read_input_tokens;
}
if (event.usage.cache_creation_input_tokens != null) {
output.usage.cacheWrite = event.usage.cache_creation_input_tokens;
}
const cacheCreation = (
event.usage as { cache_creation?: { ephemeral_1h_input_tokens?: number | null } }
).cache_creation;
if (cacheCreation?.ephemeral_1h_input_tokens != null) {
output.usage.cacheWrite1h = cacheCreation.ephemeral_1h_input_tokens;
}
// Anthropic reports reasoning tokens as a subset of output tokens.
const thinkingTokens = event.usage.output_tokens_details?.thinking_tokens;
if (thinkingTokens != null) {
output.usage.reasoning = thinkingTokens;
}
}
// Anthropic doesn't provide total_tokens, compute from components
output.usage.totalTokens =
output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
calculateCost(usageModel, output.usage);
}
}
if (options?.signal?.aborted) {
throw new Error("Request was aborted");
}
if (!receivedStopReason) {
throw new Error("Anthropic stream ended without a stop reason");
}
if (invalidToolInput && output.stopReason !== "length") throw invalidToolInput;
output.content = output.content.filter(
(block) => block.type !== "toolCall" || completeCalls.has(block),
);
if (output.stopReason === "aborted" || output.stopReason === "error") {
throw new Error(output.errorMessage || "An unknown error occurred");
}
if (inputTransformations && inputTransformations.length > 0) {
appendAssistantMessageDiagnostic(output, {
type: "anthropic_input_transformations",
timestamp: Date.now(),
details: {
transformations: inputTransformations.map((transformation) => ({
type: transformation.type ?? undefined,
path: transformation.path ?? undefined,
reason: transformation.reason ?? undefined,
})),
},
});
}
stream.push({ type: "done", reason: output.stopReason, message: output });
stream.end();
} catch (error) {
output.content = output.content.filter(
(block) => block.type !== "toolCall" || completeCalls.has(block),
);
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 = redactProviderError(formatProviderError(normalizeProviderError(error)), [
options?.apiKey,
...Object.values(model.headers ?? {}),
...Object.values(options?.headers ?? {}),
]);
output.errorDetails = modelFailure(error, options?.signal);
stream.push({ type: "error", reason: output.stopReason, error: output });
stream.end();
}
})();
return stream;
};
/**
* Map ThinkingLevel to Anthropic effort levels for adaptive thinking.
* Note: effort "max" is available on all adaptive-thinking Claude models, while native
* "xhigh" is only available on Opus 4.7/4.8, Sonnet 5, and Fable 5.
*/
function mapThinkingLevelToEffort(
model: Model<"anthropic-messages">,
level: SimpleStreamOptions["reasoning"],
): AnthropicEffort {
const mapped = level ? model.thinkingLevelMap?.[level] : undefined;
if (typeof mapped === "string") return mapped as AnthropicEffort;
switch (level) {
case "minimal":
case "low":
return "low";
case "medium":
return "medium";
case "high":
return "high";
default:
return "high";
}
}
export const streamSimple: StreamFunction<"anthropic-messages", SimpleStreamOptions> = (
model: Model<"anthropic-messages">,
context: TranscriptContext,
options?: SimpleStreamOptions,
): AssistantMessageEventStream => {
assertRequestAuth(model.provider, options?.apiKey, mergeHeaders(model.headers, options?.headers));
const base = {
...buildBaseOptions(model, context, options, options?.apiKey),
toolChoice: options?.toolChoice,
} satisfies AnthropicOptions;
if (
model.reasoningConfig ||
(options?.reasoning && isJsonObject(model.thinkingLevelMap?.[options.reasoning]))
) {
const mapped =
reasoningParameters(
model,
options?.reasoning ?? model.defaultThinkingLevel ?? model.reasoningConfig?.levels[0],
) ?? {};
const thinking = mapped.thinking as JsonObject | undefined;
const outputConfig = mapped.output_config as JsonObject | undefined;
return stream(model, context, {
...base,
thinkingEnabled: thinking?.type === "enabled" || thinking?.type === "adaptive",
thinkingBudgetTokens: typeof thinking?.budget_tokens === "number" ? thinking.budget_tokens : undefined,
effort: typeof outputConfig?.effort === "string" ? (outputConfig.effort as AnthropicEffort) : undefined,
onPayload: async (payload, requestModel) => {
const request = payload as MessageCreateParamsStreaming;
const next = { ...request, ...mapped };
if (thinking?.type === "enabled" && typeof thinking.budget_tokens === "number") {
const budget = boundedThinkingBudget(thinking.budget_tokens, request.max_tokens);
next.thinking =
budget === undefined
? { type: "disabled" }
: {
...thinking,
type: "enabled",
budget_tokens: budget,
};
}
return (await options?.onPayload?.(next, requestModel)) ?? next;
},
});
}
if (!options?.reasoning || options.reasoning === "off") {
return stream(model, context, {
...base,
thinkingEnabled: false,
} satisfies AnthropicOptions);
}
// For models with adaptive thinking: use an effort level.
// For older models: use budget-based thinking.
if (model.compat?.forceAdaptiveThinking === true) {
const effort = mapThinkingLevelToEffort(model, options.reasoning);
return stream(model, context, {
...base,
thinkingEnabled: true,
effort,
} satisfies AnthropicOptions);
}
// Undefined means the caller did not request an output cap; let the helper use the model cap.
// Do not coerce to 0 here, or the thinking budget would become the entire max_tokens value.
const adjusted = adjustMaxTokensForThinking(
base.maxTokens,
model.maxTokens,
options.reasoning,
options.thinkingBudgets,
);
const maxTokens = clampMaxTokensToContext(model, context, adjusted.maxTokens);
return stream(model, context, {
...base,
maxTokens,
thinkingEnabled: true,
thinkingBudgetTokens: Math.min(adjusted.thinkingBudget, Math.max(0, maxTokens - 1024)),
} satisfies AnthropicOptions);
};
function isOAuthToken(apiKey: string): boolean {
return apiKey.includes("sk-ant-oat");
}
function createClient(
model: Model<"anthropic-messages">,
apiKey: string | undefined,
optionsHeaders?: ProviderHeaders,
fetch?: typeof globalThis.fetch,
dynamicHeaders?: Record<string, string>,
sessionId?: string,
): { client: Anthropic; isOAuthToken: boolean } {
// Copilot: Bearer auth.
if (model.provider === "github-copilot") {
const client = new Anthropic({
apiKey: null,
authToken: apiKey ?? null,
baseURL: model.baseUrl,
dangerouslyAllowBrowser: true,
fetch,
defaultHeaders: mergeClientHeaders(
{
accept: "application/json",
"anthropic-dangerous-direct-browser-access": "true",
},
model.headers,
dynamicHeaders,
optionsHeaders,
),
});
return { client, isOAuthToken: false };
}
// OAuth: Bearer auth, Claude Code identity headers
if (apiKey && isOAuthToken(apiKey)) {
const client = new Anthropic({
apiKey: null,
authToken: apiKey,
baseURL: model.baseUrl,
dangerouslyAllowBrowser: true,
fetch,
defaultHeaders: mergeClientHeaders(
{
accept: "application/json",
"anthropic-dangerous-direct-browser-access": "true",
"user-agent": `claude-cli/${claudeCodeVersion}`,
"x-app": "cli",
},
model.headers,
optionsHeaders,
),
});
return { client, isOAuthToken: true };
}
// API key or header-owned auth.
const compat = getAnthropicCompat(model);
const sessionAffinityHeaders: ProviderHeaders = {};
if (sessionId && compat.sendSessionAffinityHeaders) {
const header = compat.sessionAffinityFormat === "openrouter" ? "x-session-id" : "x-session-affinity";
sessionAffinityHeaders[header] = sessionId;
}
const defaultHeaders = mergeClientHeaders(
{
accept: "application/json",
"anthropic-dangerous-direct-browser-access": "true",
},
sessionAffinityHeaders,
model.headers,
optionsHeaders,
);
const client = new Anthropic({
apiKey: apiKey ?? null,
authToken: null,
baseURL: model.baseUrl,
dangerouslyAllowBrowser: true,
fetch,
defaultHeaders,
});
return { client, isOAuthToken: false };
}
function getBetaFeatures(
model: Model<"anthropic-messages">,
context: TranscriptContext,
isOAuthToken: boolean,
nativeToolChanges: boolean,
options?: AnthropicOptions,
): NonNullable<MessageCreateParamsStreaming["betas"]> {
let configuredFeatures: string | null | undefined;
for (const headers of [model.headers, options?.headers]) {
for (const [name, value] of Object.entries(headers ?? {})) {
if (name.toLowerCase() === "anthropic-beta") configuredFeatures = value;
}
}
if (configuredFeatures === null) return [];
if (configuredFeatures !== undefined) {
return [
...new Set(
configuredFeatures
.split(",")
.map((feature) => feature.trim())
.filter((feature) => feature.length > 0),
),
];
}
const features: NonNullable<MessageCreateParamsStreaming["betas"]> = [];
if (isOAuthToken) features.push("claude-code-20250219", "oauth-2025-04-20");
if (shouldUseFineGrainedToolStreamingBeta(model, context)) features.push(FINE_GRAINED_TOOL_STREAMING_BETA);
if (
model.reasoning &&
options?.thinkingEnabled === true &&
(options.interleavedThinking ?? true) &&
model.compat?.forceAdaptiveThinking !== true
) {
features.push(INTERLEAVED_THINKING_BETA);
}
if (shouldUseServerSideFallbackBeta(model)) features.push(SERVER_SIDE_FALLBACK_BETA);
if (model.compat?.supportsMidConvoEffort === true) {
features.push(MID_CONVERSATION_OUTPUT_CONFIG_BETA, THINKING_BINDING_CONTROLS_BETA);
}
if (nativeToolChanges) features.push(MID_CONVERSATION_TOOL_CHANGES_BETA);
return [...new Set(features)];
}
function boundedThinkingBudget(budget: number, maxTokens: number): number | undefined {
const available = maxTokens - 1024;
return available < 1024 ? undefined : Math.max(1024, Math.min(budget, available));
}
function buildParams(
model: Model<"anthropic-messages">,
context: TranscriptContext,
isOAuthToken: boolean,
options?: AnthropicOptions,
): MessageCreateParamsStreaming {
const { cacheControl } = getCacheControl(model, options?.cacheRetention, options?.env);
const compat = getAnthropicCompat(model);
const initialSystemMessage = getInitialSystemMessage(context.messages);
const initialSystemText = initialSystemMessage ? renderSystemMessageUpdate(initialSystemMessage) : "";
const initialSystemSegments = initialSystemMessage
? [
contentText(initialSystemMessage.content),
renderSystemMessageUpdate({ ...initialSystemMessage, content: "" }),
]
.filter(Boolean)
.map((text) => ({
type: "text" as const,
text: sanitizeSurrogates(text),
...(cacheControl ? { cache_control: cacheControl } : {}),
}))
: [];
const transformedMessages = transformMessages(
finalizeProviderCacheControl(context.messages, { skipCacheWrite: options?.skipCacheWrite }),
model,
normalizeToolCallId,
);
const conversationMessages = initialSystemMessage ? transformedMessages.slice(1) : transformedMessages;
// Native tool changes reference tools by name, so a redefined name cannot be expressed,
// and Anthropic rejects a tool list where every tool is deferred, so there must be an
// initial active tool to anchor the deferred ones. Otherwise the current tool list is sent.
const initialTools = initialSystemMessage?.toolsAdded ?? [];
const nativeToolChanges =
compat.supportsMidConvoSystemMessages &&
compat.supportsMidConvoToolChanges &&
initialTools.length > 0 &&
!hasToolRedefinitions(context.messages);
const converted = convertMessages(
conversationMessages,
isOAuthToken,
cacheControl,
compat.allowEmptySignature,
model.compat?.supportsMidConvoEffort === true ? model.provider : undefined,
nativeToolChanges,
compat.supportsMidConvoSystemMessages,
);
const activeEffort = options?.effort ?? "high";
const betaFeatures = getBetaFeatures(model, context, isOAuthToken, nativeToolChanges, options);
const params: MessageCreateParamsStreaming = {
model: model.id,
messages:
model.compat?.supportsMidConvoEffort === true
? insertThinkingLevelMessages(converted, activeEffort)
: converted.messages,
max_tokens: options?.maxTokens ?? model.maxTokens,
stream: true,
...(betaFeatures.length > 0 ? { betas: betaFeatures } : {}),
};
// For OAuth tokens, we MUST include Claude Code identity
if (isOAuthToken) {
params.system = [
{
type: "text",
text: "You are Claude Code, Anthropic's official CLI for Claude.",
...(cacheControl && initialSystemSegments.length < 2 ? { cache_control: cacheControl } : {}),
},
];
if (initialSystemText) {
params.system.push(...initialSystemSegments);
}
} else if (initialSystemText) {
// Add cache control to system prompt for non-OAuth tokens
params.system = initialSystemSegments;
}
// Temperature is incompatible with extended thinking and unsupported on Claude Opus 4.7+.
if (
options?.temperature !== undefined &&
!options?.thinkingEnabled &&
model.compat?.supportsMidConvoEffort !== true &&
compat.supportsTemperature
) {
params.temperature = options.temperature;
}
const toolCacheControl = compat.supportsCacheControlOnTools ? cacheControl : undefined;
if (nativeToolChanges) {
// Initial tools stay active with the cache breakpoint on the last one. Every later
// declaration is deferred and only surfaced by its `tool_addition` block; removed
// tools stay declared and are withdrawn by `tool_removal`. The request-level list
// therefore only grows, keeping the cached prefix intact across tool changes.
const initialNames = new Set(initialTools.map((tool) => tool.name));
const laterTools = getDeclaredTools(context.messages).filter((tool) => !initialNames.has(tool.name));
params.tools = [
...convertTools(
initialTools,
isOAuthToken,
compat.supportsEagerToolInputStreaming,
compat.supportsStrictTools,
toolCacheControl,
),
DEFERRED_TOOL_PLACEHOLDER,
...convertTools(
laterTools,
isOAuthToken,
compat.supportsEagerToolInputStreaming,
compat.supportsStrictTools,
).map((tool) => ({ ...tool, defer_loading: true })),
];
} else {
const tools = getCurrentTools(context.messages);
if (tools.length > 0) {
params.tools = convertTools(
tools,
isOAuthToken,
compat.supportsEagerToolInputStreaming,
compat.supportsStrictTools,
toolCacheControl,
);
}
}
// Managed effort models always use adaptive thinking so prefix mismatches can
// be dropped instead of surfacing as persistent 400 responses.
if (model.compat?.supportsMidConvoEffort === true) {
params.thinking = {
type: "adaptive",
display: options?.thinkingDisplay ?? "summarized",
block_binding: { prefix_mismatch_behavior: "drop_block" },
};
params.output_config = { effort: "high" };
} else if (model.reasoning) {
if (options?.thinkingEnabled) {
// Default to "summarized" so Opus 4.7 and Mythos Preview behave like
// older Claude 4 models (whose API default is also "summarized").
const display: AnthropicThinkingDisplay = options.thinkingDisplay ?? "summarized";
if (model.compat?.forceAdaptiveThinking === true) {
// Adaptive thinking: Claude decides when and how much to think.
params.thinking = { type: "adaptive", display };
if (options.effort) {
params.output_config = { effort: options.effort };
}
} else {
// Budget-based thinking for older models
const budget = boundedThinkingBudget(options.thinkingBudgetTokens ?? 1024, params.max_tokens);
params.thinking =
budget === undefined
? { type: "disabled" }
: {
type: "enabled",
budget_tokens: budget,
display,
};
}
} else if (options?.thinkingEnabled === false && model.thinkingLevelMap?.off !== null) {
params.thinking = { type: "disabled" };
}
}
if (options?.metadata) {
const userId = options.metadata.user_id;
if (typeof userId === "string") {
params.metadata = { user_id: userId };
}
}
if (options?.toolChoice) {
if (typeof options.toolChoice === "string") {
params.tool_choice = { type: options.toolChoice };
} else {
params.tool_choice = options.toolChoice;
}
}
const allowedFallbackModels = model.compat?.allowedFallbackModels;
if (allowedFallbackModels && allowedFallbackModels.length > 0) {
params.fallbacks = allowedFallbackModels.map((fallback) => ({ model: fallback.model }));
}
assertProviderCacheControlLimit(params);
return params;
}
// Normalize tool call IDs to match Anthropic's required pattern and length
function normalizeToolCallId(id: string): string {
return id.replace(/[^a-zA-Z0-9_-]/g, "_").slice(0, 64);
}
function convertToolResult(msg: ToolResultMessage): Extract<ContentBlockParam, { type: "tool_result" }> {
return {
type: "tool_result",
tool_use_id: msg.toolCallId,
content: convertContentBlocks(msg.content),
is_error: msg.isError,
};
}
interface ConvertedAnthropicMessages {
messages: MessageParam[];
assistantLevels: Map<number, AnthropicEffort>;
}
function convertMessages(
transformedMessages: Message[],
isOAuthToken: boolean,
cacheControl?: CacheControlEphemeral,
allowEmptySignature = false,
managedProvider?: string,
nativeToolChanges = false,
nativeSystemMessages = false,
): ConvertedAnthropicMessages {
const params: MessageParam[] = [];
const assistantLevels = new Map<number, AnthropicEffort>();
for (let i = 0; i < transformedMessages.length; i++) {
const msg = transformedMessages[i];
const firstParamIndex = params.length;
if (msg.role === "system") {
const text = renderSystemMessageUpdate(msg);
const blocks: ContentBlockParam[] = [];
if (text.length > 0) blocks.push({ type: "text", text: sanitizeSurrogates(text) });
if (nativeToolChanges) {
for (const tool of msg.toolsRemoved ?? []) {
blocks.push({
type: "tool_removal",
tool: { type: "tool_reference", name: isOAuthToken ? toClaudeCodeName(tool.name) : tool.name },
});
}
for (const tool of msg.toolsAdded ?? []) {
blocks.push({
type: "tool_addition",
tool: { type: "tool_reference", name: isOAuthToken ? toClaudeCodeName(tool.name) : tool.name },
});
}
}
if (blocks.length > 0) {
const pending: MessageParam = { role: nativeSystemMessages ? "system" : "user", content: blocks };
if (msg.cacheControl && cacheControl) markMessageCacheControl(pending, cacheControl);
params.push(pending);
}
} else if (msg.role === "user") {
if (typeof msg.content === "string") {
if (msg.content.trim().length > 0) {
params.push({
role: "user",
content: [{ type: "text", text: sanitizeSurrogates(msg.content) }],
});
}
} else {
const blocks: ContentBlockParam[] = msg.content.map((item) => {
if (item.type === "text") {
return {
type: "text",
text: sanitizeSurrogates(item.text),
};
} else {
return {
type: "image",
source: {
type: "base64",
media_type: item.mimeType as "image/jpeg" | "image/png" | "image/gif" | "image/webp",
data: item.data,
},
};
}
});
const filteredBlocks = blocks.filter((b) => {
if (b.type === "text") {
return b.text.trim().length > 0;
}
return true;
});
if (filteredBlocks.length === 0) continue;
params.push({
role: "user",
content: filteredBlocks,
});
}
} else if (msg.role === "assistant") {
const blocks: ContentBlockParam[] = [];
for (const block of msg.content) {
if (block.type === "text") {
if (block.text.trim().length === 0) continue;
blocks.push({
type: "text",
text: sanitizeSurrogates(block.text),
});
} else if (block.type === "thinking") {
// Redacted thinking: pass the opaque payload back as redacted_thinking
if (block.redacted) {
blocks.push({
type: "redacted_thinking",
data: block.thinkingSignature ?? "",
});
continue;
}
const thinkingSignature = block.thinkingSignature;
const hasThinkingSignature = !!thinkingSignature && thinkingSignature.trim().length > 0;
if (block.thinking.trim().length === 0 && !hasThinkingSignature) continue;
// If thinking signature is missing/empty (e.g., from aborted stream),
// convert to plain text for Anthropic. Some compatible providers emit
// and accept empty signatures, so let marked models preserve the block.
if (!hasThinkingSignature) {
blocks.push(
allowEmptySignature
? {
type: "thinking",
thinking: sanitizeSurrogates(block.thinking),
signature: "",
}
: {
type: "text",
text: sanitizeSurrogates(block.thinking),
},
);
} else {
blocks.push({
type: "thinking",
thinking: sanitizeSurrogates(block.thinking),
signature: thinkingSignature,
});
}
} else if (block.type === "toolCall") {
blocks.push({
type: "tool_use",
id: block.id,
name: isOAuthToken ? toClaudeCodeName(block.name) : block.name,
input: block.arguments ?? {},
});
}
}
if (blocks.length === 0) continue;
const messageIndex = params.length;
params.push({
role: "assistant",
content: blocks,
});
if (
managedProvider !== undefined &&
msg.api === "anthropic-messages" &&
msg.provider === managedProvider &&
isAnthropicEffort(msg.providerThinkingLevel)
) {
assistantLevels.set(messageIndex, msg.providerThinkingLevel);
}
} else if (msg.role === "toolResult") {
// Collect all consecutive toolResult messages, needed for z.ai Anthropic endpoint.
const toolResults: ContentBlockParam[] = [];
let j = i;
while (j < transformedMessages.length && transformedMessages[j].role === "toolResult") {
const result = transformedMessages[j] as ToolResultMessage;
const block = convertToolResult(result);
if (result.cacheControl && cacheControl) block.cache_control = cacheControl;
toolResults.push(block);
j++;
}
// Skip the messages we've already processed.
i = j - 1;
params.push({
role: "user",
content: toolResults,
});
}
if (msg.role !== "toolResult" && msg.cacheControl && cacheControl && params.length > firstParamIndex)
markMessageCacheControl(params[params.length - 1], cacheControl);
}
return { messages: params, assistantLevels };
}
function isAnthropicEffort(value: unknown): value is AnthropicEffort {
return value === "low" || value === "medium" || value === "high" || value === "xhigh" || value === "max";
}
function insertThinkingLevelMessages(
converted: ConvertedAnthropicMessages,
activeEffort: AnthropicEffort,
): MessageParam[] {
const messages: MessageParam[] = [];
for (let index = 0; index < converted.messages.length; index++) {
const historicalEffort = converted.assistantLevels.get(index);
if (historicalEffort !== undefined) {
messages.push({ role: "system", content: [], output_config: { effort: historicalEffort } });
}
messages.push(converted.messages[index]);
}
messages.push({ role: "system", content: [], output_config: { effort: activeEffort } });
return messages;
}
function shouldUseFineGrainedToolStreamingBeta(
model: Model<"anthropic-messages">,
context: TranscriptContext,
): boolean {
return (
getCurrentTools(context.messages).length > 0 && !getAnthropicCompat(model).supportsEagerToolInputStreaming
);
}
function convertTools(
tools: Tool[],
isOAuthToken: boolean,
supportsEagerToolInputStreaming: boolean,
supportsStrictTools: boolean,
cacheControl?: CacheControlEphemeral,
): BetaTool[] {
if (!tools) return [];
return tools.map((tool, index) => {
const strict = resolveJsonSchemaStrictSampling(tool, supportsStrictTools);
const parameters = getJsonSchemaToolParameters(tool, strict);
const schema = parameters as { properties?: unknown; required?: string[] };
const legacyInputSchema = {
type: "object" as const,
properties: schema.properties ?? {},
required: schema.required ?? [],
};
const inputSchema =
strict === true
? {
...(parameters as Record<string, unknown>),
...legacyInputSchema,
}
: legacyInputSchema;
return {
name: isOAuthToken ? toClaudeCodeName(tool.name) : tool.name,
description: tool.description,
...(supportsEagerToolInputStreaming ? { eager_input_streaming: true } : {}),
...(strict === true ? { strict: true } : {}),
input_schema: inputSchema,
...(cacheControl && index === tools.length - 1 ? { cache_control: cacheControl } : {}),
};
});
}
function mapStopReason(
reason: BetaStopReason | string,
stopDetails?: RefusalStopDetails | null,
): { stopReason: StopReason; errorMessage?: string } {
switch (reason) {
case "end_turn":
return { stopReason: "stop" };
case "max_tokens":
case "model_context_window_exceeded":
return { stopReason: "length" };
case "tool_use":
return { stopReason: "toolUse" };
case "refusal":
return {
stopReason: "error",
errorMessage: stopDetails?.explanation || `The model refused to complete the request`,
};
case "pause_turn": // Stop is good enough -> resubmit
return { stopReason: "stop" };
case "stop_sequence":
return { stopReason: "stop" }; // We don't supply stop sequences, so this should never happen
case "sensitive": // Content flagged by safety filters (not yet in SDK types)
return { stopReason: "error", errorMessage: "Provider stopped with: sensitive" };
default:
// Unknown reasons must not turn an incomplete or rejected request into success.
throw new Error(`Unhandled stop reason: ${reason}`);
}
}
function markMessageCacheControl(message: MessageParam, cacheControl: CacheControlEphemeral): void {
if (typeof message.content === "string")
message.content = [{ type: "text", text: message.content, cache_control: cacheControl }];
else {
const block = message.content.findLast(
(block) => !["thinking", "redacted_thinking"].includes(block.type),
);
if (block)
(block as ContentBlockParam & { cache_control?: CacheControlEphemeral }).cache_control = cacheControl;
}
}