第 30 章《Python SDK + OpenAI 兼容 API + 最佳实践》· nanobot SDK + OpenAI 兼容 API 源码解析：Nanobot 门面 + /v1/chat/co

本章是专栏收尾篇：解析 nanobot/nanobot.py 的 Nanobot 门面类，以及 nanobot/api/server.py 暴露的 OpenAI 兼容 HTTP API。读完本章你能用"任意能调 OpenAI SDK 的客户端"调用你自己的 nanobot 实例。

1. 整体定位：为什么需要 SDK + OpenAI 兼容 API

两种集成场景：

两者都基于同一个 AgentLoop（第 10 章），只是入口不同。本章给"外部集成者"一个完整视图。

2. 核心数据结构：Nanobot 门面

2.1 类签名（已 Read 核对）

# 来源：nanobot/nanobot.py L64-L80（简化）
class Nanobot:
"""Programmatic facade for running the nanobot agent.

Usage::

bot = Nanobot.from_config()
result = await bot.run("Summarize this repo", hooks=[MyHook()])
print(result.content)
"""

def __init__(self, loop: AgentLoop, *, config: Config | None = None) –> None:
self._loop = loop
self._config = config
self.sessions = SessionClient(loop)
self.memory = MemoryClient(loop)
self.runtime = RuntimeClient(loop)

2.2 字段表

2.3 from_config 类方法

# 来源：nanobot/nanobot.py L81-L100（简化）
@classmethod
def from_config(
cls,
config_path: str | Path | None = None,
*,
workspace: str | Path | None = None,
model: str | None = None,
model_preset: str | None = None,
) –> Nanobot:
"""Create a Nanobot instance from a config file.

Args:
config_path: Path to `config.json`. Defaults to
`~/.nanobot/config.json`.
workspace: Override the workspace directory from config.
model: Override the instance default model.
model_preset: Override the instance default model preset.
"""
from nanobot.config.loader import load_config, resolve_config_env_vars
# … 实际构造逻辑

2.4 __all__ 导出（已 Read 核对）

# 来源：nanobot/nanobot.py L42-L61
__all__ = [
"Nanobot",
"RunResult",
"RunStream",
"SessionInfo",
"SessionSnapshot",
"STREAM_EVENT_REASONING_COMPLETED",
"STREAM_EVENT_REASONING_DELTA",
"STREAM_EVENT_RUN_COMPLETED",
"STREAM_EVENT_RUN_FAILED",
"STREAM_EVENT_RUN_STARTED",
"STREAM_EVENT_TEXT_COMPLETED",
"STREAM_EVENT_TEXT_DELTA",
"STREAM_EVENT_TOOL_COMPLETED",
"STREAM_EVENT_TOOL_FAILED",
"STREAM_EVENT_TOOL_STARTED",
"STREAM_EVENT_TYPES",
"StreamEvent",
"StreamEventType",
]

STREAM_EVENT_* 常量是流式回调时 event["type"] 的取值，配合 use_nanobot_stream() 等前端 hook 使用（webui/src/hooks/useNanobotStream.ts）。

3. 流式事件协议

nanobot.sdk.streaming（sdk/streaming.py）定义了 SDK 与 AgentLoop 的流式事件：

# 事件类型常量（来源 nanobot/sdk/types.py）
STREAM_EVENT_REASONING_DELTA = "reasoning_delta" # 模型思维链片段
STREAM_EVENT_REASONING_COMPLETED = "reasoning_completed"
STREAM_EVENT_TEXT_DELTA = "text_delta" # 回复文本片段
STREAM_EVENT_TEXT_COMPLETED = "text_completed"
STREAM_EVENT_TOOL_STARTED = "tool_started" # 工具调用开始
STREAM_EVENT_TOOL_COMPLETED = "tool_completed"
STREAM_EVENT_TOOL_FAILED = "tool_failed"
STREAM_EVENT_RUN_STARTED = "run_started" # 整轮开始
STREAM_EVENT_RUN_COMPLETED = "run_completed" # 整轮结束
STREAM_EVENT_RUN_FAILED = "run_failed"

WebUI 客户端按这些事件类型渲染"思维链 → 工具调用 → 文本"的分层时间线。

4. OpenAI 兼容 HTTP API

4.1 入口

nanobot api # 启动 HTTP 服务（默认 :8765）
nanobot gateway # 启动 gateway（包含 API + WebUI）

4.2 端点速查

4.3 客户端调用示例

curl http://localhost:8765/v1/chat/completions \\
-H "Content-Type: application/json" \\
-d '{
"model": "claude-sonnet-4.5",
"messages": [{"role": "user", "content": "用一句话介绍 nanobot"}],
"stream": true
}'

# Python 客户端（任何 OpenAI SDK）
from openai import OpenAI
client = OpenAI(base_url="http: # localhost:8765/v1", api_key="not-needed")
resp = client.chat.completions.create(
model="claude-sonnet-4.5",
messages=[{"role": "user", "content": "Hello"}],
)
print(resp.choices[0].message.content)

5. 关键源码摘录

5.1 Nanobot.run 入口

