deer-flow/backend/app/gateway/routers/assistants_compat.py
Aari 0d4d0cb17d
feat(agents): database-backed storage for custom agent definitions (#4359)
* feat(agents): database-backed storage for custom agent definitions

Add an agent_storage.backend switch (default file, behaviour-unchanged) with a
db backend that stores each custom agent as a row in the shared SQL persistence
layer, so a multi-instance deployment sees the same agents on every node
(#4331, #4357). Introduces an AgentStore interface routing all read/write
surfaces, an agents table + migration 0006, startup validation, and a file->db
importer. Follows the thread_meta store / run_events backend-switch /
0003_scheduled_tasks migration patterns; no new dependency.

* fix(agents): make db storage path production-ready (review round 1)

Addresses review feedback on the db/sync agent-storage path:

- sql.py: mirror the async engine's per-connection SQLite PRAGMAs on the sync
  engine (busy_timeout=30000, synchronous=NORMAL, foreign_keys=ON, WAL) so both
  engines behave identically against the shared DB; guard the engine cache with
  a lock (double-checked) so concurrent first-touch cannot build duplicate
  engines or register the connect listener twice.
- routers/agents.py + routers/assistants_compat.py: offload the sync-store reads
  that ran on the event loop (list/get/check, update's pre-read + legacy guard +
  refresh, and assistants_compat's four list routes) via asyncio.to_thread — on
  db+postgres each was a network round trip stalling the loop. Writes were
  already offloaded.
- file.py: translate the create() mkdir(exist_ok=False) race FileExistsError
  into AgentExistsError (router 409, matching SqlAgentStore's IntegrityError
  path); correct the _write docstring — per-file atomic replace, two commits
  sequential not transactional.

Tests: sync-engine PRAGMA + engine-cache reuse assertions; file create-race ->
AgentExistsError; strict Blockbuster anchor over the read endpoints so a
regression back onto the loop fails CI.

* fix(agents): address round-2 review on the db store path

- update_agent tool: align the docstring/inline comment with FileAgentStore._write.
  Cross-field write atomicity is db-only; the file backend commits config then
  soul via two sequential os.replace (a crash between them can leave a fresh
  config.yaml beside a stale SOUL.md). The dropped partial-write *reporting* is
  an intentional tradeoff — the stage-then-replace safety is preserved
  (test_update_agent_soul_failure_does_not_replace_config still holds).
- SqlAgentStore.update(): true upsert. Catch IntegrityError on the
  insert-on-missing branch, re-fetch and apply, so two concurrent first-time
  writes (e.g. two setup_agent handshakes) converge instead of surfacing a raw
  UNIQUE(user_id, name) violation as a 500. Symmetric with create().
- get_agent_store(): document the graph-subprocess config-resolution invariant
  (the except->file fallback is a genuine no-config path, not a mask for a
  misconfigured graph process) and pin it with two tests driving the real
  get_app_config() file resolution: db resolves from an on-disk config.yaml,
  file fallback when config is unresolvable.

* test(agents): cover SqlAgentStore.update() write-race upsert recovery

Mandatory-TDD test for the round-2 fix in 0680340a: two concurrent first-time
update()s where the loser's insert hits UNIQUE(user_id, name). Deterministically
forces the IntegrityError recovery path by making the first _row probe miss the
committed winner, and asserts last-writer-wins instead of a surfaced 500.
2026-07-23 08:03:21 +08:00

153 lines
4.9 KiB
Python

"""Assistants compatibility endpoints.
Provides LangGraph Platform-compatible assistants API backed by the
``langgraph.json`` graph registry and ``config.yaml`` agent definitions.
This is a minimal stub that satisfies the ``useStream`` React hook's
initialization requirements (``assistants.search()`` and ``assistants.get()``).
"""
from __future__ import annotations
import asyncio
import logging
from datetime import UTC, datetime
from typing import Any
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/assistants", tags=["assistants-compat"])
class AssistantResponse(BaseModel):
assistant_id: str
graph_id: str
name: str
config: dict[str, Any] = Field(default_factory=dict)
metadata: dict[str, Any] = Field(default_factory=dict)
description: str | None = None
created_at: str = ""
updated_at: str = ""
version: int = 1
class AssistantSearchRequest(BaseModel):
graph_id: str | None = None
name: str | None = None
metadata: dict[str, Any] | None = None
limit: int = 10
offset: int = 0
def _get_default_assistant() -> AssistantResponse:
"""Return the default lead_agent assistant."""
now = datetime.now(UTC).isoformat()
return AssistantResponse(
assistant_id="lead_agent",
graph_id="lead_agent",
name="lead_agent",
config={},
metadata={"created_by": "system"},
description="DeerFlow lead agent",
created_at=now,
updated_at=now,
version=1,
)
def _list_assistants() -> list[AssistantResponse]:
"""List all available assistants from config."""
assistants = [_get_default_assistant()]
# Also include custom agents from config.yaml agents directory
try:
from deerflow.config.agents_config import list_custom_agents
for agent_cfg in list_custom_agents():
now = datetime.now(UTC).isoformat()
assistants.append(
AssistantResponse(
assistant_id=agent_cfg.name,
graph_id="lead_agent", # All agents use the same graph
name=agent_cfg.name,
config={},
metadata={"created_by": "user"},
description=agent_cfg.description or "",
created_at=now,
updated_at=now,
version=1,
)
)
except Exception:
logger.debug("Could not load custom agents for assistants list")
return assistants
@router.post("/search", response_model=list[AssistantResponse])
async def search_assistants(body: AssistantSearchRequest | None = None) -> list[AssistantResponse]:
"""Search assistants.
Returns all registered assistants (lead_agent + custom agents from config).
"""
# _list_assistants reads custom agents through the sync agent store (file IO
# or a DB round trip), so keep it off the event loop.
assistants = await asyncio.to_thread(_list_assistants)
if body and body.graph_id:
assistants = [a for a in assistants if a.graph_id == body.graph_id]
if body and body.name:
assistants = [a for a in assistants if body.name.lower() in a.name.lower()]
offset = body.offset if body else 0
limit = body.limit if body else 10
return assistants[offset : offset + limit]
@router.get("/{assistant_id}", response_model=AssistantResponse)
async def get_assistant_compat(assistant_id: str) -> AssistantResponse:
"""Get an assistant by ID."""
for a in await asyncio.to_thread(_list_assistants):
if a.assistant_id == assistant_id:
return a
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
@router.get("/{assistant_id}/graph")
async def get_assistant_graph(assistant_id: str) -> dict:
"""Get the graph structure for an assistant.
Returns a minimal graph description. Full graph introspection is
not supported in the Gateway — this stub satisfies SDK validation.
"""
found = any(a.assistant_id == assistant_id for a in await asyncio.to_thread(_list_assistants))
if not found:
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
return {
"graph_id": "lead_agent",
"nodes": [],
"edges": [],
}
@router.get("/{assistant_id}/schemas")
async def get_assistant_schemas(assistant_id: str) -> dict:
"""Get JSON schemas for an assistant's input/output/state.
Returns empty schemas — full introspection not supported in Gateway.
"""
found = any(a.assistant_id == assistant_id for a in await asyncio.to_thread(_list_assistants))
if not found:
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
return {
"graph_id": "lead_agent",
"input_schema": {},
"output_schema": {},
"state_schema": {},
"config_schema": {},
}