共计 2543 个字符,预计需要花费 7 分钟才能阅读完成。
前置准备
获取 API 密钥
- 登录 Claude 开发者平台
- 进入「API 管理」页面创建新密钥
- 复制生成的 API Key(注意:密钥只显示一次)
环境配置
-
Python 3.8+ 环境检查:
python --version
-
安装必要依赖:
pip install requests python-dotenv -
环境变量配置(推荐使用.env 文件):
# .env 示例 CLAUDE_API_KEY=your_api_key_here CLAUDE_API_VERSION=2023-06-01
核心实现
基础请求示例
import os
from dotenv import load_dotenv
import requests
from typing import Dict, Any
# 加载环境变量
load_dotenv()
# 类型注解增强可读性
def create_claude_request(prompt: str) -> Dict[str, Any]:
"""
构建 Claude API 请求
:param prompt: 输入的提示文本
:return: 包含 headers 和 payload 的字典
"""return {"headers": {"x-api-key": os.getenv("CLAUDE_API_KEY"),"Content-Type":"application/json","anthropic-version": os.getenv("CLAUDE_API_VERSION")
},
"json": {"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
"max_tokens_to_sample": 300,
"temperature": 0.7
}
}
def call_claude_api(prompt: str) -> str:
"""
调用 Claude API 并处理响应
:param prompt: 用户输入的提示
:return: API 返回的文本结果
"""
try:
request_data = create_claude_request(prompt)
response = requests.post(
"https://api.anthropic.com/v1/complete",
headers=request_data["headers"],
json=request_data["json"]
)
response.raise_for_status() # 自动处理 4xx/5xx 错误
return response.json()["completion"]
except requests.exceptions.RequestException as e:
print(f"API 请求失败: {str(e)}")
return ""
关键点解析
- 认证头生成:
x-api-key必须从环境变量读取-
版本号需要与 API 文档保持一致
-
请求体规范:
- 必须包含 Human/Assistant 对话标记
-
temperature控制生成随机性(0- 1 范围) -
错误处理:
- 使用 raise_for_status() 捕获 HTTP 错误
- 建议记录完整的错误响应用于调试
避坑指南
常见错误代码
- 401 Unauthorized:
- 检查 API 密钥是否过期
-
验证请求头是否完整
-
429 Too Many Requests:
- 实现指数退避重试机制
-
参考 API 文档的速率限制说明
-
400 Bad Request:
- 检查 prompt 格式是否符合要求
- 验证参数类型和取值范围
进阶优化
批处理实现
from typing import List
def batch_process(prompts: List[str]) -> List[str]:
"""
批量处理多个 prompt(注意速率限制):param prompts: 提示文本列表
:return: 结果列表
"""
return [call_claude_api(prompt) for prompt in prompts]
异步调用模式
import aiohttp
import asyncio
async def async_call(prompt: str) -> str:
"""异步调用示例(需安装 aiohttp)"""
async with aiohttp.ClientSession() as session:
request_data = create_claude_request(prompt)
async with session.post(
"https://api.anthropic.com/v1/complete",
headers=request_data["headers"],
json=request_data["json"]
) as response:
return await response.json()
测试验证
单元测试示例
import unittest
from unittest.mock import patch
class TestClaudeAPI(unittest.TestCase):
@patch('requests.post')
def test_api_call(self, mock_post):
# 配置模拟响应
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {"completion": "测试响应"}
result = call_claude_api("测试 prompt")
self.assertEqual(result, "测试响应")
接入方式对比
| 特性 | REST | WebSocket |
|---|---|---|
| 连接方式 | 短连接 | 长连接 |
| 适用场景 | 低频请求 | 实时交互 |
| 复杂度 | 低 | 中 |
| 吞吐量 | 受限于 HTTP | 更高 |
最佳实践
- 密钥管理:
- 永远不要硬编码 API 密钥
-
使用密钥轮换策略
-
性能优化:
- 合理设置 max_tokens_to_sample
-
对稳定流量使用连接池
-
监控建议:
- 记录请求延迟和错误率
- 设置速率限制告警
延伸思考
- 如何设计一个自动重试机制来处理临时性 API 故障?
- 当需要处理超长文本时,应该采用什么分块策略?
- 如何利用流式响应实现打字机效果的用户体验?
通过本文的实践示例,你应该已经掌握了 Claude API 的基础接入方法。建议从简单对话场景开始,逐步尝试更复杂的参数组合和业务集成。遇到问题时,善用官方文档和社区支持资源将大大提高开发效率。
正文完

