Merge branch 'main' into dev

lidge-jun · lidge-jun · commit 0c551d4e6c90 · 2026-06-21T02:08:36.000+09:00
diff --git a/bun.lock b/bun.lock
diff --git a/package.json b/package.json
@@ -29,6 +29,7 @@
     "release:watch": "bun scripts/release.ts watch"
   },
   "dependencies": {
+    "@anthropic-ai/sdk": "^0.105.0",
     "zod": "^4.0.0"
   },
   "devDependencies": {
diff --git a/src/adapters/anthropic-sdk.ts b/src/adapters/anthropic-sdk.ts
@@ -0,0 +1,229 @@
+import Anthropic from "@anthropic-ai/sdk";
+import type { RawMessageStreamEvent } from "@anthropic-ai/sdk/resources/messages.mjs";
+import type { ProviderAdapter } from "./base";
+import type { AdapterEvent, OcxAssistantMessage, OcxContentPart, OcxParsedRequest, OcxProviderConfig, OcxTextContent, OcxThinkingContent, OcxToolCall, OcxUsage } from "../types";
+import { ANTHROPIC_OAUTH_BETA, CLAUDE_CODE_SYSTEM_INSTRUCTION, applyClaudeToolPrefix, stripClaudeToolPrefix } from "../oauth/anthropic";
+import { parseDataUrl } from "./image";
+
+function toContentPart(p: OcxContentPart): unknown {
+  if (p.type === "image") {
+    const data = parseDataUrl(p.imageUrl);
+    return data
+      ? { type: "image", source: { type: "base64", media_type: data.mediaType, data: data.base64 } }
+      : { type: "image", source: { type: "url", url: p.imageUrl } };
+  }
+  return { type: "text", text: p.text };
+}
+
+const DEFAULT_MAX_TOKENS = 8192;
+const REASONING_MAX_TOKENS_CEILING = 32_000;
+const MIN_THINKING_BUDGET = 1024;
+const OUTPUT_HEADROOM = 8192;
+const OUTPUT_FLOOR = 4096;
+
+function reasoningBudget(effort: string): number {
+  switch (effort) {
+    case "minimal": return 1024;
+    case "low": return 4096;
+    case "high": return 16384;
+    case "xhigh": return 24576;
+    case "max": return 32000;
+    case "medium":
+    default: return 8192;
+  }
+}
+
+function usageFromSdk(usage: { output_tokens: number } | undefined): OcxUsage | undefined {
+  if (!usage) return undefined;
+  return { inputTokens: 0, outputTokens: usage.output_tokens };
+}
+
+function buildMessages(parsed: OcxParsedRequest, isOAuth: boolean): { system: unknown; messages: unknown[] } {
+  const systemText = parsed.context.systemPrompt?.join("\n\n") || undefined;
+  const system = isOAuth
+    ? [{ type: "text", text: CLAUDE_CODE_SYSTEM_INSTRUCTION }, ...(systemText ? [{ type: "text", text: systemText }] : [])]
+    : systemText;
+
+  const messages: unknown[] = [];
+  for (const msg of parsed.context.messages) {
+    switch (msg.role) {
+      case "user":
+      case "developer": {
+        const content = typeof msg.content === "string"
+          ? msg.content
+          : (msg.content as OcxContentPart[]).map(toContentPart);
+        messages.push({ role: "user", content });
+        break;
+      }
+      case "assistant": {
+        const aMsg = msg as OcxAssistantMessage;
+        const content: unknown[] = [];
+        for (const part of aMsg.content) {
+          if (part.type === "text") content.push({ type: "text", text: (part as OcxTextContent).text });
+          else if (part.type === "thinking") {
+            const t = part as OcxThinkingContent;
+            content.push({ type: "thinking", thinking: t.thinking, ...(t.signature ? { signature: t.signature } : {}) });
+          } else if (part.type === "toolCall") {
+            const tc = part as OcxToolCall;
+            content.push({ type: "tool_use", id: tc.id, name: isOAuth ? applyClaudeToolPrefix(tc.name) : tc.name, input: tc.arguments });
+          }
+        }
+        messages.push({ role: "assistant", content });
+        break;
+      }
+      case "toolResult": {
+        const trContent = typeof msg.content === "string"
+          ? msg.content
+          : (msg.content as OcxContentPart[]).map(toContentPart);
+        messages.push({ role: "user", content: [{ type: "tool_result", tool_use_id: msg.toolCallId, content: trContent }] });
+        break;
+      }
+    }
+  }
+  return { system, messages };
+}
+
+function buildTools(parsed: OcxParsedRequest, isOAuth: boolean): unknown[] | undefined {
+  if (!parsed.context.tools?.length) return undefined;
+  return parsed.context.tools.map(t => ({
+    name: isOAuth ? applyClaudeToolPrefix(t.name) : t.name,
+    description: t.description,
+    input_schema: t.parameters,
+  }));
+}
+
+export function createAnthropicSdkAdapter(provider: OcxProviderConfig): ProviderAdapter {
+  const isOAuth = provider.authMode === "oauth";
+
+  return {
+    name: "anthropic-sdk",
+
+    buildRequest(parsed: OcxParsedRequest) {
+      const { system, messages } = buildMessages(parsed, isOAuth);
+      const tools = buildTools(parsed, isOAuth);
+      const body: Record<string, unknown> = {
+        model: parsed.modelId, messages, stream: parsed.stream,
+        max_tokens: parsed.options.maxOutputTokens ?? DEFAULT_MAX_TOKENS,
+      };
+      if (system) body.system = system;
+      if (tools) body.tools = tools;
+      if (parsed.options.temperature !== undefined) body.temperature = parsed.options.temperature;
+      if (parsed.options.topP !== undefined) body.top_p = parsed.options.topP;
+      if (parsed.options.stopSequences) body.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));
+        body.max_tokens = maxTokens;
+        body.thinking = { type: "enabled", budget_tokens: budget };
+        delete body.temperature;
+        delete body.top_p;
+      }
+      if (parsed.options.toolChoice) {
+        const tc = parsed.options.toolChoice;
+        if (tc === "auto") body.tool_choice = { type: "auto" };
+        else if (tc === "none") body.tool_choice = { type: "none" };
+        else if (tc === "required") body.tool_choice = { type: "any" };
+        else if (typeof tc === "object" && "name" in tc) body.tool_choice = { type: "tool", name: isOAuth ? applyClaudeToolPrefix(tc.name) : tc.name };
+      }
+      const url = `${provider.baseUrl}/v1/messages`;
+      const headers: Record<string, string> = { "Content-Type": "application/json", "anthropic-version": "2023-06-01" };
+      if (isOAuth) {
+        if (provider.apiKey) headers["Authorization"] = `Bearer ${provider.apiKey}`;
+        headers["anthropic-beta"] = ANTHROPIC_OAUTH_BETA;
+      } else if (provider.apiKey) {
+        headers["x-api-key"] = provider.apiKey;
+      }
+      if (provider.headers) Object.assign(headers, provider.headers);
+      return { url, method: "POST", headers, body: JSON.stringify(body) };
+    },
+
+    async *parseStream(): AsyncGenerator<AdapterEvent> {
+      yield { type: "error", message: "anthropic-sdk adapter uses executeStream; parseStream should not be called directly" };
+    },
+
+    async *executeStream(parsed: OcxParsedRequest, signal?: AbortSignal): AsyncGenerator<AdapterEvent> {
+      const { system, messages } = buildMessages(parsed, isOAuth);
+      const tools = buildTools(parsed, isOAuth);
+
+      const params: Record<string, unknown> = {
+        model: parsed.modelId, messages, stream: true,
+        max_tokens: parsed.options.maxOutputTokens ?? DEFAULT_MAX_TOKENS,
+      };
+      if (system) params.system = system;
+      if (tools) params.tools = tools;
+      if (parsed.options.temperature !== undefined) params.temperature = parsed.options.temperature;
+      if (parsed.options.topP !== undefined) params.top_p = parsed.options.topP;
+      if (parsed.options.stopSequences) params.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));
+        params.max_tokens = maxTokens;
+        params.thinking = { type: "enabled", budget_tokens: budget };
+        delete params.temperature;
+        delete params.top_p;
+      }
+
+      if (parsed.options.toolChoice) {
+        const tc = parsed.options.toolChoice;
+        if (tc === "auto") params.tool_choice = { type: "auto" };
+        else if (tc === "none") params.tool_choice = { type: "none" };
+        else if (tc === "required") params.tool_choice = { type: "any" };
+        else if (typeof tc === "object" && "name" in tc) params.tool_choice = { type: "tool", name: isOAuth ? applyClaudeToolPrefix(tc.name) : tc.name };
+      }
+
+      const sdkHeaders: Record<string, string> = {};
+      if (isOAuth) sdkHeaders["anthropic-beta"] = ANTHROPIC_OAUTH_BETA;
+      if (provider.headers) Object.assign(sdkHeaders, provider.headers);
+
+      const client = new Anthropic({
+        apiKey: provider.apiKey ?? "",
+        baseURL: `${provider.baseUrl}/v1`,
+        maxRetries: 2,
+        defaultHeaders: sdkHeaders,
+        ...(isOAuth ? { authToken: provider.apiKey } : {}),
+      });
+
+      let stream: AsyncIterable<RawMessageStreamEvent>;
+      try {
+        stream = await client.messages.create(
+          params as unknown as Parameters<typeof client.messages.create>[0],
+          { signal },
+        ) as AsyncIterable<RawMessageStreamEvent>;
+      } catch (err) {
+        yield { type: "error", message: err instanceof Error ? err.message : String(err) };
+        return;
+      }
+
+      let inToolUse = false;
+      for await (const event of stream) {
+        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;
+              inToolUse = true;
+              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 (inToolUse) { yield { type: "tool_call_end" }; inToolUse = false; }
+            break;
+          case "message_delta":
+            yield { type: "done", usage: usageFromSdk(event.usage) };
+            break;
+          default:
+            break;
+        }
+      }
+    },
+  };
+}
diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts
@@ -312,5 +312,6 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda
       });
       return events;
     },
