anthropic

Anthropic SDK - 官方 Claude AI 集成

---

progressive_disclosure: entry_point: summary: "Claude AI 官方 SDK — 支持聊天、流式传输、函数调用和视觉" when_to_use: - "将 Claude AI 集成到应用程序时" - "使用 Claude 模型构建 AI 驱动的功能时" - "使用函数调用／工具使用模式时" - "使用视觉模型处理图像时" - "实现流式聊天界面时" quick_start: - "pip install anthropic（Python）或 npm install @anthropic-ai/sdk（TypeScript）" - "设置 ANTHROPIC_API_KEY 环境变量" - "创建客户端，并使用 Messages API 发送消息" - "使用流式传输获取实时响应" installation: python: "pip install anthropic" typescript: "npm install @anthropic-ai/sdk" config: - "ANTHROPIC_API_KEY：从 console.anthropic.com 获取的 API 密钥" - "Model：claude-3-5-sonnet-20241022（推荐）" - "Max tokens：响应使用 1024-8192" token_estimate: entry: 85 full: 5000

安装与设置

Python

pip install anthropic

TypeScript

npm install @anthropic-ai/sdk

API 密钥配置

export ANTHROPIC_API_KEY='your-api-key-here'

从以下位置获取 API 密钥：https://console.anthropic.com/settings/keys

---

Messages API - 基本用法

Python - 简单消息

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY")
)

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Explain quantum computing in simple terms"}
    ]
)

print(message.content[0].text)

TypeScript - 简单消息

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

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

const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: 'Explain quantum computing in simple terms' }
  ],
});

console.log(message.content[0].text);

系统提示词

# Python - System prompt for context
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system="You are a helpful coding assistant specializing in Python and TypeScript.",
    messages=[
        {"role": "user", "content": "How do I handle errors in async functions?"}
    ]
)

// TypeScript - System prompt
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  system: 'You are a helpful coding assistant specializing in Python and TypeScript.',
  messages: [
    { role: 'user', content: 'How do I handle errors in async functions?' }
  ],
});

---

流式响应

Python - 流式传输

# Real-time streaming responses
with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Write a short poem about coding"}
    ]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Python - 异步流式传输

import asyncio

async def stream_response():
    async with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Explain recursion"}
        ]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(stream_response())

TypeScript - 流式传输

// Streaming with event handlers
const stream = await client.messages.stream({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: 'Write a short poem about coding' }
  ],
});

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

---

函数调用／工具使用

Python - 函数调用

# Define tools (functions)
tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name, e.g., San Francisco, CA"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["location"]
        }
    }
]

# Initial request
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "What's the weather in San Francisco?"}
    ]
)

# Check for tool use
if message.stop_reason == "tool_use":
    tool_use = next(block for block in message.content if block.type == "tool_use")
    tool_name = tool_use.name
    tool_input = tool_use.input

    # Execute function (mock example)
    if tool_name == "get_weather":
        weather_result = {
            "temperature": 72,
            "unit": "fahrenheit",
            "conditions": "sunny"
        }

    # Send result back to Claude
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=[
            {"role": "user", "content": "What's the weather in San Francisco?"},
            {"role": "assistant", "content": message.content},
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_use.id,
                        "content": str(weather_result)
                    }
                ]
            }
        ]
    )
    print(response.content[0].text)

TypeScript - 函数调用

// Define tools
const tools: Anthropic.Tool[] = [
  {
    name: 'get_weather',
    description: 'Get the current weather for a location',
    input_schema: {
      type: 'object',
      properties: {
        location: {
          type: 'string',
          description: 'City name, e.g., San Francisco, CA',
        },
        unit: {
          type: 'string',
          enum: ['celsius', 'fahrenheit'],
          description: 'Temperature unit',
        },
      },
      required: ['location'],
    },
  },
];

// Initial request
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  tools,
  messages: [
    { role: 'user', content: "What's the weather in San Francisco?" },
  ],
});

