Anthropic Streaming Patterns

Anthropic Claude API Streaming Patterns

Overview

Claude API integration with streaming, tool execution, and cost tracking using Anthropic SDK.

Core principle: Stream (don't buffer). Track costs. Handle tools correctly.

Announce at start: "I'm using the anthropic-streaming-patterns skill for Claude API integration."

When to Use

• Implementing Claude API service (Task 3.4)
• Implementing streaming responses
• Implementing tool execution within streams
• Tracking API costs
• Debugging streaming issues

Quick Reference

Streaming Pattern (Complete)

const stream = client.messages.stream({
  model: 'claude-sonnet-4-20250514',
  max_tokens: 8192,
  messages: messageHistory,
  tools: toolDefinitions,
});

let currentToolUse = null;
let accumulatedInput = '';

// Text deltas → forward to client
stream.on('text', (text) => {
  sendToClient({type: 'content_delta', delta: text});
});

// Tool use started
stream.on('contentBlockStart', (block) => {
  if (block.type === 'tool_use') {
    currentToolUse = {name: block.name, id: block.id};
    accumulatedInput = '';
    sendToClient({type: 'tool_execution', tool: block.name});
  }
});

// Tool input accumulation
stream.on('contentBlockDelta', (delta) => {
  if (delta.type === 'input_json_delta' && currentToolUse) {
    accumulatedInput = delta.partial_json;
  }
});

// Tool execution
stream.on('contentBlockStop', async () => {
  if (currentToolUse) {
    const input = JSON.parse(accumulatedInput);
    const result = await executeTool(currentToolUse.name, input);
    
    sendToClient({type: 'tool_result', result});
    currentToolUse = null;
  }
});

// Stream complete with usage
stream.on('message', (message) => {
  if (message.usage) {
    const inputCost = (message.usage.input_tokens / 1000) * 0.003;
    const outputCost = (message.usage.output_tokens / 1000) * 0.015;
    
    saveSessionCost(sessionId, {
      inputTokens: message.usage.input_tokens,
      outputTokens: message.usage.output_tokens,
      cost: inputCost + outputCost
    });
  }
});

stream.on('error', (error) => {
  logger.error('Streaming error:', error);
  sendToClient({type: 'error', error: error.message});
});

await stream.finalMessage(); // Wait for completion

Cost Tracking (MANDATORY)

const PRICING = {
  input: 0.003,  // $0.003 per 1k tokens
  output: 0.015, // $0.015 per 1k output tokens
};

// Calculate per message
const cost = {
  input: (inputTokens / 1000) * PRICING.input,
  output: (outputTokens / 1000) * PRICING.output,
  total: inputCost + outputCost
};

// Aggregate per session
sessionCosts.push(cost);
const sessionTotal = sessionCosts.reduce((sum, c) => sum + c.total, 0);

Error Handling

try {
  const stream = await client.messages.stream({...});
} catch (error) {
  if (error.status === 429) {
    // Rate limit - wait and retry
    await delay(60000);
    return retry();
  } else if (error.status === 401) {
    // Auth error
    throw new Error('Invalid API key');
  } else {
    logger.error(error);
    throw error;
  }
}

Common Mistakes

❌ WRONG: Buffering

const response = await client.messages.create({...}); // Buffering
const fullText = response.content[0].text;
sendToClient(fullText);

✅ CORRECT: Streaming

const stream = await client.messages.stream({...});
stream.on('text', (delta) => sendToClient({type: 'content_delta', delta}));

Red Flags

• "Buffering is easier" → WRONG. Stream for real-time.
• "Cost tracking is overhead" → WRONG. Mandatory feature.
• "Skip error handling" → WRONG. Streams fail often.

Integration

• Use FOR: Task 3.4 (claude.service.ts)
• Use WITH: @claude-mobile-cost-tracking
• Integrate: Task 3.11 (cost.service.ts)