[agent] feat: Anthropic SDK adapter — executeStream via @anthropic-ai/sdk

- Add @anthropic-ai/sdk@0.105.0 dependency
- base.ts: add optional executeStream() to ProviderAdapter interface
- anthropic.ts: implement executeStream() using SDK messages.create(stream: true)
  - Type-safe params, auto-retry on 429/5xx, SDK-managed SSE parsing
  - Existing buildRequest/parseStream kept as fallback for web-search sidecar
- server.ts: prefer executeStream when available, fall back to fetch+parseStream

Author: lidge-jun

bun.lock

@@ -29,6 +29,7 @@
     "release:watch": "bun scripts/release.ts watch"
   },
   "dependencies": {
+    "@anthropic-ai/sdk": "^0.105.0",
     "zod": "^4.0.0"
   },
   "devDependencies": {

package.json

@@ -1,3 +1,5 @@
+import Anthropic from "@anthropic-ai/sdk";
+import type { RawMessageStreamEvent } from "@anthropic-ai/sdk/resources/messages.mjs";
 import type { ProviderAdapter } from "./base";
 import { debugDroppedFrame } from "../debug";
 import type {
@@ -312,5 +314,106 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda
       });
       return events;
     },
+
+    async *executeStream(parsed: OcxParsedRequest, signal?: AbortSignal): AsyncGenerator<AdapterEvent> {
+      const { system, messages } = messagesToAnthropicFormat(parsed, isOAuth);
+      const tools = toolsToAnthropicFormat(parsed, isOAuth);
+
+      const sdkParams: Record<string, unknown> = {
+        model: parsed.modelId,
+        messages,
+        max_tokens: parsed.options.maxOutputTokens ?? DEFAULT_MAX_TOKENS,
+        stream: true,
+      };
+      if (isOAuth) {
+        sdkParams.system = [
+          { type: "text", text: CLAUDE_CODE_SYSTEM_INSTRUCTION },
+          ...(system ? [{ type: "text", text: system }] : []),
+        ];
+      } else if (system) {
+        sdkParams.system = system;
+      }
+      if (tools) sdkParams.tools = tools;
+      if (parsed.options.temperature !== undefined) sdkParams.temperature = parsed.options.temperature;
+      if (parsed.options.topP !== undefined) sdkParams.top_p = parsed.options.topP;
+      if (parsed.options.stopSequences) sdkParams.stop_sequences = parsed.options.stopSequences;
+
+      if (parsed.options.reasoning) {
+        const maxOut = parsed.options.maxOutputTokens ?? DEFAULT_MAX_TOKENS;
+        const wantBudget = reasoningBudget(parsed.options.reasoning);
+        const maxTokens = Math.min(REASONING_MAX_TOKENS_CEILING, Math.max(maxOut, wantBudget + OUTPUT_HEADROOM));
+        const budget = Math.max(MIN_THINKING_BUDGET, Math.min(wantBudget, maxTokens - OUTPUT_FLOOR));
+        sdkParams.max_tokens = maxTokens;
+        sdkParams.thinking = { type: "enabled", budget_tokens: budget };
+        delete sdkParams.temperature;
+        delete sdkParams.top_p;
+      }
+
+      if (parsed.options.toolChoice) {
+        const tc = parsed.options.toolChoice;
+        if (tc === "auto") sdkParams.tool_choice = { type: "auto" };
+        else if (tc === "none") sdkParams.tool_choice = { type: "none" };
+        else if (tc === "required") sdkParams.tool_choice = { type: "any" };
+        else if (typeof tc === "object" && "name" in tc) sdkParams.tool_choice = { type: "tool", name: isOAuth ? applyClaudeToolPrefix(tc.name) : tc.name };
+      }
+
+      const headers: Record<string, string> = {};
+      if (isOAuth) headers["anthropic-beta"] = ANTHROPIC_OAUTH_BETA;
+      if (provider.headers) Object.assign(headers, provider.headers);
+
+      const client = new Anthropic({
+        apiKey: provider.apiKey ?? "",
+        baseURL: `${provider.baseUrl}/v1`,
+        maxRetries: 2,
+        defaultHeaders: headers,
+        ...(isOAuth ? { authToken: provider.apiKey } : {}),
+      });
+
+      const stream = await client.messages.create(
+        sdkParams as unknown as Parameters<typeof client.messages.create>[0],
+        { signal },
+      );
+
+      let currentToolName = "";
+      for await (const event of stream as AsyncIterable<RawMessageStreamEvent>) {
+        switch (event.type) {
+          case "content_block_start": {
+            if (event.content_block.type === "tool_use") {
+              const name = isOAuth ? stripClaudeToolPrefix(event.content_block.name) : event.content_block.name;
+              currentToolName = name;
+              yield { type: "tool_call_start", id: event.content_block.id, name };
+            }
+            break;
+          }
+          case "content_block_delta": {
+            if (event.delta.type === "text_delta") {
+              yield { type: "text_delta", text: event.delta.text };
+            } else if (event.delta.type === "thinking_delta") {
+              yield { type: "thinking_delta", thinking: event.delta.thinking };
+            } else if (event.delta.type === "input_json_delta") {
+              yield { type: "tool_call_delta", arguments: event.delta.partial_json };
+            }
+            break;
+          }
+          case "content_block_stop": {
+            if (currentToolName) {
+              yield { type: "tool_call_end" };
+              currentToolName = "";
+            }
+            break;
+          }
+          case "message_delta": {
+            const u = event.usage;
+            yield {
+              type: "done",
+              usage: u ? usageFromAnthropic({ output_tokens: u.output_tokens }) : undefined,
+            };
+            break;
+          }
+          default:
+            break;
+        }
+      }
+    },
   };
 }

