Anthropic Python SDK 入门笔记

基于 s04_subagent.py 的 Messages API、工具调用与子代理实现整理。

1. Anthropic SDK 是什么

Anthropic Python SDK 是对 Claude API 的 Python 封装，核心入口是：

1	from anthropic import Anthropic

它主要提供：

普通文本对话
多轮消息管理
同步与异步请求
流式输出
工具调用 Tool Use
图像、文档等多模态输入
结构化输出
Token 统计
Prompt Caching
Batch、服务端工具等进阶功能

当前 Python SDK 要求 Python 3.9 或更高版本，并同时提供同步客户端 Anthropic 和异步客户端 AsyncAnthropic。

2. 安装与环境配置

2.1 安装依赖

1	pip install -U anthropic python-dotenv

其中：

anthropic：官方 Python SDK
python-dotenv：从 .env 文件读取环境变量

2.2 创建 `.env`

ANTHROPIC_API_KEY=你的API密钥
MODEL_ID=claude-sonnet-5

# 使用兼容接口或代理时再配置
# ANTHROPIC_BASE_URL=https://your-api.example.com

模型名称不要散落在代码各处，最好统一放进环境变量。Anthropic 较新的模型 ID 是固定版本，而不是自动指向未来模型的“最新版别名”。

2.3 创建客户端

import os

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("ANTHROPIC_API_KEY")
model = os.getenv("MODEL_ID")

if not api_key:
    raise RuntimeError("缺少 ANTHROPIC_API_KEY")

if not model:
    raise RuntimeError("缺少 MODEL_ID")

client_kwargs = {
    "api_key": api_key,
    "timeout": 60.0,
    "max_retries": 2,
}

if base_url := os.getenv("ANTHROPIC_BASE_URL"):
    client_kwargs["base_url"] = base_url

client = Anthropic(**client_kwargs)

原示例通过 Anthropic(base_url=...) 创建客户端，并用 MODEL_ID 环境变量指定模型。

3. 第一次调用 Claude

response = client.messages.create(
    model=model,
    max_tokens=1024,
    system="你是一位耐心的 Python 老师。",
    messages=[
        {
            "role": "user",
            "content": "请用简单的话解释 Python 装饰器。",
        }
    ],
)

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

3.1 主要参数

`model`

指定使用的 Claude 模型：

1	model=model

建议通过环境变量管理。

`max_tokens`

允许模型最多生成多少 Token：

1	max_tokens=1024

它是输出上限，不代表模型一定会用满。

`system`

系统提示词必须通过顶层 system 参数传入：

1	system="你是一位 Python 老师。"

Anthropic Messages API 中不存在：

1	{"role": "system"}

也就是说，下面的写法是不正确的：

1
2
3

messages=[
    {"role": "system", "content": "你是一位老师"}
]

Messages API 是无状态接口，每次请求都需要提交本轮所需的历史消息；系统提示则单独放在顶层 system 参数中。

`messages`

消息通常包含两种角色：

1 2	{"role": "user", "content": "用户消息"} {"role": "assistant", "content": "模型历史回复"}

content 可以直接是字符串：

1	{"role": "user", "content": "你好"}

也可以是内容块数组：

{
    "role": "user",
    "content": [
        {
            "type": "text",
            "text": "你好",
        }
    ],
}

4. 正确读取响应

response.content 不是字符串，而是内容块列表：

1	print(response.content)

可能得到类似结构：

[
    TextBlock(
        type="text",
        text="Python 装饰器是一种包装函数的方式……"
    )
]

因此不要过度依赖：

1	response.content[0].text

因为返回内容还可能包含：

text
tool_use
thinking
服务端工具结果
其他内容块

更稳妥的文本提取函数是：

def get_text(response) -> str:
    return "".join(
        block.text
        for block in response.content
        if block.type == "text"
    )

使用：

1 2	text = get_text(response) print(text)

4.1 查看停止原因

1	print(response.stop_reason)

常见值包括：

end_turn：模型正常结束回复
max_tokens：达到输出上限
stop_sequence：遇到自定义停止字符串
tool_use：模型请求调用工具
refusal：模型拒绝完成请求

