Anthropic 客户端 SDK 与 Messages API

import anthropic

client = anthropic.Anthropic()  # 默认从环境变量 ANTHROPIC_API_KEY 读取密钥
# 异步版本：client = anthropic.AsyncAnthropic()

response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,  # 必填：本次最多输出多少 token
    system="你是一个资深后端工程师，回答要简洁。",
    messages=[{"role": "user", "content": "用一句话解释什么是幂等性。"}],
)
for block in response.content:  # content 是一组 content block
    if block.type == "text":
        print(block.text)
print(response.stop_reason, response.usage.input_tokens, response.usage.output_tokens)

from anthropic import Anthropic
from anthropic.types import MessageParam

client = Anthropic()
history: list[MessageParam] = []


def chat(user_text: str) -> str:
    history.append({"role": "user", "content": user_text})
    response = client.messages.create(model="claude-opus-5-5", max_tokens=16000, messages=history)
    # 把完整的 content（包括 thinking 块）原样追加进历史，而不只是文本
    history.append({"role": "assistant", "content": response.content})
    return "".join(b.text for b in response.content if b.type == "text")


print(chat("我叫小王。"))
print(chat("我叫什么？"))  # 模型能记住，是因为我们把历史都发了过去

from anthropic import Anthropic

client = Anthropic()
response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    # Opus 5.5 的 thinking 始终开启；display=summarized 表示返回可读的推理摘要
    thinking={"type": "adaptive", "display": "summarized"},
    output_config={"effort": "high"},  # 可选值：low / medium / high / xhigh / max
    messages=[{"role": "user", "content": "一步一步地证明 √2 是无理数。"}],
)
for block in response.content:
    if block.type == "thinking":
        print("[推理摘要]", block.thinking)
    elif block.type == "text":
        print("[回答]", block.text)

from anthropic import Anthropic, beta_tool

client = Anthropic()


@beta_tool
def get_weather(city: str) -> str:
    """查询城市今天的天气。

    Args:
        city: 城市名，例如 北京。
    """
    return {"北京": "晴，25°C", "上海": "小雨，22°C"}.get(city, "暂无数据")


@beta_tool
def calculate_sum(a: int, b: int) -> str:
    """计算两个整数之和。

    Args:
        a: 第一个数
        b: 第二个数
    """
    return str(a + b)


runner = client.beta.messages.tool_runner(
    model="claude-opus-5-5",
    max_tokens=16000,
    tools=[get_weather, calculate_sum],  # 装饰器会根据函数签名和 docstring 生成 JSON Schema
    messages=[{"role": "user", "content": "北京天气如何？另外 15 + 27 等于多少？"}],
)
final_message = runner.until_done()  # 自动循环，直到模型不再调用工具
print("".join(b.text for b in final_message.content if b.type == "text"))

from anthropic import Anthropic
from anthropic.types import ToolParam

client = Anthropic()
book_flight: ToolParam = {
    "name": "book_flight",
    "description": "预订航班",
    "strict": True,  # 保证 tool_use.input 严格符合下面的 Schema
    "input_schema": {
        "type": "object",
        "properties": {
            "destination": {"type": "string"},
            "date": {"type": "string", "format": "date"},
            "passengers": {"type": "integer", "enum": [1, 2, 3, 4, 5, 6]},
        },
        "required": ["destination", "date", "passengers"],
        "additionalProperties": False,
    },
}
response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    tools=[book_flight],
    messages=[{"role": "user", "content": "帮我订 3 月 15 日去东京的机票，2 个人。请使用 book_flight 工具。"}],
)
calls = [b for b in response.content if b.type == "tool_use"]
print(calls[0].input if calls else "模型没有调用工具，需要再引导一次")

from anthropic import Anthropic
from anthropic.types import MessageParam

client = Anthropic()
messages: list[MessageParam] = [{"role": "user", "content": "火星车最近有什么新进展？"}]


def ask(msgs: list[MessageParam]):
    return client.messages.create(
        model="claude-opus-5-5",
        max_tokens=16000,
        tools=[{"type": "web_search_20260209", "name": "web_search", "max_uses": 5}],
        messages=msgs,
    )


response = ask(messages)
for _ in range(5):  # 给 pause_turn 续跑设一个上限
    if response.stop_reason != "pause_turn":
        break
    # 服务端工具的循环暂停了：把这一轮追加进历史后重新发送，模型会接着做
    messages.append({"role": "assistant", "content": response.content})
    response = ask(messages)

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

from pydantic import BaseModel

from anthropic import Anthropic

client = Anthropic()


class ContactInfo(BaseModel):
    name: str
    email: str
    plan: str
    demo_requested: bool


response = client.messages.parse(
    model="claude-opus-5-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "抽取信息：Jane Doe（jane@co.com）想要 Enterprise 套餐，并希望看演示。"}],
    output_format=ContactInfo,  # SDK 会把它转换成 API 的 output_config.format
)
contact = response.parsed_output  # 已经校验过的 ContactInfo 实例
print(contact)

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=64000,  # 流式请求可以设更大的上限，不用担心 HTTP 超时
    messages=[{"role": "user", "content": "写一个 300 字的科幻小故事。"}],
) as stream:
    for text in stream.text_stream:  # 只关心文本增量
        print(text, end="", flush=True)
    final = stream.get_final_message()  # 流结束后拿到完整的 Message

print("\n", final.stop_reason, final.usage.output_tokens)

from anthropic import Anthropic

client = Anthropic()
LONG_DOC = "（这里是一份很长的产品手册……）" * 500  # 示意：被缓存的前缀需要足够长

for question in ["退货政策是什么？", "保修多久？"]:
    response = client.messages.create(
        model="claude-opus-5-5",
        max_tokens=16000,
        system=[
            {
                "type": "text",
                "text": f"你是客服助手。以下是产品手册：\n{LONG_DOC}",
                "cache_control": {"type": "ephemeral"},  # 在这里打一个缓存断点
            }
        ],
        messages=[{"role": "user", "content": question}],
    )
    u = response.usage
    print(question, "写入缓存:", u.cache_creation_input_tokens, "命中缓存:", u.cache_read_input_tokens)

import anthropic

client = anthropic.Anthropic()

try:
    response = client.messages.create(
        model="claude-opus-5-5", max_tokens=1024, messages=[{"role": "user", "content": "hi"}]
    )
    print(response._request_id)  # 排查问题时把这个 ID 提供给 Anthropic
except anthropic.BadRequestError as e:  # 400：参数有误，不要重试
    print("请求参数错误：", e.message)
except anthropic.RateLimitError as e:  # 429：退避后重试
    print("限流，建议等待：", e.response.headers.get("retry-after"))
except anthropic.APIStatusError as e:  # 其他非 2xx 状态码
    print("API 错误：", e.status_code)
except anthropic.APIConnectionError:  # 网络问题
    print("网络连接失败")

from anthropic import Anthropic

client = Anthropic()

response = client.beta.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "给我一份 Linux 服务器加固清单。"}],
    fallbacks="default",  # 被拒绝时，由服务端自动换成推荐的备用模型重试
    betas=["server-side-fallback-2026-07-01"],
)

if response.stop_reason == "refusal":  # 备用模型也拒绝了
    print("请求被拒绝：", response.stop_details)
else:
    print("实际作答的模型：", response.model)
    print("".join(b.text for b in response.content if b.type == "text"))