Anthropic Python SDK 教程：安装、流式与工具调用（2026）

Anthropic Python SDK 是官方维护的 anthropic 包：pip 安装后，客户端默认从环境变量 ANTHROPIC_API_KEY 读取密钥，调用 client.messages.create() 传入模型、最大 token 数和消息列表即可拿到回复；流式、异步、工具调用、重试超时和 base_url 中转配置都在同一个客户端上。

安装 Anthropic Python SDK

这个包发布在 PyPI，源码托管在 GitHub。建议装进虚拟环境，避免升级时影响系统里其他项目：

python -m venv .venv
source .venv/bin/activate
pip install -U anthropic
python -c "import anthropic; print(anthropic.__version__)"

SDK 要求较新的 Python 3 版本。生产环境建议在 requirements 或 pyproject 里锁定具体版本号，而不是始终拉最新版，这样升级时可以先在测试环境跑一遍再放行。如果项目锁定了旧解释器，升级前先看一眼仓库 README 里的版本要求，因为 SDK 的大版本升级曾经抬高过最低 Python 版本。

构造客户端：API Key 从哪里来

客户端类是 anthropic.Anthropic。不传任何参数时，它会去读环境变量 ANTHROPIC_API_KEY。这是推荐做法：密钥不出现在源码里，同一份代码在本地、CI 和生产环境都不用改。

export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic

client = anthropic.Anthropic()  # 自动读取 ANTHROPIC_API_KEY

也可以显式传 anthropic.Anthropic(api_key=...)，适合密钥来自密钥管理服务而不是进程环境的场景；但不要把密钥写成字面量硬编码。要注意，环境变量缺失或格式不对时，构造客户端这一步不会报错，第一次真正发请求时才会抛认证异常。常见原因和排查步骤见Anthropic API Key 不可用怎么排查。

第一次调用 messages.create

所有请求都走 Messages API。三个必填参数是 model、max_tokens 和 messages；system 提示词是可选的，放在消息列表之外。

import anthropic

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-5"  # 换成你实际要用的模型 id

response = client.messages.create(
    model=MODEL,
    max_tokens=1024,
    system="你是一个面向 Python 开发者的简洁助手。",
    messages=[
        {"role": "user", "content": "用两句话解释什么是上下文管理器。"}
    ],
)

print(response.content[0].text)

messages 列表按 user、assistant 角色交替，第一条必须是 user。接口本身是无状态的：多轮对话要把上一轮的助手回复和新的用户消息都追加进列表，再把完整历史重新发一次。关于请求结构更完整的讲解可以看Claude API 使用指南。

读取响应里的内容块

response.content 是一个带类型的内容块列表，而不是一个字符串。纯文本请求通常只有一个 TextBlock，但一旦用上工具调用或扩展思考，列表里会出现多种类型的块。写代码时先判断 block.type 再读 block.text，后续加功能时不用改读取逻辑：

for block in response.content:
    if block.type == "text":
        print(block.text)

print(response.stop_reason)         # "end_turn"、"max_tokens"、"tool_use" 等
print(response.usage.input_tokens, response.usage.output_tokens)

另外，response.content[0].text 这种直接取下标的写法只适合示例，正式代码应按类型遍历，否则遇到非文本块时会报属性错误。两个字段值得从第一天就记日志。stop_reason 说明生成为什么结束，值为 max_tokens 表示回答被截断，应该调大上限；usage 给出输入输出 token 数，计费和限流都以它为依据。

用 stream 辅助方法做流式输出

只要回答不是几句话就能结束，就应该用流式：用户立刻能看到输出，HTTP 连接也不会长时间空闲。SDK 提供了上下文管理器形式的 client.messages.stream()，它会帮你累积完整消息，并暴露一个 text_stream 迭代器：

