claude-sdk-integration-patterns

Claude SDK 集成模式

将 Claude API 和 TypeScript SDK 集成到应用中的生产就绪模式。掌握流式响应、工具执行、错误处理和 AI 驱动功能的优化策略。

何时使用此技能

在以下情况下使用此技能：

• 将 Claude API 集成到 Node.js/TypeScript 应用中
• 实现可实时响应的流式对话
• 构建使用 Claude 工具调用（函数调用）的应用
• 通过重试逻辑妥善处理 API 错误
• 优化 token 用量并管理成本
• 将 Claude 驱动的功能部署到生产环境
• 构建带有上下文管理的多轮对话
• 为高并发处理实现消息批处理

核心概念

Messages API 基础

Claude Messages API 是对话式 AI 的主要接口：

关键组件：

• 模型选择：选择合适的模型（Opus、Sonnet、Haiku）
• 消息数组：包含 user/assistant 角色的对话历史
• 最大 token 数：控制响应长度
• 系统提示词：引导模型行为
• 流式传输：实时生成响应

SDK 架构

TypeScript SDK 提供：

• 类型安全的 API 客户端
• 用于实时响应的流式辅助工具
• 工具执行框架
• 错误处理实用工具
• 消息批处理
• 事件驱动架构

安装和设置

# Install the SDK
npm install @anthropic-ai/sdk

# Or with yarn
yarn add @anthropic-ai/sdk

import Anthropic from '@anthropic-ai/sdk';

// Initialize client
const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

Messages API 模式

模式 1：创建基本消息

const message = await anthropic.messages.create({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: 'Hello, Claude!' }
  ],
});

console.log(message.content);

适用场景：

• 简单的问答交互
• 一次性 API 调用
• 同步工作流

模式 2：多轮对话

const messages = [
  { role: 'user', content: 'What is TypeScript?' },
  { role: 'assistant', content: 'TypeScript is a typed superset of JavaScript...' },
  { role: 'user', content: 'Give me an example' },
];

const response = await anthropic.messages.create({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1024,
  messages,
});

适用场景：

• 聊天机器人和对话式 UI
• 多步骤工作流
• 依赖上下文的交互

模式 3：系统提示词

const response = await anthropic.messages.create({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1024,
  system: 'You are a helpful Python programming assistant. Provide concise, tested code examples.',
  messages: [
    { role: 'user', content: 'How do I read a CSV file?' }
  ],
});

适用场景：

• 专业助手
• 角色扮演场景
• 跨对话保持一致的行为

流式传输模式

模式 4：基本流式传输

const stream = await anthropic.messages.create({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Write a story' }],
  stream: true,
});

for await (const event of stream) {
  if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
    process.stdout.write(event.delta.text);
  }
}

适用场景：

• 实时用户界面
• 长篇内容生成
• 交互式体验

模式 5：使用事件处理器进行流式传输

const stream = anthropic.messages.stream({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Explain quantum computing' }],
})
  .on('text', (text) => {
    console.log(text);
  })
  .on('message', (message) => {
    console.log('Complete message:', message);
  })
  .on('error', (error) => {
    console.error('Stream error:', error);
  });

const finalMessage = await stream.finalMessage();

适用场景：

• 实时 UI（聊天机器人、实时编辑器）
• 进度指示器
• 部分结果处理

模式 6：带中止控制的流式传输

const stream = anthropic.messages.stream({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Long task...' }],
});

// Abort after timeout
setTimeout(() => stream.abort(), 5000);

try {
  await stream.done();
} catch (error) {
  if (error instanceof Anthropic.APIUserAbortError) {
    console.log('Stream aborted by user');
  }
}

适用场景：

• 用户可取消的操作
• 超时处理
• 资源管理

工具调用模式

模式 7：使用 Zod 定义工具

import { betaZodTool } from '@anthropic-ai/sdk/helpers/zod';
import { z } from 'zod';

const weatherTool = betaZodTool({
  name: 'get_weather',
  inputSchema: z.object({
    location: z.string(),
    unit: z.enum(['celsius', 'fahrenheit']).default('fahrenheit'),
  }),
  description: 'Get current weather for a location',
  run: async (input) => {
    // Call weather API
    return `Weather in ${input.location}: 72°F, sunny`;
  },
});

