简体中文 | English

matterloop-integration-fastapi

这是一个薄 HTTP 适配层：校验请求、执行鉴权依赖、调用 Runtime、把领域错误映射成稳定状态码。 Loop 编排、持久化和 Worker 不会进入路由代码。

pip install matterloop-integration-fastapi

挂载路由

from contextlib import asynccontextmanager

from fastapi import FastAPI, Header, HTTPException
from matterloop_integration_fastapi import create_router


async def authenticate(authorization: str | None = Header(default=None)) -> str:
    principal = await identity_service.authenticate(authorization)
    if principal is None:
        raise HTTPException(status_code=401, detail="unauthorized")
    return principal.subject


@asynccontextmanager
async def lifespan(_: FastAPI):
    yield
    await runtime.aclose()


app = FastAPI(lifespan=lifespan)
app.include_router(
    create_router(runtime=runtime, auth_dependency=authenticate, prefix="/loops")
)

create_router(runtime, auth_dependency, prefix) 只创建 APIRouter。Runtime 和外部客户端应在 lifespan 中构造、共享和关闭。

路由

生产 HTTP 服务通常应传入 QueueRuntimeProtocol。DirectRuntimeProtocol 会占用请求连接，也没有 可查询的运行目录。两种协议都是结构协议，不要求继承 MatterLoop 类型。

请求约束

创建请求的形状是：

{
  "goal": "生成并验证发布说明",
  "acceptance_criteria": ["包含变更摘要", "验证链接有效"],
  "limits": {
    "max_cycles": 5,
    "max_attempts": 20,
    "max_steps_per_plan": 20,
    "timeout_seconds": 300
  },
  "metadata": {"tenant_id": "acme", "trace_id": "..."},
  "run_id": "release-note-2026-07-16"
}

HTTP DTO 使用严格 Pydantic 配置，未知字段直接拒绝。run_id 最长 128 字符，并限制为可安全用于 内部标识的字符；它仍然不是授权凭据。metadata 只允许 JSON 值，但不会自动做租户校验或脱敏。

恢复请求 ResumeLoopRequest(mode) 默认 continue；传 replan 才丢弃当前计划重新规划。

响应不是内部对象透传

RunResponse 只公开运行状态、输出和步骤记录，不返回任意 Core metadata。模型输出、验证证据和 artifact URI 仍可能敏感，API 网关应按数据分类继续过滤。

错误契约

路由错误使用固定文案，不回传捕获异常的文本。但成功响应里的 RunResponse.error 来自运行记录， 应用在持久化前仍需清理供应商正文、内部路径和凭据。

上线前需要补上的部分

• auth_dependency 只是认证挂载点。每个端点还要校验当前主体是否拥有对应 run_id。
• 网关负责请求体大小、速率限制、CORS/CSRF、审计和租户级预算。
• 当前没有提交人工反馈的 HTTP 路由，RunResponse 也不暴露 pending_interaction。仅靠本包无法完成 HTTP HITL 闭环。
• 当前没有 SSE/WebSocket、数据库迁移、Worker 或管理后台。

队列部署拓扑见企业集成指南。