nanobot 源码解读 · 09 | 接入与产品化：一个核心如何接进 20+ 平台与 API

class BaseChannel(ABC):
    name: str = "base"
    display_name: str = "Base"
    send_progress: bool = True
    show_reasoning: bool = True

    @abstractmethod
    async def start(self) -> None: ...
    @abstractmethod
    async def stop(self) -> None: ...
    @abstractmethod
    async def send(self, msg: OutboundMessage) -> None: ...

    # 流式是可选的：子类实现 send_delta 才开启
    async def send_delta(self, chat_id, delta, *, stream_id=None, ...) -> None: ...
    async def send_reasoning_delta(self, chat_id, delta, ...) -> None: ...  # 思考过程
    async def send_file_edit_events(self, chat_id, edits, ...) -> None: ...

    @property
    def supports_streaming(self) -> bool:
        cfg = ...  # 配置里 streaming 开关
        return bool(streaming) and type(self).send_delta is not BaseChannel.send_delta

async def _handle_message(self, sender_id, chat_id, content, media=None,
                          session_key=None, is_dm=False, authorization_id=None):
    permission_id = authorization_id if authorization_id is not None else sender_id
    if not self.is_allowed(permission_id):
        if is_dm:
            code = generate_code(self.name, str(sender_id))     # 生成配对码
            await self.send(OutboundMessage(content=format_pairing_reply(code), ...))
            return
        else:
            logger.warning("Access denied for sender {}", sender_id)
            return
    meta = metadata or {}
    if self.supports_streaming:
        meta = {**meta, "_wants_stream": True}                  # 声明想要流式
    msg = InboundMessage(
        channel=self.name, sender_id=str(sender_id), chat_id=str(chat_id),
        content=content, media=media or [], metadata=meta,
        session_key_override=session_key,
    )
    await self.bus.publish_inbound(msg)                        # 丢回主总线！

class SessionClient:       # bot.sessions
    async def ingest(session_key, messages, ...)   # 导入历史对话，不跑模型
    def get(session_key) -> SessionSnapshot        # 只读快照
    def list() -> list[SessionInfo]                # 列出所有会话
    async def restore(snapshot, ...)                # 恢复会话
    def clear / delete / flush(...)

class MemoryClient:        # bot.memory
    def read() -> str                               # 读 MEMORY.md
    def write(text) / append_history(text, ...)    # 改写/追加长期记忆

class RuntimeClient:       # bot.runtime
    @property def model / workspace
    def add_context_provider(p) -> unsubscribe     # 每轮注入上下文
    async def compact_session(session_key)          # 手动触发压缩

app.router.add_post("/v1/chat/completions", handle_chat_completions)
app.router.add_get("/v1/models", handle_models)
app.router.add_get("/health", handle_health)

async def handle_chat_completions(request):
    # 1) 解析：支持 JSON 和 multipart/form-data（可带文件/图片）
    if content_type.startswith("multipart/"):
        text, media_paths, session_id, requested_model = await _parse_multipart(request)
    else:
        body = await request.json()
        stream = body.get("stream", False)
        text, media_paths = _parse_json_content(body)
        session_id = body.get("session_id")
    # 2) 只允许配置的单一模型名（保持简单）
    if requested_model and requested_model != model_name:
        return _error_json(400, f"Only configured model '{model_name}' is available")
    session_key = f"api:{session_id}" if session_id else API_SESSION_KEY
    # 3) 按会话加锁（呼应 02 篇同会话串行）
    session_lock = session_locks.setdefault(session_key, asyncio.Lock())

    if stream:
        # SSE 流式：起一个 task 跑 agent，token 通过 queue 传给 SSE writer
        resp = web.StreamResponse(); resp.content_type = "text/event-stream"
        await resp.prepare(request)
        queue: asyncio.Queue[str | None] = asyncio.Queue()
        async def _on_stream(token): await queue.put(token)
        async def _run():
            async with session_lock:
                response = await asyncio.wait_for(
                    agent_loop.process_direct(content=text, media=..., session_key=session_key,
                                               channel="api", chat_id=API_CHAT_ID,
                                               on_stream=_on_stream, on_stream_end=_on_stream_end),
                    timeout=timeout_s)
                ...
        task = asyncio.create_task(_run())
        while True:
            token = await queue.get()
            if token is None: break
            await resp.write(_sse_chunk(token, model_name, chunk_id))   # 拼 OpenAI SSE 块
        ...
    else:
        # 非流式：直接拿结果，包成 OpenAI 的 chat.completion 响应
        async with session_lock:
            response = await asyncio.wait_for(agent_loop.process_direct(...), timeout=timeout_s)
        return web.json_response(_chat_completion_response(response_text, model_name, ...))

def _compute_next_run(schedule, now_ms):
    if schedule.kind == "at":       return schedule.at_ms if future else None   # 一次性
    if schedule.kind == "every":    return now_ms + schedule.every_ms          # 每 N 毫秒
    if schedule.kind == "cron":     # 标准 cron 表达式 + 时区
        cron = croniter(schedule.expr, base_dt); return int(cron.get_next(datetime).timestamp()*1000)

async def run_bound_cron_job(job, *, agent, cron):
    prompt = render_template("agent/cron_reminder.md", message=job.payload.message)
    channel, chat_id, metadata = _bound_session_delivery_context(job, ...)
    metadata[CRON_TRIGGER_META] = {"job_id": job.id, "job_name": job.name, "run_id": run_id, ...}
    metadata[CRON_DEFER_UNTIL_IDLE_META] = True     # 等会话空闲再注入
    resp = await agent.submit_cron_turn(            # ← 又是入站一条消息！
        InboundMessage(channel=channel, chat_id=chat_id, content=prompt,
                       session_key=job.payload.session_key, metadata=metadata))