适用场景：

• 类型安全的工具定义
• 输入验证
• 清晰的工具约定

模式 8：用于自动执行的工具运行器

const finalMessage = await anthropic.beta.messages.toolRunner({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1000,
  messages: [
    { role: 'user', content: 'What\'s the weather in San Francisco?' }
  ],
  tools: [weatherTool],
});

console.log(finalMessage.content);

适用场景：

• 自动化工具执行
• 使用函数调用的 AI Agent
• 复杂的多步骤工作流

模式 9：流式工具执行

const runner = anthropic.beta.messages.toolRunner({
  model: 'claude-sonnet-4-5-20250929',
  max_tokens: 1000,
  messages: [{ role: 'user', content: 'Check weather and calculate travel time' }],
  tools: [weatherTool, travelTimeTool],
  stream: true,
});

for await (const messageStream of runner) {
  for await (const event of messageStream) {
    console.log('Event:', event);
  }
  console.log('Message:', await messageStream.finalMessage());
}

适用场景：

• 实时工具执行反馈
• 多工具工作流
• 交互式 AI Agent

错误处理模式

模式 10：全面的错误处理

async function createMessage(prompt: string) {
  try {
    const message = await anthropic.messages.create({
      model: 'claude-sonnet-4-5-20250929',
      max_tokens: 1024,
      messages: [{ role: 'user', content: prompt }],
    });

    return message;
  } catch (error) {
    if (error instanceof Anthropic.APIError) {
      console.error('API Error:', {
        status: error.status,
        name: error.name,
        message: error.message,
        headers: error.headers,
      });

      // Handle specific errors
      if (error.status === 429) {
        // Rate limit - implement backoff
        console.log('Rate limited, waiting...');
        await new Promise(resolve => setTimeout(resolve, 60000));
        return createMessage(prompt); // Retry
      } else if (error.status === 401) {
        throw new Error('Invalid API key');
      } else if (error.status === 400) {
        throw new Error(`Bad request: ${error.message}`);
      }
    }

    throw error;
  }
}

适用场景：

• 生产应用
• 稳健的错误恢复
• 面向用户的应用

模式 11：指数退避重试

async function createWithRetry(
  params: Anthropic.MessageCreateParams,
  maxRetries = 3,
  baseDelay = 1000
): Promise<Anthropic.Message> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await anthropic.messages.create(params);
    } catch (error) {
      if (error instanceof Anthropic.APIError && error.status === 429) {
        // Rate limit - exponential backoff
        const delay = baseDelay * Math.pow(2, attempt);
        console.log(`Retry attempt ${attempt + 1} after ${delay}ms`);
        await new Promise(resolve => setTimeout(resolve, delay));
      } else {
        throw error;
      }
    }
  }

  throw new Error('Max retries exceeded');
}

适用场景：

• 高并发应用
• 速率限制处理
• 网络可靠性

Token 优化模式

模式 12：Token 计数

import { encode } from 'gpt-tokenizer';

function estimateTokens(text: string): number {
  // Approximate token count (Claude uses similar tokenization to GPT)
  return encode(text).length;
}

function optimizePrompt(userMessage: string, maxTokens: number = 1024): string {
  const tokens = estimateTokens(userMessage);

  if (tokens > maxTokens) {
    // Truncate or summarize
    return userMessage.slice(0, maxTokens * 4); // Rough char estimate
  }

  return userMessage;
}

适用场景：

• 成本优化
• Token 预算管理
• 大型输入处理

模式 13：上下文窗口管理

interface ConversationMessage {
  role: 'user' | 'assistant';
  content: string;
  tokens?: number;
}

class ConversationManager {
  private messages: ConversationMessage[] = [];
  private maxContextTokens = 100000;

  addMessage(role: 'user' | 'assistant', content: string) {
    const tokens = estimateTokens(content);
    this.messages.push({ role, content, tokens });
    this.pruneIfNeeded();
  }

