Anthropic Python SDK 完全指南：从安装到生产级集成的最佳实践

pip install anthropic

# 可选：异步支持（已内置，无需额外安装）
# 可选：流式输出（已内置）

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-你的key")
# 或设置环境变量 ANTHROPIC_API_KEY，无需传参

message = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "解释量子纠缠，用通俗语言"}
    ]
)

print(message.content[0].text)
print(f"输入 Token: {message.usage.input_tokens}")
print(f"输出 Token: {message.usage.output_tokens}")

message = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system="你是一位专业的 Python 代码审查员。用中文回复，重点关注性能和安全问题。",
    messages=[
        {"role": "user", "content": "帮我审查这段代码：
```python
[代码]
```"}
    ]
)

conversation = []

def chat(user_message: str) -> str:
    conversation.append({"role": "user", "content": user_message})
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=conversation
    )
    
    assistant_message = response.content[0].text
    conversation.append({"role": "assistant", "content": assistant_message})
    return assistant_message

# 多轮对话
print(chat("我叫张三"))
print(chat("我叫什么名字？"))  # Claude 会记住「张三」

# 同步流式
with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "写一首关于春天的诗"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# 获取最终完整消息（含 usage）
final_message = stream.get_final_message()
print(f"
总 Token: {final_message.usage.input_tokens + final_message.usage.output_tokens}")

import asyncio
import anthropic

async def main():
    client = anthropic.AsyncAnthropic()
    
    # 异步调用
    message = await client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "你好"}]
    )
    print(message.content[0].text)
    
    # 异步流式
    async with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "写一段代码"}]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(main())

tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}]
)

# 检查是否调用了工具
if response.stop_reason == "tool_use":
    tool_use = next(b for b in response.content if b.type == "tool_use")
    print(f"调用工具: {tool_use.name}")
    print(f"参数: {tool_use.input}")
    
    # 执行工具（你自己的函数）
    result = get_weather(**tool_use.input)
    
    # 把结果返回给 Claude
    final = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        tools=tools,
        messages=[
            {"role": "user", "content": "北京今天天气怎么样？"},
            {"role": "assistant", "content": response.content},
            {"role": "user", "content": [
                {"type": "tool_result", "tool_use_id": tool_use.id, "content": str(result)}
            ]}
        ]
    )
    print(final.content[0].text)

# 长系统提示词或文档缓存，节省 90% 读取成本
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一位代码助手...",  # 短提示词，不缓存
        },
        {
            "type": "text",
            "text": 长文档内容,  # 超长内容，启用缓存
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[{"role": "user", "content": "问题1"}],
    extra_headers={"anthropic-beta": "prompt-caching-2024-07-31"}
)

# 查看缓存命中情况
print(response.usage.cache_creation_input_tokens)  # 首次：缓存写入
print(response.usage.cache_read_input_tokens)       # 后续：从缓存读取

import anthropic

def safe_call(messages, retries=3):
    for attempt in range(retries):
        try:
            return client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=1024,
                messages=messages
            )
        except anthropic.RateLimitError:
            if attempt < retries - 1:
                import time
                time.sleep(2 ** attempt)  # 指数退避
            else:
                raise
        except anthropic.AuthenticationError:
            raise  # API Key 错误，不重试
        except anthropic.BadRequestError as e:
            print(f"请求参数错误: {e}")
            raise
        except anthropic.APIConnectionError:
            if attempt < retries - 1:
                import time
                time.sleep(1)
            else:
                raise

from fastapi import FastAPI, Depends
import anthropic

app = FastAPI()

# 依赖注入：每个请求共享同一个 client 实例
def get_claude_client() -> anthropic.AsyncAnthropic:
    return anthropic.AsyncAnthropic()

@app.post("/chat")
async def chat(
    message: str,
    client: anthropic.AsyncAnthropic = Depends(get_claude_client)
):
    response = await client.messages.create(
        model="claude-haiku-3-5",  # 接口用 Haiku，速度快
        max_tokens=512,
        messages=[{"role": "user", "content": message}]
    )
    return {"reply": response.content[0].text}

# 流式响应
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(message: str):
    async def generate():
        async with anthropic.AsyncAnthropic().messages.stream(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": message}]
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {text}

"
    
    return StreamingResponse(generate(), media_type="text/event-stream")