📌 内容摘要
- 使用 Anthropic 官方 Python SDK，10行代码完成第一次调用。
- 覆盖7个核心场景：基础调用、System Prompt、多轮对话、流式输出、异步调用、图片理解、错误处理。
- 所有代码示例均可直接复制运行，附详细注释说明每个参数的作用。
- 文末附生产环境最佳实践：重试机制、并发控制、成本优化。
一、环境准备
Python 版本要求
Anthropic Python SDK 要求 Python 3.8 及以上版本。推荐使用 Python 3.11+。
python --version # 确认版本 ≥ 3.8
安装 SDK
pip install anthropic # 如果需要异步支持（已包含在主包中，无需单独安装） # 如果使用图片处理，建议同时安装 pip install httpx
配置 API Key
推荐使用环境变量存储 Key，不要写死在代码里：
# 方式一：终端临时设置（仅当前会话有效） export ANTHROPIC_API_KEY="sk-ant-api03-你的key" # 方式二：写入 .env 文件（推荐，配合 python-dotenv） # .env 文件内容： # ANTHROPIC_API_KEY=sk-ant-api03-你的key # 安装 dotenv pip install python-dotenv
from dotenv import load_dotenv
import os
load_dotenv() # 自动读取 .env 文件
api_key = os.environ.get("ANTHROPIC_API_KEY")
二、基础调用
最简示例（10行代码）
import anthropic
# 初始化客户端（自动读取环境变量 ANTHROPIC_API_KEY）
client = anthropic.Anthropic()
# 发送消息
message = client.messages.create(
model="claude-sonnet-4-6", # 模型名称
max_tokens=1024, # 最大输出 token 数
messages=[
{"role": "user", "content": "你好，请用一句话自我介绍"}
]
)
# 获取回复文本
print(message.content[0].text)
完整响应对象说明
print(message.id) # 请求唯一ID print(message.model) # 实际使用的模型 print(message.role) # 始终是 "assistant" print(message.content[0].text) # 回复文本内容 print(message.stop_reason) # 停止原因：end_turn / max_tokens print(message.usage.input_tokens) # 输入消耗 token 数 print(message.usage.output_tokens) # 输出消耗 token 数
三、System Prompt 设置
System Prompt 在整个对话中持续生效，用于设定角色、风格、规则：
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system="""你是一名资深 Python 工程师，专注于代码质量和最佳实践。
回答规则：
- 优先给出可运行的代码示例
- 代码加上类型注解
- 指出潜在的性能问题或安全隐患
- 回答简洁，避免不必要的废话""",
messages=[
{"role": "user", "content": "如何安全地读取一个可能不存在的配置文件？"}
]
)
print(message.content[0].text)
四、多轮对话
Claude API 是无状态的，每次请求需要把完整对话历史传入：
import anthropic
from typing import List, Dict
client = anthropic.Anthropic()
class ChatSession:
def __init__(self, system: str = ""):
self.system = system
self.history: List[Dict] = []
def chat(self, user_input: str) -> str:
# 追加用户消息到历史
self.history.append({
"role": "user",
"content": user_input
})
# 发送完整历史
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=self.system,
messages=self.history
)
assistant_reply = response.content[0].text
# 追加 AI 回复到历史
self.history.append({
"role": "assistant",
"content": assistant_reply
})
return assistant_reply
def clear(self):
self.history = []
# 使用示例
session = ChatSession(system="你是一个友善的 AI 助手，记住用户说过的信息。")
print(session.chat("我叫小明，是一名后端工程师"))
print(session.chat("我主要用什么语言？")) # Claude 会记住上文
print(session.chat("给我推荐一个适合我的项目管理工具"))
五、流式输出（Streaming）
流式输出让内容逐字符实时显示，适合构建聊天界面：
import anthropic
client = anthropic.Anthropic()
# 方式一：使用 stream 上下文管理器（推荐）
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=2048,
messages=[
{"role": "user", "content": "写一篇300字的关于秋天的散文"}
]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print() # 换行
# 获取完整响应统计（流结束后可用）
final_message = stream.get_final_message()
print(f"\n输入 tokens: {final_message.usage.input_tokens}")
print(f"输出 tokens: {final_message.usage.output_tokens}")
# 方式二：手动处理事件流（更精细的控制）
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "写一首短诗"}]
) as stream:
for event in stream:
if event.type == "content_block_delta":
if event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
elif event.type == "message_stop":
print("\n[生成完毕]")
六、异步调用（AsyncAnthropic）
在异步应用（如 FastAPI、aiohttp）中使用异步客户端：
import asyncio
import anthropic
client = anthropic.AsyncAnthropic()
async def ask_claude(question: str) -> str:
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": question}]
)
return message.content[0].text
async def main():
# 并发发送多个请求（比顺序发送快得多）
questions = [
"Python 的 GIL 是什么？",
"解释一下装饰器",
"什么是上下文管理器？"
]
tasks = [ask_claude(q) for q in questions]
answers = await asyncio.gather(*tasks)
for q, a in zip(questions, answers):
print(f"Q: {q}")
print(f"A: {a[:100]}...") # 只打印前100字
print()
asyncio.run(main())
七、图片理解（Vision）
import anthropic
import base64
from pathlib import Path
client = anthropic.Anthropic()
# 方式一：传入本地图片（base64编码）
def ask_about_image(image_path: str, question: str) -> str:
image_data = Path(image_path).read_bytes()
base64_image = base64.standard_b64encode(image_data).decode("utf-8")
# 根据文件扩展名判断媒体类型
ext = Path(image_path).suffix.lower()
media_type_map = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp"
}
media_type = media_type_map.get(ext, "image/jpeg")
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": media_type,
"data": base64_image,
},
},
{
"type": "text",
"text": question
}
],
}
],
)
return message.content[0].text
# 使用示例
# result = ask_about_image("screenshot.png", "图片里有什么内容？")
# print(result)
# 方式二：传入图片 URL（更简单）
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "url",
"url": "https://example.com/image.jpg",
},
},
{"type": "text", "text": "描述这张图片"}
],
}
],
)
print(message.content[0].text)
八、错误处理
import anthropic
import time
client = anthropic.Anthropic()
def call_claude_with_retry(
messages: list,
model: str = "claude-sonnet-4-6",
max_tokens: int = 1024,
max_retries: int = 3
) -> str:
"""带重试机制的 Claude 调用"""
for attempt in range(max_retries):
try:
message = client.messages.create(
model=model,
max_tokens=max_tokens,
messages=messages
)
return message.content[0].text
except anthropic.AuthenticationError:
# API Key 无效，不需要重试
raise Exception("API Key 无效，请检查 ANTHROPIC_API_KEY 环境变量")
except anthropic.RateLimitError as e:
# 频率限制，等待后重试
wait_time = 2 ** attempt # 指数退避：1s, 2s, 4s
print(f"频率限制，{wait_time}秒后重试（第{attempt+1}次）")
time.sleep(wait_time)
except anthropic.APIStatusError as e:
if e.status_code == 529: # 服务器过载
wait_time = 2 ** attempt
print(f"服务器过载，{wait_time}秒后重试")
time.sleep(wait_time)
else:
raise # 其他 API 错误直接抛出
except anthropic.APIConnectionError:
# 网络连接问题
if attempt < max_retries - 1:
print(f"网络连接失败，重试中...")
time.sleep(1)
else:
raise Exception("网络连接持续失败，请检查网络环境")
raise Exception(f"重试{max_retries}次后仍然失败")
九、生产环境最佳实践
1. Prompt Caching（节省 50-90% 输入费用）
# 对重复使用的长 System Prompt 开启缓存
# 缓存命中时，输入 token 费用降低 90%
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=[
{
"type": "text",
"text": "你的长 System Prompt（几百到几千字）...",
"cache_control": {"type": "ephemeral"} # 标记为可缓存
}
],
messages=[{"role": "user", "content": "用户的问题"}]
)
2. 并发请求限速
import asyncio
import anthropic
client = anthropic.AsyncAnthropic()
async def batch_process(texts: list, concurrency: int = 5) -> list:
"""并发批量处理，限制同时请求数量避免超出速率限制"""
semaphore = asyncio.Semaphore(concurrency)
async def process_one(text: str) -> str:
async with semaphore:
msg = await client.messages.create(
model="claude-haiku-4-5-20251001", # 批量任务用 Haiku 省钱
max_tokens=512,
messages=[{"role": "user", "content": text}]
)
return msg.content[0].text
tasks = [process_one(t) for t in texts]
return await asyncio.gather(*tasks)
3. Token 使用量监控
class TokenTracker:
"""简单的 token 用量追踪器"""
PRICES = {
"claude-opus-4-6": {"input": 5.0, "output": 25.0},
"claude-sonnet-4-6": {"input": 3.0, "output": 15.0},
"claude-haiku-4-5-20251001": {"input": 1.0, "output": 5.0},
}
def __init__(self):
self.total_input = 0
self.total_output = 0
self.total_cost = 0.0
def track(self, response) -> None:
model = response.model
input_t = response.usage.input_tokens
output_t = response.usage.output_tokens
self.total_input += input_t
self.total_output += output_t
if model in self.PRICES:
p = self.PRICES[model]
cost = (input_t * p["input"] + output_t * p["output"]) / 1_000_000
self.total_cost += cost
def report(self) -> None:
print(f"总输入 tokens: {self.total_input:,}")
print(f"总输出 tokens: {self.total_output:,}")
print(f"预估费用: ${self.total_cost:.4f}")
# 使用示例
tracker = TokenTracker()
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "你好"}]
)
tracker.track(response)
tracker.report()
常见问题
Q：max_tokens 和实际输出长度是什么关系？
max_tokens 是输出的上限，实际输出可能更短。如果 stop_reason 是 max_tokens，说明输出被截断，需要增大这个值。费用只按实际输出的 token 计算，设得大一些不会多收费。
Q：对话历史太长怎么办？
有两个方向：一是定期总结历史（让 Claude 把前几轮对话总结成一段文字，替换掉原始历史）；二是使用滑动窗口，只保留最近 N 轮对话。Claude Sonnet 和 Opus 支持 100 万 token 上下文，Haiku 支持 20 万，一般场景很难触及上限。
Q：Python SDK 和直接调用 HTTP API 有什么区别？
SDK 封装了认证、重试、错误处理、流式响应等细节，推荐使用 SDK。如果有特殊需求（如在不支持 pip 的环境），也可以用 requests 或 httpx 直接调用 https://api.anthropic.com/v1/messages 端点，请求格式和 SDK 一致。
总结
本文覆盖了 Python 调用 Claude API 的7个核心场景。日常开发建议从基础调用开始，按需加入 System Prompt 和多轮对话；需要更好用户体验时加上流式输出；处理大量任务时用异步并发提速。生产环境务必加入错误处理和重试机制，同时开启 Prompt Caching 降低成本。