anthropic 是 Anthropic 官方的 Python SDK，封装了 Claude API 的所有功能。
本文从安装到生产最佳实践，全面讲解 SDK 的使用方法。
安装
bash
pip install anthropic
# 可选：异步支持（已内置，无需额外安装）
# 可选：流式输出（已内置）基础用法
python
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}")系统提示词
python
message = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
system="你是一位专业的 Python 代码审查员。用中文回复，重点关注性能和安全问题。",
messages=[
{"role": "user", "content": "帮我审查这段代码：
```python
[代码]
```"}
]
)多轮对话
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 会记住「张三」流式输出（Streaming）
python
# 同步流式
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}")异步客户端
python
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())工具调用（Tool Use）
python
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)Prompt Caching
python
# 长系统提示词或文档缓存，节省 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) # 后续：从缓存读取错误处理
python
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在 FastAPI 中集成
python
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")来源：Anthropic Python SDK 官方文档