fix(ai): support Anthropic eager tool streaming compat

closes #3575

packages/ai/test/anthropic-eager-tool-input-compat.test.ts

@@ -0,0 +1,122 @@
+import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
+import type { AddressInfo } from "node:net";
+import { Type } from "typebox";
+import { describe, expect, it } from "vitest";
+import { streamAnthropic } from "../src/providers/anthropic.js";
+import type { Context, Model, Tool } from "../src/types.js";
+
+interface CapturedRequest {
+	headers: IncomingMessage["headers"];
+	body: Record<string, unknown>;
+}
+
+function createModel(baseUrl: string, compat?: Model<"anthropic-messages">["compat"]): Model<"anthropic-messages"> {
+	return {
+		id: "claude-opus-4-7",
+		name: "Claude Opus 4.7",
+		api: "anthropic-messages",
+		provider: "test-anthropic",
+		baseUrl,
+		reasoning: true,
+		input: ["text"],
+		cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+		contextWindow: 200000,
+		maxTokens: 32000,
+		compat,
+	};
+}
+
+const tool: Tool = {
+	name: "lookup",
+	description: "Look up a value",
+	parameters: Type.Object({ value: Type.String() }),
+};
+
+function createContext(tools: Tool[] = [tool]): Context {
+	return {
+		messages: [{ role: "user", content: "Use the tool", timestamp: Date.now() }],
+		...(tools.length > 0 ? { tools } : {}),
+	};
+}
+
+async function readRequestBody(request: IncomingMessage): Promise<Record<string, unknown>> {
+	const chunks: Buffer[] = [];
+	for await (const chunk of request) {
+		chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+	}
+	return JSON.parse(Buffer.concat(chunks).toString("utf8")) as Record<string, unknown>;
+}
+
+function writeEmptySseResponse(response: ServerResponse): void {
+	response.writeHead(200, { "content-type": "text/event-stream" });
+	response.end();
+}
+
+async function captureAnthropicRequest(
+	compat: Model<"anthropic-messages">["compat"],
+	context: Context,
+): Promise<CapturedRequest> {
+	let capturedRequest: CapturedRequest | undefined;
+
+	const server = createServer(async (request, response) => {
+		capturedRequest = {
+			headers: request.headers,
+			body: await readRequestBody(request),
+		};
+		writeEmptySseResponse(response);
+	});
+
+	await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+	const address = server.address() as AddressInfo;
+
+	try {
+		const stream = streamAnthropic(createModel(`http://127.0.0.1:${address.port}`, compat), context, {
+			apiKey: "test-key",
+			cacheRetention: "none",
+		});
+
+		for await (const event of stream) {
+			if (event.type === "done" || event.type === "error") break;
+		}
+	} finally {
+		await new Promise<void>((resolve, reject) => {
+			server.close((error) => (error ? reject(error) : resolve()));
+		});
+	}
+
+	if (!capturedRequest) {
+		throw new Error("Anthropic request was not captured");
+	}
+	return capturedRequest;
+}
+
+function getFirstTool(body: Record<string, unknown>): Record<string, unknown> {
+	const tools = body.tools;
+	if (!Array.isArray(tools) || typeof tools[0] !== "object" || tools[0] === null) {
+		throw new Error("Expected first tool in request body");
+	}
+	return tools[0] as Record<string, unknown>;
+}
+
+describe("Anthropic eager tool input streaming compatibility", () => {
+	it("sends per-tool eager_input_streaming by default", async () => {
+		const request = await captureAnthropicRequest(undefined, createContext());
+
+		expect(getFirstTool(request.body).eager_input_streaming).toBe(true);
+		expect(request.headers["anthropic-beta"]).toBeUndefined();
+	});
+
+	it("uses the legacy fine-grained tool streaming beta when eager tool input streaming is disabled", async () => {
+		const request = await captureAnthropicRequest({ supportsEagerToolInputStreaming: false }, createContext());
+
+		expect(getFirstTool(request.body).eager_input_streaming).toBeUndefined();
+		expect(request.headers["anthropic-beta"]).toBe("fine-grained-tool-streaming-2025-05-14");
+	});
+
+	it("does not send the legacy fine-grained tool streaming beta when there are no tools", async () => {
+		const request = await captureAnthropicRequest({ supportsEagerToolInputStreaming: false }, createContext([]));
+
+		expect(request.body.tools).toBeUndefined();
+		expect(request.headers["anthropic-beta"]).toBeUndefined();
+	});
+});

packages/ai/test/anthropic-eager-tool-input-e2e.test.ts

