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 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.public_knowledge_cors import PublicKnowledgeCORSMiddleware from app.gateway.routers import ( admin_active_runs, admin_users, agents, ai_writing, artifacts, assistant_knowledge, assistants_compat, auth, browser_context, business_mapping, channels, checkpoint_migration, compact_menu_overrides, dashboard_sessions, deep_research, external_llmwiki, feedback, fixed_questions, flow_extract, html_page_favorites, intent, knowledge, knowledge_ingest, light_apps, llm_metrics, llmwiki, llmwiki_index, mcp, memory, menu_overrides, model_management, models, multi_agent, notifications, open_3qfx, open_chat, parallel_agent_tasks, parallel_agents, position_roles, position_roundtable, positions, public_config, public_embed, public_knowledge_vector_search, public_leaderboard, public_scheduled_tasks, recommend, recommended_questions, report_collaboration, report_structures, roundtable_artifact_submissions, roundtable_chains, roundtable_diagnostics, roundtable_draft_shares, roundtable_drafts, roundtable_jobs, roundtable_task_drafts, runs, scheduled_tasks, sensitive_words, sentiment_agent, skill_addresses, skill_knowledge, skills, suggestions, system_settings, tags, task_buttons, task_reports, taskcop_tasks, thread_runs, thread_shares, threads, tool_metrics, uploads, user_preferences, user_prompts, workflow_data_sources, workflow_embed, workflow_planning, workflow_resources, workflow_runs, workflow_studio, workflows, workflows_coze_compat, 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 @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") # One-time repair switch for a failed, unused Workflow Studio schema. # The helper is intentionally MySQL-only and drops an exact allow-list # in the configured current database; it never expands ``workflow_%``. # Keep this opt-in so established workflow data is never cleared during # an ordinary restart. if os.getenv("DEER_FLOW_REBUILD_WORKFLOW_SCHEMA", "").lower() in {"1", "true", "yes"}: from deerflow.persistence.engine import rebuild_workflow_schema_for_mysql logger.warning( "DEER_FLOW_REBUILD_WORKFLOW_SCHEMA is enabled; rebuilding the unused workflow schema" ) await rebuild_workflow_schema_for_mysql() # Run Alembic migrations before anything else touches the schema. await _run_db_migrations() # Local Wiki mirror/index scans are opt-in and run outside request # handling. Per-library database leases keep multi-worker startup safe. app.state.llmwiki_index_scheduler_task = None local_wiki_index = app.state.config.llmwiki.local_wiki_index if ( local_wiki_index.enabled and local_wiki_index.auto_sync and getattr(app.state, "llmwiki_sync_service", None) is not None ): from app.gateway.llmwiki_index_scheduler import local_wiki_index_scheduler_loop app.state.llmwiki_index_scheduler_task = asyncio.create_task( local_wiki_index_scheduler_loop(app) ) # ?? 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)" ) # 实验性「页面研报设计师」：在 _sync_legacy_agents 前补齐 agent 目录与 # 公共页面生成技能，首次启动即可在智能体管理页和普通对话中复用。 try: from app.gateway.routers._page_report_designer_seed import ( ensure_page_report_designer_agent, ) ensure_page_report_designer_agent() except Exception: logger.exception( "Page-report designer agent seeding failed at startup (non-fatal)" ) # 深度研究「收集员」内置 agent(对话式信息收集)。同样必须在 # _sync_legacy_agents 之前补建目录,由后者 upsert 进 agents 表, # 深度研究页的收集对话与智能体管理页即可直接使用。 try: from app.gateway.routers._deep_research_seed import ( ensure_deep_research_collector_agent, ) ensure_deep_research_collector_agent() except Exception: logger.exception( "Deep-research collector agent seeding failed at startup (non-fatal)" ) # Workflow Studio owns a separate planning controller. It must be # seeded before _sync_legacy_agents, so the planner is available to # every user's candidate-workflow request on a fresh deployment. try: from app.gateway.routers._workflow_planner_seed import ( ensure_workflow_planner_agent, ) ensure_workflow_planner_agent() except Exception: logger.exception( "Workflow planner agent seeding failed at startup (non-fatal)" ) # 内置「强制检索输出助手」：必须在 _sync_legacy_agents 之前补建， # 使首次启动就能在智能体管理/广场中使用。每轮检索门禁由 harness # 的 ForcedResearchMiddleware 执行，而不是仅依赖 SOUL 提示词。 try: from app.gateway.routers._forced_research_seed import ( ensure_forced_research_agent, ) ensure_forced_research_agent() except Exception: logger.exception( "Forced-research responder agent seeding failed at startup (non-fatal)" ) # 内置「任务研判报告助手」（agentfx 任务深链固定使用）：同样必须在 # _sync_legacy_agents 之前补建目录 + 补建配套 task-report-import 技能 # （skills/ 被 .gitignore 排除），由后者 upsert 进 agents 表。 try: from app.gateway.routers._agentfx_seed import ensure_agentfx_agent ensure_agentfx_agent() except Exception: logger.exception( "Agentfx analyst 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)") # 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 llmwiki_index_scheduler_task = getattr(app.state, "llmwiki_index_scheduler_task", None) if llmwiki_index_scheduler_task is not None: llmwiki_index_scheduler_task.cancel() try: await llmwiki_index_scheduler_task except asyncio.CancelledError: pass except Exception: logger.exception("Local Wiki index scheduler shutdown error") llmwiki_index_jobs = list(getattr(app.state, "llmwiki_index_jobs", set())) for job in llmwiki_index_jobs: job.cancel() if llmwiki_index_jobs: await asyncio.gather(*llmwiki_index_jobs, return_exceptions=True) # ?? 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") # ?? 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() # Keep standard local frontend origins available even when the startup # script supplies an explicit production/development allow-list. Vite can # choose 5173 or move to 5174 when that port is already occupied; excluding # the former makes every browser preflight fail after a backend restart. local_vite_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", ] # The legacy TaskCOP Vue application uses its own Vite development port. taskcop_vite_origins = [ "http://localhost:8080", "http://127.0.0.1:8080", ] local_dev_origins = [*local_vite_origins, *taskcop_vite_origins] 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, *local_dev_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 = local_dev_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, ) # This one anonymous API is intentionally callable by browser code from # every origin. Keep the override route-scoped so authenticated APIs retain # the deployment's normal CORS allow-list. app.add_middleware(PublicKnowledgeCORSMiddleware) # Include routers # Models API is mounted at /api/models app.include_router(models.router) # Administrator-managed config.yaml model settings and connection tests. app.include_router(model_management.router) # MCP API is mounted at /api/mcp app.include_router(mcp.router) # DeerFlow-native LLMWiki management/search BFF (conditionally backed by WeKnora). app.include_router(llmwiki.router) app.include_router(llmwiki_index.router) app.include_router(external_llmwiki.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) # Admin skill distillation matrix + global assistant knowledge base. app.include_router(skill_knowledge.router) app.include_router(assistant_knowledge.router) app.include_router(assistant_knowledge.export_router) from app.gateway.routers import wiki_packages app.include_router(wiki_packages.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) # Task-scoped artifact submission status; stores references only, never artifact content. app.include_router(roundtable_artifact_submissions.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) # Workflow Studio: definitions / runs / SSE / data sources / embed / Coze compat. # Static prefixes (node-types, resources, planning, runs, data-sources, embed) must be # registered before the /{workflow_id} catch-alls in workflows.router. app.include_router(workflow_resources.router) app.include_router(workflow_data_sources.router) app.include_router(workflow_embed.router) app.include_router(workflow_planning.router) app.include_router(workflow_runs.router) app.include_router(workflow_studio.router) app.include_router(workflows.router) app.include_router(workflows_coze_compat.router) app.include_router(workflows_coze_compat.chrome_router) # Sidebar-menu overrides (菜单管理) at /api/menu-overrides app.include_router(menu_overrides.router) # Compact/sidebar-summary menu overrides at /api/compact-menu-overrides app.include_router(compact_menu_overrides.router) # Sensitive-word map (敏感词管理 / 脱敏词) at /api/sensitive-words app.include_router(sensitive_words.router) # Sentiment-analysis virtual agent AG-UI SSE proxy at /api/sentiment-agent app.include_router(sentiment_agent.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) # Admin exact-match question/answer short-circuit at /api/fixed-questions app.include_router(fixed_questions.router) # AgentScope report-collaboration workbench at /api/report-collaboration app.include_router(report_collaboration.router) # Deep-research report structures (工作流配置 / 报告结构) at /api/report-structures app.include_router(report_structures.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) # Deep Research (深度研究) API at /api/deep-research app.include_router(deep_research.router) # The independent enterprise-research workbench is temporarily disabled. # Do not mount /api/enterprise-research while its MySQL schema is inactive. # 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) # Anonymous, section-level vector search over every published Wiki base # except the system conversation-deposit base. app.include_router(public_knowledge_vector_search.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? # Health probes are always unauthenticated (see # auth_middleware.PUBLIC_HEALTH_PATHS). Support the bare Gateway path, # API base URL, and LangGraph proxy base URL for deployment monitors. @app.api_route("/health", methods=["GET", "HEAD"], tags=["health"]) @app.api_route("/api/health", methods=["GET", "HEAD"], tags=["health"]) @app.api_route("/api/langgraph/health", methods=["GET", "HEAD"], tags=["health"]) async def health_check() -> dict: """Health check endpoint. Returns: Service health status information. """ return {"status": "healthy", "service": "deer-flow-gateway"} # WeKnora's embedded frontend uses root-relative routes such as /platform, # /assets, /locales and /api/v1/*. Register that proxy last so it stays a # true fallback and cannot intercept DeerFlow APIs. # Register the public-prefix routes before the unprefixed static fallback. # Otherwise proxy_router's /{proxied_path:path} route consumes every # /deerflow/* request before the prefixed platform/assets routes can match. app.include_router(llmwiki.proxy_router, prefix="/deerflow") app.include_router(llmwiki.proxy_router) return app # Create app instance for uvicorn app = create_app()