with client.messages.stream(
    model=MODEL,
    max_tokens=4096,
    messages=[{"role": "user", "content": "写一篇 Python dataclass 的简短指南。"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final = stream.get_final_message()

print()
print(final.usage.output_tokens)

get_final_message() 返回的 Message 对象和非流式调用拿到的完全一样，所以下游代码不需要维护两套逻辑。如果只想拿原始事件流，可以给 messages.create() 传 stream=True 自己迭代事件。事件类型和 SSE 细节见Claude API 流式输出详解。

异步客户端

Web 框架和并发流水线应该用 anthropic.AsyncAnthropic。它的方法与同步客户端一一对应，只是需要 await：

import asyncio
import anthropic

async def main() -> None:
    client = anthropic.AsyncAnthropic()

    response = await client.messages.create(
        model=MODEL,
        max_tokens=512,
        messages=[{"role": "user", "content": "给一条写 async Python 的建议。"}],
    )
    print(response.content[0].text)

    async with client.messages.stream(
        model=MODEL,
        max_tokens=1024,
        messages=[{"role": "user", "content": "把这条建议展开成一段话。"}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(main())

一个进程里建一个异步客户端反复复用即可，底层的连接池正是并发效率的来源。在事件循环里混用同步客户端会阻塞整个循环，所以同一条代码路径只选一种风格。

工具调用入门

工具调用让模型可以请求你的程序执行某个函数。你用名称、描述和一段 JSON Schema 描述每个工具的输入；当模型决定调用时，响应的 stop_reason 为 "tool_use"，内容里会有一个带参数的 ToolUseBlock。你执行函数，再把结果以 tool_result 块的形式发回去。

import json

tools = [
    {
        "name": "get_weather",
        "description": "查询某个城市的当前天气。",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    }
]

def get_weather(city: str) -> str:
    return json.dumps({"city": city, "condition": "clear", "temp_c": 21})

messages = [{"role": "user", "content": "里斯本现在天气怎么样？"}]

response = client.messages.create(
    model=MODEL, max_tokens=1024, tools=tools, messages=messages
)

while response.stop_reason == "tool_use":
    messages.append({"role": "assistant", "content": response.content})
    results = []
    for block in response.content:
        if block.type == "tool_use":
            output = get_weather(**block.input)
            results.append(
                {"type": "tool_result", "tool_use_id": block.id, "content": output}
            )
    messages.append({"role": "user", "content": results})
    response = client.messages.create(
        model=MODEL, max_tokens=1024, tools=tools, messages=messages
    )

for block in response.content:
    if block.type == "text":
        print(block.text)

有三个细节容易踩坑。第一，追加历史时要放入助手回复的完整 response.content，不能只放文本，因为接口要靠其中的 tool_use 块按 tool_use_id 匹配 tool_result。第二，如果模型一轮里请求了多个工具，所有结果要放在同一条 user 消息里一起返回。第三，函数执行失败时，把错误文本放进 tool result 并加上 "is_error": True，而不是直接丢掉，这样模型才有机会自行纠正。

异常类、重试与超时

SDK 抛出的都是带类型的异常，全部继承自 anthropic.APIError。先捕获能针对性处理的具体异常，再兜底捕获通用异常：

import anthropic

try:
    response = client.messages.create(
        model=MODEL, max_tokens=256,
        messages=[{"role": "user", "content": "ping"}],
    )
except anthropic.AuthenticationError:
    print("API key 缺失或被拒绝")
except anthropic.RateLimitError as e:
    print("触发限流，retry-after:", e.response.headers.get("retry-after"))
except anthropic.APIStatusError as e:
    print("HTTP", e.status_code, e.message)
except anthropic.APIConnectionError:
    print("网络层无法连到端点")

重试是内置的。默认情况下，客户端会对连接错误以及 HTTP 408、409、429 和 5xx 响应做两次指数退避重试；默认请求超时是十分钟。两者都能在客户端级别设置，也能针对单次调用覆盖：

client = anthropic.Anthropic(max_retries=3, timeout=60.0)

quick = client.with_options(max_retries=0, timeout=10.0).messages.create(
    model=MODEL, max_tokens=64,  # 这里的覆盖只对本次请求生效
    messages=[{"role": "user", "content": "ok?"}],
)

因为超时也会触发重试，最坏情况下的总耗时大约是 timeout 乘以 max_retries + 1。对延迟敏感的接口把两个值都调低；长文本生成则改用流式，而不是一味拉高超时。

用 base_url 把 SDK 指向网关或中转端点

默认情况下客户端把请求发往 Anthropic 官方 API 主机。base_url 参数只改这个主机地址，请求的其他部分完全不变，团队借此把流量导向内部网关、日志代理，或任何实现了 Anthropic 协议的第三方中转端点：

client = anthropic.Anthropic(
    base_url="https://your-endpoint.example.com",
    api_key="key-issued-by-that-endpoint",
)

同样的设置也有环境变量形式 ANTHROPIC_BASE_URL，切换环境时不用动代码：

export ANTHROPIC_BASE_URL="https://your-endpoint.example.com"
export ANTHROPIC_API_KEY="key-issued-by-that-endpoint"

这种做法的好处是业务代码与网络拓扑解耦：开发机直连、生产走网关，只需要换一个环境变量。这样做时要核对两点。一是密钥必须是目标端点签发的那把，而不是你的 Anthropic 密钥，除非该中转会原样转发；二是端点要能接受 SDK 自动拼上的 /v1/messages 路径，所以 base_url 填的是协议加主机名（可带路径前缀），不要填以 /messages 结尾的完整 URL。如果你更习惯 OpenAI 客户端的写法，有些端点同时暴露两种协议，那是另一套基于 OpenAI 客户端的接法，不在本文范围内。

ROIBest AI 是一个兼容 Anthropic 与 OpenAI 两种协议的 API 中转端点。如果你使用它，配置位置就是上面展示的 base_url，配合该服务签发的密钥即可，本文其余内容无需改动。详情见 ai.roibest.com。

常见问题

Anthropic Python SDK 会自动读取 API Key 吗？

会。不带参数调用 anthropic.Anthropic() 时，它从环境变量 ANTHROPIC_API_KEY 读取密钥。只有当密钥来自密钥管理服务等其他来源时，才需要显式传 api_key=。

messages.create 和 messages.stream 有什么区别？

messages.create() 在生成完成后一次性返回完整消息；messages.stream() 是上下文管理器，边生成边产出文本，结束后仍可通过 get_final_message() 拿到完整消息对象。

收到 tool_use 响应后该怎么处理？

执行 ToolUseBlock 里指定的函数，把助手消息和一条包含 tool_result 块（带对应 tool_use_id）的 user 消息追加进历史，再次调用接口；循环直到 stop_reason 不再是 tool_use。

这个 SDK 能接非 Anthropic 官方的端点吗？

可以，前提是该端点实现了 Anthropic Messages API。在客户端设置 base_url 或导出 ANTHROPIC_BASE_URL，并使用该端点签发的密钥。