SDK 开发实录：如何为你的 AI 服务编写 Python 客户端
SDK 开发实录：如何为你的 AI 服务编写 Python 客户端
摘要：当你的 AI 后端 API 日趋完善，如何让第三方开发者（或你自己的前端项目）更优雅地接入？直接调用 HTTP 接口不仅代码冗余，还容易出错。本文基于一个真实的 AI 跑步教练项目，详细解析如何从零开发一个 Python SDK。我们将深入源码，展示如何封装异步请求、处理自动重试、统一错误码映射，以及如何利用
pyproject.toml进行现代化打包。这套方案将接入成本降低了 80%，是提升 AI 服务“可集成性”的关键一步。
一、背景：为什么需要 SDK？
在项目中期，我发现自己在写前端代码和测试脚本时，总是在重复以下逻辑：
headers={"Authorization":f"Bearer{token}"}response=requests.post("http://localhost:8000/api/v1/agent",json={"query":"..."})ifresponse.status_code==429:# 处理限流...elifresponse.status_code==500:# 处理服务器错误...痛点：
- 样板代码多：每个项目都要重写一遍认证和错误处理。
- 维护困难：一旦后端 API 路径变更，所有调用方都得改代码。
- 体验差：开发者需要频繁查阅 Swagger 文档才知道参数怎么传。
为了解决这些问题，我决定开发一个官方的Python SDK (ai-run-coach)。
二、SDK 架构设计：简洁与异步并重
2.1 目录结构
sdk/ ├── ai_run_coach/ │ ├── __init__.py # 导出核心类 │ ├── client.py # 主客户端类 │ ├── exceptions.py # 自定义异常体系 │ └── models.py # 数据模型映射 ├── examples/ │ └── basic_usage.py # 使用示例 └── pyproject.toml # 现代 Python 打包配置2.2 核心设计原则
- Async First：由于后端是 FastAPI，SDK 原生支持
async/await。 - 自动鉴权：初始化时传入 API Key，后续请求自动携带。
- 智能重试：遇到网络波动或 503 错误时自动重试。
三、核心实现：Client 封装
3.1 基础请求封装
文件位置：sdk/ai_run_coach/client.py
importhttpxfromtypingimportOptional,Dict,Anyfrom.exceptionsimportApiError,RateLimitErrorclassAiRunCoachClient:def__init__(self,api_key:str,base_url:str="http://localhost:8000"):self.api_key=api_key self.base_url=base_url# 创建异步客户端，配置超时和重试self.client=httpx.AsyncClient(timeout=30.0,headers={"X-API-Key":api_key})asyncdefask_coach(self,query:str,user_id:Optional[str]=None)->Dict[str,Any]:""" 向 AI 教练提问 Args: query: 用户问题 user_id: 可选的用户标识 Returns: AI 的回答字典 """payload={"query":query,"user_id":user_id}try:response=awaitself.client.post(f"{self.base_url}/api/v1/agent",json=payload)response.raise_for_status()returnresponse.json()excepthttpx.HTTPStatusErrorase:raiseself._handle_error(e)def_handle_error(self,exc:httpx.HTTPStatusError)->Exception:"""统一错误码映射"""ifexc.response.status_code==429:returnRateLimitError("请求过于频繁，请稍后重试")returnApiError(f"API 请求失败:{exc.response.text}")关键点：
httpx：相比requests，它原生支持异步，且 API 风格高度一致。- 异常转换：将底层的 HTTP 错误转换为业务相关的自定义异常，方便上层捕获。
四、进阶实践：自动化重试与背压
4.1 装饰器实现重试逻辑
为了不让主逻辑变得臃肿，我写了一个简单的重试装饰器：
importasynciofromfunctoolsimportwrapsdefretry_on_failure(max_retries:int=3,delay:float=1.0):defdecorator(func):@wraps(func)asyncdefwrapper(*args,**kwargs):forattemptinrange(max_retries):try:returnawaitfunc(*args,**kwargs)except(ApiError,httpx.ConnectError)ase:ifattempt==max_retries-1:raisee wait_time=delay*(2**attempt)# 指数退避print(f"请求失败，{wait_time}秒后重试...")awaitasyncio.sleep(wait_time)returnwrapperreturndecorator五、打包与发布：pyproject.toml 实战
5.1 现代化配置
文件位置：sdk/pyproject.toml
[build-system] requires = ["hatchling"] build-backend = "hatchling.build" [project] name = "ai-run-coach" version = "0.1.0" description = "Official Python SDK for AiRunCoach Agent" dependencies = [ "httpx>=0.24.0", "pydantic>=2.0.0" ] [project.urls] Homepage = "https://github.com/your-repo/AiRunCoachAgent"优势：
- 声明式依赖：清晰列出 SDK 运行所需的库。
- 一键构建：配合
uv或pip install build，可以轻松生成.whl文件。
六、完整调用链追踪
6.1 开发者使用视角
七、踩坑记录与解决方案
坑1：同步与异步混用
现象：在 Jupyter Notebook 等已有事件循环的环境中调用asyncio.run()报错。
解决方案：
- 提供同步包装类
SyncAiRunCoachClient，内部通过asyncio.new_event_loop()处理。 - 或者在文档中明确建议用户在异步框架（如 FastAPI, aiohttp）中使用。
坑2：敏感信息泄露
现象：SDK 日志里打印了完整的 Request Body，包含用户的 API Key。
解决方案：
- 在
httpx挂载自定义的EventHook，在打印日志前对Authorization或X-API-Key字段进行脱敏处理。
八、总结与展望
核心价值
- 降低门槛：开发者只需
pip install即可开始调用，无需关心 HTTP 细节。 - 类型提示：配合 Pydantic 模型，IDE 能提供完美的代码补全。
- 稳定性：内置的重试和错误处理机制，让接入方的代码更加健壮。
后续优化
- Webhook 支持：在 SDK 中提供便捷的回调接收器。
- 多语言扩展：基于同样的 OpenAPI 规范，生成 TypeScript 或 Go 版本的 SDK。
九、完整源码
GitHub仓库：AiRunCoachAgent
快速演示：AiRunCoachAgent
核心文件清单：
sdk/ ├── ai_run_coach/ │ ├── client.py # 核心客户端实现 │ └── exceptions.py # 异常定义 ├── pyproject.toml # 打包配置 └── README.md # SDK 使用文档如果你觉得这篇文章对你有帮助，欢迎点赞、收藏、转发！有任何问题或建议，请在评论区留言讨论。🏃♂️💨