// Check for tool use
if (message.stop_reason === 'tool_use') {
  const toolUse = message.content.find(
    (block): block is Anthropic.ToolUseBlock => block.type === 'tool_use'
  );

  if (toolUse && toolUse.name === 'get_weather') {
    // Execute function
    const weatherResult = {
      temperature: 72,
      unit: 'fahrenheit',
      conditions: 'sunny',
    };

    // Send result back
    const response = await client.messages.create({
      model: 'claude-3-5-sonnet-20241022',
      max_tokens: 1024,
      tools,
      messages: [
        { role: 'user', content: "What's the weather in San Francisco?" },
        { role: 'assistant', content: message.content },
        {
          role: 'user',
          content: [
            {
              type: 'tool_result',
              tool_use_id: toolUse.id,
              content: JSON.stringify(weatherResult),
            },
          ],
        },
      ],
    });

    console.log(response.content[0].text);
  }
}

---

视觉模型 - 图像输入

Python - 图像分析

import base64

# Load image
with open("image.jpg", "rb") as image_file:
    image_data = base64.standard_b64encode(image_file.read()).decode("utf-8")

# Send image to Claude
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/jpeg",
                        "data": image_data,
                    },
                },
                {
                    "type": "text",
                    "text": "Describe this image in detail"
                }
            ],
        }
    ],
)

print(message.content[0].text)

TypeScript - 图像分析

import * as fs from 'fs';

// Load image
const imageData = fs.readFileSync('image.jpg').toString('base64');

// Send image to Claude
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    {
      role: 'user',
      content: [
        {
          type: 'image',
          source: {
            type: 'base64',
            media_type: 'image/jpeg',
            data: imageData,
          },
        },
        {
          type: 'text',
          text: 'Describe this image in detail',
        },
      ],
    },
  ],
});

console.log(message.content[0].text);

---

提示词缓存（Beta）

通过缓存重复的提示词内容来降低成本。

Python - 提示词缓存

# Cache system prompt and long context
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are an expert Python developer...",
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "How do I use async/await?"
        }
    ]
)

# Subsequent requests reuse cached system prompt

TypeScript - 提示词缓存

const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  system: [
    {
      type: 'text',
      text: 'You are an expert TypeScript developer...',
      cache_control: { type: 'ephemeral' },
    },
  ],
  messages: [
    { role: 'user', content: 'How do I use async/await?' },
  ],
});

缓存优势：

• 降低重复内容的延迟
• 降低成本（缓存的 token 按优惠费率计费）
• 适用于较长的系统提示词、文档和示例

---

FastAPI 集成（Python）

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import os

app = FastAPI()
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

class ChatRequest(BaseModel):
    message: str
    stream: bool = False

@app.post("/chat")
async def chat(request: ChatRequest):
    try:
        if request.stream:
            # Streaming response
            async def generate():
                async with client.messages.stream(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=1024,
                    messages=[{"role": "user", "content": request.message}]
                ) as stream:
                    async for text in stream.text_stream:
                        yield text

            return StreamingResponse(generate(), media_type="text/plain")
        else:
            # Non-streaming response
            message = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": request.message}]
            )
            return {"response": message.content[0].text}

    except anthropic.APIError as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/chat/tools")
async def chat_with_tools(request: ChatRequest):
    tools = [
        {
            "name": "search_database",
            "description": "Search the knowledge database",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    ]

    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=[{"role": "user", "content": request.message}]
    )

    return {"response": message.content, "stop_reason": message.stop_reason}

---

Express 集成（TypeScript）

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

const app = express();
app.use(express.json());

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

interface ChatRequest {
  message: string;
  stream?: boolean;
}

app.post('/chat', async (req, res) => {
  const { message, stream }: ChatRequest = req.body;

  try {
    if (stream) {
      // Streaming response
      res.setHeader('Content-Type', 'text/plain');
      res.setHeader('Transfer-Encoding', 'chunked');

      const streamResponse = await client.messages.stream({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: message }],
      });

      for await (const chunk of streamResponse) {
        if (chunk.type === 'content_block_delta' &&
            chunk.delta.type === 'text_delta') {
          res.write(chunk.delta.text);
        }
      }
      res.end();
    } else {
      // Non-streaming response
      const response = await client.messages.create({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: message }],
      });

      res.json({ response: response.content[0].text });
    }
  } catch (error) {
    if (error instanceof Anthropic.APIError) {
      res.status(500).json({ error: error.message });
    } else {
      res.status(500).json({ error: 'Internal server error' });
    }
  }
});