特别需要注意：

1 2	if response.stop_reason == "max_tokens": print("回答可能被截断")

4.2 查看 Token 用量

1 2	print(response.usage.input_tokens) print(response.usage.output_tokens)

也可以直接查看：

1	print(response.usage)

SDK 还支持在正式请求前计算输入 Token：

count = client.messages.count_tokens(
    model=model,
    system="你是一位 Python 老师。",
    messages=[
        {
            "role": "user",
            "content": "解释一下生成器。",
        }
    ],
)

print(count.input_tokens)

4.3 记录 Request ID

1	print(response._request_id)

线上报错时，最好记录：

Request ID
模型名称
时间
stop_reason
错误类型

_request_id 虽然以下划线开头，但它是 SDK 明确公开的调试属性。

5. 多轮对话

Messages API 不会替我们保存聊天记录，需要应用程序自己维护历史。

history = []

while True:
    query = input("你：").strip()

    if query.lower() in {"q", "exit"}:
        break

    history.append({
        "role": "user",
        "content": query,
    })

    response = client.messages.create(
        model=model,
        max_tokens=2048,
        system="你是一位耐心的编程老师。",
        messages=history,
    )

    history.append({
        "role": "assistant",
        "content": response.content,
    })

    print("Claude：", get_text(response))

原代码也是通过 history 保存父代理的历史消息，并在每轮请求后追加模型的完整内容块。

5.1 持久化历史记录

SDK 返回的是 Pydantic 对象。需要写入 JSON 时，可以转换为普通字典：

assistant_content = [
    block.model_dump()
    for block in response.content
]

history.append({
    "role": "assistant",
    "content": assistant_content,
})

保存：

import json

with open("history.json", "w", encoding="utf-8") as file:
    json.dump(history, file, ensure_ascii=False, indent=2)

5.2 不要无限保存历史

聊天时间越长，请求输入越大，延迟和费用通常也会增长。

常见处理方式：

只保留最近若干轮。
将早期对话总结为一条摘要。
把长期信息存入数据库或向量数据库。
使用 Prompt Caching。
对智能体上下文进行压缩。

简单裁剪示例：

MAX_MESSAGES = 20

if len(history) > MAX_MESSAGES:
    history = history[-MAX_MESSAGES:]

不过这种方式可能直接裁掉重要背景。真实项目更适合“摘要 + 最近消息”。

6. 常用生成参数

response = client.messages.create(
    model=model,
    max_tokens=2048,
    temperature=0.3,
    stop_sequences=["<END>"],
    system="你是一位代码审查助手。",
    messages=[
        {
            "role": "user",
            "content": "检查下面的代码……",
        }
    ],
)

`temperature`

控制输出随机性：

1	temperature=0.0

适合：

信息抽取
分类
代码修改
格式严格的任务

1	temperature=0.7

适合：

创意写作
头脑风暴
多样化表达

不要把它理解为“正确率旋钮”。温度低只是让输出更加稳定，并不保证事实正确。

`stop_sequences`

指定停止字符串：

1	stop_sequences=["<END>"]

模型生成该字符串时会停止。

大部分普通聊天并不需要设置。

7. 流式输出

普通请求需要等模型生成完成后一次性返回。流式输出会在生成过程中不断返回增量内容，适合：

聊天界面
命令行助手
长回答
长时间运行的请求
需要尽快显示首段内容的场景

Anthropic 使用 SSE，也就是 Server-Sent Events 传输流式响应。官方 SDK 同时提供底层事件流和更方便的流式辅助方法。

7.1 推荐方式：`messages.stream()`

