import { query } from '@anthropic-ai/claude-agent-sdk'; import type { SDKMessage, SDKAssistantMessage, SDKUserMessage, SDKResultMessage, SDKPartialAssistantMessage, SDKSystemMessage, SDKToolProgressMessage, Options, McpStdioServerConfig, NotificationHookInput, PostToolUseHookInput, } from '@anthropic-ai/claude-agent-sdk'; import type { ClaudeStreamOptions, SSEEvent, TokenUsage, MCPServerConfig, PermissionRequestEvent } from '@/types'; import { registerPendingPermission } from './permission-registry'; import { execFileSync } from 'child_process'; import os from 'os'; import path from 'path'; let cachedClaudePath: string | null | undefined; function findClaudePath(): string | undefined { if (cachedClaudePath !== undefined) return cachedClaudePath || undefined; const home = os.homedir(); const candidates = [ '/usr/local/bin/claude', '/opt/homebrew/bin/claude', path.join(home, '.npm-global', 'bin', 'claude'), path.join(home, '.local', 'bin', 'claude'), path.join(home, '.claude', 'bin', 'claude'), ]; for (const p of candidates) { try { execFileSync(p, ['--version'], { timeout: 3000, stdio: 'pipe' }); cachedClaudePath = p; return p; } catch { // not found } } // Fallback: which claude const extra = ['/usr/local/bin', '/opt/homebrew/bin', '/usr/bin', '/bin', path.join(home, '.npm-global', 'bin'), path.join(home, '.nvm', 'current', 'bin'), path.join(home, '.local', 'bin'), path.join(home, '.claude', 'bin')]; const current = (process.env.PATH || '').split(':'); for (const p of extra) { if (!current.includes(p)) current.push(p); } try { const result = execFileSync('/usr/bin/which', ['claude'], { timeout: 3000, stdio: 'pipe', env: { ...process.env, PATH: current.join(':') }, }); const found = result.toString().trim(); if (found) { cachedClaudePath = found; return found; } } catch { /* not found */ } cachedClaudePath = null; return undefined; } /** * Convert our MCPServerConfig to the SDK's McpStdioServerConfig format */ function toSdkMcpConfig( servers: Record ): Record { const result: Record = {}; for (const [name, config] of Object.entries(servers)) { result[name] = { command: config.command, args: config.args, env: config.env, }; } return result; } /** * Format an SSE line from an event object */ function formatSSE(event: SSEEvent): string { return `data: ${JSON.stringify(event)}\n\n`; } /** * Extract text content from an SDK assistant message */ function extractTextFromMessage(msg: SDKAssistantMessage): string { const parts: string[] = []; for (const block of msg.message.content) { if (block.type === 'text') { parts.push(block.text); } } return parts.join(''); } /** * Extract token usage from an SDK result message */ function extractTokenUsage(msg: SDKResultMessage): TokenUsage | null { if (!msg.usage) return null; return { input_tokens: msg.usage.input_tokens, output_tokens: msg.usage.output_tokens, cache_read_input_tokens: msg.usage.cache_read_input_tokens ?? 0, cache_creation_input_tokens: msg.usage.cache_creation_input_tokens ?? 0, cost_usd: 'total_cost_usd' in msg ? msg.total_cost_usd : undefined, }; } /** * Stream Claude responses using the Agent SDK. * Returns a ReadableStream of SSE-formatted strings. */ export function streamClaude(options: ClaudeStreamOptions): ReadableStream { const { prompt, sdkSessionId, model, systemPrompt, workingDirectory, mcpServers, abortController, permissionMode, } = options; return new ReadableStream({ async start(controller) { try { const queryOptions: Options = { cwd: workingDirectory || process.cwd(), abortController, includePartialMessages: true, permissionMode: (permissionMode as Options['permissionMode']) || 'acceptEdits', }; // Find claude binary for packaged app where PATH is limited const claudePath = findClaudePath(); if (claudePath) { queryOptions.pathToClaudeCodeExecutable = claudePath; } if (model) { queryOptions.model = model; } if (systemPrompt) { queryOptions.systemPrompt = systemPrompt; } if (mcpServers && Object.keys(mcpServers).length > 0) { queryOptions.mcpServers = toSdkMcpConfig(mcpServers); } // Resume session if we have an SDK session ID from a previous conversation turn if (sdkSessionId) { queryOptions.resume = sdkSessionId; } // Permission handler: sends SSE event and waits for user response queryOptions.canUseTool = async (toolName, input, opts) => { const permissionRequestId = `perm-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; const permEvent: PermissionRequestEvent = { permissionRequestId, toolName, toolInput: input, suggestions: opts.suggestions as PermissionRequestEvent['suggestions'], decisionReason: opts.decisionReason, blockedPath: opts.blockedPath, toolUseId: opts.toolUseID, description: undefined, }; // Send permission_request SSE event to the client controller.enqueue(formatSSE({ type: 'permission_request', data: JSON.stringify(permEvent), })); // Wait for user response (resolved by POST /api/chat/permission) // Store original input so registry can inject updatedInput on allow return registerPendingPermission(permissionRequestId, input, opts.signal); }; // Hooks: capture notifications and tool completion events queryOptions.hooks = { Notification: [{ hooks: [async (input) => { const notif = input as NotificationHookInput; controller.enqueue(formatSSE({ type: 'status', data: JSON.stringify({ notification: true, title: notif.title, message: notif.message, }), })); return {}; }], }], PostToolUse: [{ hooks: [async (input) => { const toolEvent = input as PostToolUseHookInput; controller.enqueue(formatSSE({ type: 'tool_result', data: JSON.stringify({ tool_use_id: toolEvent.tool_use_id, content: typeof toolEvent.tool_response === 'string' ? toolEvent.tool_response : JSON.stringify(toolEvent.tool_response), is_error: false, }), })); return {}; }], }], }; // Capture real-time stderr output from Claude Code process queryOptions.stderr = (data: string) => { // Diagnostic: log raw stderr data length to server console console.log(`[stderr] received ${data.length} bytes, first 200 chars:`, data.slice(0, 200).replace(/[\x00-\x1F\x7F]/g, '?')); // Strip ANSI escape codes, OSC sequences, and control characters // but preserve tabs (\x09) and carriage returns (\x0D) const cleaned = data .replace(/\x1B\[[0-9;]*[a-zA-Z]/g, '') // CSI sequences (colors, cursor) .replace(/\x1B\][^\x07\x1B]*(?:\x07|\x1B\\)/g, '') // OSC sequences .replace(/\x1B\([A-Z]/g, '') // Character set selection .replace(/\x1B[=>]/g, '') // Keypad mode .replace(/[\x00-\x08\x0B\x0C\x0E-\x1F]/g, '') // Control chars (keep \t \n \r) .replace(/\r\n/g, '\n') // Normalize CRLF .replace(/\r/g, '\n') // Convert remaining CR to LF .replace(/\n{3,}/g, '\n\n') // Collapse multiple blank lines .trim(); if (cleaned) { controller.enqueue(formatSSE({ type: 'tool_output', data: cleaned, })); } }; const conversation = query({ prompt, options: queryOptions, }); let lastAssistantText = ''; let tokenUsage: TokenUsage | null = null; for await (const message of conversation) { if (abortController?.signal.aborted) { break; } switch (message.type) { case 'assistant': { const assistantMsg = message as SDKAssistantMessage; const text = extractTextFromMessage(assistantMsg); if (text && text !== lastAssistantText) { // Send the new delta const delta = text.slice(lastAssistantText.length); if (delta) { controller.enqueue(formatSSE({ type: 'text', data: delta })); } lastAssistantText = text; } // Check for tool use blocks for (const block of assistantMsg.message.content) { if (block.type === 'tool_use') { controller.enqueue(formatSSE({ type: 'tool_use', data: JSON.stringify({ id: block.id, name: block.name, input: block.input, }), })); } } break; } case 'user': { // Tool execution results come back as user messages with tool_result blocks const userMsg = message as SDKUserMessage; const content = userMsg.message.content; if (Array.isArray(content)) { for (const block of content) { if (block.type === 'tool_result') { const resultContent = typeof block.content === 'string' ? block.content : Array.isArray(block.content) ? block.content .filter((c: { type: string }) => c.type === 'text') .map((c: { text: string }) => c.text) .join('\n') : String(block.content ?? ''); controller.enqueue(formatSSE({ type: 'tool_result', data: JSON.stringify({ tool_use_id: block.tool_use_id, content: resultContent, is_error: block.is_error || false, }), })); } } } break; } case 'stream_event': { const streamEvent = message as SDKPartialAssistantMessage; const evt = streamEvent.event; if (evt.type === 'content_block_delta' && 'delta' in evt) { const delta = evt.delta; if ('text' in delta && delta.text) { controller.enqueue(formatSSE({ type: 'text', data: delta.text })); } } break; } case 'system': { const sysMsg = message as SDKSystemMessage; if ('subtype' in sysMsg) { if (sysMsg.subtype === 'init') { controller.enqueue(formatSSE({ type: 'status', data: JSON.stringify({ session_id: sysMsg.session_id, model: sysMsg.model, tools: sysMsg.tools, }), })); } } break; } case 'tool_progress': { const progressMsg = message as SDKToolProgressMessage; controller.enqueue(formatSSE({ type: 'tool_output', data: JSON.stringify({ _progress: true, tool_use_id: progressMsg.tool_use_id, tool_name: progressMsg.tool_name, elapsed_time_seconds: progressMsg.elapsed_time_seconds, }), })); break; } case 'result': { const resultMsg = message as SDKResultMessage; tokenUsage = extractTokenUsage(resultMsg); controller.enqueue(formatSSE({ type: 'result', data: JSON.stringify({ subtype: resultMsg.subtype, is_error: resultMsg.is_error, num_turns: resultMsg.num_turns, duration_ms: resultMsg.duration_ms, usage: tokenUsage, session_id: resultMsg.session_id, }), })); break; } } } controller.enqueue(formatSSE({ type: 'done', data: '' })); controller.close(); } catch (error) { const errorMessage = error instanceof Error ? error.message : 'Unknown error'; controller.enqueue(formatSSE({ type: 'error', data: errorMessage })); controller.enqueue(formatSSE({ type: 'done', data: '' })); controller.close(); } }, cancel() { abortController?.abort(); }, }); }