app.listen(3000, () => {
  console.log('Server running on port 3000');
});

---

错误处理与重试

Python - 错误处理

from anthropic import (
    APIError,
    APIConnectionError,
    RateLimitError,
    APITimeoutError
)
import time

def chat_with_retry(message_content: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": message_content}]
            )
            return message.content[0].text

        except RateLimitError as e:
            if attempt < max_retries - 1:
                # Exponential backoff
                wait_time = 2 ** attempt
                print(f"Rate limit hit, waiting {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise

        except APIConnectionError as e:
            if attempt < max_retries - 1:
                print(f"Connection error, retrying...")
                time.sleep(1)
            else:
                raise

        except APITimeoutError as e:
            if attempt < max_retries - 1:
                print(f"Timeout, retrying...")
                time.sleep(2)
            else:
                raise

        except APIError as e:
            # Don't retry on general API errors
            print(f"API error: {e}")
            raise

TypeScript - 错误处理

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

async function chatWithRetry(
  messageContent: string,
  maxRetries: number = 3
): Promise<string> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const message = await client.messages.create({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: messageContent }],
      });

      return message.content[0].text;
    } catch (error) {
      if (error instanceof Anthropic.RateLimitError) {
        if (attempt < maxRetries - 1) {
          const waitTime = Math.pow(2, attempt) * 1000;
          console.log(`Rate limit hit, waiting ${waitTime}ms...`);
          await new Promise(resolve => setTimeout(resolve, waitTime));
        } else {
          throw error;
        }
      } else if (error instanceof Anthropic.APIConnectionError) {
        if (attempt < maxRetries - 1) {
          console.log('Connection error, retrying...');
          await new Promise(resolve => setTimeout(resolve, 1000));
        } else {
          throw error;
        }
      } else {
        // Don't retry on other errors
        throw error;
      }
    }
  }

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

---

Token 计数与成本管理

Python - Token 计数

# Get token usage from response
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

print(f"Input tokens: {message.usage.input_tokens}")
print(f"Output tokens: {message.usage.output_tokens}")

# Calculate cost (example rates)
INPUT_COST_PER_1K = 0.003  # $3 per million tokens
OUTPUT_COST_PER_1K = 0.015  # $15 per million tokens

input_cost = (message.usage.input_tokens / 1000) * INPUT_COST_PER_1K
output_cost = (message.usage.output_tokens / 1000) * OUTPUT_COST_PER_1K
total_cost = input_cost + output_cost

print(f"Total cost: ${total_cost:.6f}")

TypeScript - Token 计数

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

console.log(`Input tokens: ${message.usage.input_tokens}`);
console.log(`Output tokens: ${message.usage.output_tokens}`);

// Calculate cost
const INPUT_COST_PER_1K = 0.003;
const OUTPUT_COST_PER_1K = 0.015;

const inputCost = (message.usage.input_tokens / 1000) * INPUT_COST_PER_1K;
const outputCost = (message.usage.output_tokens / 1000) * OUTPUT_COST_PER_1K;
const totalCost = inputCost + outputCost;

console.log(`Total cost: $${totalCost.toFixed(6)}`);

---

最佳实践

温度与参数

# Low temperature (0.0-0.3) for factual, deterministic responses
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    temperature=0.1,  # More focused
    messages=[{"role": "user", "content": "What is 2+2?"}]
)

# Higher temperature (0.7-1.0) for creative responses
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=2048,
    temperature=0.9,  # More creative
    messages=[{"role": "user", "content": "Write a creative story"}]
)

# Top-p (nucleus sampling)
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    top_p=0.9,  # Consider top 90% probability mass
    messages=[{"role": "user", "content": "Brainstorm ideas"}]
)

速率限制策略