with client.messages.stream(
    model=model,
    max_tokens=2048,
    system="你是一位 Python 老师。",
    messages=[
        {
            "role": "user",
            "content": "详细解释 Python 的 async 和 await。",
        }
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    print()

    final_message = stream.get_final_message()

这里的：

1	stream.text_stream

只返回文本增量，不需要自己判断每一种 SSE 事件。

流结束后：

1	final_message = stream.get_final_message()

可以得到完整的最终消息，包括：

content
usage
stop_reason
Request ID 等信息

1 2	print(final_message.stop_reason) print(final_message.usage)

流式辅助器内部会把增量事件累积成完整消息，并提供 text_stream、get_final_message() 和 get_final_text() 等方法。

7.2 只获取最终文本

with client.messages.stream(
    model=model,
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "写一段关于月亮的短文。",
        }
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final_text = stream.get_final_text()

get_final_text() 适合确定响应只需要普通文本的场景。

如果模型可能调用工具，优先使用：

1	stream.get_final_message()

因为工具调用不是文本块。

7.3 底层事件流：`stream=True`

stream = client.messages.create(
    model=model,
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "讲一个短故事。",
        }
    ],
    stream=True,
)

for event in stream:
    print(event.type)

常见事件包括：

message_start
content_block_start
content_block_delta
content_block_stop
message_delta
message_stop

只处理文本增量：

stream = client.messages.create(
    model=model,
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "讲一个短故事。",
        }
    ],
    stream=True,
)

for event in stream:
    if event.type != "content_block_delta":
        continue

    if event.delta.type == "text_delta":
        print(event.delta.text, end="", flush=True)

普通应用优先选择 messages.stream()。

只有在需要精确处理以下内容时，才更适合直接读取底层事件：

工具参数的增量 JSON
Thinking 增量
引用增量
自定义事件转发
WebSocket/SSE 网关

7.4 流式输出与历史记录

不要把每个文本碎片都添加进历史。

正确流程是：

history.append({
    "role": "user",
    "content": query,
})

with client.messages.stream(
    model=model,
    max_tokens=2048,
    messages=history,
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    response = stream.get_final_message()

history.append({
    "role": "assistant",
    "content": response.content,
})

必须等完整消息生成完，再把完整 response.content 放入历史。

8. 异步调用

Web 服务、聊天后端或需要同时处理多个请求时，通常使用 AsyncAnthropic。

import asyncio
import os

from anthropic import AsyncAnthropic

client = AsyncAnthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"]
)

model = os.environ["MODEL_ID"]


async def main() -> None:
    response = await client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "解释什么是协程。",
            }
        ],
    )

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


asyncio.run(main())

8.1 异步流式输出

import asyncio
import os

from anthropic import AsyncAnthropic

client = AsyncAnthropic()
model = os.environ["MODEL_ID"]


async def main() -> None:
    async with client.messages.stream(
        model=model,
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": "写一篇关于 asyncio 的入门教程。",
            }
        ],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

        print()

        final_message = await stream.get_final_message()
        print(final_message.usage)


asyncio.run(main())

不要在同步函数中直接调用：

1	asyncio.run(...)

如果本身已经运行在 FastAPI、Jupyter 或其他事件循环中，应直接：

1	response = await client.messages.create(...)

9. Tool Use：让 Claude 调用函数

Claude 不会直接执行本地 Python 函数。

完整过程是：

程序向 Claude 描述工具。
Claude返回 tool_use 内容块。
程序执行真正的 Python 函数。
程序把结果作为 tool_result 发回。
Claude根据结果继续回答。

这是一份“调用协议”，模型负责选择工具和生成参数，真正执行客户端工具的是应用程序。

9.1 定义工具

TOOLS = [
    {
        "name": "get_weather",
        "description": "查询指定城市当前的天气。",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，例如：杭州",
                }
            },
            "required": ["city"],
        },
    }
]

工具定义主要包含：

name：工具名称
description：工具什么时候使用
input_schema：输入参数的 JSON Schema

描述不要只写：

1	"description": "查询天气"

更好的写法是：

1	"description": "查询指定城市当前天气。当用户询问实时天气、温度或降雨情况时使用。"

9.2 编写真正的工具函数

def get_weather(city: str) -> str:
    data = {
        "北京": "晴，31°C",
        "杭州": "多云，29°C",
        "上海": "小雨，28°C",
    }

    return data.get(city, f"暂时没有 {city} 的天气数据")

建立处理器映射：

1
2
3

TOOL_HANDLERS = {
    "get_weather": get_weather,
}

