""" API Server - FastAPI 应用入口 聚合所有模块的 API 路由: - GET /api/traces — 查询(trace/api.py) - POST /api/traces — 执行控制(trace/run_api.py,需配置 Runner) - WS /api/traces/{id}/watch — 实时推送(trace/websocket.py) - GET /api/experiences — 经验查询(trace/run_api.py,需配置 Runner) """ import json import logging import os import fcntl from ipaddress import ip_address from pathlib import Path from dotenv import load_dotenv load_dotenv() from fastapi import FastAPI, Request, WebSocket from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.trustedhost import TrustedHostMiddleware import uvicorn from cyber_agent.trace import FileSystemTraceStore from cyber_agent.trace.api import router as api_router, set_trace_store as set_api_trace_store from cyber_agent.trace.run_api import ( router as run_router, experiences_router, set_application_runtime, set_runner, ) from cyber_agent.trace.websocket import router as ws_router, set_trace_store as set_ws_trace_store from cyber_agent.trace.examples_api import router as examples_router from cyber_agent.trace.logs_websocket import router as logs_router, setup_websocket_logging from cyber_agent.trace.upload_api import router as upload_router, set_trace_store as set_upload_trace_store # ===== 日志配置 ===== logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s" ) logger = logging.getLogger(__name__) # ===== 单用户本地开发边界 ===== LOCAL_DEVELOPMENT_ORIGINS = ( "http://127.0.0.1:3000", "http://localhost:3000", ) LOCAL_DEVELOPMENT_HOSTS = ("127.0.0.1", "localhost") _writer_lock_file = None def _env_flag(name: str, *, default: bool = False) -> bool: """解析显式开关;未配置和未识别的值均按关闭处理。""" value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def _is_local_websocket_origin(websocket: WebSocket) -> bool: """浏览器 WebSocket 必须来自本地前端。""" return websocket.headers.get("origin") in LOCAL_DEVELOPMENT_ORIGINS def _is_loopback_host(host: str | None) -> bool: """Only numeric IPv4/IPv6 loopback peers are trusted.""" if not host: return False try: return ip_address(host).is_loopback except ValueError: return False def _is_loopback_websocket_peer(websocket: WebSocket) -> bool: return _is_loopback_host(websocket.client.host if websocket.client else None) def _acquire_single_writer_lock(): """Fail startup when another API process owns this Trace directory.""" global _writer_lock_file if _writer_lock_file is not None: return lock_path = Path(".trace").resolve() / ".api-writer.lock" lock_path.parent.mkdir(parents=True, exist_ok=True) handle = lock_path.open("a+") try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: handle.close() raise RuntimeError( f"another Agent API writer already owns {lock_path.parent}" ) from exc _writer_lock_file = handle def _release_single_writer_lock(): global _writer_lock_file if _writer_lock_file is None: return fcntl.flock(_writer_lock_file.fileno(), fcntl.LOCK_UN) _writer_lock_file.close() _writer_lock_file = None # 设置WebSocket日志推送 setup_websocket_logging(level=logging.INFO) # ===== FastAPI 应用 ===== app = FastAPI( title="Agent API", description="Agent 查询 + 执行 API", version="1.0.0" ) # CORS 配置(允许前端跨域访问) app.add_middleware( CORSMiddleware, allow_origins=list(LOCAL_DEVELOPMENT_ORIGINS), allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # 拒绝非本地 Host,防止误暴露和 DNS rebinding。 app.add_middleware( TrustedHostMiddleware, allowed_hosts=list(LOCAL_DEVELOPMENT_HOSTS), ) @app.middleware("http") async def reject_non_local_browser_origins(request: Request, call_next): """Fail closed unless the TCP peer and optional browser Origin are local.""" if not _is_loopback_host(request.client.host if request.client else None): return JSONResponse( status_code=403, content={"detail": "Loopback client required"}, ) origin = request.headers.get("origin") if origin is not None and origin not in LOCAL_DEVELOPMENT_ORIGINS: return JSONResponse( status_code=403, content={"detail": "Local development origin required"}, ) return await call_next(request) # ===== 初始化存储 ===== # 使用文件系统存储(支持跨进程和持久化) trace_store = FileSystemTraceStore(base_path=".trace") # 注入到 step_tree 模块 set_api_trace_store(trace_store) set_ws_trace_store(trace_store) set_upload_trace_store(trace_store) # ===== 可选:配置 Runner(启用执行 API)===== # 如需启用 POST /api/traces(新建/运行/停止/反思),取消以下注释并配置 LLM: from cyber_agent.core.runner import AgentRunner from cyber_agent.application.bootstrap import build_runtime_from_environment from cyber_agent.llm import create_openrouter_llm_call llm_call = create_openrouter_llm_call(model="anthropic/claude-sonnet-4.5") runner = AgentRunner( trace_store=trace_store, llm_call=llm_call, ) set_runner(runner) application_runtime = build_runtime_from_environment( trace_store=trace_store, llm_call=llm_call, ) set_application_runtime(application_runtime) # ===== 注册路由 ===== # Examples API(GET /api/examples) app.include_router(examples_router) # Trace 上传 API(POST /api/traces/upload) app.include_router(upload_router) # Trace 执行 API(POST + GET /running,需配置 Runner) # 注意:run_router 必须在 api_router 之前注册,否则 GET /running 会被 /{trace_id} 捕获 app.include_router(run_router) # 经验 API(GET /api/experiences,需配置 Runner) app.include_router(experiences_router) # Trace 查询 API(GET) app.include_router(api_router) # Trace WebSocket(实时推送) app.include_router(ws_router) # Logs WebSocket(日志推送) app.include_router(logs_router) @app.on_event("startup") async def on_startup(): """服务器启动时执行状态对齐""" _acquire_single_writer_lock() for trace in await trace_store.list_traces(limit=10_000): await trace_store.reconcile_trace_integrity(trace.trace_id) from cyber_agent.trace.run_api import reconcile_traces await reconcile_traces() @app.on_event("shutdown") async def on_shutdown(): _release_single_writer_lock() @app.websocket("/ws_ping") async def ws_ping(websocket: WebSocket): if not _is_loopback_websocket_peer(websocket): await websocket.close(code=1008, reason="Loopback client required") return if not _is_local_websocket_origin(websocket): await websocket.close(code=1008, reason="Local development origin required") return await websocket.accept() await websocket.send_text("pong") await websocket.close() # ===== 健康检查 ===== @app.get("/health") async def health_check(): """健康检查""" return { "status": "ok", "service": "Agent Step Tree API", "version": "1.0.0", "deployment_mode": "local_development", "network_scope": "loopback_only", "authentication": "none", } # ===== 启动服务 ===== if __name__ == "__main__": logger.info("Starting API server...") uvicorn.run( "api_server:app", host="127.0.0.1", port=8000, reload=_env_flag("TRACE_API_RELOAD"), log_level="info" )