# 简化示意（具体行号见 nanobot/nanobot.py）
async def run(
self,
message: str,
*,
session_key: str | None = None,
hooks: list[AgentHook] | None = None,
stream: bool = False,
) –> RunResult | RunStream:
"""Run a single agent turn.

Args:
message: User message content.
session_key: Optional override for session key (default: derive from message).
hooks: Per-turn hooks (pre_turn / post_turn).
stream: If True, return RunStream for incremental consumption.

Returns:
RunResult on non-streaming, RunStream on streaming.
"""
...

5.2 api/server.py 路由（简化）

# 简化示意（具体行号见 nanobot/api/server.py）
from fastapi import FastAPI, Request
from nanobot.nanobot import Nanobot

app = FastAPI()
_bot: Nanobot | None = None

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
body = await request.json()
# 1. 校验 model / messages
# 2. 调用 Nanobot.run(message=…, stream=body.get("stream", False))
# 3. 把 RunResult 转换为 OpenAI ChatCompletion 格式
...

6. 为什么这样设计：5 个核心决策

决策 1 · 为什么 SDK 门面叫 Nanobot 而非 Client？

仿照 openai.OpenAI / anthropic.Anthropic 的"厂商名 = 客户端类名"惯例。Nanobot.from_config() 暗示"工厂方法模式"，便于测试 mock。

决策 2 · 为什么 SDK 暴露 sessions / memory / runtime 子客户端？

nanobot 的会话 / 记忆 / 运行时是独立子系统，每个有独立 API（CRUD / 读取 / 控制）。把它们做成子客户端（SessionClient / MemoryClient / RuntimeClient）让 bot.sessions.list() 比 bot.list_sessions() 更清晰。

决策 3 · 为什么 OpenAI 兼容而非自创协议？

直接复用 OpenAI SDK 生态——任何语言、任何客户端（LangChain / LlamaIndex / Cursor / Continue / 自研脚本）零成本接入。

决策 4 · 为什么 STREAM_EVENT_* 用字符串常量而非 Enum？

前端 React 用字符串 event.type === "text_delta" 判断；后端 Python 也要序列化到 JSON。str 常量比 Enum 兼容性更好（无需 value 访问）。

决策 5 · 为什么 SDK 提供 RunStream（流）和 RunResult（非流）两种返回？

非流简单（一次 await 拿结果）；流适合长任务 + UI 增量渲染 + 工具调用可观测。两个类型明确区分让调用方选型无歧义。

7. 最佳实践（专栏收尾）

实践 1 · 脚本调用首选 SDK

import asyncio
from nanobot.nanobot import Nanobot

async def main():
bot = Nanobot.from_config()
result = await bot.run("总结当前目录的 Python 文件")
print(result.content)
await bot.close()

asyncio.run(main())

实践 2 · 多客户端接入用 gateway

nanobot gateway # 一次启动，所有通道 + WebUI + API

不要为每个客户端起一个 nanobot 进程——浪费内存 + session 隔离。

实践 3 · 自研前端用 OpenAI 兼容 API

任何能调 https://api.openai.com/v1/chat/completions 的客户端（前端 SDK / IDE 插件 / Slack bot）都能用 http://your-nanobot:8765/v1 替换 endpoint，零代码改动。

实践 4 · 生产部署关注 3 个指标

实践 5 · 升级前看 CHANGELOG + .agent/

仓库根有：

• AGENTS.md —— 给 AI Agent 的开发指南
• .agent/design.md —— 架构约束
• .agent/security.md —— 安全边界
• .agent/gotchas.md —— 常见坑

升级前通读这 4 个文档能避开 90% 的兼容性问题。

8. 常见问题 / 避坑

Q：SDK 和 HTTP API 同时用会 session 冲突吗？

A：会。两者共享同一个 ~/.nanobot/sessions/<key>.jsonl。建议：

• 调试时 SDK + HTTP 二选一
• 生产时统一走 gateway，多客户端共享同一 loop

Q：怎么禁用 HTTP API？

A：在 ~/.nanobot/config.json：

{ "api": { "enabled": false } }

或者干脆用 nanobot gateway（含 API）而 nanobot run（仅 CLI）。

Q：流式调用中途断开会怎样？

A:HTTP SSE 默认 90 秒空闲视为断开(见第 17 章 LLMProvider 抽象)。SDK 的 RunStream 会在断开时抛 asyncio.IncompleteReadError,由调用方决定重试或回退到非流。

9. 小结

• Nanobot 门面 = from_config() + sessions/memory/runtime 子客户端 + run() 入口
• SDK 流式事件 11 种类型，覆盖 reasoning / 工具 / 文本 / 回合边界
• HTTP API 端点：/v1/models / /v1/chat/completions / /v1/audio/transcriptions / /v1/images/generations
• 最佳实践：脚本用 SDK / 多客户端用 gateway / 自研前端用 OpenAI 兼容

专栏收尾

至此 30 章全部完成。回顾全专栏覆盖的 6 大主题群：

下一步建议：

• 想看代码 → 读 nanobot/agent/loop.py 主体 + nanobot/agent/runner.py
• 想跑起来 → nanobot onboard + nanobot gateway + 打开 WebUI
• 想接自家 LLM → 读第 19 章 Provider 实现对比(主题群"LLM Provider 适配")