通过 FastAPI 提供 Agent 服务

为 Agent 提供 HTTP 接口，让其他程序可以向它提交任务。每个请求指定会话名称；使用相同名称的请求会继续磁盘上保存的会话。

本指南接续构建自己的 Agent，直接使用 core.Agent。FastAPI 管理模型的生命周期，每个请求负责管理自己的 Agent 输出流。

安装

需要 Python 3.12+ 和 OpenAI API key。创建项目，安装 Bub 与 Web 服务依赖：

uv init bub-api
cd bub-api
uv add bub fastapi uvicorn

这个服务使用基础 SDK 依赖，以及 FastAPI 和 Uvicorn。

如果要接入已有的 Bub 插件，见复用应用插件。加入插件工具或指定 hooks 后，仍可沿用下面的接口与请求生命周期。

编写服务

保存为 app.py。订单工具返回示例数据，接入自己的服务时替换函数体即可。

import asyncio
from collections import defaultdict
from contextlib import asynccontextmanager
from pathlib import Path

import republic
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel

from bub.core import Agent
from bub.core.store import FileTapeStore
from bub.core.tools import Tool


async def lookup_order(order_id: str) -> dict[str, str]:
    """Look up the status of an order."""
    return {"order_id": order_id, "status": "shipped"}


def create_agent() -> Agent:
    return Agent(
        store=FileTapeStore(Path(__file__).resolve().parent / "sessions"),
        tools=[Tool.from_callable(lookup_order)],
        system_prompt=[
            "You are an order assistant. Reply directly to the user.",
            "Use lookup_order to check order status before answering.",
        ],
    )


@asynccontextmanager
async def lifespan(app: FastAPI):
    # The key comes from REPUBLIC_OPENAI_API_KEY.
    async with republic.get_model("openai:gpt-5-mini") as model:
        app.state.agent = create_agent()
        app.state.model = model
        app.state.session_locks = defaultdict(asyncio.Lock)
        yield


app = FastAPI(lifespan=lifespan)


class Task(BaseModel):
    session_id: str
    prompt: str


@app.post("/tasks")
async def run_task(task: Task, request: Request):
    state = request.app.state
    # Runs on one session's tape must not overlap; different sessions run concurrently.
    async with state.session_locks[task.session_id]:
        parts: list[str] = []
        error: str | None = None
        async with state.agent.run([task.prompt], model=state.model, tape=task.session_id) as stream:
            async for event in stream:
                if event.kind == "text":
                    parts.append(str(event.data.get("delta", "")))
                elif event.kind == "error":
                    error = str(event.data.get("message", "Agent failed"))
        if error is not None:
            raise HTTPException(status_code=502, detail=error)
        return {"session_id": task.session_id, "output": "".join(parts)}

FastAPI lifespan 在启动时打开 Republic 模型，并在关闭时释放。每个请求都用 async with 打开 Agent 流，退出时关闭流并合并 Tape 分支，Agent 运行失败时也会执行这些操作。

会话锁避免两个请求同时基于同一份尚未完成的历史开始执行。不同会话可以并发运行。

运行

在 shell 中设置 REPUBLIC_OPENAI_API_KEY，然后启动服务：

uv run uvicorn app:app

在另一个终端中，用相同会话 ID 发送两次请求：

curl -s localhost:8000/tasks -H 'content-type: application/json' \
  -d '{"session_id": "customer-42", "prompt": "Where is order A123?"}'

curl -s localhost:8000/tasks -H 'content-type: application/json' \
  -d '{"session_id": "customer-42", "prompt": "And when did it ship?"}'

两次运行都读取 sessions/ 下的 customer-42 Tape，因此第二次回答可以参考第一次交互。示例订单只有状态，没有发货日期，Agent 无法从中查到日期。

支持更多会话

示例为每个会话保留一个内存锁。长期创建大量会话的服务应清理空闲锁。这些锁只在单进程内生效；多个 worker 共享 Tape 存储时，需要跨进程协调。协调范围应覆盖整轮运行，包括模型调用和 Tape 写入。

FileTapeStore 将历史保存在一台主机上。使用其他后端时，实现 TapeStore 或 AsyncTapeStore，再传给 Agent(store=...)，详见选择历史存储位置。如果存储连接需要打开和关闭，也应纳入应用的 lifespan。

扩展服务

通过工具或 CodeMode增加动作，用事件处理器观察调用或在执行前作出决定，通过 TapeContext选择交给模型的内容。