《AI Agent 工程实战》3.3 幂等、分页与统一错误契约

import hashlib
from fastapi import Header
from fastapi.responses import JSONResponse

_idem: dict[str, tuple[str, str]] = {}   # Idempotency-Key -> (body_hash, run_id)


@app.post("/sessions/{sid}/runs", status_code=201)
def create_run(sid: str, body: RunCreate, idempotency_key: str | None = Header(default=None)):
    if sid not in _sessions:
        raise ApiError(404, "session_not_found", "会话不存在", {"session_id": sid})

    if idempotency_key:
        body_hash = hashlib.sha256(body.model_dump_json().encode()).hexdigest()[:12]
        if idempotency_key in _idem:
            old_hash, old_rid = _idem[idempotency_key]
            if old_hash != body_hash:
                raise ApiError(409, "idempotency_conflict",
                               "同一 Idempotency-Key 携带了不同请求体",
                               {"idempotency_key": idempotency_key})
            return JSONResponse(status_code=200, content=_runs[old_rid])   # 重放
        rid = "r_" + uuid.uuid4().hex[:8]
        _idem[idempotency_key] = (body_hash, rid)
    else:
        rid = "r_" + uuid.uuid4().hex[:8]

    _runs[rid] = {"id": rid, "session_id": sid, "status": "queued",
                  "input": body.input, "created_at": now()}
    return _runs[rid]

key = "idem-abc-001"
r1 = client.post(f"/sessions/{sid}/runs", json={"input": "对比 pgvector 与 Qdrant"},
                 headers={"Idempotency-Key": key})
r2 = client.post(f"/sessions/{sid}/runs", json={"input": "对比 pgvector 与 Qdrant"},
                 headers={"Idempotency-Key": key})
print("first :", r1.status_code, r1.json()["id"])
print("second:", r2.status_code, r2.json()["id"])
print("same id?", r1.json()["id"] == r2.json()["id"], " run 总数:", len(_runs))

# 同一 key 换 body
r3 = client.post(f"/sessions/{sid}/runs", json={"input": "换个问题"},
                 headers={"Idempotency-Key": key})
print("status:", r3.status_code, r3.json())

first : 201 r_d5b4e816
second: 200 r_d5b4e816
same id? True  run 总数: 1
status: 409 {'code': 'idempotency_conflict', 'message': '同一 Idempotency-Key 携带了不同请求体', 'details': {'idempotency_key': 'idem-abc-001'}, 'request_id': 'req_8c8d11aa5a12'}

class ApiError(Exception):
    def __init__(self, status: int, code: str, message: str, details: dict | None = None):
        self.status = status
        self.code = code
        self.message = message
        self.details = details or {}


@app.exception_handler(ApiError)
async def api_error_handler(request: Request, exc: ApiError) -> JSONResponse:
    return JSONResponse(status_code=exc.status, content={
        "code": exc.code,               # 机器可判别的稳定标识
        "message": exc.message,         # 给人读的一句话
        "details": exc.details,         # 结构化补充，便于前端定位
        "request_id": request.state.request_id,   # 排障时的关联 id
    })

@app.middleware("http")
async def attach_request_id(request: Request, call_next):
    request.state.request_id = "req_" + uuid.uuid4().hex[:12]
    resp = await call_next(request)
    resp.headers["X-Request-Id"] = request.state.request_id
    return resp

status: 404 {'code': 'session_not_found', 'message': '会话不存在', 'details': {'session_id': 's_nope'}, 'request_id': 'req_e4deddd821ba'}

import base64


def encode_cursor(sid: str) -> str:
    return base64.urlsafe_b64encode(sid.encode()).decode()


@app.get("/sessions")
def list_sessions(limit: int = 2, cursor: str | None = None) -> dict:
    ids = list(_sessions.keys())
    start = 0
    if cursor:
        after = base64.urlsafe_b64decode(cursor.encode()).decode()
        if after in ids:
            start = ids.index(after) + 1
    page = ids[start:start + limit]
    has_more = len(ids) > start + limit
    return {
        "items": [_sessions[i] for i in page],
        "next_cursor": encode_cursor(page[-1]) if has_more and page else None,
    }

page1: ['研究课题 0', '研究课题 1']  next_cursor=c18zNzllYjQy...
page2: ['研究课题 2', '研究课题 3']  next_cursor=c19jZmNkODFi...
page3: ['研究课题 4']  next_cursor=None