src/adapters/anthropic.ts

@@ -17,4 +17,8 @@ export interface ProviderAdapter {
 
   parseStream(response: Response): AsyncGenerator<AdapterEvent>;
   parseResponse?(response: Response): Promise<AdapterEvent[]>;
+
+  /** Optional: adapter handles its own fetch + streaming via a provider SDK. When present,
+   *  server.ts uses this instead of buildRequest → fetch → parseStream. */
+  executeStream?(parsed: OcxParsedRequest, signal?: AbortSignal): AsyncGenerator<AdapterEvent>;
 }

src/adapters/base.ts

@@ -204,19 +204,35 @@ async function handleResponses(
     });
   }
 
-  const request = adapter.buildRequest(parsed, { headers: req.headers });
-
-  // Abort the upstream fetch if the client (Codex) disconnects mid-stream, so a cancelled turn does
-  // not leak the upstream connection or keep draining tokens. The bridge's cancel() fires upstream.abort() (RC2).
   const upstream = new AbortController();
   linkAbortSignal(upstream, options.abortSignal);
+
+  if (parsed.stream && adapter.executeStream) {
+    const eventStream = adapter.executeStream(parsed, upstream.signal);
+    const toolNsMap = new Map<string, { namespace: string; name: string }>();
+    const freeformToolNames = new Set<string>();
+    const toolSearchToolNames = new Set<string>();
+    for (const t of parsed.context.tools ?? []) {
+      if (t.namespace) toolNsMap.set(namespacedToolName(t.namespace, t.name), { namespace: t.namespace, name: t.name });
+      if (t.freeform) freeformToolNames.add(t.name);
+      if (t.toolSearch) toolSearchToolNames.add(t.name);
+    }
+    const sseStream = bridgeToResponsesSSE(
+      eventStream, parsed.modelId, toolNsMap, freeformToolNames, toolSearchToolNames,
+      () => upstream.abort(), 2_000,
+      options.forceEmptyResponseId ? { responseId: "" } : undefined,
+    );
+    return new Response(sseStream, {
+      headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" },
+    });
+  }
+
+  // Fallback: buildRequest → fetch → parseStream (used by all non-SDK adapters and web-search sidecar)
+  const request = adapter.buildRequest(parsed, { headers: req.headers });
   let upstreamResponse: Response;
   try {
     upstreamResponse = await fetch(request.url, {
-      method: request.method,
-      headers: request.headers,
-      body: request.body,
-      signal: upstream.signal,
+      method: request.method, headers: request.headers, body: request.body, signal: upstream.signal,
     });
   } catch (err) {
     return formatErrorResponse(502, "upstream_error", `Provider unreachable: ${err instanceof Error ? err.message : String(err)}`);
@@ -229,8 +245,6 @@ async function handleResponses(
 
   if (parsed.stream) {
     const eventStream = adapter.parseStream(upstreamResponse);
-    // Map flattened MCP tool names back to {namespace, name} so the bridge can restore the
-    // namespace field Codex needs to route the call to the right MCP server.
     const toolNsMap = new Map<string, { namespace: string; name: string }>();
     const freeformToolNames = new Set<string>();
     const toolSearchToolNames = new Set<string>();
@@ -240,31 +254,19 @@ async function handleResponses(
       if (t.toolSearch) toolSearchToolNames.add(t.name);
     }
     const sseStream = bridgeToResponsesSSE(
-      eventStream,
-      parsed.modelId,
-      toolNsMap,
-      freeformToolNames,
-      toolSearchToolNames,
-      () => upstream.abort(),
-      2_000,
+      eventStream, parsed.modelId, toolNsMap, freeformToolNames, toolSearchToolNames,
+      () => upstream.abort(), 2_000,
       options.forceEmptyResponseId ? { responseId: "" } : undefined,
     );
     return new Response(sseStream, {
-      headers: {
-        "Content-Type": "text/event-stream",
-        "Cache-Control": "no-cache",
-        "Connection": "keep-alive",
-        "X-Accel-Buffering": "no",
-      },
+      headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" },
     });
   }
 
   if (adapter.parseResponse) {
     const events = await adapter.parseResponse(upstreamResponse);
     const json = buildResponseJSON(events, parsed.modelId);
-    return new Response(JSON.stringify(json), {
-      headers: { "Content-Type": "application/json" },
-    });
+    return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } });
   }
 
   return formatErrorResponse(500, "internal_error", "Non-streaming not supported by this adapter");