+
   };
 }
diff --git a/src/adapters/base.ts b/src/adapters/base.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>;
 }
diff --git a/src/server.ts b/src/server.ts
@@ -1,6 +1,7 @@
 import { existsSync, readFileSync } from "node:fs";
 import { extname, join } from "node:path";
 import { createAnthropicAdapter } from "./adapters/anthropic";
+import { createAnthropicSdkAdapter } from "./adapters/anthropic-sdk";
 import { createAzureAdapter } from "./adapters/azure";
 import { createGoogleAdapter } from "./adapters/google";
 import { createOpenAIChatAdapter } from "./adapters/openai-chat";
@@ -104,6 +105,8 @@ export function resolveAdapter(providerConfig: OcxProviderConfig) {
       return createOpenAIChatAdapter(providerConfig);
     case "anthropic":
       return createAnthropicAdapter(providerConfig);
+    case "anthropic-sdk":
+      return createAnthropicSdkAdapter(providerConfig);
     case "openai-responses":
       return createResponsesPassthroughAdapter(providerConfig);
     case "google":
@@ -218,19 +221,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)}`);
@@ -243,8 +262,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>();
@@ -254,31 +271,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");

Original file line number	Diff line number	Diff line change
`@@ -312,5 +312,6 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda`
`312`	`312`	`});`
`313`	`313`	`return events;`
`314`	`314`	`},`
	`315`	`+`
`315`	`316`	`};`
`316`	`317`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,4 +17,8 @@ export interface ProviderAdapter {`
`17`	`17`
`18`	`18`	`parseStream(response: Response): AsyncGenerator<AdapterEvent>;`
`19`	`19`	`parseResponse?(response: Response): Promise<AdapterEvent[]>;`
	`20`	`+`
	`21`	`+ /** Optional: adapter handles its own fetch + streaming via a provider SDK. When present,`
	`22`	`+ * server.ts uses this instead of buildRequest → fetch → parseStream. */`
	`23`	`+ executeStream?(parsed: OcxParsedRequest, signal?: AbortSignal): AsyncGenerator<AdapterEvent>;`
`20`	`24`	`}`