Add bidirectional communication between SDK and consumer

apps/claude-sdk-cli/src/main.ts

@@ -1,23 +1,26 @@
-import { AnthropicAgent, AnthropicBeta } from '@shellicar/claude-sdk';
+import { AnthropicBeta, createAnthropicAgent, type SdkMessage } from '@shellicar/claude-sdk';
 import { logger } from './logger';
 import { editConfirmTool } from './tools/edit/editConfirmTool';
 import { editTool } from './tools/edit/editTool';
 
 const main = async () => {
-  const agent = new AnthropicAgent({
-    apiKey: process.env.CLAUDE_CODE_API_KEY ?? 'no-key',
+  const apiKey = process.env.CLAUDE_CODE_API_KEY;
+  if (!apiKey) {
+    logger.error('CLAUDE_CODE_API_KEY is not set');
+    process.exit(1);
+  }
+
+  const agent = createAnthropicAgent({
+    apiKey,
     logger,
   });
 
-  agent.on('message_start', () => process.stdout.write('> '));
-  agent.on('message_text', (x) => process.stdout.write(x));
-  agent.on('message_end', () => process.stdout.write('\n'));
-
-  await agent.runAgent({
+  const { port, done } = agent.runAgent({
     model: 'claude-sonnet-4-6',
     maxTokens: 8096,
     messages: ['Please add a comment "// hello world" on line 1344 of the file /Users/stephen/repos/@shellicar/claude-cli/node_modules/.pnpm/@anthropic-ai+sdk@0.80.0_zod@4.3.6/node_modules/@anthropic-ai/sdk/src/resources/messages/messages.ts'],
     tools: [editTool, editConfirmTool],
+    requireToolApproval: true,
     betas: {
       [AnthropicBeta.InterleavedThinking]: true,
       [AnthropicBeta.ContextManagement]: true,
@@ -27,5 +30,31 @@ const main = async () => {
       [AnthropicBeta.TokenEfficientTools]: true,
     },
   });
+
+  port.on('message', (msg: SdkMessage) => {
+    switch (msg.type) {
+      case 'message_start':
+        process.stdout.write('> ');
+        break;
+      case 'message_text':
+        process.stdout.write(msg.text);
+        break;
+      case 'message_end':
+        process.stdout.write('\n');
+        break;
+      case 'tool_approval_request':
+        logger.info('tool_approval_request', { name: msg.name, input: msg.input });
+        port.postMessage({ type: 'tool_approval_response', requestId: msg.requestId, approved: true });
+        break;
+      case 'done':
+        logger.info('done', { stopReason: msg.stopReason });
+        break;
+      case 'error':
+        logger.error('error', { message: msg.message });
+        break;
+    }
+  });
+
+  await done;
 };
 main();

apps/claude-sdk-cli/src/tools/edit/editConfirmTool.ts

@@ -13,7 +13,7 @@ export const editConfirmTool: ToolDefinition<EditConfirmInputType, EditConfirmOu
       patchId: '2b9cfd39-7f29-4911-8cb2-ef4454635e51',
     },
   ],
-  handler: ({ patchId }, store) => {
+  handler: async ({ patchId }, store) => {
     const input = store.get(patchId);
     if (input == null) {
       throw new Error('edit_confirm requires a staged edit from the edit tool');

apps/claude-sdk-cli/src/tools/edit/editTool.ts

@@ -32,7 +32,7 @@ export const editTool: ToolDefinition<EditInputType, EditOutputType> = {
       ],
     },
   ],
-  handler: (input, store) => {
+  handler: async (input, store) => {
     const originalContent = readFileSync(input.file, 'utf-8');
     const originalHash = createHash('sha256').update(originalContent).digest('hex');
     const originalLines = originalContent.split('\n');

packages/claude-sdk/src/index.ts

@@ -1,6 +1,7 @@
-import { AnthropicAgent } from './public/AnthropicAgent';
-import type { AgentEvents, AnthropicAgentOptions, AnthropicBetaFlags, AnyToolDefinition, ChainedToolStore, ILogger, JsonObject, JsonValue, RunAgentQuery, ToolDefinition } from './public/types';
-import { AnthropicBeta } from './public/types';
+import { createAnthropicAgent } from './public/createAnthropicAgent';
+import { AnthropicBeta } from './public/enums';
+import { IAnthropicAgent } from './public/interfaces';
+import type { AnthropicAgentOptions, AnthropicBetaFlags, AnyToolDefinition, ChainedToolStore, ConsumerMessage, ILogger, JsonObject, JsonValue, RunAgentQuery, RunAgentResult, SdkMessage, ToolDefinition } from './public/types';
 
-export type { AgentEvents, AnthropicAgentOptions, AnthropicBetaFlags, AnyToolDefinition, ChainedToolStore, ILogger, JsonObject, JsonValue, RunAgentQuery, ToolDefinition };
-export { AnthropicAgent, AnthropicBeta };
+export type { AnthropicAgentOptions, AnthropicBetaFlags, AnyToolDefinition, ChainedToolStore, ConsumerMessage, ILogger, JsonObject, JsonValue, RunAgentQuery, RunAgentResult, SdkMessage, ToolDefinition };
+export { AnthropicBeta, createAnthropicAgent, IAnthropicAgent };

packages/claude-sdk/src/private/AgentChannel.ts

@@ -0,0 +1,22 @@
+import { MessageChannel, type MessagePort } from 'node:worker_threads';
+import type { ConsumerMessage, SdkMessage } from '../public/types';
+
+export class AgentChannel {
+  readonly #port: MessagePort;
+  public readonly consumerPort: MessagePort;
+
+  public constructor(onMessage: (msg: ConsumerMessage) => void) {
+    const { port1, port2 } = new MessageChannel();
+    this.#port = port1;
+    this.consumerPort = port2;
+    port1.on('message', onMessage);
+  }
+
+  public send(msg: SdkMessage): void {
+    this.#port.postMessage(msg);
+  }
+
+  public close(): void {
+    this.#port.close();
+  }
+}

packages/claude-sdk/src/private/AgentRun.ts

@@ -0,0 +1,210 @@
+import { randomUUID } from 'node:crypto';
+import type { RequestOptions } from 'node:http';
+import type { MessagePort } from 'node:worker_threads';
+import type { Anthropic } from '@anthropic-ai/sdk';
+import type { BetaMessageStreamParams } from '@anthropic-ai/sdk/resources/beta/messages.js';
+import type { BetaCacheControlEphemeral } from '@anthropic-ai/sdk/resources/beta.mjs';
+import { z } from 'zod';
+import type { AnyToolDefinition, ChainedToolStore, ILogger, RunAgentQuery, SdkMessage } from '../public/types';
+import { AgentChannel } from './AgentChannel';
+import { ApprovalState } from './ApprovalState';
+import { AGENT_SDK_PREFIX } from './consts';
+import { MessageStream } from './MessageStream';
+import type { ToolUseResult } from './types';
+
+export class AgentRun {
+  readonly #client: Anthropic;
+  readonly #logger: ILogger | undefined;
+  readonly #options: RunAgentQuery;
+  readonly #channel: AgentChannel;
+  readonly #approval: ApprovalState;
+
+  public constructor(client: Anthropic, logger: ILogger | undefined, options: RunAgentQuery) {
+    this.#client = client;
+    this.#logger = logger;
+    this.#options = options;
+    this.#approval = new ApprovalState();
+    this.#channel = new AgentChannel((msg) => this.#approval.handle(msg));
+  }
+
+  public get port(): MessagePort {
+    return this.#channel.consumerPort;
+  }
+
+  public async execute(): Promise<void> {
+    const messages: Anthropic.Beta.Messages.BetaMessageParam[] = this.#options.messages.map((content) => ({
+      role: 'user',
+      content,
+    }));
+    const store: ChainedToolStore = new Map<string, unknown>();
+
+    try {
+      while (!this.#approval.cancelled) {
+        this.#logger?.debug('messages', { messages });
+        const stream = this.#getMessageStream(messages);
+        this.#logger?.info('Processing messages');
+
+        const messageStream = new MessageStream(this.#logger);
+        messageStream.on('message_start', () => this.#channel.send({ type: 'message_start' }));
+        messageStream.on('message_text', (text) => this.#channel.send({ type: 'message_text', text }));
+        messageStream.on('message_stop', () => this.#channel.send({ type: 'message_end' }));
+
+        let result: Awaited<ReturnType<MessageStream['process']>>;
+        try {
+          result = await messageStream.process(stream);
+        } catch (err) {
+          if (err instanceof Error) {
+            this.#channel.send({ type: 'error', message: err.message });
+          }
+          return;
+        }
+
+        if (result.stopReason !== 'tool_use' || result.toolUses.length === 0) {
+          this.#channel.send({ type: 'done', stopReason: result.stopReason ?? 'end_turn' });
+          break;
+        }
+
+        const toolResults = await this.#handleTools(result.toolUses, store);
+
+        messages.push({
+          role: 'assistant',
+          content: [
+            ...(result.text.length > 0 ? [{ type: 'text' as const, text: result.text }] : []),
+            ...result.toolUses.map((t) => ({
+              type: 'tool_use' as const,
+              id: t.id,
+              name: t.name,
+              input: t.input,
+            })),
+          ],
+        });
+        messages.push({ role: 'user', content: toolResults });
+      }
+    } finally {
+      this.#channel.close();
+    }
+  }
+
+  #getMessageStream(messages: Anthropic.Beta.Messages.BetaMessageParam[]) {
+    const body = {
+      model: this.#options.model,
+      max_tokens: this.#options.maxTokens,
+      tools: this.#options.tools.map((t) => ({
+        name: t.name,
+        description: t.description,
+        input_schema: z.toJSONSchema(t.input_schema) as Anthropic.Tool['input_schema'],
+        input_examples: t.input_examples,
+      })),
+      cache_control: { type: 'ephemeral', scope: 'global' } as BetaCacheControlEphemeral,
+      system: [{ type: 'text', text: AGENT_SDK_PREFIX }],
+      messages,
+      thinking: { type: 'adaptive' },
+      stream: true,
+    } satisfies BetaMessageStreamParams;
+
+    const betas = Object.entries(this.#options.betas ?? {})
+      .filter(([, enabled]) => enabled)
+      .map(([beta]) => beta)
+      .join(',');
+
+    const requestOptions = {
+      headers: { 'anthropic-beta': betas },
+    } satisfies RequestOptions;
+
+    this.#logger?.info('Sending request', {
+      model: this.#options.model,
+      max_tokens: this.#options.maxTokens,
+      tools: this.#options.tools.map((t) => ({ name: t.name, description: t.description })),
+      cache_control: { type: 'ephemeral', scope: 'global' } as BetaCacheControlEphemeral,
+      thinking: { type: 'adaptive' },
+      stream: true,
+      headers: requestOptions.headers,
+    });
+
+    return this.#client.beta.messages.stream(body, requestOptions);
+  }
+
+  async #handleTools(toolUses: ToolUseResult[], store: ChainedToolStore): Promise<Anthropic.Beta.Messages.BetaToolResultBlockParam[]> {
+    const requireApproval = this.#options.requireToolApproval ?? false;
+    const toolResults: Anthropic.Beta.Messages.BetaToolResultBlockParam[] = [];
+
+    // Resolve tools and validate input first. Error immediately without requesting approval.
+    const resolved = [];
+    for (const toolUse of toolUses) {
+      const tool = this.#options.tools.find((t) => t.name === toolUse.name);
+      if (tool == null) {
+        const content = `Tool not found: ${toolUse.name}`;
+        this.#logger?.debug('tool_result_error', { name: toolUse.name, content });
+        toolResults.push({ type: 'tool_result', tool_use_id: toolUse.id, is_error: true, content });
+        continue;
+      }
+      const parseResult = tool.input_schema.safeParse(toolUse.input);
+      if (!parseResult.success) {
+        this.#logger?.debug('tool_parse_error', { name: toolUse.name, error: parseResult.error });
+        toolResults.push({ type: 'tool_result', tool_use_id: toolUse.id, is_error: true, content: `Invalid input: ${parseResult.error.message}` });
+        continue;
+      }
+      resolved.push({ toolUse, tool, input: parseResult.data });
+    }
+
+    if (requireApproval) {
+      // Send all approval requests to the consumer at once
+      const pending = resolved.map(({ toolUse, tool, input }) => {
+        const requestId = randomUUID();
+        return {
+          toolUse,
+          tool,
+          input,
+          promise: this.#approval.request(requestId, () => {
+            this.#channel.send({ type: 'tool_approval_request', requestId, name: toolUse.name, input: toolUse.input } satisfies SdkMessage);
+          }),
+        };
+      });
+
+      // Execute tools in the order approvals arrive
+      while (pending.length > 0) {
+        if (this.#approval.cancelled) {
+          break;
+        }
+        const { toolUse, tool, input, response, index } = await Promise.race(pending.map((item, idx) => item.promise.then((response) => ({ toolUse: item.toolUse, tool: item.tool, input: item.input, response, index: idx }))));
+        pending.splice(index, 1);
+
+        if (!response.approved) {
+          const content = response.reason ?? 'Tool use rejected';
+          this.#logger?.debug('tool_rejected', { name: toolUse.name, reason: content });
+          toolResults.push({ type: 'tool_result', tool_use_id: toolUse.id, is_error: true, content });
+          continue;
+        }
+
+        toolResults.push(await this.#executeTool(toolUse, tool, input, store));
+      }
+    } else {
+      for (const { toolUse, tool, input } of resolved) {
+        if (this.#approval.cancelled) {
+          break;
+        }
+        toolResults.push(await this.#executeTool(toolUse, tool, input, store));
+      }
+    }
+
+    return toolResults;
+  }
+
+  async #executeTool(toolUse: ToolUseResult, tool: AnyToolDefinition, input: unknown, store: ChainedToolStore): Promise<Anthropic.Beta.Messages.BetaToolResultBlockParam> {
+    this.#logger?.debug('tool_call', { name: toolUse.name, input: toolUse.input });
+    const handler = tool.handler as (input: unknown, store: Map<string, unknown>) => Promise<unknown>;
+    try {
+      const toolOutput = await handler(input, store);
+      this.#logger?.debug('tool_result', { name: toolUse.name, output: toolOutput });
+      return {
+        type: 'tool_result',
+        tool_use_id: toolUse.id,
+        content: typeof toolOutput === 'string' ? toolOutput : JSON.stringify(toolOutput),
+      };
+    } catch (err) {
+      const message = err instanceof Error ? err.message : String(err);
+      this.#logger?.debug('tool_handler_error', { name: toolUse.name, error: message });
+      return { type: 'tool_result', tool_use_id: toolUse.id, is_error: true, content: message };
+    }
+  }
+}

