FastAPI для агентских API:
streaming, задачи, WebSocket

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

async def agent_events(question: str, request: Request):
    try:
        async for chunk in agent.astream(question):        # агент стримит шаги/токены
            if await request.is_disconnected():            # клиент ушёл — прекращаем
                break                                       # (не жжём токены впустую)
            payload = {"type": chunk.type, "content": chunk.content}
            yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"   # формат SSE
    except Exception as e:
        yield f"data: {json.dumps({'type': 'error', 'content': str(e)})}\n\n"
    finally:
        yield "data: [DONE]\n\n"                            # явный маркер конца

@app.post("/chat/stream")
async def chat_stream(question: str, request: Request):
    return StreamingResponse(
        agent_events(question, request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},  # не буферизовать
    )

from fastapi import BackgroundTasks, HTTPException
import uuid

@app.post("/research", status_code=202)          # 202 Accepted — «принято в работу»
async def start_research(topic: str, bg: BackgroundTasks):
    job_id = str(uuid.uuid4())
    await store.set(job_id, {"status": "queued", "topic": topic})
    bg.add_task(run_research, job_id, topic)     # выполнится ПОСЛЕ отправки ответа
    return {"job_id": job_id, "status_url": f"/research/{job_id}"}

async def run_research(job_id: str, topic: str):
    await store.update(job_id, {"status": "running", "progress": 0})
    try:
        result = await agent.deep_research(topic,
                    on_progress=lambda p: store.update(job_id, {"progress": p}))
        await store.update(job_id, {"status": "done", "result": result})
    except Exception as e:
        await store.update(job_id, {"status": "failed", "error": str(e)})

@app.get("/research/{job_id}")
async def get_status(job_id: str):
    job = await store.get(job_id)
    if not job:
        raise HTTPException(404, "job not found")
    return job          # клиент опрашивает (или подписывается на SSE со статусом)

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/agent")
async def agent_ws(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            msg = await ws.receive_json()          # ждём сообщение клиента

            if msg["type"] == "cancel":            # ← клиент может прервать в ЛЮБОЙ момент
                await current_task.cancel()
                continue

            async for chunk in agent.astream(msg["content"]):
                await ws.send_json({"type": "token", "content": chunk})

                if chunk.needs_approval:           # ← агент СПРАШИВАЕТ пользователя (HITL)
                    await ws.send_json({"type": "approval_request",
                                        "action": chunk.action})
                    answer = await ws.receive_json()      # ждём решения человека
                    await agent.resume(approved=answer["approved"])

            await ws.send_json({"type": "done"})
    except WebSocketDisconnect:
        await cleanup(ws)                          # обязательная очистка ресурсов

ПОЧЕМУ НЕ ОБЫЧНЫЙ REST: ответ ДОЛГИЙ (10–60с) · идёт ПО КУСОЧКАМ ·
  часть задач ДОЛЬШЕ HTTP · нужен прогресс

3 ПАТТЕРНА:
  ① SSE-СТРИМИНГ (по умолчанию) — StreamingResponse + async-генератор
     media_type="text/event-stream" · "data: {...}\\n\\n" · маркер [DONE]
     ✓ TTFT, просто · ✗ односторонний, живёт пока висит запрос
     ⚠️ ПРОКСИ БУФЕРИЗУЕТ → X-Accel-Buffering: no + проверять на ПРОДЕ, не локально
     ✓ request.is_disconnected() → клиент ушёл, не жжём токены
  ② JOB-ПАТТЕРН (дольше HTTP) — POST → 202 + job_id, работа в фоне, GET /jobs/{id}
     ✓ переживает закрытие вкладки/таймауты · ✗ нужно хранилище состояния
     ⚠️ BackgroundTasks = В ТОМ ЖЕ ПРОЦЕССЕ: погибнет при рестарте, не распределится
        → прод: внешняя очередь (Celery/RQ/Dramatiq/ARQ) + Redis/БД (НЕ словарь в памяти)
  ③ WEBSOCKET (нужна двусторонность) — HITL, отмена, уточнения в любой момент
     ✗ reconnect, состояние сессии, sticky-балансировка → только если ДЕЙСТВИТЕЛЬНО нужно

ГЛАВНАЯ ЛОВУШКА — БЛОКИРУЮЩИЙ КОД в async def:
  ❌ time.sleep · requests · синхронный SDK · тяжёлый CPU → встаёт ВЕСЬ event loop
  ✓ await asyncio.sleep · httpx.AsyncClient · async-клиенты LLM
  ✓ легаси-синхронное → asyncio.to_thread(...) ИЛИ обычный def (FastAPI сам в пул потоков)

ГРАБЛИ: синхронный REST · блокирующий код · BackgroundTasks для важного ·
        не проверять отключение клиента · состояние задач в памяти процесса