原代码同样使用 TOOL_HANDLERS 将 Claude 返回的工具名称映射到实际 Python 函数。

9.3 完整工具调用循环

from typing import Any


def extract_text(content: list[Any]) -> str:
    return "".join(
        block.text
        for block in content
        if block.type == "text"
    )


def run_agent(query: str) -> str:
    messages = [
        {
            "role": "user",
            "content": query,
        }
    ]

    for _ in range(20):
        response = client.messages.create(
            model=model,
            max_tokens=2048,
            system="你是一位助手。必要时调用工具获取信息。",
            messages=messages,

            tools=TOOLS,
        )

        # 必须保留 Claude 返回的完整内容，包括 tool_use 块。
        messages.append({
            "role": "assistant",
            "content": response.content,
        })

        tool_calls = [
            block
            for block in response.content
            if block.type == "tool_use"
        ]

        if not tool_calls:
            return extract_text(response.content)

        tool_results = []

        for call in tool_calls:
            handler = TOOL_HANDLERS.get(call.name)

            if handler is None:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": f"未知工具：{call.name}",
                    "is_error": True,
                })
                continue

            try:
                result = handler(**call.input)

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": str(result),
                })

            except Exception as exc:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": f"工具执行失败：{exc}",
                    "is_error": True,
                })

        # 所有工具结果放进同一条 user 消息。
        messages.append({
            "role": "user",
            "content": tool_results,
        })

    raise RuntimeError("工具调用轮数超过限制")

使用：

1 2	answer = run_agent("杭州今天的天气怎么样？") print(answer)

9.4 工具循环的三个关键点

第一，保留完整 assistant 内容

错误做法：

messages.append({
    "role": "assistant",
    "content": get_text(response),
})

这会丢失 tool_use 块。

正确做法：

messages.append({
    "role": "assistant",
    "content": response.content,
})

第二，`tool_use_id` 必须对应

{
    "type": "tool_result",
    "tool_use_id": call.id,
    "content": result,
}

Claude 可能一轮调用多个工具，因此不能自己随便生成 ID。

第三，一轮的多个工具结果一起返回

messages.append({
    "role": "user",
    "content": tool_results,
})

原示例也是先执行本轮全部工具，再把所有 tool_result 作为同一条用户消息反馈给模型。

10. 流式输出结合工具调用

最简单可靠的实现方式是：

流式显示文本。
流结束后获取完整 Message。
从完整 Message 中处理工具调用。
再开始下一轮流。

def stream_agent(query: str) -> str:
    messages = [
        {
            "role": "user",
            "content": query,
        }
    ]

    for _ in range(20):
        with client.messages.stream(
            model=model,
            max_tokens=2048,
            system="必要时调用工具。",
            messages=messages,
            tools=TOOLS,
        ) as stream:
            for text in stream.text_stream:
                print(text, end="", flush=True)

            response = stream.get_final_message()

        messages.append({
            "role": "assistant",
            "content": response.content,
        })

        tool_calls = [
            block
            for block in response.content
            if block.type == "tool_use"
        ]

        if not tool_calls:
            print()
            return extract_text(response.content)

        tool_results = []

        for call in tool_calls:
            handler = TOOL_HANDLERS.get(call.name)

            try:
                if handler is None:
                    raise ValueError(f"未知工具：{call.name}")

                output = handler(**call.input)

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": str(output),
                })

            except Exception as exc:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": str(exc),
                    "is_error": True,
                })

        messages.append({
            "role": "user",
            "content": tool_results,
        })

    raise RuntimeError("智能体循环次数超过限制")

底层 SSE 会把工具参数作为增量 JSON 返回，但入门阶段不建议手动拼接这些 JSON。等待流结束，然后读取 get_final_message() 更简单。

11. 子代理与上下文隔离

原代码实现了一个很实用的模式：

sub_messages = [
    {
        "role": "user",
        "content": prompt,
    }
]

子代理拥有全新的消息列表，不会复制父代理的完整历史。

它执行完成后，只返回最终摘要：

return "".join(
    block.text
    for block in response.content
    if hasattr(block, "text")
)

这样可以避免：

