feat(ai): Anthropic provider over fetch with streaming tool use
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Lucas Winther
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
bccbc692e9
commit
3c2220145b
@@ -0,0 +1,139 @@
|
||||
import type { AiProvider, ChatMessage, ToolSpec } from "@stc/contracts";
|
||||
import type { Effort } from "../settings";
|
||||
import { errorMessage, readSse } from "./sse";
|
||||
|
||||
export const DEFAULT_ANTHROPIC_MODEL = "claude-opus-5-5";
|
||||
const FIRST_PARTY = "https://api.anthropic.com";
|
||||
|
||||
export interface AnthropicOptions {
|
||||
apiKey: string;
|
||||
model?: string;
|
||||
baseUrl?: string;
|
||||
effort?: Effort;
|
||||
maxTokens?: number;
|
||||
fetch?: typeof fetch;
|
||||
}
|
||||
|
||||
type Block = Record<string, unknown> & { type: string };
|
||||
type ApiMessage = { role: "user" | "assistant"; content: Block[] };
|
||||
|
||||
/**
|
||||
* Anthropic Messages API over plain fetch with streaming and tool use.
|
||||
* - Thinking is left to the model default (adaptive on current models); no temperature or forced tool_choice.
|
||||
* - The frozen ChatMessage contract has no slot for thinking blocks, so the provider remembers every
|
||||
* response's raw content blocks by tool_use id and replays them unchanged when the caller sends back
|
||||
* the matching assistant tool-call message (required for thinking + tool use).
|
||||
* - On the first-party endpoint, server-side refusal fallbacks are on ("default" routing).
|
||||
*/
|
||||
export function createAnthropicProvider(o: AnthropicOptions): AiProvider {
|
||||
const doFetch = o.fetch ?? fetch;
|
||||
const base = (o.baseUrl || FIRST_PARTY).replace(/\/+$/, "");
|
||||
const firstParty = base === FIRST_PARTY;
|
||||
const rawByToolId = new Map<string, Block[]>();
|
||||
|
||||
const toApi = (messages: ChatMessage[]): { system: string; messages: ApiMessage[] } => {
|
||||
const system = messages.filter((m) => m.role === "system").map((m) => m.content).join("\n\n");
|
||||
const out: ApiMessage[] = [];
|
||||
const push = (role: "user" | "assistant", block: Block) => {
|
||||
const last = out[out.length - 1];
|
||||
if (last && last.role === role) last.content.push(block);
|
||||
else out.push({ role, content: [block] });
|
||||
};
|
||||
for (const m of messages) {
|
||||
if (m.role === "system") continue;
|
||||
if (m.role === "tool") push("user", { type: "tool_result", tool_use_id: m.toolCallId ?? "", content: m.content });
|
||||
else if (m.role === "assistant" && m.toolCallId) {
|
||||
let input: unknown = {};
|
||||
try { input = JSON.parse(m.content || "{}"); } catch { input = { value: m.content }; }
|
||||
push("assistant", { type: "tool_use", id: m.toolCallId, name: m.name ?? "", input });
|
||||
} else if (m.content) push(m.role === "assistant" ? "assistant" : "user", { type: "text", text: m.content });
|
||||
}
|
||||
// Replace assistant turns we produced with their exact original blocks (thinking signatures intact).
|
||||
for (const msg of out) {
|
||||
if (msg.role !== "assistant") continue;
|
||||
const id = msg.content.find((b) => b.type === "tool_use")?.id as string | undefined;
|
||||
const raw = id ? rawByToolId.get(id) : undefined;
|
||||
if (raw) msg.content = structuredClone(raw);
|
||||
}
|
||||
return { system, messages: out };
|
||||
};
|
||||
|
||||
return {
|
||||
id: "anthropic",
|
||||
async *chat(messages: ChatMessage[], opts: { tools?: ToolSpec[]; signal?: AbortSignal }) {
|
||||
const { system, messages: apiMessages } = toApi(messages);
|
||||
const body: Record<string, unknown> = {
|
||||
model: o.model || DEFAULT_ANTHROPIC_MODEL,
|
||||
max_tokens: o.maxTokens ?? 64000,
|
||||
stream: true,
|
||||
messages: apiMessages,
|
||||
};
|
||||
if (system) body.system = system;
|
||||
if (opts.tools?.length) body.tools = opts.tools.map((t) => ({ name: t.name, description: t.description, input_schema: t.inputSchema }));
|
||||
if (o.effort) body.output_config = { effort: o.effort };
|
||||
const headers: Record<string, string> = {
|
||||
"content-type": "application/json",
|
||||
"x-api-key": o.apiKey,
|
||||
"anthropic-version": "2023-06-01",
|
||||
};
|
||||
if (firstParty) {
|
||||
body.fallbacks = "default";
|
||||
headers["anthropic-beta"] = "server-side-fallback-2026-07-01";
|
||||
}
|
||||
const res = await doFetch(`${base}/v1/messages`, { method: "POST", headers, body: JSON.stringify(body), signal: opts.signal });
|
||||
if (!res.ok || !res.body) throw new Error(`Anthropic API error ${await errorMessage(res)}`);
|
||||
|
||||
const blocks: Block[] = [];
|
||||
const partialJson = new Map<number, string>();
|
||||
const toolIds: string[] = [];
|
||||
let stopReason: string | null = null;
|
||||
for await (const { data } of readSse(res.body)) {
|
||||
let ev: any;
|
||||
try { ev = JSON.parse(data); } catch { continue; }
|
||||
switch (ev.type) {
|
||||
case "content_block_start": {
|
||||
const b = { ...ev.content_block } as Block;
|
||||
blocks[ev.index] = b;
|
||||
if (b.type === "tool_use") partialJson.set(ev.index, "");
|
||||
break;
|
||||
}
|
||||
case "content_block_delta": {
|
||||
const b = blocks[ev.index];
|
||||
const d = ev.delta ?? {};
|
||||
if (!b) break;
|
||||
if (d.type === "text_delta") {
|
||||
b.text = String(b.text ?? "") + d.text;
|
||||
yield { text: d.text as string };
|
||||
} else if (d.type === "input_json_delta") partialJson.set(ev.index, (partialJson.get(ev.index) ?? "") + d.partial_json);
|
||||
else if (d.type === "thinking_delta") b.thinking = String(b.thinking ?? "") + d.thinking;
|
||||
else if (d.type === "signature_delta") b.signature = String(b.signature ?? "") + d.signature;
|
||||
else if (d.type === "citations_delta") b.citations = [...((b.citations as unknown[]) ?? []), d.citation];
|
||||
break;
|
||||
}
|
||||
case "content_block_stop": {
|
||||
const b = blocks[ev.index];
|
||||
if (b?.type === "tool_use") {
|
||||
const json = partialJson.get(ev.index) ?? "";
|
||||
let input: unknown = {};
|
||||
try { input = json ? JSON.parse(json) : (b.input ?? {}); } catch { input = { _invalidJson: json }; }
|
||||
b.input = input;
|
||||
toolIds.push(b.id as string);
|
||||
yield { toolCall: { id: b.id as string, name: b.name as string, input } };
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "message_delta":
|
||||
if (ev.delta?.stop_reason) stopReason = ev.delta.stop_reason;
|
||||
break;
|
||||
case "error":
|
||||
throw new Error(`Anthropic API error: ${ev.error?.message ?? "stream error"}`);
|
||||
}
|
||||
}
|
||||
const content = blocks.filter(Boolean);
|
||||
for (const id of toolIds) rawByToolId.set(id, content);
|
||||
if (stopReason === "refusal" && !content.some((b) => b.type === "text" && b.text)) {
|
||||
yield { text: "The model declined to answer this request." };
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
/** Minimal Server-Sent Events reader: yields one {event, data} per blank-line-terminated record. */
|
||||
export async function* readSse(body: ReadableStream<Uint8Array>): AsyncIterable<{ event: string; data: string }> {
|
||||
const decoder = new TextDecoder();
|
||||
let buf = "";
|
||||
let event = "message";
|
||||
let data: string[] = [];
|
||||
const flushLine = function* (line: string): Generator<{ event: string; data: string }> {
|
||||
if (line === "") {
|
||||
if (data.length) yield { event, data: data.join("\n") };
|
||||
event = "message";
|
||||
data = [];
|
||||
return;
|
||||
}
|
||||
if (line.startsWith(":")) return;
|
||||
const i = line.indexOf(":");
|
||||
const field = i === -1 ? line : line.slice(0, i);
|
||||
let value = i === -1 ? "" : line.slice(i + 1);
|
||||
if (value.startsWith(" ")) value = value.slice(1);
|
||||
if (field === "event") event = value;
|
||||
else if (field === "data") data.push(value);
|
||||
};
|
||||
for await (const chunk of body as unknown as AsyncIterable<Uint8Array>) {
|
||||
buf += decoder.decode(chunk, { stream: true });
|
||||
let nl: number;
|
||||
while ((nl = buf.search(/\r\n|\n|\r/)) !== -1) {
|
||||
const line = buf.slice(0, nl);
|
||||
buf = buf.slice(nl + (buf.startsWith("\r\n", nl) ? 2 : 1));
|
||||
yield* flushLine(line);
|
||||
}
|
||||
}
|
||||
buf += decoder.decode();
|
||||
if (buf) yield* flushLine(buf);
|
||||
yield* flushLine("");
|
||||
}
|
||||
|
||||
export async function errorMessage(res: Response): Promise<string> {
|
||||
const text = await res.text().catch(() => "");
|
||||
try {
|
||||
const j = JSON.parse(text);
|
||||
const m = j?.error?.message ?? j?.message ?? j?.error;
|
||||
if (typeof m === "string") return `${res.status}: ${m}`;
|
||||
} catch {}
|
||||
return `${res.status}: ${text.slice(0, 300) || res.statusText}`;
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
import { expect, test } from "bun:test";
|
||||
import type { ChatMessage } from "@stc/contracts";
|
||||
import { createAnthropicProvider } from "../src/providers/anthropic";
|
||||
import { collect, fakeFetch, sseBody } from "./helpers";
|
||||
|
||||
const ok = (events: [string | null, unknown][]) => () => new Response(sseBody(events), { headers: { "content-type": "text/event-stream" } });
|
||||
|
||||
const textTurn: [string, unknown][] = [
|
||||
["message_start", { type: "message_start", message: { id: "m1", content: [] } }],
|
||||
["content_block_start", { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }],
|
||||
["ping", { type: "ping" }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello " } }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "world" } }],
|
||||
["content_block_stop", { type: "content_block_stop", index: 0 }],
|
||||
["message_delta", { type: "message_delta", delta: { stop_reason: "end_turn" } }],
|
||||
["message_stop", { type: "message_stop" }],
|
||||
];
|
||||
|
||||
const toolTurn: [string, unknown][] = [
|
||||
["message_start", { type: "message_start", message: { id: "m2", content: [] } }],
|
||||
["content_block_start", { type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "", signature: "" } }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "" } }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "signature_delta", signature: "SIG-abc" } }],
|
||||
["content_block_stop", { type: "content_block_stop", index: 0 }],
|
||||
["content_block_start", { type: "content_block_start", index: 1, content_block: { type: "text", text: "" } }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 1, delta: { type: "text_delta", text: "Let me look." } }],
|
||||
["content_block_stop", { type: "content_block_stop", index: 1 }],
|
||||
["content_block_start", { type: "content_block_start", index: 2, content_block: { type: "tool_use", id: "toolu_1", name: "search_course", input: {} } }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: "{\"que" } }],
|
||||
["content_block_delta", { type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: "ry\": \"add\"}" } }],
|
||||
["content_block_stop", { type: "content_block_stop", index: 2 }],
|
||||
["message_delta", { type: "message_delta", delta: { stop_reason: "tool_use" } }],
|
||||
["message_stop", { type: "message_stop" }],
|
||||
];
|
||||
|
||||
test("streams text and sends a well-formed request", async () => {
|
||||
const ff = fakeFetch([ok(textTurn)]);
|
||||
const p = createAnthropicProvider({ apiKey: "sk-test", fetch: ff.fetch });
|
||||
const msgs: ChatMessage[] = [{ role: "system", content: "You are kind." }, { role: "user", content: "Hi" }];
|
||||
const r = await collect(p.chat(msgs, { tools: [{ name: "search_course", description: "Search", inputSchema: { type: "object", properties: {} } }] }));
|
||||
expect(r.text).toBe("Hello world");
|
||||
const c = ff.calls[0]!;
|
||||
expect(c.url).toBe("https://api.anthropic.com/v1/messages");
|
||||
const h = new Headers(c.init.headers);
|
||||
expect(h.get("x-api-key")).toBe("sk-test");
|
||||
expect(h.get("anthropic-version")).toBe("2023-06-01");
|
||||
expect(c.body.model).toBe("claude-opus-5-5");
|
||||
expect(c.body.stream).toBe(true);
|
||||
expect(c.body.system).toBe("You are kind.");
|
||||
expect(c.body.messages).toEqual([{ role: "user", content: [{ type: "text", text: "Hi" }] }]);
|
||||
expect(c.body.tools).toEqual([{ name: "search_course", description: "Search", input_schema: { type: "object", properties: {} } }]);
|
||||
expect(c.body.thinking).toBeUndefined();
|
||||
expect(c.body.temperature).toBeUndefined();
|
||||
expect(c.body.tool_choice).toBeUndefined();
|
||||
expect(c.body.fallbacks).toBe("default");
|
||||
expect(h.get("anthropic-beta")).toBe("server-side-fallback-2026-07-01");
|
||||
});
|
||||
|
||||
test("assembles tool input and replays thinking blocks unchanged on the next call", async () => {
|
||||
const ff = fakeFetch([ok(toolTurn), ok(textTurn)]);
|
||||
const p = createAnthropicProvider({ apiKey: "k", fetch: ff.fetch });
|
||||
const msgs: ChatMessage[] = [{ role: "user", content: "How do I add?" }];
|
||||
const first = await collect(p.chat(msgs, {}));
|
||||
expect(first.text).toBe("Let me look.");
|
||||
expect(first.toolCalls).toEqual([{ id: "toolu_1", name: "search_course", input: { query: "add" } }]);
|
||||
|
||||
msgs.push({ role: "assistant", content: "Let me look." });
|
||||
msgs.push({ role: "assistant", content: JSON.stringify({ query: "add" }), toolCallId: "toolu_1", name: "search_course" });
|
||||
msgs.push({ role: "tool", content: "lessons/04-exercise.md: ...", toolCallId: "toolu_1", name: "search_course" });
|
||||
await collect(p.chat(msgs, {}));
|
||||
const sent = ff.calls[1]!.body.messages;
|
||||
expect(sent).toHaveLength(3);
|
||||
expect(sent[1]).toEqual({
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "thinking", thinking: "", signature: "SIG-abc" },
|
||||
{ type: "text", text: "Let me look." },
|
||||
{ type: "tool_use", id: "toolu_1", name: "search_course", input: { query: "add" } },
|
||||
],
|
||||
});
|
||||
expect(sent[2]).toEqual({ role: "user", content: [{ type: "tool_result", tool_use_id: "toolu_1", content: "lessons/04-exercise.md: ..." }] });
|
||||
});
|
||||
|
||||
test("tool calls it did not produce are mapped from the message fields", async () => {
|
||||
const ff = fakeFetch([ok(textTurn)]);
|
||||
const p = createAnthropicProvider({ apiKey: "k", fetch: ff.fetch });
|
||||
await collect(p.chat([
|
||||
{ role: "user", content: "q" },
|
||||
{ role: "assistant", content: "{\"x\":1}", toolCallId: "t9", name: "list_notes" },
|
||||
{ role: "tool", content: "none", toolCallId: "t9" },
|
||||
], {}));
|
||||
expect(ff.calls[0]!.body.messages[1]).toEqual({ role: "assistant", content: [{ type: "tool_use", id: "t9", name: "list_notes", input: { x: 1 } }] });
|
||||
});
|
||||
|
||||
test("errors carry the API message", async () => {
|
||||
const ff = fakeFetch([() => Response.json({ type: "error", error: { type: "authentication_error", message: "invalid x-api-key" } }, { status: 401 })]);
|
||||
const p = createAnthropicProvider({ apiKey: "bad", fetch: ff.fetch });
|
||||
await expect(collect(p.chat([{ role: "user", content: "x" }], {}))).rejects.toThrow(/invalid x-api-key/);
|
||||
const ff2 = fakeFetch([ok([["error", { type: "error", error: { type: "overloaded_error", message: "Overloaded" } }]])]);
|
||||
await expect(collect(createAnthropicProvider({ apiKey: "k", fetch: ff2.fetch }).chat([{ role: "user", content: "x" }], {}))).rejects.toThrow(/Overloaded/);
|
||||
});
|
||||
|
||||
test("a refusal yields a plain sentence", async () => {
|
||||
const ff = fakeFetch([ok([["message_start", { type: "message_start", message: { content: [] } }], ["message_delta", { type: "message_delta", delta: { stop_reason: "refusal" } }], ["message_stop", { type: "message_stop" }]])]);
|
||||
const r = await collect(createAnthropicProvider({ apiKey: "k", fetch: ff.fetch }).chat([{ role: "user", content: "x" }], {}));
|
||||
expect(r.text).toContain("declined");
|
||||
});
|
||||
|
||||
test("custom base URL, model and effort", async () => {
|
||||
const ff = fakeFetch([ok(textTurn)]);
|
||||
await collect(createAnthropicProvider({ apiKey: "k", baseUrl: "http://proxy.local/", model: "claude-sonnet-5-5", effort: "high", fetch: ff.fetch }).chat([{ role: "user", content: "x" }], {}));
|
||||
expect(ff.calls[0]!.url).toBe("http://proxy.local/v1/messages");
|
||||
expect(ff.calls[0]!.body.model).toBe("claude-sonnet-5-5");
|
||||
expect(ff.calls[0]!.body.output_config).toEqual({ effort: "high" });
|
||||
expect(ff.calls[0]!.body.fallbacks).toBeUndefined();
|
||||
});
|
||||
@@ -0,0 +1,53 @@
|
||||
import type { AiProvider, ChatMessage, ToolSpec } from "@stc/contracts";
|
||||
|
||||
/** SSE body from a list of [event, data] pairs, split into awkward chunks to exercise buffering. */
|
||||
export function sseBody(events: [string | null, unknown][], chunk = 7): ReadableStream<Uint8Array> {
|
||||
const text = events.map(([e, d]) => `${e ? `event: ${e}\n` : ""}data: ${typeof d === "string" ? d : JSON.stringify(d)}\n\n`).join("");
|
||||
const bytes = new TextEncoder().encode(text);
|
||||
let i = 0;
|
||||
return new ReadableStream({
|
||||
pull(c) {
|
||||
if (i >= bytes.length) return c.close();
|
||||
c.enqueue(bytes.slice(i, i + chunk));
|
||||
i += chunk;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export type Captured = { url: string; init: RequestInit; body: any };
|
||||
/** A fake fetch that records requests and answers from a queue of responses. */
|
||||
export function fakeFetch(responses: (() => Response)[]) {
|
||||
const calls: Captured[] = [];
|
||||
const f = (async (url: string | URL | Request, init?: RequestInit) => {
|
||||
calls.push({ url: String(url), init: init ?? {}, body: init?.body ? JSON.parse(String(init.body)) : null });
|
||||
const next = responses.shift();
|
||||
if (!next) throw new Error("fakeFetch: no response queued");
|
||||
return next();
|
||||
}) as unknown as typeof fetch;
|
||||
return { fetch: f, calls };
|
||||
}
|
||||
|
||||
export async function collect(it: AsyncIterable<{ text?: string; toolCall?: { id: string; name: string; input: unknown } }>) {
|
||||
let text = "";
|
||||
const toolCalls: { id: string; name: string; input: unknown }[] = [];
|
||||
for await (const ev of it) {
|
||||
if (ev.text) text += ev.text;
|
||||
if (ev.toolCall) toolCalls.push(ev.toolCall);
|
||||
}
|
||||
return { text, toolCalls };
|
||||
}
|
||||
|
||||
/** Scripted fake provider: each call to chat() plays the next turn and records the messages it got. */
|
||||
export function fakeProvider(turns: ((messages: ChatMessage[], tools?: ToolSpec[]) => { text?: string; toolCalls?: { id: string; name: string; input: unknown }[] })[]) {
|
||||
const seen: { messages: ChatMessage[]; tools?: ToolSpec[] }[] = [];
|
||||
const provider: AiProvider = {
|
||||
id: "fake",
|
||||
async *chat(messages, opts) {
|
||||
seen.push({ messages: structuredClone(messages), tools: opts.tools });
|
||||
const turn = turns[Math.min(seen.length - 1, turns.length - 1)]!(messages, opts.tools);
|
||||
if (turn.text) for (const part of turn.text.match(/.{1,8}/gs) ?? []) yield { text: part };
|
||||
for (const tc of turn.toolCalls ?? []) yield { toolCall: tc };
|
||||
},
|
||||
};
|
||||
return { provider, seen };
|
||||
}
|
||||
Reference in New Issue
Block a user