第 3 章
Token 经济学：Input/Output/Thinking/Cache Token 精确计算与成本预估
第三章：API 快速上手：鉴权、限流、SDK 安装与第一个请求
3.1 获取 API Key
在调用 Claude API 之前，你需要一个 API Key。获取流程如下：
- 访问 console.anthropic.com
- 注册账户（需要邮箱验证）
- 进入 API Keys 页面
- 点击 Create Key，命名你的 key（建议按用途命名，如
production-chatbot、dev-testing） - 复制并立即保存 API Key——它只会显示一次
API Key 的安全管理
API Key 拥有访问你账户的完整权限，包括消耗你的信用额度。永远不要：
- 将 API Key 硬编码到源代码中
- 将包含 API Key 的文件提交到 Git 仓库
- 在客户端（浏览器/移动端）代码中使用 API Key
正确的做法是通过环境变量传递：
# Linux/macOS
export ANTHROPIC_API_KEY="sk-ant-api03-..."
# Windows PowerShell
$env:ANTHROPIC_API_KEY = "sk-ant-api03-..."
# .env 文件（配合 python-dotenv 使用）
ANTHROPIC_API_KEY=sk-ant-api03-...
对于生产环境，推荐使用密钥管理服务（AWS Secrets Manager、GCP Secret Manager、HashiCorp Vault 等），避免在任何配置文件中明文存储 key。
Key 的格式
Anthropic API Key 以 sk-ant-api03- 开头，后接约 90 个字符的随机字符串。如果你看到其他格式，可能使用了旧版本的 key 或者来自其他服务。
3.2 理解限流机制
Anthropic 的限流（Rate Limiting）有两个独立维度，需要分别理解：
速率限制（RPM / TPM）
每个账户有两种速率限制：
- RPM（Requests Per Minute）：每分钟最大请求数
- TPM（Tokens Per Minute）：每分钟最大 token 数（输入+输出）
- TPD（Tokens Per Day）：每日最大 token 数（部分账户层级有此限制）
不同账户等级的限制不同。以 Sonnet 为例（2025 年数据，以官方文档为准）：
账户等级 RPM Input TPM Output TPM
──────────── ──── ───────── ──────────
Free / Tier 1 50 40,000 8,000
Tier 2 1,000 80,000 16,000
Tier 3 2,000 160,000 32,000
Tier 4 4,000 400,000 80,000
Tier 5 4,000 400,000 80,000（+TPD 限制）
升级账户等级需要在控制台中完成充值或绑定账单，并等待 Anthropic 审核（通常在 24-48 小时内）。
限流响应的处理
当触发限流时，API 返回 HTTP 429 状态码：
{
"type": "error",
"error": {
"type": "rate_limit_error",
"message": "Rate limit exceeded. Please wait 30 seconds before retrying."
}
}
响应头中包含限流相关信息：
anthropic-ratelimit-requests-limit: 1000
anthropic-ratelimit-requests-remaining: 0
anthropic-ratelimit-requests-reset: 2024-01-15T10:30:00Z
anthropic-ratelimit-tokens-limit: 80000
anthropic-ratelimit-tokens-remaining: 0
anthropic-ratelimit-tokens-reset: 2024-01-15T10:30:30Z
retry-after: 30
正确的重试策略是指数退避（Exponential Backoff）：
import time
import anthropic
from anthropic import RateLimitError
def call_with_retry(client, max_retries=5, **kwargs):
"""
带指数退避的 API 调用
"""
for attempt in range(max_retries):
try:
return client.messages.create(**kwargs)
except RateLimitError as e:
if attempt == max_retries - 1:
raise # 最后一次重试也失败，则抛出异常
# 从响应头读取等待时间，或使用指数退避
wait_time = (2 ** attempt) + (0.1 * attempt) # 1, 2.1, 4.2, 8.3, 16.4 秒
print(f"限流触发，{wait_time:.1f}秒后重试（第{attempt+1}次）")
time.sleep(wait_time)
并发请求的 Token 预算管理
在高并发场景下，TPM 限制可能比 RPM 更先触发。以下是一个简单的 token 预算管理器：
import threading
import time
from collections import deque
class TokenBudgetManager:
"""
滑动窗口 token 预算管理器
确保每分钟 token 消耗不超过限制
"""
def __init__(self, tokens_per_minute: int):
self.tpm_limit = tokens_per_minute
self.window = deque() # (timestamp, token_count) 队列
self.lock = threading.Lock()
def can_request(self, estimated_tokens: int) -> bool:
with self.lock:
now = time.time()
# 移除 60 秒之前的记录
while self.window and now - self.window[0][0] > 60:
self.window.popleft()
current_usage = sum(t for _, t in self.window)
return current_usage + estimated_tokens <= self.tpm_limit
def record_usage(self, token_count: int):
with self.lock:
self.window.append((time.time(), token_count))
def wait_for_budget(self, estimated_tokens: int, timeout: float = 120):
"""等待直到有足够的 token 预算"""
start = time.time()
while not self.can_request(estimated_tokens):
if time.time() - start > timeout:
raise TimeoutError("等待 token 预算超时")
time.sleep(1)
3.3 安装 SDK
Python SDK
Anthropic 官方 Python SDK 是最完整的实现，支持所有 API 特性。
pip install anthropic
# 或者用 poetry
poetry add anthropic
# 或者用 uv
uv add anthropic
推荐版本固定策略：
# pyproject.toml
[tool.poetry.dependencies]
python = "^3.9"
anthropic = "^0.34.0" # 允许小版本更新，锁定大版本
SDK 依赖项很轻量：主要是 httpx（HTTP 客户端）、pydantic（数据验证）和 typing-extensions。
TypeScript/Node.js SDK
npm install @anthropic-ai/sdk
# 或
yarn add @anthropic-ai/sdk
# 或
pnpm add @anthropic-ai/sdk
TypeScript 版本提供完整的类型定义，在 TypeScript 项目中享有完整的 IDE 提示。
// package.json 版本参考
{
"dependencies": {
"@anthropic-ai/sdk": "^0.26.0"
}
}
直接使用 HTTP（无 SDK）
对于不使用 Python 或 Node.js 的环境，可以直接调用 REST API：
curl https://api.anthropic.com/v1/messages \
-H "x-api-key: $ANTHROPIC_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "content-type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{"role": "user", "content": "Hello, Claude!"}
]
}'
注意两个必须的请求头：
x-api-key：你的 API Keyanthropic-version：API 版本标识符，目前固定使用2023-06-01
3.4 第一个请求：从零到运行
Python 版本
import anthropic
# 客户端自动从 ANTHROPIC_API_KEY 环境变量读取 key
client = anthropic.Anthropic()
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "用一句话解释量子纠缠。"}
]
)
print(message.content[0].text)
# 输出类似：量子纠缠是指两个或多个粒子之间存在关联，
# 当测量一个粒子的状态时，无论相距多远，另一个粒子的状态会即刻确定。
理解响应对象结构
# message 对象的完整结构
print(message)
# Message(
# id='msg_01XFDUDYJgAACzvnptvVoYEL',
# type='message',
# role='assistant',
# content=[
# TextBlock(
# text='量子纠缠是指...',
# type='text'
# )
# ],
# model='claude-sonnet-4-6',
# stop_reason='end_turn',
# stop_sequence=None,
# usage=Usage(
# input_tokens=21,
# output_tokens=58
# )
# )
# 常用字段访问
text = message.content[0].text # 响应文本
input_tokens = message.usage.input_tokens # 输入 token 数
output_tokens = message.usage.output_tokens # 输出 token 数
stop_reason = message.stop_reason # 停止原因
model_used = message.model # 实际使用的模型版本
TypeScript 版本
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
// API key 自动从 process.env.ANTHROPIC_API_KEY 读取
async function main() {
const message = await client.messages.create({
model: "claude-sonnet-4-6",
max_tokens: 1024,
messages: [
{ role: "user", content: "用一句话解释量子纠缠。" }
],
});
const textBlock = message.content[0];
if (textBlock.type === "text") {
console.log(textBlock.text);
}
console.log(`Token 使用：输入 ${message.usage.input_tokens}，输出 ${message.usage.output_tokens}`);
}
main();
3.5 流式响应（Streaming）
对于需要实时显示响应的应用（如聊天界面），使用流式响应可以显著改善用户体验。
Python 流式示例
import anthropic
client = anthropic.Anthropic()
# 方法一：使用上下文管理器（推荐）
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "写一首关于春天的五言绝句。"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print() # 换行
# 获取最终消息（包含完整的 usage 信息）
final_message = stream.get_final_message()
print(f"\n总计：{final_message.usage.input_tokens} 输入 + "
f"{final_message.usage.output_tokens} 输出 tokens")
# 方法二：直接使用 stream 对象，处理所有事件类型
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=2048,
messages=[{"role": "user", "content": "详细解释 TCP 三次握手"}]
) 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[流结束]")
TypeScript 流式示例
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
async function streamResponse() {
const stream = await client.messages.stream({
model: "claude-sonnet-4-6",
max_tokens: 1024,
messages: [{ role: "user", content: "写一首关于春天的五言绝句。" }],
});
for await (const chunk of stream) {
if (
chunk.type === "content_block_delta" &&
chunk.delta.type === "text_delta"
) {
process.stdout.write(chunk.delta.text);
}
}
const finalMessage = await stream.finalMessage();
console.log(`\nTokens: ${finalMessage.usage.input_tokens} in, ${finalMessage.usage.output_tokens} out`);
}
streamResponse();
3.6 带 System Prompt 的完整示例
System prompt 是定义 Claude 行为和角色的核心工具。以下是一个实用的完整示例：
import anthropic
import os
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
def create_technical_assistant():
"""创建一个技术文档助手"""
system_prompt = """你是一位专业的技术文档工程师，专注于为开发者提供清晰、准确的技术解释。
你的工作原则：
1. 始终提供可运行的代码示例
2. 解释每个代码块的作用
3. 指出常见的陷阱和注意事项
4. 在不确定时，明确说明你的不确定性
输出格式：
- 使用 Markdown 格式
- 代码块需要指定语言
- 重要概念用粗体标注
"""
return system_prompt
def ask(question: str, conversation_history: list = None) -> tuple[str, list]:
"""
向技术助手提问
返回：(响应文本, 更新后的对话历史)
"""
if conversation_history is None:
conversation_history = []
# 添加用户消息
conversation_history.append({
"role": "user",
"content": question
})
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=create_technical_assistant(),
messages=conversation_history
)
assistant_reply = response.content[0].text
# 将助手响应加入历史
conversation_history.append({
"role": "assistant",
"content": assistant_reply
})
return assistant_reply, conversation_history
# 使用示例
history = []
reply, history = ask("Python 中的 GIL 是什么？", history)
print(reply)
reply, history = ask("那异步编程如何绕过 GIL 的限制？", history)
print(reply)
3.7 错误处理完全指南
生产代码必须处理各类 API 错误。Anthropic SDK 提供了结构化的异常类：
import anthropic
from anthropic import (
APIError,
APIConnectionError,
APITimeoutError,
AuthenticationError,
BadRequestError,
NotFoundError,
PermissionDeniedError,
RateLimitError,
InternalServerError,
UnprocessableEntityError,
)
def robust_api_call(client, **kwargs):
"""
带完整错误处理的 API 调用
"""
try:
return client.messages.create(**kwargs)
except AuthenticationError as e:
# API Key 无效或已过期
print(f"认证失败：{e.message}")
print("请检查 ANTHROPIC_API_KEY 环境变量")
raise
except BadRequestError as e:
# 请求参数错误（如 max_tokens 超出限制）
print(f"请求参数错误：{e.message}")
print(f"错误代码：{e.status_code}")
raise
except RateLimitError as e:
# 触发限流
print(f"限流触发：{e.message}")
# 此处应实现重试逻辑（见 3.2 节）
raise
except APIConnectionError as e:
# 网络连接问题
print(f"连接失败：{e}")
# 可安全重试
raise
except APITimeoutError as e:
# 请求超时
print(f"请求超时：{e}")
# 可安全重试
raise
except InternalServerError as e:
# Anthropic 服务端错误（5xx）
print(f"服务端错误：{e.message}")
# 短暂等待后可重试
raise
except UnprocessableEntityError as e:
# 422 错误：通常是 prompt 内容违反使用政策
print(f"内容策略错误：{e.message}")
# 不应重试，需要修改请求内容
raise
except APIError as e:
# 其他 API 错误
print(f"API 错误（{e.status_code}）：{e.message}")
raise
可重试 vs 不可重试错误
可安全重试（等待后重试）：
429 RateLimitError - 等待 retry-after 头指定的时间
500 InternalServerError - 指数退避，最多 3 次
APIConnectionError - 检查网络后重试
APITimeoutError - 增加超时时间后重试
不可重试（需要修复代码或请求）：
401 AuthenticationError - 检查 API Key
400 BadRequestError - 修复请求参数
422 UnprocessableEntityError - 修改请求内容
404 NotFoundError - 检查模型 ID 是否正确
3.8 配置 HTTP 客户端
自定义超时
import anthropic
import httpx
# 设置自定义超时（默认是 600 秒）
client = anthropic.Anthropic(
timeout=httpx.Timeout(
connect=5.0, # 连接超时：5 秒
read=120.0, # 读取超时：2 分钟
write=10.0, # 写入超时：10 秒
pool=5.0 # 连接池超时：5 秒
)
)
使用代理
import anthropic
import httpx
client = anthropic.Anthropic(
http_client=httpx.Client(
proxy="http://your-proxy.example.com:8080"
)
)
连接池配置（高并发场景）
import anthropic
import httpx
# 为高并发场景配置更大的连接池
client = anthropic.Anthropic(
http_client=httpx.Client(
limits=httpx.Limits(
max_connections=100,
max_keepalive_connections=20
)
)
)
3.9 异步客户端
对于基于 asyncio 的应用（FastAPI、aiohttp 等），使用异步客户端可以避免阻塞事件循环：
import asyncio
import anthropic
async def main():
# 使用 AsyncAnthropic 而不是 Anthropic
client = anthropic.AsyncAnthropic()
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "解释异步编程的核心概念"}]
)
print(message.content[0].text)
# FastAPI 中的使用示例
from fastapi import FastAPI
app = FastAPI()
async_client = anthropic.AsyncAnthropic()
@app.post("/chat")
async def chat_endpoint(user_message: str):
response = await async_client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": user_message}]
)
return {"reply": response.content[0].text}
小结
本章覆盖了从零开始使用 Claude API 的所有基础知识：
- API Key 获取与安全管理：始终通过环境变量传递，永不硬编码
- 限流机制：理解 RPM 和 TPM 两个维度，用指数退避处理 429 错误
- SDK 安装：Python（
pip install anthropic）和 TypeScript（npm install @anthropic-ai/sdk） - 第一个请求：5 行代码完成基本调用，理解响应对象结构
- 流式响应：使用
stream()上下文管理器实现实时输出 - 错误处理：区分可重试和不可重试错误，实现生产级的健壮性
下一章将深入 Prompt 工程，学习如何通过系统性的角色设定、指令结构和上下文管理，最大化 Claude 的输出质量。