探索过程污染父代理上下文
大量工具输出进入主对话
主代理上下文迅速膨胀
子任务细节影响最终回答

原代码正是通过独立的 sub_messages 保存子代理临时历史，并在任务结束后只向父代理返回文本摘要。

11.1 一个需要纠正的概念

这个实现属于：

消息历史隔离，或者说上下文隔离。

它并不是真正的操作系统进程隔离。

因为父代理和子代理：

运行在同一个 Python 进程
共享全局 client
共享 WORKDIR
共享 Python 内存空间
都通过普通函数调用执行

代码里的 subprocess.run() 只用于执行 Bash 命令，并不是用来启动子代理。

因此更加准确的说法是：

使用独立的 messages 列表实现模型上下文隔离，并与父代理共享应用程序资源。

真正的进程隔离需要使用：

multiprocessing
独立 Worker
容器
子进程 RPC
独立服务

11.2 子代理基础实现

def run_subagent(prompt: str) -> str:
    messages = [
        {
            "role": "user",
            "content": prompt,
        }
    ]

    for _ in range(20):
        response = client.messages.create(
            model=model,
            max_tokens=4096,
            system=(
                "你是一个子代理。请独立完成任务，"
                "最后只返回结论和重要依据。"
            ),
            messages=messages,
            tools=CHILD_TOOLS,
        )

        messages.append({
            "role": "assistant",
            "content": response.content,
        })

        tool_calls = [
            block
            for block in response.content
            if block.type == "tool_use"
        ]

        if not tool_calls:
            return extract_text(response.content)

        results = []

        for call in tool_calls:
            handler = TOOL_HANDLERS.get(call.name)

            try:
                if handler is None:
                    raise ValueError(f"未知工具：{call.name}")

                output = handler(**call.input)

                results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": str(output),
                })

            except Exception as exc:
                results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": str(exc),
                    "is_error": True,
                })

        messages.append({
            "role": "user",
            "content": results,
        })

    raise RuntimeError("子代理超过最大循环次数")

12. 工具安全

原示例的 run_bash() 使用：

subprocess.run(
    command,
    shell=True,
    ...
)

并通过字符串黑名单拦截部分危险命令。原文件也明确说明这只是教学演示，不等同于完整安全沙箱。

生产环境不能只依赖：

dangerous = [
    "rm -rf /",
    "sudo",
    "shutdown",
]

因为命令有很多绕过方式。

至少需要考虑：

不使用 shell=True
使用命令白名单
参数分别传递
限定工作目录
限制文件权限
容器隔离
CPU、内存和执行时间限制
网络访问控制
高风险操作人工确认
记录完整审计日志

例如：

import subprocess


ALLOWED_COMMANDS = {
    "python",
    "pytest",
    "git",
}


def safe_run(args: list[str]) -> str:
    if not args:
        raise ValueError("命令不能为空")

    if args[0] not in ALLOWED_COMMANDS:
        raise ValueError(f"不允许执行：{args[0]}")

    result = subprocess.run(
        args,
        cwd=WORKDIR,
        capture_output=True,
        text=True,
        timeout=30,
        shell=False,
    )

    return (result.stdout + result.stderr)[:20_000]

即使使用白名单，也不能把它视为真正的安全沙箱。

13. 结构化输出

当程序需要稳定的 JSON 数据时，不要只在提示词里写：

1	请严格输出 JSON。

更推荐使用结构化输出和 Pydantic。

from pydantic import BaseModel


class StudentInfo(BaseModel):
    name: str
    major: str
    skills: list[str]
    graduation_year: int | None


response = client.messages.parse(
    model=model,
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": (
                "提取信息：小明是计算机专业学生，"
                "会 Python、Java，预计 2027 年毕业。"
            ),
        }
    ],
    output_format=StudentInfo,
)

student = response.parsed_output

print(student.name)
print(student.skills)

client.messages.parse() 会根据 Pydantic 模型生成结构要求、验证响应，并把结果放在 parsed_output 中。

适合：

信息抽取
API 参数生成
数据清洗
分类
表单填写
数据库写入前校验