packages/claude-sdk/src/private/AnthropicAgent.ts

@@ -0,0 +1,20 @@
+import { Anthropic } from '@anthropic-ai/sdk';
+import { IAnthropicAgent } from '../public/interfaces';
+import type { AnthropicAgentOptions, ILogger, RunAgentQuery, RunAgentResult } from '../public/types';
+import { AgentRun } from './AgentRun';
+
+export class AnthropicAgent extends IAnthropicAgent {
+  readonly #client: Anthropic;
+  readonly #logger: ILogger | undefined;
+
+  public constructor(options: AnthropicAgentOptions) {
+    super();
+    this.#logger = options.logger;
+    this.#client = new Anthropic({ apiKey: options.apiKey });
+  }
+
+  public runAgent(options: RunAgentQuery): RunAgentResult {
+    const run = new AgentRun(this.#client, this.#logger, options);
+    return { port: run.port, done: run.execute() };
+  }
+}

packages/claude-sdk/src/private/ApprovalState.ts

@@ -0,0 +1,30 @@
+import type { ConsumerMessage } from '../public/types';
+import type { ApprovalResponse } from './types';
+
+export class ApprovalState {
+  readonly #pending = new Map<string, (response: ApprovalResponse) => void>();
+  #cancelled = false;
+
+  public get cancelled(): boolean {
+    return this.#cancelled;
+  }
+
+  public handle(msg: ConsumerMessage): void {
+    if (msg.type === 'tool_approval_response') {
+      const resolve = this.#pending.get(msg.requestId);
+      if (resolve != null) {
+        this.#pending.delete(msg.requestId);
+        resolve({ approved: msg.approved, reason: msg.reason });
+      }
+    } else if (msg.type === 'cancel') {
+      this.#cancelled = true;
+    }
+  }
+
+  public request(requestId: string, onRequest: () => void): Promise<ApprovalResponse> {
+    return new Promise<ApprovalResponse>((resolve) => {
+      this.#pending.set(requestId, resolve);
+      onRequest();
+    });
+  }
+}

packages/claude-sdk/src/private/types.ts

@@ -4,6 +4,11 @@ export type ToolUseAccumulator = {
   partialJson: string;
 };
 
+export type ApprovalResponse = {
+  approved: boolean;
+  reason?: string;
+};
+
 export type ToolUseResult = {
   id: string;
   name: string;