AI Agent Engineer

AI Agent Engineer

HTTP API 与 Python SDK

5 min read

HTTP API 与 Python SDK

对应源码

主题源码
HTTP 路由和 OpenAI-compatible responsenanobot/api/server.py::handle_chat_completions、create_app
Python facadenanobot/nanobot.py::Nanobot
SDK 客户端nanobot/sdk/clients.py、nanobot/sdk/runtime.py
SDK streamnanobot/sdk/streaming.py::RunStream、SDKStreamingHook
SDK 类型nanobot/sdk/types.py

学习目标

读完后，你应当能够：

  • 说明 HTTP API 和 SDK 如何复用 process_direct()；
  • 解释 API session lock 和 SSE 生命周期；
  • 区分 SDK run、run_streamed 和 stream；
  • 判断问题属于入口适配还是核心 Agent。

前置知识

  • 理解 AgentLoop direct path；
  • 知道 HTTP JSON、SSE 和 Python async iterator 的基本概念。

1. HTTP API

实现：

  • nanobot/api/server.py
  • 使用 aiohttp.web。

主要端点：

端点作用
POST /v1/chat/completions普通或 stream 对话
GET /v1/models当前模型信息
GET /health健康检查

2. 请求解析

Chat Completions 支持：

  • JSON；
  • multipart/form-data；
  • 文本；
  • media；
  • session_id；
  • model override；
  • stream。

输入层负责把 OpenAI-compatible 请求转换成 nanobot 的 direct call 参数，而不是直接调用 Provider SDK。

3. API Session

默认：

api:default

提供 session_id 后：

api:<session_id>

API server 维护：

session_locks: dict[str, asyncio.Lock]

同一 API Session 串行，不同 Session 可并发。AgentLoop 自身也有 Session lock，API 这层锁保护 HTTP 请求和 SSE adapter 的完整生命周期。

4. 非流式 API

sequenceDiagram
    participant C as HTTP Client
    participant A as API server
    participant L as AgentLoop

    C->>A: POST JSON
    A->>A: 解析 + session lock
    A->>L: process_direct
    L-->>A: response
    A->>A: 转为 chat.completion
    A-->>C: JSON

API 会把 process_direct() 的返回值归一化为 assistant text，并封装 OpenAI-compatible response 和 usage。

5. SSE

stream 模式：

  1. 创建 SSE response；
  2. 为 on_stream 注册 delta callback；
  3. 后台运行 process_direct()；
  4. 每个 delta 写出 chat.completion.chunk；
  5. 运行完成后写 finish chunk 和 [DONE]；
  6. 关闭连接。

关键事实：

HTTP SSE 不能只等待 on_stream_end 就关闭，因为 stream segment 结束后 Agent 可能继续调用工具。必须等待完整 process_direct() 返回。

6. API 的空响应重试

非流式路径对空响应有入口层补救逻辑。调试空回复时要同时区分：

  • Runner 的空响应恢复；
  • AgentLoop 的统一空响应文本；
  • API 层的再次 direct call；
  • response text 归一化。

不要只在最外层加无限重试。

7. Python SDK

高层类：

  • nanobot/nanobot.py::Nanobot

创建：

bot = Nanobot.from_config(...)

内部：

Config
  → AgentLoop.from_config
  → Nanobot(loop)

8. run()

run()：

  • 获取 SDK 侧 Session lock；
  • 构造 process_direct kwargs；
  • 支持 session/channel/chat/sender/media；
  • 支持 ephemeral；
  • 支持 hooks；
  • 支持单次 model/preset override；
  • 返回统一 RunResult。

适合只关心最终结果的调用方。

9. run_streamed()

返回 RunStream：

  • 后台 Task 运行完整 Turn；
  • Queue 接收 StreamEvent；
  • 事件包括 run started、text/reasoning/tool 等流事件、run completed 或 failed；
  • 可以同时消费事件和最终等待结果；
  • aclose() 可取消未完成运行。

10. stream()

stream() 是更简洁的 async iterator：

async for event in bot.stream("Hello"):
    ...

内部仍调用 run_streamed()，并在 finally 中确保未完成运行被关闭。

11. 资源清理

SDK 使用完成后应调用：

await bot.aclose()

或：

async with Nanobot.from_config(...) as bot:
    ...

这会释放 MCP 等由 AgentLoop 持有的资源。

API 与 SDK 对照

特性HTTP APIPython SDK
传输HTTP JSON/SSEPython 调用/async iterator
核心入口process_directprocess_direct
Sessionapi:<id>调用方指定，默认 SDK key
StreamSSE chunktyped StreamEvent
错误HTTP/JSONPython exception + failed event
生命周期server 管理宿主程序管理
适合跨语言调用Python 内嵌

调试边界

如果 CLI direct 正常：

  • API 400：检查 body parser；
  • API stream 卡住：检查 SSE writer 和完整 Turn 等待；
  • SDK event 缺失：检查 emitter/adapter；
  • 两者最终回答都错：回到 AgentLoop/Runner；
  • Session 串线：检查入口 key 和 lock。

源码与测试

  • nanobot/api/server.py
  • nanobot/nanobot.py
  • nanobot/sdk/streaming.py
  • nanobot/sdk/types.py
  • nanobot/sdk/runtime.py
  • tests/test_openai_api.py
  • tests/test_api_stream.py
  • tests/test_api_attachment.py
  • 使用 rg -n "Nanobot\\(" tests 定位 SDK 行为测试

自检

  1. API 为什么不是直接调用 Provider？
  2. SSE 为什么必须等待 process_direct() 返回？
  3. API session lock 与 AgentLoop lock 各保护什么？
  4. run_streamed() 与 stream() 有什么关系？
  5. SDK 为什么需要 aclose()？

下一步

阅读 WebSocket 与 WebUI。

Backlinks