@@ -0,0 +1,143 @@
+import { Type } from "typebox";
+import { describe, expect, it } from "vitest";
+import { getEnvApiKey } from "../src/env-api-keys.js";
+import { getModels, getProviders } from "../src/models.js";
+import { complete } from "../src/stream.js";
+import type { Api, KnownProvider, Model, ProviderStreamOptions, Tool } from "../src/types.js";
+import { resolveApiKey } from "./oauth.js";
+
+const githubCopilotToken = await resolveApiKey("github-copilot");
+
+const echoToolSchema = Type.Object({
+	value: Type.String({ description: "The value to echo" }),
+});
+
+const echoTool: Tool<typeof echoToolSchema> = {
+	name: "echo_value",
+	description: "Echo a string value",
+	parameters: echoToolSchema,
+};
+
+interface AnthropicEagerE2ECase {
+	name: string;
+	provider: KnownProvider;
+	model: Model<"anthropic-messages">;
+	apiKey: string | undefined;
+}
+
+function getE2EApiKey(provider: KnownProvider): string | undefined {
+	if (provider === "github-copilot") {
+		return githubCopilotToken;
+	}
+	return getEnvApiKey(provider);
+}
+
+function getAnthropicMessagesModels(provider: KnownProvider): Model<"anthropic-messages">[] {
+	const models = getModels(provider) as Model<Api>[];
+	return models.filter((model) => model.api === "anthropic-messages") as Model<"anthropic-messages">[];
+}
+
+const anthropicMessagesCases: AnthropicEagerE2ECase[] = getProviders().flatMap((provider) =>
+	getAnthropicMessagesModels(provider).map((model) => ({
+		name: `${provider}/${model.id}`,
+		provider,
+		model,
+		apiKey: getE2EApiKey(provider),
+	})),
+);
+
+function getProbePriority(model: Model<"anthropic-messages">): number {
+	const modelId = model.id.toLowerCase();
+	const cost = model.cost.input + model.cost.output;
+	let priority = cost;
+
+	// Prefer current Claude 4 Haiku routes when present: they are cheap and avoid
+	// stale Claude 3.x aliases that can remain in catalogs after upstream removal.
+	if (modelId.includes("haiku") && (modelId.includes("4-5") || modelId.includes("4.5"))) {
+		priority -= 1000;
+	} else if (modelId.includes("claude") && (modelId.includes("4-") || modelId.includes("4."))) {
+		priority -= 500;
+	}
+
+	return priority;
+}
+
+function selectOneCasePerProvider(cases: AnthropicEagerE2ECase[]): AnthropicEagerE2ECase[] {
+	const byProvider = new Map<KnownProvider, AnthropicEagerE2ECase[]>();
+	for (const testCase of cases) {
+		const providerCases = byProvider.get(testCase.provider) ?? [];
+		providerCases.push(testCase);
+		byProvider.set(testCase.provider, providerCases);
+	}
+
+	return Array.from(byProvider.values()).map(
+		(providerCases) =>
+			providerCases.sort(
+				(a, b) => getProbePriority(a.model) - getProbePriority(b.model) || a.model.id.localeCompare(b.model.id),
+			)[0],
+	);
+}
+
+const probeCases = selectOneCasePerProvider(anthropicMessagesCases);
+
+function withEagerToolInputStreaming(model: Model<"anthropic-messages">): Model<"anthropic-messages"> {
+	return {
+		...model,
+		compat: {
+			...model.compat,
+			supportsEagerToolInputStreaming: true,
+		},
+	};
+}
+
+async function expectToolEnabledRequestAccepted(
+	model: Model<"anthropic-messages">,
+	apiKey: string | undefined,
+): Promise<void> {
+	const options: ProviderStreamOptions = {
+		apiKey,
+		maxTokens: 128,
+		thinkingEnabled: false,
+	};
+	const response = await complete(
+		model,
+		{
+			systemPrompt: "You are a concise assistant. Use tools when useful.",
+			messages: [
+				{
+					role: "user",
+					content: "Call echo_value with value set to eager-input-streaming-compat.",
+					timestamp: Date.now(),
+				},
+			],
+			tools: [echoTool],
+		},
+		options,
+	);
+
+	expect(response.errorMessage, response.errorMessage).toBeFalsy();
+	expect(response.stopReason, response.errorMessage).not.toBe("error");
+}
+
+describe("Anthropic Messages eager tool input streaming E2E", () => {
+	it("covers every generated anthropic-messages model", () => {
+		const expectedModels = getProviders().flatMap((provider) =>
+			getAnthropicMessagesModels(provider).map((model) => `${provider}/${model.id}`),
+		);
+		expect(anthropicMessagesCases.map((testCase) => testCase.name).sort()).toEqual(expectedModels.sort());
+	});
+
+	describe("forced eager_input_streaming probe", () => {
+		for (const testCase of probeCases) {
+			const model = withEagerToolInputStreaming(testCase.model);
+
+			it.skipIf(!testCase.apiKey)(
+				`${testCase.name} accepts forced eager_input_streaming`,
+				{ retry: 2 },
+				async () => {
+					await expectToolEnabledRequestAccepted(model, testCase.apiKey);
+				},
+			);
+		}
+	});
+});