  private pruneIfNeeded() {
    const totalTokens = this.messages.reduce((sum, msg) => sum + (msg.tokens || 0), 0);

    if (totalTokens > this.maxContextTokens) {
      // Remove oldest messages (keep system prompt if present)
      this.messages = this.messages.slice(-10); // Keep last 10 messages
    }
  }

  getMessages() {
    return this.messages.map(({ role, content }) => ({ role, content }));
  }
}

适用场景：

• 长对话
• 上下文管理
• 成本控制

消息批处理模式

模式 14：批处理

const batchResult = await anthropic.messages.batches.create({
  requests: [
    {
      custom_id: 'request-1',
      params: {
        model: 'claude-sonnet-4-5-20250929',
        max_tokens: 1024,
        messages: [{ role: 'user', content: 'Summarize this article...' }],
      },
    },
    {
      custom_id: 'request-2',
      params: {
        model: 'claude-sonnet-4-5-20250929',
        max_tokens: 1024,
        messages: [{ role: 'user', content: 'Translate this text...' }],
      },
    },
  ],
});

// Poll for results
const batch = await anthropic.messages.batches.retrieve(batchResult.id);

适用场景：

• 批量处理
• 离线工作流
• 成本优化（更低的批处理定价）

生产模式

模式 15：请求超时配置

const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
  timeout: 60000, // 60 seconds
  maxRetries: 3,
});

适用场景：

• 生产环境部署
• 网络可靠性
• 资源管理

模式 16：自定义标头和元数据

const message = await anthropic.messages.create(
  {
    model: 'claude-sonnet-4-5-20250929',
    max_tokens: 1024,
    messages: [{ role: 'user', content: 'Hello' }],
  },
  {
    headers: {
      'anthropic-beta': 'max-tokens-3-5-sonnet-2024-07-15',
    },
  }
);

适用场景：

• Beta 功能
• 自定义跟踪
• A/B 测试

最佳实践

API 密钥管理

1. 切勿将 API 密钥硬编码

   // ❌ Bad
   const anthropic = new Anthropic({ apiKey: 'sk-ant-...' });
   
   // ✅ Good
   const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
   

2. 使用环境变量

   # .env file
   ANTHROPIC_API_KEY=sk-ant-your-key-here
   

模型选择

1. 根据任务选择合适的模型：

   • Opus：能力最强，适用于复杂推理
   • Sonnet：性能与速度均衡
   • Haiku：速度最快，适用于简单任务

2. 使用最新的模型版本：

   model: 'claude-sonnet-4-5-20250929' // Latest
   

流式传输注意事项

1. 以下情况始终使用流式传输：

   • 面向用户的应用
   • 长篇内容（>500 个 token）
   • 实时交互

2. 以下情况避免使用流式传输：

   • 后端批处理
   • 需要预先获得确定性输出
   • 简单的 API 集成

错误处理

1. 始终捕获 API 错误
2. 为瞬时故障实现重试逻辑
3. 记录错误以便调试
4. 提供易于用户理解的错误消息

Token 管理

1. 在调用 API 前估算 token 数量
2. 设置合适的 max_tokens
3. 监控 token 用量
4. 为长对话实现上下文裁剪

故障排除

问题：速率限制错误（429）

解决方案：

• 实现指数退避
• 对批量操作使用消息批处理
• 升级到更高的速率限制等级

问题：超时错误

解决方案：

• 对较长的响应使用流式传输
• 增加超时配置
• 减少 max_tokens

问题：API 密钥无效（401）

解决方案：

• 确认已设置环境变量
• 检查 API 密钥格式（以 sk-ant- 开头）
• 确认密钥尚未轮换

问题：超出上下文长度

解决方案：

• 实现对话裁剪
• 总结较早的消息
• 使用合适的上下文窗口限制

资源

• Anthropic API 文档
• TypeScript SDK GitHub
• Claude API 定价
• 模型对比

---

技能版本：1.0.0 上次更新：2025 年 10 月 兼容性：@anthropic-ai/sdk 0.30.0+ 研究来源：Context7 - Anthropic TypeScript SDK（106 个代码片段，信任评分 8.8）