即便使用结构化输出，仍然应该捕获验证或 API 异常。

14. 错误处理、超时与重试

import anthropic


try:
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "你好",
            }
        ],
    )

except anthropic.AuthenticationError:
    print("API Key 无效或没有权限")

except anthropic.RateLimitError:
    print("触发速率限制，请稍后重试")

except anthropic.APITimeoutError:
    print("请求超时")

except anthropic.APIConnectionError as exc:
    print("无法连接到 API")
    print(exc.__cause__)

except anthropic.APIStatusError as exc:
    print(f"API 返回错误状态：{exc.status_code}")
    print(exc.response)

常见异常包括：

AuthenticationError
PermissionDeniedError
BadRequestError
RateLimitError
APITimeoutError
APIConnectionError
InternalServerError
APIStatusError

官方 SDK 默认会对连接错误、408、409、429 和部分服务端错误进行两次自动重试，并采用退避策略。重试次数可以通过 max_retries 修改。

14.1 配置重试

1
2
3

client = Anthropic(
    max_retries=4,
)

单次请求覆盖：

response = client.with_options(
    max_retries=0,
).messages.create(
    model=model,
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "你好",
        }
    ],
)

工具如果具有副作用，例如：

扣款
发邮件
创建订单
删除文件

就不能盲目重试。否则可能重复执行。

这类工具应使用：

幂等键
操作状态查询
去重记录
明确的事务设计

14.2 配置超时

1
2
3

client = Anthropic(
    timeout=60.0,
)

更细粒度的设置：

import httpx

client = Anthropic(
    timeout=httpx.Timeout(
        timeout=60.0,
        connect=5.0,
        read=45.0,
        write=10.0,
    )
)

单次请求覆盖：

response = client.with_options(
    timeout=120.0,
).messages.create(
    model=model,
    max_tokens=4096,
    messages=[
        {
            "role": "user",
            "content": "完成一个较长的分析任务。",
        }
    ],
)

长输出通常优先考虑流式请求，而不是简单地不断增大普通请求超时时间。

15. Prompt Caching

如果很多请求都会重复发送相同的：

长系统提示
工具定义
项目文档
固定示例
长对话前缀

可以考虑 Prompt Caching。

最简单的自动缓存方式：

response = client.messages.create(
    model=model,
    max_tokens=1024,
    cache_control={
        "type": "ephemeral",
    },
    system=(
        "你是一位专业代码审查员。"
        "以下包含完整的团队编码规范……"
    ),
    messages=[
        {
            "role": "user",
            "content": "检查这段代码。",
        }
    ],
)

查看缓存使用情况：

1 2	print(response.usage.cache_creation_input_tokens) print(response.usage.cache_read_input_tokens)

Prompt Caching 会复用相同提示前缀，常用于减少重复长上下文的处理延迟和成本。自动缓存可以直接在请求顶层添加 cache_control；默认缓存周期为 5 分钟，也支持 1 小时 TTL。

缓存并不代表应用程序可以不保存聊天记录。它只是优化相同前缀的处理过程。

16. 自定义 API 地址

原代码支持：

1
2
3

client = Anthropic(
    base_url=os.getenv("ANTHROPIC_BASE_URL")
)

常用于：

公司 API 网关
本地代理
兼容 Anthropic Messages API 的服务
请求审计和统一鉴权

建议仅在变量存在时传入：

kwargs = {}

if api_key := os.getenv("ANTHROPIC_API_KEY"):
    kwargs["api_key"] = api_key

if base_url := os.getenv("ANTHROPIC_BASE_URL"):
    kwargs["base_url"] = base_url

client = Anthropic(**kwargs)

需要注意：

使用第三方兼容接口时，模型名称、鉴权方式、流式事件、工具调用和扩展字段不一定与 Anthropic 官方接口完全一致。

出现问题时应先确认：

请求是否真正发往 Anthropic 官方 API。
服务是否完整兼容 Messages API。
是否支持 Tool Use。
是否支持 SSE 流式传输。
模型 ID 是否由该服务提供。
鉴权变量究竟是 ANTHROPIC_API_KEY 还是其他名称。

