# ruff: noqa: E402 # load_dotenv() must run before application imports because they read environment # configuration during import-time initialisation. import asyncio import logging import os import queue import sys import threading from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from datetime import datetime from time import perf_counter from typing import Any from dotenv import load_dotenv from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware # Load .env BEFORE any module that reads os.environ at import time. # Without this, GATEWAY_CORS_ORIGINS / AUTH_JWT_SECRET / DEER_FLOW_AUTH_DISABLED # placed in /workspace/.env are silently ignored when the container is started # without explicit `-e` flags, leading to ephemeral JWT secrets, missing CORS # origins, and "Failed to fetch" in the browser. load_dotenv() from app.gateway.auth.disabled_mode import is_auth_disabled # noqa: E402 from app.gateway.auth_middleware import AuthMiddleware from app.gateway.config import get_gateway_config from app.gateway.csrf_middleware import CSRFMiddleware from app.gateway.deps import langgraph_runtime from app.gateway.routers import ( admin_active_runs, admin_users, agents, ai_writing, artifacts, assistants_compat, auth, browser_context, business_mapping, channels, checkpoint_migration, dashboard_sessions, feedback, flow_extract, html_page_favorites, intent, knowledge, knowledge_ingest, light_apps, llm_metrics, maf_conversations, maf_roundtable, mcp, memory, menu_overrides, models, multi_agent, notifications, open_3qfx, open_chat, parallel_agent_tasks, parallel_agents, position_roles, position_roundtable, positions, public_config, public_embed, public_leaderboard, public_scheduled_tasks, recommend, recommended_questions, roundtable_chains, roundtable_diagnostics, roundtable_draft_shares, roundtable_drafts, roundtable_jobs, roundtable_task_drafts, runs, scheduled_tasks, sensitive_words, skill_addresses, skills, suggestions, system_settings, tags, task_buttons, task_reports, taskcop_tasks, thread_runs, thread_shares, threads, tool_metrics, uploads, user_preferences, user_prompts, weknora_sources, writing, ) from deerflow.config import app_config as deerflow_app_config from deerflow.config.app_config import apply_logging_level from deerflow.runtime.scheduler import ScheduledTaskService, set_scheduled_task_service AppConfig = deerflow_app_config.AppConfig get_app_config = deerflow_app_config.get_app_config # Default logging; lifespan overrides from config.yaml log_level. logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger(__name__) # Upper bound (seconds) each lifespan shutdown hook is allowed to run. # Bounds worker exit time so uvicorn's reload supervisor does not keep # firing signals into a worker that is stuck waiting for shutdown cleanup. _SHUTDOWN_HOOK_TIMEOUT_SECONDS = 5.0 async def _ensure_admin_user(app: FastAPI) -> None: """Startup hook: handle first boot and migrate orphan threads otherwise. After admin creation, migrate orphan threads from the LangGraph store (metadata.user_id unset) to the admin account. This is the "no-auth ? with-auth" upgrade path: users who ran DeerFlow without authentication have existing LangGraph thread data that needs an owner assigned. First boot (no admin exists): - Does NOT create any user accounts automatically. - The operator must visit ``/setup`` to create the first admin. Subsequent boots (admin already exists): - Runs the one-time "no-auth ? with-auth" orphan thread migration for existing LangGraph thread metadata that has no owner_id. No SQL persistence migration is needed: the four user_id columns (threads_meta, runs, run_events, feedback) only come into existence alongside the auth module via create_all, so freshly created tables never contain NULL-owner rows. """ from sqlalchemy import select from app.gateway.deps import get_local_provider from deerflow.persistence.engine import get_session_factory from deerflow.persistence.user.model import UserRow try: provider = get_local_provider() except RuntimeError: # Auth persistence may not be initialized in some test/boot paths. # Skip admin migration work rather than failing gateway startup. logger.warning("Auth persistence not ready; skipping admin bootstrap check") return sf = get_session_factory() if sf is None: return admin_count = await provider.count_admin_users() if admin_count == 0: logger.info("=" * 60) logger.info(" First boot detected ? no admin account exists.") logger.info(" Visit /setup to complete admin account creation.") logger.info("=" * 60) return # Admin already exists ? run orphan thread migration for any # LangGraph thread metadata that pre-dates the auth module. async with sf() as session: stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: return # Should not happen (admin_count > 0 above), but be safe. admin_id = str(row.id) # LangGraph store orphan migration ? non-fatal. # This covers the "no-auth ? with-auth" upgrade path for users # whose existing LangGraph thread metadata has no user_id set. store = getattr(app.state, "store", None) if store is not None: try: migrated = await _migrate_orphaned_threads(store, admin_id) if migrated: logger.info("Migrated %d orphan LangGraph thread(s) to admin", migrated) except Exception: logger.exception("LangGraph thread migration failed (non-fatal)") async def _iter_store_items(store, namespace, *, page_size: int = 500): """Paginated async iterator over a LangGraph store namespace. Replaces the old hardcoded ``limit=1000`` call with a cursor-style loop so that environments with more than one page of orphans do not silently lose data. Terminates when a page is empty OR when a short page arrives (indicating the last page). """ offset = 0 while True: batch = await store.asearch(namespace, limit=page_size, offset=offset) if not batch: return for item in batch: yield item if len(batch) < page_size: return offset += page_size async def _sync_legacy_skills(app: FastAPI) -> None: """Idempotently register existing on-disk ``skills/custom/*`` directories in the ``skills`` DB table, owned by the first admin and published=True. Without this step, after upgrading to per-user skill ownership the existing custom skill folders would not be visible to anyone (the SkillStore visibility filter relies on a DB row). Running this once on every boot is cheap (folder scan + one ``ensure_legacy`` per name) and idempotent. Non-fatal: a failure here MUST NOT block server startup. """ from sqlalchemy import select from deerflow.persistence.engine import get_session_factory from deerflow.persistence.user.model import UserRow from deerflow.skills.storage import get_or_new_skill_storage from deerflow.skills.types import SkillCategory sf = get_session_factory() if sf is None: return skill_store = getattr(app.state, "skill_store", None) if skill_store is None: return async with sf() as session: stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: # No admin yet ? leave ownership null; first admin is responsible for # claiming via the UI once they exist. admin_id: str | None = None else: admin_id = str(row.id) config = getattr(app.state, "config", None) storage = get_or_new_skill_storage(app_config=config) try: skills = storage.load_skills(enabled_only=False) except Exception: logger.exception("_sync_legacy_skills: could not list on-disk skills") return inserted = 0 for skill in skills: if skill.category != SkillCategory.CUSTOM: continue try: existing = await skill_store.get_any(skill.name) if existing is not None: continue await skill_store.ensure_legacy( {"name": skill.name, "owner_user_id": admin_id, "published": True} ) inserted += 1 except Exception: logger.warning("_sync_legacy_skills: could not register %s", skill.name, exc_info=True) if inserted: logger.info("Skill ownership migration: registered %d legacy custom skill(s)", inserted) async def _migrate_orphaned_threads(store, admin_user_id: str) -> int: """Migrate LangGraph store threads with no user_id to the given admin. Uses cursor pagination so all orphans are migrated regardless of count. Returns the number of rows migrated. """ migrated = 0 async for item in _iter_store_items(store, ("threads",)): metadata = item.value.get("metadata", {}) if not metadata.get("user_id"): metadata["user_id"] = admin_user_id item.value["metadata"] = metadata await store.aput(("threads",), item.key, item.value) migrated += 1 return migrated async def _run_db_migrations() -> None: """Run Alembic migrations at startup so schema changes are applied automatically.""" from pathlib import Path from alembic import command from alembic.config import Config import deerflow.persistence as _persistence_pkg from deerflow.persistence.engine import get_engine engine = get_engine() if engine is None: logger.info("Persistence backend=memory ? skipping Alembic migrations") return if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}: logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations") return db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%") migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations") result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1) def _run_sync() -> None: try: cfg = Config() cfg.set_main_option("script_location", migrations_dir) cfg.set_main_option("sqlalchemy.url", db_url) command.upgrade(cfg, "head") except BaseException as exc: result_queue.put(exc) else: result_queue.put(None) try: try: timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60")) except ValueError: timeout_s = 60.0 worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True) worker.start() try: deadline = asyncio.get_running_loop().time() + timeout_s while result_queue.empty(): if asyncio.get_running_loop().time() >= deadline: raise TimeoutError await asyncio.sleep(0.2) result = result_queue.get_nowait() except TimeoutError: logger.error( "Database migration did not finish within %.0fs; continuing startup on create_all schema. " "Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 " "to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.", timeout_s, ) return if result is not None: raise result logger.info("Database migrations applied (head)") except Exception: logger.exception("Database migration failed ? schema may be out of date") # Skill curator ????? # - ????:?????????"??????";??????? should_run_now() ??? # - ????:?????????????,??????? _CURATOR_CHECK_INTERVAL_SECONDS = 3600 # 1 ?? _CURATOR_STARTUP_DELAY_SECONDS = 300 # 5 ?? def _acquire_singleton_lock(lock_path): """???????????,??? worker ?? leader ??? ?? -> ??????????(?????????,??????/????); ????(???? worker ??)????? -> ?? None? ??????? OS ???????? """ try: lock_path.parent.mkdir(parents=True, exist_ok=True) handle = open(lock_path, "a+") except OSError: logger.exception("Curator scheduler: ???????") return None try: if sys.platform == "win32": import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: handle.close() return None return handle def _release_singleton_lock(handle) -> None: """????? _acquire_singleton_lock ?????????""" try: if sys.platform == "win32": import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) else: import fcntl fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except OSError: pass finally: try: handle.close() except OSError: pass async def _curator_scheduler_loop() -> None: """??????:?? _CURATOR_CHECK_INTERVAL_SECONDS ??????????? ? worker ???????? leader ?? ?? ???????? worker ???? ????,?? worker ????,?? curator ?????????? LLM? ? cancel ? CancelledError ??? ``except Exception``(?? BaseException ??,?????),?????????? finally ????? """ from deerflow.config.runtime_paths import runtime_home from deerflow.skills.curator import maybe_run_curator lock_handle = _acquire_singleton_lock(runtime_home() / ".curator_scheduler.lock") if lock_handle is None: logger.info("Curator scheduler: ??? worker ??????,? worker ??") return logger.info("Curator scheduler: ?????,? worker ?? curator ??") try: # ?????????????,?????? await asyncio.sleep(_CURATOR_STARTUP_DELAY_SECONDS) while True: try: await maybe_run_curator() except Exception: logger.exception("Curator scheduler tick failed (non-fatal)") await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS) finally: _release_singleton_lock(lock_handle) def _env_bool(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() not in {"0", "false", "no", "off"} def _env_int(name: str, default: int, *, alias: str | None = None, minimum: int = 1) -> int: raw = os.getenv(name) if raw is None and alias: raw = os.getenv(alias) if raw is None: raw = str(default) try: value = int(raw) except ValueError: return default return max(minimum, value) def _env_float(name: str, default: float, *, alias: str | None = None, minimum: float = 0.0) -> float: raw = os.getenv(name) if raw is None and alias: raw = os.getenv(alias) if raw is None: raw = str(default) try: value = float(raw) except ValueError: return default return max(minimum, value) _LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED = _env_bool("ADMIN_LEADERBOARD_BACKGROUND_ENABLED", True) _LEADERBOARD_SNAPSHOT_PREWARM_DAYS = _env_int( "ADMIN_LEADERBOARD_BACKGROUND_PREWARM_DAYS", 30, ) _LEADERBOARD_SNAPSHOT_BATCH_SIZE = _env_int( "ADMIN_LEADERBOARD_BACKGROUND_BATCH_SIZE", 1, ) _LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS = _env_float( "ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS", 300.0, ) _LEADERBOARD_SNAPSHOT_SLEEP_SECONDS = _env_float( "ADMIN_LEADERBOARD_BACKGROUND_SLEEP_SECONDS", 2.0, ) _LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS = _env_int( "ADMIN_LEADERBOARD_BACKGROUND_RUNNING_TTL_SECONDS", 1800, ) _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float( "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS", 60.0, ) # Concurrency (system-pressure) sampler: how often to snapshot the in-flight run # count, and how long to keep samples (drives the admin 并发量 line chart, which # aggregates per minute/hour). Sampling is in-memory cheap (one tiny INSERT). _CONCURRENCY_SAMPLE_ENABLED = _env_bool("CONCURRENCY_SAMPLE_ENABLED", True) _CONCURRENCY_SAMPLE_INTERVAL_SECONDS = _env_float( "CONCURRENCY_SAMPLE_INTERVAL_SECONDS", 15.0, ) _CONCURRENCY_SAMPLE_RETENTION_HOURS = _env_float( "CONCURRENCY_SAMPLE_RETENTION_HOURS", 168.0, # 7 days ) # Hour (local Beijing time, 0-23) after which the scheduler force-recomputes the # *previous* day's snapshot once, so records that landed near midnight are fully # captured. Default 1 ? "???? 1 ????????". _LEADERBOARD_FINALIZE_HOUR = min( 23, _env_int("ADMIN_LEADERBOARD_FINALIZE_HOUR", 1, minimum=0), ) def _leaderboard_scheduler_sql_for_log(stmt) -> str: try: return str(stmt.compile(compile_kwargs={"literal_binds": True})) except Exception: return str(stmt) async def _execute_leaderboard_scheduler_query(session, step: str, stmt, **context): sql = _leaderboard_scheduler_sql_for_log(stmt) logger.info( "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s", step, context.get("stat_date") or context.get("stat_dates") or "", context.get("job_key") or "", context.get("settings_hash") or "", datetime.now().isoformat(), sql, ) start = perf_counter() try: result = await session.execute(stmt) except Exception: logger.exception( "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f", step, context.get("stat_date") or context.get("stat_dates") or "", context.get("job_key") or "", context.get("settings_hash") or "", (perf_counter() - start) * 1000, ) raise logger.info( "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f", step, context.get("stat_date") or context.get("stat_dates") or "", context.get("job_key") or "", context.get("settings_hash") or "", datetime.now().isoformat(), (perf_counter() - start) * 1000, ) return result async def _time_leaderboard_scheduler_db_operation(step: str, operation, *, sql: str = "", **context): logger.info( "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s", step, context.get("stat_date") or context.get("stat_dates") or "", context.get("job_key") or "", context.get("settings_hash") or "", datetime.now().isoformat(), sql, ) start = perf_counter() try: result = await operation() except Exception: logger.exception( "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f", step, context.get("stat_date") or context.get("stat_dates") or "", context.get("job_key") or "", context.get("settings_hash") or "", (perf_counter() - start) * 1000, ) raise logger.info( "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f", step, context.get("stat_date") or context.get("stat_dates") or "", context.get("job_key") or "", context.get("settings_hash") or "", datetime.now().isoformat(), (perf_counter() - start) * 1000, ) return result def _recent_leaderboard_stat_dates(days: int) -> list[str]: from datetime import datetime, timedelta from deerflow.persistence.types import BEIJING_TZ today = datetime.now(BEIJING_TZ).date() start = today - timedelta(days=max(1, days) - 1) return [(start + timedelta(days=i)).isoformat() for i in range((today - start).days + 1)] async def _leaderboard_snapshot_due_dates( session_factory, *, settings_hash: str, limit: int, running_ttl_seconds: int, ) -> list[str]: from datetime import UTC, datetime, timedelta from sqlalchemy import and_, or_, select from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow cutoff = datetime.now(UTC) - timedelta(seconds=running_ttl_seconds) async with session_factory() as session: stmt = ( select(AdminLeaderboardDailyStatRow.stat_date) .where( AdminLeaderboardDailyStatRow.settings_hash == settings_hash, or_( AdminLeaderboardDailyStatRow.status.in_(["queued", "error"]), and_( AdminLeaderboardDailyStatRow.status == "running", AdminLeaderboardDailyStatRow.updated_at < cutoff, ), ), ) .order_by(AdminLeaderboardDailyStatRow.stat_date.desc()) .limit(limit) ) rows = ( await _execute_leaderboard_scheduler_query( session, "scheduler.due_dates", stmt, settings_hash=settings_hash, ) ).all() return [str(row[0]) for row in rows] async def _claim_leaderboard_snapshot( session_factory, *, stat_date: str, settings_hash: str, running_ttl_seconds: int, ) -> bool: from datetime import UTC, datetime from app.gateway.routers.admin_users import _snapshot_is_fresh from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow from deerflow.persistence.types import BEIJING_TZ async with session_factory() as session: row = await _time_leaderboard_scheduler_db_operation( "scheduler.claim.get", lambda: session.get( AdminLeaderboardDailyStatRow, {"stat_date": stat_date, "settings_hash": settings_hash}, ), sql="SELECT admin_leaderboard_daily_stats by primary key", stat_date=stat_date, settings_hash=settings_hash, ) if row is None or _snapshot_is_fresh(row, stat_date): return False if row.status == "running": updated_at = row.updated_at if isinstance(updated_at, datetime): if updated_at.tzinfo is None: updated_at = updated_at.replace(tzinfo=BEIJING_TZ) age = (datetime.now(BEIJING_TZ) - updated_at.astimezone(BEIJING_TZ)).total_seconds() if age < running_ttl_seconds: return False row.status = "running" row.message = "?????????????" row.error = None row.updated_at = datetime.now(UTC) await _time_leaderboard_scheduler_db_operation( "scheduler.claim.commit", lambda: session.commit(), sql="COMMIT", stat_date=stat_date, settings_hash=settings_hash, ) return True async def _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash: str) -> bool: """Once per day after ``_LEADERBOARD_FINALIZE_HOUR``, force-recompute yesterday. A ``ready`` past-day snapshot is otherwise treated as fresh forever and never rebuilt. But the previous day's snapshot is usually generated before midnight, so it can miss runs/tool-calls that only land in the DB around the day boundary. This re-queues yesterday (``force=True``) a single time; the rebuild stamps ``generated_at`` to today, which stops it from re-firing for the rest of the day. """ from datetime import timedelta from app.gateway.routers.admin_users import ( _as_beijing, _queue_daily_leaderboard_snapshots, ) from deerflow.persistence.admin_stats.finalize import should_finalize_previous_day from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow from deerflow.persistence.types import BEIJING_TZ now = datetime.now(BEIJING_TZ) yesterday = (now.date() - timedelta(days=1)).isoformat() async with session_factory() as session: row = await session.get( AdminLeaderboardDailyStatRow, {"stat_date": yesterday, "settings_hash": settings_hash}, ) generated_at = _as_beijing(row.generated_at) if (row is not None and row.generated_at is not None) else None if not should_finalize_previous_day( status=(row.status if row is not None else None), generated_at=generated_at, now=now, finalize_hour=_LEADERBOARD_FINALIZE_HOUR, ): return False queued = await _queue_daily_leaderboard_snapshots( session_factory, [yesterday], settings_hash, force=True, message="???????????????????????????", ) if queued: logger.info( "Leaderboard snapshot finalize: queued previous day %s for full recompute (hash=%s)", yesterday, settings_hash[:10], ) return True return False async def _run_leaderboard_snapshot_tick() -> tuple[int, int]: from app.gateway.routers.admin_users import ( _daily_job_key, _queue_daily_leaderboard_snapshots, _refresh_daily_leaderboard_snapshot, _resolve_leaderboard_snapshot_scope, ) from deerflow.config.system_settings import load_system_settings from deerflow.persistence.engine import get_session_factory session_factory = get_session_factory() if session_factory is None: return 0, 0 settings = load_system_settings().leaderboard excluded, include_scheduled, include_admins, include_failed, settings_hash = _resolve_leaderboard_snapshot_scope(settings) stat_dates = _recent_leaderboard_stat_dates(_LEADERBOARD_SNAPSHOT_PREWARM_DAYS) queued_dates = await _queue_daily_leaderboard_snapshots( session_factory, stat_dates, settings_hash, message="?????????????/?????", ) # ??????? 1 ???????"???"???????????????????? try: await _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash) except Exception: logger.exception("Leaderboard snapshot finalize check failed (non-fatal)") due_dates = await _leaderboard_snapshot_due_dates( session_factory, settings_hash=settings_hash, limit=_LEADERBOARD_SNAPSHOT_BATCH_SIZE, running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS, ) processed = 0 for stat_date in due_dates: claimed = await _claim_leaderboard_snapshot( session_factory, stat_date=stat_date, settings_hash=settings_hash, running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS, ) if not claimed: continue logger.info("Leaderboard snapshot scheduler: building date=%s hash=%s", stat_date, settings_hash[:10]) await _refresh_daily_leaderboard_snapshot( _daily_job_key(stat_date, settings_hash), session_factory=session_factory, settings=settings, stat_date=stat_date, settings_hash=settings_hash, excluded=excluded, include_scheduled=include_scheduled, include_admins=include_admins, include_failed=include_failed, ) processed += 1 if _LEADERBOARD_SNAPSHOT_SLEEP_SECONDS > 0 and processed < len(due_dates): await asyncio.sleep(_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS) return processed, len(queued_dates) async def _leaderboard_snapshot_scheduler_loop() -> None: """Background task inside the existing gateway process for daily snapshots.""" if not _LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED: logger.info("Leaderboard snapshot scheduler disabled by ADMIN_LEADERBOARD_BACKGROUND_ENABLED") return from deerflow.config.runtime_paths import runtime_home lock_handle = _acquire_singleton_lock(runtime_home() / ".leaderboard_snapshot_scheduler.lock") if lock_handle is None: logger.info("Leaderboard snapshot scheduler: another gateway worker holds the lock; this worker skips") return logger.info( "Leaderboard snapshot scheduler started: prewarm_days=%d batch_size=%d interval=%.1fs", _LEADERBOARD_SNAPSHOT_PREWARM_DAYS, _LEADERBOARD_SNAPSHOT_BATCH_SIZE, _LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS, ) try: if _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS > 0: await asyncio.sleep(_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS) while True: try: processed, queued = await _run_leaderboard_snapshot_tick() if processed or queued: logger.info( "Leaderboard snapshot scheduler tick: queued=%d processed=%d", queued, processed, ) except Exception: logger.exception("Leaderboard snapshot scheduler tick failed (non-fatal)") await asyncio.sleep(_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS) finally: _release_singleton_lock(lock_handle) async def _concurrency_sampler_loop(app: FastAPI) -> None: """Periodically record the in-flight run count for the admin pressure chart. Runs in the gateway process, singleton-locked so only one worker samples. Each tick records ``len(run_manager.list_active())`` and, roughly hourly, purges samples older than the retention window. Best-effort throughout — a failed tick is logged and the loop continues. """ if not _CONCURRENCY_SAMPLE_ENABLED: logger.info("Concurrency sampler disabled by CONCURRENCY_SAMPLE_ENABLED") return store = getattr(app.state, "concurrency_sample_store", None) run_mgr = getattr(app.state, "run_manager", None) concurrency_gate = getattr(app.state, "concurrency_gate", None) if store is None or run_mgr is None: logger.info("Concurrency sampler not started (no store / run manager — memory backend?)") return from deerflow.config.runtime_paths import runtime_home lock_handle = _acquire_singleton_lock(runtime_home() / ".concurrency_sampler.lock") if lock_handle is None: logger.info("Concurrency sampler: another gateway worker holds the lock; this worker skips") return interval = max(1.0, _CONCURRENCY_SAMPLE_INTERVAL_SECONDS) purge_every_ticks = max(1, int(3600 / interval)) # ~once per hour logger.info("Concurrency sampler started: interval=%.1fs retention=%.1fh", interval, _CONCURRENCY_SAMPLE_RETENTION_HOURS) ticks = 0 try: while True: try: # Runtime toggle (admin can turn sampling off without a restart). # mtime-cached read, so this is cheap every tick. from deerflow.config.system_settings import get_concurrency_monitor_settings if get_concurrency_monitor_settings().enabled: active_count = None if concurrency_gate is not None: active_count = await concurrency_gate.count_active() if active_count is None: active = await run_mgr.list_active() active_count = len(active) await store.record(active_count) ticks += 1 if ticks % purge_every_ticks == 0: from datetime import UTC, datetime, timedelta cutoff = datetime.now(UTC) - timedelta(hours=_CONCURRENCY_SAMPLE_RETENTION_HOURS) await store.purge_older_than(cutoff) except Exception: logger.exception("Concurrency sampler tick failed (non-fatal)") await asyncio.sleep(interval) finally: _release_singleton_lock(lock_handle) def _warn_if_multi_worker() -> None: """?? uvicorn / gunicorn ? worker ???? ERROR ??? ? agent / canvas / AI ??????? Gateway ????? ``StreamBridge`` ?? run_id ? SSE ?????? map?? worker ????? worker A ?? ``/runs/stream``???? worker B ? join/resume??? in-process map ?? ???? 404 / ??? ????? startup ? ERROR ????**???? fail-fast** ?? ?? ???????????????? sticky session?????????? """ # uvicorn / gunicorn ??????? worker ????????????? # ?????? sibling ??????????? WEB_CONCURRENCY?gunicorn # / uvicorn ???? web_concurrency = os.environ.get("WEB_CONCURRENCY", "").strip() if web_concurrency and web_concurrency.isdigit() and int(web_concurrency) > 1: logger.error( "? ??? WEB_CONCURRENCY=%s?? worker ????AI ?? / ? agent " "??????????????? worker ????????? / " "SSE ?? / resume 400??????????? WEB_CONCURRENCY=1 " "????????? sticky session??? AIWritingPage.tsx ????", web_concurrency, ) return # ???? /proc ??? uvicorn ?????? Linux ?????? if not sys.platform.startswith("linux"): return try: from pathlib import Path peers = 0 for entry in Path("/proc").iterdir(): if not entry.name.isdigit(): continue try: cmdline = (entry / "cmdline").read_bytes().replace(b"\x00", b" ").decode("utf-8", "ignore") except OSError: continue if "uvicorn" in cmdline and "app.gateway.app" in cmdline: peers += 1 if peers > 1: logger.error( "? ??? %d ??? uvicorn ???? worker ????? AI ?? " "????? / SSE ??????????? worker ??????? " "?? sticky session?", peers, ) except Exception: # ????????? ?? ?????? pass async def _scan_maf_approval_expiry(app: Any) -> list[str]: """Expire due MAF approvals once, then resume affected workflows. The store owns the conditional state transition and critical audit event. Only the worker that wins that CAS receives an affected run, so repeated scans cannot resume the same expiry twice. """ service = getattr(app.state, "maf_roundtable_service", None) if service is None or not getattr(service, "enabled", False): return [] expired = await service.store.expire_due_approvals() run_ids = list(dict.fromkeys( str(item["run_id"]) for item in expired if item.get("run_id") )) if not run_ids: return [] from app.maf_roundtable.dispatcher import get_dispatcher dispatcher = get_dispatcher(app) resumed: list[str] = [] for run_id in run_ids: run = await service.store.runs.get_run(run_id) if run is None: continue session = await service.store.sessions.get_session(run["session_id"], user_id=None) if session is None: continue # Dispatcher.resume is a no-op for a terminal/non-waiting run; keeping # this call centralized preserves the restart and run-level CAS path. await dispatcher.resume(run_id, user_id=str(session.get("user_id") or "default")) resumed.append(run_id) return resumed async def _maf_approval_scanner_loop(app: Any) -> None: """Continuously fail closed expired MAF approvals until shutdown.""" service = getattr(app.state, "maf_roundtable_service", None) config = getattr(service, "config", None) interval = max(5, int(getattr(config, "approval_scan_interval_seconds", 30) or 30)) while True: try: await _scan_maf_approval_expiry(app) except asyncio.CancelledError: raise except Exception: logger.exception("MAF approval expiry scan failed (non-fatal)") await asyncio.sleep(interval) def _start_maf_approval_scanner(app: Any) -> asyncio.Task[None] | None: """Start the MAF-only approval scanner when the feature is enabled.""" service = getattr(app.state, "maf_roundtable_service", None) if service is None or not getattr(service, "enabled", False): return None task = asyncio.create_task(_maf_approval_scanner_loop(app)) app.state.maf_approval_scanner_task = task logger.info( "MAF approval expiry scanner started (interval=%ss)", getattr(service.config, "approval_scan_interval_seconds", 30), ) return task async def _stop_maf_approval_scanner(app: Any) -> None: """Cancel and await the MAF approval scanner during Gateway shutdown.""" task = getattr(app.state, "maf_approval_scanner_task", None) if task is None: return task.cancel() try: await task except asyncio.CancelledError: pass except Exception: logger.exception("MAF approval expiry scanner shutdown error") @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """Application lifespan handler.""" # ?????? worker ?? ?? ?? AI ?? / ? agent ??????? # ??????? worker ???????? ERROR ????? _warn_if_multi_worker() # Load config and check necessary environment variables at startup try: app.state.config = get_app_config() apply_logging_level(app.state.config.log_level) logger.info("Configuration loaded successfully") except Exception as e: error_msg = f"Failed to load configuration during gateway startup: {e}" logger.exception(error_msg) raise RuntimeError(error_msg) from e config = get_gateway_config() logger.info(f"Starting API Gateway on {config.host}:{config.port}") # Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store) async with langgraph_runtime(app): logger.info("LangGraph runtime initialised") # Run Alembic migrations before anything else touches the schema. await _run_db_migrations() # ?? recall_budget cache?? DB ????,? get_effective_memory_config ????? try: from deerflow.config.memory_config import load_recall_budget_cache _mc_store = getattr(app.state, "memory_config_store", None) if _mc_store is not None: _recall_budgets = await _mc_store.load_all() load_recall_budget_cache(_recall_budgets) logger.info("Memory config cache warmed: %d user override(s)", len(_recall_budgets)) except Exception: logger.exception("Memory config cache warm-up failed (non-fatal)") # Ensure admin user exists (auto-create on first boot) # Must run AFTER langgraph_runtime so app.state.store is available for thread migration await _ensure_admin_user(app) # Migrate "ownerless" on-disk skills into the new SkillStore table ? # makes the multitenancy switch transparent for existing deployments. try: await _sync_legacy_skills(app) except Exception: logger.exception("Skill ownership migration failed (non-fatal)") # ????? agent(intent / recommender / coordinator)??????? # ??? _sync_legacy_agents **??**?????????????????? # ??? upsert ? agents ?,?????? /api/agents ?????, # ????????????????.deer-flow/ ? gitignore,??? clone # ????????,seeder ????? _roundtable_seed_assets/ ?????? try: from app.gateway.routers._roundtable_seed import ( ensure_roundtable_functional_agents, ) ensure_roundtable_functional_agents() except Exception: logger.exception( "Roundtable functional agent seeding failed at startup (non-fatal)" ) # AI ????? agent(researcher / outliner / writer / editor)??????? # ???:??? _sync_legacy_agents ??,????????????? # upsert ? agents ?,admin ?????????????????? agent? try: from app.gateway.routers._ai_writing_seed import ( ensure_ai_writing_functional_agents, ) ensure_ai_writing_functional_agents() except Exception: logger.exception( "AI-writing functional agent seeding failed at startup (non-fatal)" ) # 内置「模板数据转换器」agent(JSON → HTML 模板填充)。同样必须在 # _sync_legacy_agents **之前**补建目录,后者把它 upsert 进 agents 表, # 第一次启动 /api/agents 与定时任务下拉框里就有它。 try: from app.gateway.routers._template_builder_seed import ( ensure_template_builder_agent, ) ensure_template_builder_agent() except Exception: logger.exception( "Template-builder agent seeding failed at startup (non-fatal)" ) # ?????????????????3Q/6BF/7BF/8BF??agent/chain ????? try: business_mapping_store = getattr(app.state, "business_mapping_store", None) if business_mapping_store is not None: inserted = await business_mapping_store.ensure_seed_defaults() if inserted: logger.info("Seeded %d business mapping rows", inserted) except Exception: logger.exception("Business mapping seeding failed at startup (non-fatal)") # ???? .deer-flow/agents/{id}/ ???????? agent ?? agents ?? # ??????????????? sync(????? + N ? agent ??? DB # SELECT/UPDATE),??????? + ?????? TTL ??,???? # /api/agents ????? try: from app.gateway.routers.agents import ( _mark_legacy_sync_fresh, _sync_legacy_agents, ) from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh agent_store = getattr(app.state, "agent_store", None) if agent_store is not None: await _sync_legacy_agents(agent_store) _mark_legacy_sync_fresh() # ``services.py`` keeps an independent TTL gate (used by # ``start_run`` ? ``_enforce_agent_access``). Without this # ack, the first ``/runs/stream`` after startup would # re-run the heavy sync even though we just did it. _services_mark_sync_fresh() logger.info("Legacy folder-backed agents synced at startup") # 内置功能型 agent(roundtable-* / ai-writing-*)按用途默认打标签, # 供管理员「内置智能体」筛选按用途归类。幂等、只追加、失败不阻塞启动。 try: from app.gateway.routers._builtin_agent_tag_seed import ( ensure_builtin_agent_purpose_tags, ) tag_store = getattr(app.state, "tag_store", None) if tag_store is not None: await ensure_builtin_agent_purpose_tags(agent_store, tag_store) except Exception: logger.exception("Built-in agent purpose-tag seeding failed at startup (non-fatal)") except Exception: logger.exception("Legacy agent sync failed at startup (non-fatal)") # ?? skill curator ???:??????,????????????? # (per-user interval_hours ??????????????)? try: app.state.curator_scheduler_task = asyncio.create_task(_curator_scheduler_loop()) except Exception: logger.exception("Skill curator scheduler failed to start (non-fatal)") # ?? AI ????????:? config.yaml ? ai_writing.cleanup ???, # ?????? 3 ????,? retention_days(?? 7 ?)?????,?? # ?? MySQL ??? + SQLite checkpointer ?? thread ???? worker # ?????? leader,?????? worker ?????? try: from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app)) except Exception: logger.exception("AI writing cleanup scheduler failed to start (non-fatal)") try: app.state.leaderboard_snapshot_scheduler_task = asyncio.create_task(_leaderboard_snapshot_scheduler_loop()) except Exception: logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)") # Concurrency (system-pressure) sampler — lightweight, singleton-locked, # runtime-toggleable via system_settings.concurrency_monitor.enabled. try: app.state.concurrency_sampler_task = asyncio.create_task(_concurrency_sampler_loop(app)) except Exception: logger.exception("Concurrency sampler failed to start (non-fatal)") # MAF-only approval expiry scanner. Legacy DeerFlow, old roundtable # and AI-writing paths never enter this loop. try: _start_maf_approval_scanner(app) except Exception: logger.exception("MAF approval expiry scanner failed to start (non-fatal)") # Start IM channel service if any channels are configured try: from app.channels.service import start_channel_service channel_service = await start_channel_service(app.state.config) logger.info("Channel service started: %s", channel_service.get_status()) except Exception: logger.exception("No IM channels configured or channel service failed to start") scheduled_tasks_enabled = os.environ.get("SCHEDULED_TASKS_ENABLED", "true").strip().lower() if scheduled_tasks_enabled in {"0", "false", "no", "off"}: logger.info("Scheduled task service disabled by SCHEDULED_TASKS_ENABLED=%s", scheduled_tasks_enabled) else: try: scheduled_task_store = getattr(app.state, "scheduled_task_store", None) if scheduled_task_store is not None: scheduler_service = ScheduledTaskService(app, scheduled_task_store) app.state.scheduled_task_service = scheduler_service set_scheduled_task_service(scheduler_service) await scheduler_service.start() logger.info("Scheduled task service started") except Exception: logger.exception("Scheduled task service failed to start") yield # ?? curator ??? curator_task = getattr(app.state, "curator_scheduler_task", None) if curator_task is not None: curator_task.cancel() try: await curator_task except asyncio.CancelledError: pass except Exception: logger.exception("Curator scheduler shutdown error") # ?? AI ??????? cleanup_task = getattr(app.state, "ai_writing_cleanup_task", None) if cleanup_task is not None: cleanup_task.cancel() try: await cleanup_task except asyncio.CancelledError: pass except Exception: logger.exception("AI writing cleanup scheduler shutdown error") leaderboard_snapshot_task = getattr(app.state, "leaderboard_snapshot_scheduler_task", None) if leaderboard_snapshot_task is not None: leaderboard_snapshot_task.cancel() try: await leaderboard_snapshot_task except asyncio.CancelledError: pass except Exception: logger.exception("Leaderboard snapshot scheduler shutdown error") concurrency_sampler_task = getattr(app.state, "concurrency_sampler_task", None) if concurrency_sampler_task is not None: concurrency_sampler_task.cancel() try: await concurrency_sampler_task except asyncio.CancelledError: pass except Exception: logger.exception("Concurrency sampler shutdown error") await _stop_maf_approval_scanner(app) # ?? 7 ????AI ?? graph ???? checkpointer / in-flight task? # thread ??? LangGraph Server ? checkpointer ???lifespan ??? # ``langgraph_runtime()`` ? contextmanager ??????????????? try: scheduler_service = getattr(app.state, "scheduled_task_service", None) if scheduler_service is not None: await scheduler_service.stop() except Exception: logger.exception("Failed to stop scheduled task service") finally: set_scheduled_task_service(None) # Stop channel service on shutdown (bounded to prevent worker hang) try: from app.channels.service import stop_channel_service await asyncio.wait_for( stop_channel_service(), timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS, ) except TimeoutError: logger.warning( "Channel service shutdown exceeded %.1fs; proceeding with worker exit.", _SHUTDOWN_HOOK_TIMEOUT_SECONDS, ) except Exception: logger.exception("Failed to stop channel service") logger.info("Shutting down API Gateway") def create_app() -> FastAPI: """Create and configure the FastAPI application. Returns: Configured FastAPI application instance. """ config = get_gateway_config() docs_kwargs = {"docs_url": "/docs", "redoc_url": "/redoc", "openapi_url": "/openapi.json"} if config.enable_docs else {"docs_url": None, "redoc_url": None, "openapi_url": None} app = FastAPI( title="DeerFlow API Gateway", description=""" ## DeerFlow API Gateway API Gateway for DeerFlow - A LangGraph-based AI agent backend with sandbox execution capabilities. ### Features - **Models Management**: Query and retrieve available AI models - **MCP Configuration**: Manage Model Context Protocol (MCP) server configurations - **Memory Management**: Access and manage global memory data for personalized conversations - **Skills Management**: Query and manage skills and their enabled status - **Artifacts**: Access thread artifacts and generated files - **Health Monitoring**: System health check endpoints ### Architecture LangGraph requests are handled by nginx reverse proxy. This gateway provides custom endpoints for models, MCP configuration, skills, and artifacts. """, version="0.1.0", lifespan=lifespan, **docs_kwargs, openapi_tags=[ { "name": "models", "description": "Operations for querying available AI models and their configurations", }, { "name": "mcp", "description": "Manage Model Context Protocol (MCP) server configurations", }, { "name": "memory", "description": "Access and manage global memory data for personalized conversations", }, { "name": "skills", "description": "Manage skills and their configurations", }, { "name": "artifacts", "description": "Access and download thread artifacts and generated files", }, { "name": "uploads", "description": "Upload and manage user files for threads", }, { "name": "threads", "description": "Manage DeerFlow thread-local filesystem data", }, { "name": "agents", "description": "Create and manage custom agents with per-agent config and prompts", }, { "name": "suggestions", "description": "Generate follow-up question suggestions for conversations", }, { "name": "channels", "description": "Manage IM channel integrations (Feishu, Slack, Telegram)", }, { "name": "assistants-compat", "description": "LangGraph Platform-compatible assistants API (stub)", }, { "name": "runs", "description": "LangGraph Platform-compatible runs lifecycle (create, stream, cancel)", }, { "name": "scheduled-tasks", "description": "Create, manage, and execute user scheduled agent tasks", }, { "name": "health", "description": "Health check and system status endpoints", }, ], ) # Auth: reject unauthenticated requests to non-public paths (fail-closed safety net) app.add_middleware(AuthMiddleware) # CSRF: Double Submit Cookie pattern for state-changing requests app.add_middleware(CSRFMiddleware) # CORS if is_auth_disabled(): logger.warning( "DEER_FLOW_AUTH_DISABLED is set ? API authentication is bypassed. " "Do not use this in production." ) def _cors_allow_any_origin() -> bool: """Mirror any browser Origin (works with credentials); cannot use literal '*'.""" v = os.environ.get("GATEWAY_CORS_ALLOW_ALL", "").strip().lower() return v in ("1", "true", "yes", "on") or is_auth_disabled() # The legacy TaskCOP Vue application uses Vite's default development port. # Keep these origins available even when deployments configure additional # origins through GATEWAY_CORS_ORIGINS. taskcop_vite_origins = [ "http://localhost:8080", "http://127.0.0.1:8080", ] cors_origins_env = os.environ.get("GATEWAY_CORS_ORIGINS", "").strip() cors_origins: list[str] = [o.strip() for o in cors_origins_env.split(",") if o.strip()] if cors_origins_env else [] if cors_origins: cors_origins = list(dict.fromkeys([*cors_origins, *taskcop_vite_origins])) # Headers the LangGraph SDK / browser clients need to read from cross-origin # responses. Without exposing them, the SDK's `response.headers.get(...)` returns # null on cross-origin streaming responses, breaking run-id extraction and reconnect. cors_expose_headers = [ "Content-Disposition", "Content-Location", "Location", "X-Run-Id", "X-Thread-Id", ] # Public pages (e.g. http://47.x.x.x:7010) calling this Gateway on localhost # trigger Chrome/Edge Private Network Access preflights. Without # allow_private_network=True the browser reports a generic CORS failure even # when Access-Control-Allow-Origin is correct. cors_kwargs = { "allow_credentials": True, "allow_methods": ["*"], "allow_headers": ["*"], "expose_headers": cors_expose_headers, "allow_private_network": True, } if _cors_allow_any_origin(): logger.warning( "Permissive CORS enabled ? any Origin is echoed back (GATEWAY_CORS_ALLOW_ALL or " "DEER_FLOW_AUTH_DISABLED). Do not expose publicly." ) app.add_middleware( CORSMiddleware, allow_origins=[], # fullmatch() on the Origin header; browsers send scheme://host[:port] allow_origin_regex=r".+", **cors_kwargs, ) elif cors_origins: if "*" in cors_origins: logger.error( "GATEWAY_CORS_ORIGINS contains '*' with allow_credentials=True ? invalid. " "Set GATEWAY_CORS_ALLOW_ALL=1 instead, or list explicit origins." ) cors_origins = [o for o in cors_origins if o != "*"] if cors_origins: app.add_middleware( CORSMiddleware, allow_origins=cors_origins, **cors_kwargs, ) else: # Safe-by-default fallback for local frontend development. Without any CORS # middleware, browser preflight OPTIONS requests will hit routes directly and # fail with 405, which blocks login and all JSON POST APIs. dev_origins = [ "http://localhost:5173", "http://127.0.0.1:5173", "http://localhost:5174", "http://127.0.0.1:5174", "http://localhost:3000", "http://127.0.0.1:3000", *taskcop_vite_origins, ] logger.warning( "No CORS configuration found; enabling local dev origins only: %s. " "Set GATEWAY_CORS_ORIGINS or GATEWAY_CORS_ALLOW_ALL for deployment.", ", ".join(dev_origins), ) app.add_middleware( CORSMiddleware, allow_origins=dev_origins, **cors_kwargs, ) # Include routers # Models API is mounted at /api/models app.include_router(models.router) # MCP API is mounted at /api/mcp app.include_router(mcp.router) # Authenticated source-context proxy for the WeKnora retrieval skill. app.include_router(weknora_sources.router) # Memory API is mounted at /api/memory app.include_router(memory.router) # Skills API is mounted at /api/skills app.include_router(skills.router) # Artifacts API is mounted at /api/threads/{thread_id}/artifacts app.include_router(artifacts.router) # Uploads API is mounted at /api/threads/{thread_id}/uploads app.include_router(uploads.router) # Optional Chrome-extension page capture API is mounted at /api/browser-context app.include_router(browser_context.router) # Thread cleanup API is mounted at /api/threads/{thread_id} app.include_router(threads.router) # Agents API is mounted at /api/agents app.include_router(agents.router) # Recommended Questions API is mounted at /api/recommended-questions app.include_router(recommended_questions.router) app.include_router(user_prompts.router) # Per-user UI preferences at /api/user-preferences app.include_router(user_preferences.router) # Roundtable-planning drafts + recommendation history at /api/roundtable-drafts app.include_router(roundtable_drafts.router) # Task-scoped roundtable drafts (taskId 深链聊天记录，独立存、不分权) at /api/roundtable-task-drafts app.include_router(roundtable_task_drafts.router) # Step3 流程图结构化抽取 + 校验 (SSE) at /api/flow-extract app.include_router(flow_extract.router) # Roundtable-planning business chains at /api/roundtable-chains app.include_router(roundtable_chains.router) # 岗位协同独立会话 / 节点状态（不使用原圆桌总控或后台作业） # Canonical position-workspace persistence API. app.include_router(position_roundtable.router, prefix="/api") # Compatibility route for intranet gateways whose allowlist only exposes # the established `/api/multi-agent/*` namespace. app.include_router( position_roundtable.router, prefix="/api/multi-agent", include_in_schema=False, ) # Global business ? agent / chain mapping at /api/business-mapping app.include_router(business_mapping.router) # Position-roundtable work-view role configuration at /api/position-roles. app.include_router(position_roles.router) # Light-app registry (?????) at /api/light-apps app.include_router(light_apps.router) # Sidebar-menu overrides (菜单管理) at /api/menu-overrides app.include_router(menu_overrides.router) # Sensitive-word map (敏感词管理 / 脱敏词) at /api/sensitive-words app.include_router(sensitive_words.router) # Skill URL/IP batch replace (技能地址管理) at /api/skill-addresses app.include_router(skill_addresses.router) # 大屏绘制智能体独立页面会话（按 taskId 存取，不分权）at /api/dashboard-sessions app.include_router(dashboard_sessions.router) # Task-workspace configurable jump buttons (按钮管理) at /api/task-buttons app.include_router(task_buttons.router) # Microsoft Agent Framework 多智能体会商实验 (independent of the legacy # roundtable; feature-flag-gated, six isolated tables) at /api/maf-roundtable app.include_router(maf_roundtable.router) # MAF roundtable V2 Conversation API (user-visible layer) at /api/maf-roundtable/conversations app.include_router(maf_conversations.router) # Temporary compatibility bridge for the TaskCOP situation-overview frontend. app.include_router(taskcop_tasks.router) # Task-scoped situation-report details used by dashboard drawing / task gates. app.include_router(task_reports.router) # Standalone parallel multi-agent panel: fan-out streaming + saved tasks app.include_router(parallel_agents.router) app.include_router(parallel_agent_tasks.router) # Roundtable-planning background jobs at /api/roundtable-jobs app.include_router(roundtable_jobs.router) app.include_router(roundtable_diagnostics.router) # Suggestions API is mounted at /api/threads/{thread_id}/suggestions app.include_router(suggestions.router) # Writing rewrite API is mounted at /api/writing app.include_router(writing.router) # AI Writing multi-agent API is mounted at /api/ai-writing app.include_router(ai_writing.router) # Scheduled task API is mounted at /api/scheduled-tasks app.include_router(scheduled_tasks.router) # Public, unauthenticated read-only scheduled task API # (/api/public/scheduled-tasks/*). Lets external systems query tasks and # surface results without a DeerFlow login. app.include_router(public_scheduled_tasks.router) # Public read-only user leaderboard publish page (/api/public/leaderboard) app.include_router(public_leaderboard.router) # Public embed Q&A status polling (/api/public/embed/*) app.include_router(public_embed.router) app.include_router(public_config.router) app.include_router(html_page_favorites.router) # Product knowledge base API is mounted at /api/knowledge app.include_router(knowledge.router) # External knowledge ingest (publish md artifacts) at /api/knowledge-ingest app.include_router(knowledge_ingest.router) # System settings API is mounted at /api/system-settings app.include_router(system_settings.router) # Admin LLM metrics audit + export app.include_router(llm_metrics.router) # Admin window into other users' threads / messages / memory app.include_router(admin_users.router) app.include_router(admin_users.self_router) app.include_router(admin_active_runs.router) # Admin tool / skill call metrics audit + export app.include_router(tool_metrics.router) # Channels API is mounted at /api/channels app.include_router(channels.router) # Admin checkpoint migration API: SQLite legacy checkpoints -> Postgres. app.include_router(checkpoint_migration.router) # Assistants compatibility API (LangGraph Platform stub) app.include_router(assistants_compat.router) # Auth API is mounted at /api/v1/auth app.include_router(auth.router) # Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback app.include_router(feedback.router) # Notifications API is mounted at /api/notifications app.include_router(notifications.router) # Tags API is mounted at /api/tags app.include_router(tags.router) # 岗位 (position) management API is mounted at /api/positions app.include_router(positions.router) # Thread Runs API (LangGraph Platform-compatible runs lifecycle) app.include_router(thread_runs.router) app.include_router(thread_shares.router) # Public, unauthenticated read-only share snapshots (/api/public/shares/*). app.include_router(thread_shares.public_router) # ?????????import ???? / view ???????? app.include_router(roundtable_draft_shares.router) # Public, unauthenticated read-only roundtable result (/api/public/roundtable-shares/*). app.include_router(roundtable_draft_shares.public_router) # Stateless Runs API (stream/wait without a pre-existing thread) app.include_router(runs.router) # 开放问答接口 /api/open/chat —— 无需登录,指定模型 + 智能体做问答,可关闭思考模式 app.include_router(open_chat.router) # 开放接口 /api/open/3qfx/ask —— 无需登录,按 taskId 直接发起 3qfx 第一次问答(后台运行,落成真实会话) app.include_router(open_3qfx.router) # --- ?????????? /page/workspace/roundtable/planning?--- # ??? app/gateway/routers/{multi_agent,intent,recommend}.py? # ??? frontend-web/docs/multi-agent-backend-dev.md app.include_router(multi_agent.router) # Step 2?init(SSE) + run/stream(leader|special) app.include_router(intent.router) # Step 1?/api/intent/init + /stream app.include_router(recommend.router) # Step 1?2?/api/recommend/stream?? init? # Step 3 ????/?????? artifacts.router?/api/threads/{id}/artifacts? # ?????????(? auth_middleware._PUBLIC_PATH_PREFIXES)? # ????? ``/health`` ?? API ??? ``/api/health``,????? # ?? base URL ????,?????????? 401? @app.get("/health", tags=["health"]) @app.get("/api/health", tags=["health"]) async def health_check() -> dict: """Health check endpoint. Returns: Service health status information. """ return {"status": "healthy", "service": "deer-flow-gateway"} return app # Create app instance for uvicorn app = create_app()