Python 调用 Claude API:安装、流式、多轮与图片输入
使用官方 Python SDK 构建文本与图片请求,正确保存历史、处理异步流和错误,明确示例验证范围。
准备环境,不把密钥写进脚本
在虚拟环境安装 anthropic,并记录锁定版本。核验时官方 SDK 要求 Python 3.10 或更新版本;旧 0.x 迁移需看官方迁移说明。ANTHROPIC_API_KEY 由安全环境提供,CLAUDE_MODEL 填写账号当前可用的模型 ID,示例不替你创建或充值账号。不要把密钥写进仓库、笔记、截图或前端。
python -m venv .venv
# 激活方式依操作系统选择;以下是 POSIX shell
. .venv/bin/activate
python -m pip install anthropic
python -m pip show anthropic单轮与多轮共用同一个消息结构
Messages 请求包含 model、max_tokens 和 messages。下面的函数使用调用者持有的历史,并且只有完整 end_turn 才追加为成功对话;未完成输出返回给上层处理。不是每个内容块都是文本,读取时按 type 分支。system 独立传递。
import os
from anthropic import Anthropic
def ask(client, history, question):
pending = history + [{"role":"user", "content":question}]
msg = client.messages.create(
model=os.environ["CLAUDE_MODEL"], max_tokens=700,
system="用中文简洁回答;不确定时说明。", messages=pending)
if msg.stop_reason == "end_turn":
history[:] = pending + [{"role":"assistant", "content":msg.content}]
return {"text":"".join(b.text for b in msg.content if b.type == "text"),
"stop_reason":msg.stop_reason, "usage":msg.usage}
# 授权测试时才执行:
# with Anthropic(max_retries=0, timeout=30) as client:
# history = []
# print(ask(client, history, "用一个例子解释幂等"))异步与流式:取得最终消息后再结算
服务端并发场景使用 AsyncAnthropic;流式片段只能改善展示等待,不能证明任务已完成。下面只演示显示与获取最终状态,不写入数据库;部署时增加取消、有限并发和客户端断线处理。
import os
from anthropic import AsyncAnthropic
async def stream_answer(question):
async with AsyncAnthropic(max_retries=0, timeout=30) as client:
async with client.messages.stream(
model=os.environ["CLAUDE_MODEL"], max_tokens=700,
messages=[{"role":"user", "content":question}]
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
final = await stream.get_final_message()
return final.stop_reason, final.usage图片输入要写对内容块
对于有权处理的本地 PNG,可读取字节、base64 编码后填入 image 块,media_type 必须与实际文件一致,随后附 text 块描述任务。不要仅靠文件扩展名信任上传类型,服务端还应检测文件与大小。示例只构造消息,不发送图片。
import base64
from pathlib import Path
def png_message(path):
raw = Path(path).read_bytes()
if not raw.startswith(bytes([137, 80, 78, 71, 13, 10, 26, 10])):
raise ValueError("expected_png")
return {"role":"user", "content":[
{"type":"image", "source":{"type":"base64", "media_type":"image/png",
"data":base64.b64encode(raw).decode("ascii")}},
{"type":"text", "text":"描述界面中的错误提示,不推断未显示的信息。"}
]}错误处理与检查
区分连接异常和带 HTTP 状态的 API 异常,日志只留请求 ID 和安全元数据。明确 SDK 与业务谁负责重试;401、权限与参数问题应修复,短暂故障才有限退避。max_tokens 截断、工具请求或空文本都要单独处理,不能直接当普通答案。
先以假客户端测试文本、多内容块、截断和异常,再用授权的小样本检查真实 usage 和上下文限制。保存历史时隔离用户,注意工具与思考内容块的原样回传要求。本文代码完成语法检查与本地假数据验证;未验证真实账号、网络或在线输出质量。
参考来源
资料核对日期:2026-10-04。涉及产品与账户条件时,请以当前官方说明为准。