"""FastAPI 应用入口。 既导出 ASGI 应用实例 ``app``(供 ``uvicorn app.main:app`` 使用),又在作为脚本执行 (``python -m app.main``)时启动 uvicorn。日志系统在模块级初始化,确保无论从哪种 入口启动,uvicorn 横幅、第三方库与应用日志都统一走 ``app.core.logging`` 出口。 """ from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager from time import perf_counter import httpx from fastapi import FastAPI, Request, Response from fastapi.middleware.cors import CORSMiddleware from sqlalchemy.exc import OperationalError from app.api.router import api_router from app.application.agent.confirmation import ConfirmationRegistry from app.application.tool_runtime.runtime import ToolRuntime from app.core.config import settings from app.core.errors import register_exception_handlers, trace_id_var from app.core.ids import new_trace_id from app.core.logging import ( clear_log_context, log_access, log_with_context, set_request_id, setup_logging, ) from app.db.init_data import seed_default_data from app.db.session import AsyncSessionLocal, close_engine from app.integrations.sandbox.client import SandboxClient from app.modules.knocking.pivot import KnockingPivot from app.modules.safety.confirmation import InMemoryConfirmationStore from app.schemas.common import ErrorResponse from app.services.dispatcher import Dispatcher from app.services.event_store import DbEventStore from app.services.personality_emotion import handle_time_event from app.services.safety import load_governor_from_db # 模块级初始化日志系统:import 本模块即完成 loguru 配置并接管标准 logging, # 覆盖 ``python -m app.main`` 与 ``uvicorn app.main:app`` 两种启动入口。 setup_logging() log_with_context( "应用日志已初始化", log_dir=str(settings.resolved_log_dir), log_level=settings.log_level, app_env=settings.app_env, ) @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: try: async with AsyncSessionLocal() as session: await seed_default_data(session) except OperationalError: log_with_context("数据库尚未完成迁移,跳过默认数据初始化。", level="WARNING") # SQLite WAL 模式:用 sqlite3 直接在文件层面设置(必须跑在事务外才能生效), # 后续所有 aiosqlite 连接自动继承 WAL 模式;允许读写并发, # 避免叩心枢后台循环与请求处理同时写时锁冲突 if settings.resolved_database_url.startswith("sqlite"): try: import sqlite3 _db_path = settings.resolved_database_url.replace("sqlite+aiosqlite:///", "") _raw_conn = sqlite3.connect(_db_path) _raw_conn.execute("PRAGMA journal_mode=WAL") _raw_conn.execute("PRAGMA busy_timeout=5000") _raw_conn.close() log_with_context("SQLite WAL 模式已启用,busy_timeout=5000ms") except Exception: # noqa: BLE001 log_with_context("SQLite WAL 模式设置失败,已跳过", level="WARNING") # 加载模型网关 endpoint 覆盖缓存(DB 优先于 yaml/env);迁移未完或读失败则降级跳过 try: from app.services.ai_provider_admin import reload_provider_manager await reload_provider_manager() except Exception: # noqa: BLE001 log_with_context("模型网关覆盖缓存加载失败,已跳过", level="WARNING") # 运行时注入:事件存储始终就绪(DbEventStore 持久化到 events 表,重启不丢); # 工具运行时依赖沙箱配置,未配置(缺 SANDBOX_AUTH_SECRET 等)则降级 None(工具 API 503)。 # 事件流与历史查询不依赖沙箱,始终可用。 _app.state.event_store = DbEventStore(AsyncSessionLocal) # Kernel(rewrite/1.0 M0):极小硬核 + 事件日志 + 前台席位;神识客户端按配置装配, # 未配置神识地址时为进程内回声桩(骨架形态,见 app.shenshi_client)。 from app.kernel import Kernel, KernelEventLog from app.shenshi_client import build_shenshi_client _app.state.kernel = Kernel( shenshi=build_shenshi_client(settings.shenshi_base_url, settings.shenshi_service_token), session_factory=AsyncSessionLocal, event_log=KernelEventLog(_app.state.event_store), timeout_seconds=float(settings.run_timeout_seconds), ) # ── MCP 进程管理器(stdio 子进程生命周期管理) ────────────── from app.integrations.mcp.process_manager import ( ProcessManager, set_process_manager, ) process_manager = ProcessManager() _app.state.mcp_process_manager = process_manager set_process_manager(process_manager) log_with_context("MCP 进程管理器就绪") # 一炁进程内订阅分发器:订阅 EventStore 全量事件,按 glob pattern 分发给注册的 handler。 # 各枢核心就绪后在 start() 前 register;lifespan 管理 start/stop。 _app.state.dispatcher = Dispatcher(_app.state.event_store) # 灵核情绪状态机:订阅辰正 time.* 事件驱动 EmotionState(设计 §11,首个真实 handler) _app.state.dispatcher.register("time.*", handle_time_event) await _app.state.dispatcher.start() # 叩心枢主动对话守护进程:后台常驻循环,在 dispatcher 就绪后启动 _app.state.knocking_pivot = KnockingPivot(_app.state.event_store) await _app.state.knocking_pivot.start() _app.state.confirmation_store = InMemoryConfirmationStore() if settings.use_db_confirmation_store: from app.modules.safety.confirmation import DbConfirmationStore _app.state.confirmation_store = DbConfirmationStore(AsyncSessionLocal) log_with_context("confirmation_store_set_to_db") # 进程内检查点确认注册表:混合唤醒(同进程 Event + 重启后 DB polling)。 _app.state.confirmation_registry = ConfirmationRegistry() from app.application.agent import AgentRunRegistry, discover_python_skills _app.state.agent_run_registry = AgentRunRegistry() _app.state.python_skill_catalog = discover_python_skills(("app.skills",)) log_with_context( "Python Skills 目录就绪", skills=[skill.name for skill in _app.state.python_skill_catalog.skills], ) sandbox_http: httpx.AsyncClient | None = None tools_http: httpx.AsyncClient | None = None if settings.enable_tool_runtime: try: if settings.sandbox_auth_enabled and not settings.sandbox_auth_secret: raise ValueError("SANDBOX_AUTH_ENABLED=true 但 SANDBOX_AUTH_SECRET 未配置") sandbox_http = httpx.AsyncClient(trust_env=False) # executor 工具(fetch_url)的长生命周期 HTTP 客户端, # 与 sandbox client 同生命周期,由 lifespan 统一 aclose。 tools_http = httpx.AsyncClient(trust_env=False) sandbox = SandboxClient(sandbox_http, settings) governor = await load_governor_from_db() _app.state.tool_runtime = ToolRuntime( sandbox, governor, _app.state.event_store, http_client=tools_http, ) log_with_context("百工工具运行时就绪") # 启动加载 MCP server 工具(按 transport 类型注册,失败不阻断启动) try: from app.services.mcp import load_registered_mcp_servers await load_registered_mcp_servers( _app.state.tool_runtime, process_manager=_app.state.mcp_process_manager, ) except Exception: # noqa: BLE001 log_with_context("MCP 启动加载失败,已跳过", level="WARNING") except Exception as exc: # noqa: BLE001 — 任意配置/构造失败均降级,不阻断启动 _app.state.tool_runtime = None log_with_context(f"百工工具运行时未就绪,工具 API 将返回 503:{exc}", level="WARNING") else: _app.state.tool_runtime = None log_with_context("ENABLE_TOOL_RUNTIME=false,工具运行时未启用", level="INFO") yield # 叩心枢先停,再停 dispatcher if getattr(_app.state, "knocking_pivot", None) is not None: await _app.state.knocking_pivot.stop() # Dispatcher 先停(取消 pump/consumer task + unsubscribe all_queue),再关 http/engine if getattr(_app.state, "dispatcher", None) is not None: await _app.state.dispatcher.stop() if sandbox_http is not None: await sandbox_http.aclose() if tools_http is not None: await tools_http.aclose() # MCP 进程管理器关闭所有 stdio 子进程 if getattr(_app.state, "mcp_process_manager", None) is not None: await _app.state.mcp_process_manager.shutdown_all() log_with_context("MCP stdio 子进程已全部关闭") await close_engine() def create_app() -> FastAPI: # OpenAPI servers 仅作展示:backend_host 为通配地址(0.0.0.0 / ::)时回退 127.0.0.1, # 避免在 Swagger 文档里暴露不可访问的地址或写死的占位域名。 server_host = ( settings.backend_host if settings.backend_host not in ("0.0.0.0", "::") else "127.0.0.1" ) app = FastAPI( title=settings.app_name, debug=settings.debug, lifespan=lifespan, servers=[ {"url": f"http://{server_host}:{settings.backend_port}", "description": "本地开发"}, ], responses={ 400: {"model": ErrorResponse, "description": "请求无效或业务前置条件不满足。"}, 404: {"model": ErrorResponse, "description": "资源不存在或不属于当前人物。"}, 422: {"model": ErrorResponse, "description": "请求字段校验失败。"}, 502: {"model": ErrorResponse, "description": "模型或外部工具调用失败。"}, }, ) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.middleware("http") async def trace_middleware( request: Request, call_next: Callable[[Request], Awaitable[Response]] ) -> Response: trace_id = request.headers.get("X-Trace-Id") or new_trace_id() token = trace_id_var.set(trace_id) set_request_id(trace_id) started_at = perf_counter() try: response = await call_next(request) response.headers["X-Trace-Id"] = trace_id duration_ms = round((perf_counter() - started_at) * 1000, 2) log_access( "HTTP 请求完成", method=request.method, path=request.url.path, status=response.status_code, duration_ms=duration_ms, ) return response finally: trace_id_var.reset(token) clear_log_context() register_exception_handlers(app) app.include_router(api_router, prefix=settings.api_prefix) return app app = create_app() if __name__ == "__main__": import uvicorn uvicorn.run( "app.main:app", host=settings.backend_host, port=settings.backend_port, reload=settings.is_development, workers=1 if settings.is_development else 4, log_config=None, # 禁用 uvicorn 默认 dictConfig,统一由 InterceptHandler 接管 )