from datetime import datetime, timedelta
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window  # seconds
        self.requests = deque()

    def can_proceed(self) -> bool:
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.time_window)

        # Remove old requests
        while self.requests and self.requests[0] < cutoff:
            self.requests.popleft()

        return len(self.requests) < self.max_requests

    def add_request(self):
        self.requests.append(datetime.now())

# Usage: 50 requests per minute
limiter = RateLimiter(max_requests=50, time_window=60)

if limiter.can_proceed():
    limiter.add_request()
    message = client.messages.create(...)
else:
    print("Rate limit reached, waiting...")

对话管理

# Multi-turn conversation
conversation = []

def chat(user_message: str):
    # Add user message
    conversation.append({"role": "user", "content": user_message})

    # Send to Claude
    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=conversation
    )

    # Add assistant response
    conversation.append({
        "role": "assistant",
        "content": message.content
    })

    return message.content[0].text

# Multi-turn usage
response1 = chat("What is Python?")
response2 = chat("Can you show me an example?")
response3 = chat("Explain the example in detail")

---

生产环境模式

连接池与超时

# Configure client with custom timeout
client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    timeout=60.0,  # 60 second timeout
    max_retries=2,
)

# For async operations
async_client = anthropic.AsyncAnthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    timeout=60.0,
)

日志记录与监控

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitored_chat(user_message: str):
    start_time = time.time()

    try:
        message = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": user_message}]
        )

        duration = time.time() - start_time

        logger.info(
            f"Chat completed - "
            f"Duration: {duration:.2f}s, "
            f"Input tokens: {message.usage.input_tokens}, "
            f"Output tokens: {message.usage.output_tokens}"
        )

        return message.content[0].text

    except Exception as e:
        logger.error(f"Chat failed: {e}")
        raise

基于环境的配置

import os
from typing import Optional

class Config:
    ANTHROPIC_API_KEY: str = os.getenv("ANTHROPIC_API_KEY", "")
    MODEL: str = os.getenv("ANTHROPIC_MODEL", "claude-3-5-sonnet-20241022")
    MAX_TOKENS: int = int(os.getenv("MAX_TOKENS", "1024"))
    TEMPERATURE: float = float(os.getenv("TEMPERATURE", "0.7"))
    TIMEOUT: float = float(os.getenv("API_TIMEOUT", "60.0"))

    @classmethod
    def validate(cls):
        if not cls.ANTHROPIC_API_KEY:
            raise ValueError("ANTHROPIC_API_KEY not set")

# Initialize client with config
Config.validate()
client = anthropic.Anthropic(
    api_key=Config.ANTHROPIC_API_KEY,
    timeout=Config.TIMEOUT,
)

---

可用模型

模型	上下文窗口	适用场景
claude-3-5-sonnet-20241022	200K tokens	通用任务、推理、代码
claude-3-5-haiku-20241022	200K tokens	快速响应、经济实惠
claude-3-opus-20240229	200K tokens	复杂任务、能力最强

推荐： 使用 claude-3-5-sonnet-20241022，在速度、成本和能力之间实现最佳平衡。

---

常见问题

1. 未处理工具使用循环：始终检查 stop_reason，并迭代处理工具使用
2. 超出 max_tokens 限制：根据预期响应长度设置合适的上限
3. 缺少错误处理：始终使用 try/catch 包装 API 调用，并指定具体错误类型
4. 忽略速率限制：在生产系统中实现指数退避
5. 将 API 密钥硬编码：始终使用环境变量
6. 未监控 token 使用量：在生产环境中跟踪成本和用量
7. 阻塞操作：高吞吐量应用应使用异步客户端

---

Additional Resources

• Official Docs: https://docs.anthropic.com/
• API Reference: https://docs.anthropic.com/en/api/
• Python SDK: https://github.com/anthropics/anthropic-sdk-python
• TypeScript SDK: https://github.com/anthropics/anthropic-sdk-typescript
• Prompt Engineering: https://docs.anthropic.com/en/docs/prompt-engineering
• Model Comparison: https://docs.anthropic.com/en/docs/models-overview