原代码在配置自定义地址时删除 ANTHROPIC_AUTH_TOKEN，属于该项目为避免鉴权冲突做的特定处理，并不是所有项目都必须照搬。

17. 一个完整的命令行聊天示例

#!/usr/bin/env python3

import os

import anthropic
from anthropic import Anthropic
from dotenv import load_dotenv


load_dotenv()

MODEL = os.getenv("MODEL_ID")
API_KEY = os.getenv("ANTHROPIC_API_KEY")
BASE_URL = os.getenv("ANTHROPIC_BASE_URL")

if not MODEL:
    raise RuntimeError("请配置 MODEL_ID")

if not API_KEY:
    raise RuntimeError("请配置 ANTHROPIC_API_KEY")

client_kwargs = {
    "api_key": API_KEY,
    "timeout": 120.0,
    "max_retries": 2,
}

if BASE_URL:
    client_kwargs["base_url"] = BASE_URL

client = Anthropic(**client_kwargs)

SYSTEM = """
你是一位耐心的编程老师。
回答时优先给出直观解释，然后给出代码示例。
不要假装执行了实际并未执行的操作。
""".strip()


def extract_text(content) -> str:
    return "".join(
        block.text
        for block in content
        if block.type == "text"
    )


def main() -> None:
    history = []

    while True:
        try:
            query = input("\n你 >> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break

        if query.lower() in {"q", "quit", "exit"}:
            break

        if not query:
            continue

        history.append({
            "role": "user",
            "content": query,
        })

        try:
            with client.messages.stream(
                model=MODEL,
                max_tokens=2048,
                system=SYSTEM,
                messages=history,
            ) as stream:
                print("Claude >> ", end="", flush=True)

                for text in stream.text_stream:
                    print(text, end="", flush=True)

                response = stream.get_final_message()

            print()

            history.append({
                "role": "assistant",
                "content": response.content,
            })

            print(
                f"[input={response.usage.input_tokens}, "
                f"output={response.usage.output_tokens}, "
                f"stop={response.stop_reason}, "
                f"request_id={response._request_id}]"
            )

        except anthropic.RateLimitError:
            print("\n请求过于频繁，请稍后重试。")
            history.pop()

        except anthropic.AuthenticationError:
            print("\nAPI Key 无效或无权访问该模型。")
            history.pop()

        except anthropic.APITimeoutError:
            print("\n请求超时。")
            history.pop()

        except anthropic.APIConnectionError as exc:
            print(f"\n网络连接失败：{exc}")
            history.pop()

        except anthropic.APIStatusError as exc:
            print(f"\nAPI 错误：HTTP {exc.status_code}")
            history.pop()


if __name__ == "__main__":
    main()

18. 常见坑速查

坑一：使用 `system` 角色

错误：

1	{"role": "system", "content": "..."}

正确：

client.messages.create(
    system="...",
    messages=[...],
)

坑二：把 `response.content` 当字符串

错误：

1	print(response.content)

虽然能打印，但得到的是内容块对象。

正确：

1
2
3

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

坑三：工具执行后没有传回结果

仅执行 Python 函数还不够，必须把结果包装为：

{
    "type": "tool_result",
    "tool_use_id": call.id,
    "content": result,
}

再请求一次 Claude。

坑四：工具调用时只保存文本

不能丢掉 Claude 返回的 tool_use 内容块。

正确：

messages.append({
    "role": "assistant",
    "content": response.content,
})

坑五：流式片段逐个写入历史

历史中应该保存完整 assistant Message，而不是大量 text_delta。

坑六：子代理无限递归创建子代理

原示例只给父代理 task 工具，子代理的 CHILD_TOOLS 中没有 task。

这是很好的限制，可以避免：

无限递归
成本失控
难以追踪调用链
上下文和工具调用爆炸

坑七：工具循环没有最大次数

必须设置：

1 2	for _ in range(20): ...

超过限制后显式报错，不能无休止调用。

坑八：只用字符串黑名单保护 Shell

字符串黑名单很容易绕过。运行模型生成的命令必须使用真正的权限隔离。

坑九：没有检查 `stop_reason`

至少关注：

1 2	if response.stop_reason == "max_tokens": print("回答被截断")

坑十：在日志中打印密钥或敏感提示

不要记录：

API Key
用户隐私信息
完整认证头
数据库密码
未脱敏的工具结果

19. 推荐学习路线

第一阶段：

完成普通 messages.create()。
理解 system、messages 和内容块。
自己维护多轮历史。
读取 stop_reason 和 usage。

第二阶段：

使用 messages.stream()。
学习 AsyncAnthropic。
加入错误处理、超时和重试。
使用 Pydantic 结构化输出。

第三阶段：

定义一个只读工具。
编写完整 Tool Use 循环。
支持一轮多个工具调用。
支持工具错误反馈。
增加最大循环次数。

第四阶段：

实现父代理和子代理。
为不同代理划分工具权限。
对历史消息做摘要和压缩。
添加日志、追踪和成本统计。
使用容器或独立 Worker 隔离高风险工具。

20. 核心总结

Anthropic SDK 最重要的不是记住某个调用函数，而是理解下面这套数据流：

用户消息
   ↓
messages.create / messages.stream
   ↓
Claude 返回 content blocks
   ↓
普通文本 ───────────────→ 展示给用户
   ↓
tool_use
   ↓
应用程序执行工具
   ↓
tool_result
   ↓
再次请求 Claude
   ↓
最终回答

多轮对话的本质是：

1	应用程序自己保存 messages

子代理隔离的本质是：

1	为子任务创建独立 messages

流式输出的本质是：

1	增量展示内容，结束后获得完整 Message

工具调用的本质是：

1 2	Claude 生成结构化调用请求，应用程序负责真正执行并反馈结果

掌握这四点，就已经越过 Anthropic SDK 最容易绕晕人的部分了。

Anthropic Python SDK 入门笔记

Anthropic Python SDK 入门笔记

1. Anthropic SDK 是什么

2. 安装与环境配置

2.1 安装依赖

2.2 创建 .env

2.3 创建客户端

3. 第一次调用 Claude

3.1 主要参数

model

max_tokens

system

messages

4. 正确读取响应

4.1 查看停止原因

4.2 查看 Token 用量

4.3 记录 Request ID

5. 多轮对话

5.1 持久化历史记录

5.2 不要无限保存历史

6. 常用生成参数

temperature

stop_sequences

7. 流式输出

7.1 推荐方式：messages.stream()

7.2 只获取最终文本

7.3 底层事件流：stream=True

7.4 流式输出与历史记录

8. 异步调用

8.1 异步流式输出

9. Tool Use：让 Claude 调用函数

9.1 定义工具

9.2 编写真正的工具函数

9.3 完整工具调用循环

9.4 工具循环的三个关键点

第一，保留完整 assistant 内容

第二，tool_use_id 必须对应

第三，一轮的多个工具结果一起返回

10. 流式输出结合工具调用

11. 子代理与上下文隔离

11.1 一个需要纠正的概念

11.2 子代理基础实现

12. 工具安全

13. 结构化输出

14. 错误处理、超时与重试

14.1 配置重试

14.2 配置超时

15. Prompt Caching

16. 自定义 API 地址

17. 一个完整的命令行聊天示例

18. 常见坑速查

坑一：使用 system 角色

坑二：把 response.content 当字符串

坑三：工具执行后没有传回结果

坑四：工具调用时只保存文本

坑五：流式片段逐个写入历史

坑六：子代理无限递归创建子代理

坑七：工具循环没有最大次数

坑八：只用字符串黑名单保护 Shell

坑九：没有检查 stop_reason

坑十：在日志中打印密钥或敏感提示

19. 推荐学习路线

20. 核心总结

2.2 创建 `.env`

`model`

`max_tokens`

`system`

`messages`

`temperature`

`stop_sequences`

7.1 推荐方式：`messages.stream()`

7.3 底层事件流：`stream=True`

第二，`tool_use_id` 必须对应

坑一：使用 `system` 角色

坑二：把 `response.content` 当字符串

坑九：没有检查 `stop_reason`