qiaoqichaoandClaude Code
fcf4d8c948
T02 第三个子卡。命名空间按 T00/04 §1 定案 P1 挂 project-assistant 之下，
复用现有鉴权与同源代理，鉴权/代理代码零改动。
新增 backend/api/prescreen.py（7 路径 / 8 操作）+ test_prescreen_api.py（27 项）。
app.py 仅新增 1 个 include_router 与 1 个异常处理器注册。
落实：
- 404 统一口径（03 §2）：不存在与无访问权一律 404，不泄露对象存在性。
既有 check_project 抛 403 且被 5 个模块共用，故不改它，由本模块 scope() 包装。
实测：无权项目返回 404 NOT_FOUND 而非 403。
- 错误信封（D2）：只在新路由输出 {error:{code,message,retryable,request_id}}。
- E-09 与 E-05 精确区分：新增 store.IdempotencyConflict，不再靠字符串比对猜码。
- 客户端清单只接受 4 个字段；id/status/received_at 由服务端生成——
状态机写权不下放。manifest 校验因此从"整对象"改为"4 字段 + 服务端补全"。
**实测发现的缺口并已修复**：Input(extra="forbid") 的校验错误由 FastAPI 校验层在
进入端点前抛出，不经过 run()，导致同一命名空间出现两种错误形状。现注册
RequestValidationError 处理器，但**只在 prescreen 路径生效**，其它路径交回
FastAPI 默认实现——既有端点形状经实测确认未变（含校验错误的形状）。
验证：
- scripts/verify.py 9/9 PASS（exit 0）；全量单测 225 项 OK（196 → 225）
- claude_use/T02/smoke_s1.py 在真实 app.py 上 10/10 PASS：覆盖路由注册、
真实前缀下的 manager 鉴权放行、404 统一、信封形状、以及既有端点未被污染
Co-Authored-By: Claude Code <noreply@anthropic.com>
118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
"""Mini-OpenClaw Backend — FastAPI Entry Point"""
|
|
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
load_dotenv()
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
"""Startup: scan skills, initialize agent, build memory index."""
|
|
from tools.skills_scanner import scan_skills
|
|
from graph.agent import agent_manager
|
|
from graph.memory_indexer import get_memory_indexer
|
|
|
|
scan_skills(BASE_DIR)
|
|
agent_manager.initialize(BASE_DIR)
|
|
|
|
# Initialize memory indexer for RAG mode
|
|
try:
|
|
indexer = get_memory_indexer(BASE_DIR)
|
|
indexer.rebuild_index()
|
|
except Exception as e:
|
|
print(f"⚠️ Memory indexer initialization skipped: {e}")
|
|
|
|
print("✅ mini OpenClaw backend ready")
|
|
from services.project_assistant import ProjectAssistantService
|
|
from services.assistant_worker import DurableWorker
|
|
from services.assistant_settings import flags
|
|
worker = DurableWorker(ProjectAssistantService()).start() if flags()["assistant"] else None
|
|
from services.operations_demo import DemoService
|
|
demo_worker = DurableWorker(DemoService()).start() if flags()["assistant"] and flags()["demo"] else None
|
|
try:
|
|
yield
|
|
finally:
|
|
if worker:
|
|
worker.stop()
|
|
if demo_worker:
|
|
demo_worker.stop()
|
|
|
|
|
|
app = FastAPI(title="mini OpenClaw", version="0.1.0", lifespan=lifespan)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
from api.chat import router as chat_router
|
|
from api.files import router as files_router
|
|
from api.sessions import router as sessions_router
|
|
from api.tokens import router as tokens_router
|
|
from api.compress import router as compress_router
|
|
from api.config_api import router as config_router
|
|
from api.xingbang_router import router as xingbang_router
|
|
from api.xingbang_v2 import router as xingbang_v2_router
|
|
from api.review_assist import router as review_assist_router
|
|
from api.xingbang_unified import router as xingbang_unified_router
|
|
from api.risk_assess import router as risk_assess_router
|
|
from api.xingbang_workspace import router as xingbang_workspace_router
|
|
from api.risk_agent_config import router as risk_agent_config_router
|
|
|
|
app.include_router(chat_router, prefix="/api")
|
|
app.include_router(files_router, prefix="/api")
|
|
app.include_router(sessions_router, prefix="/api")
|
|
app.include_router(tokens_router, prefix="/api")
|
|
app.include_router(compress_router, prefix="/api")
|
|
app.include_router(config_router, prefix="/api")
|
|
app.include_router(xingbang_router)
|
|
app.include_router(xingbang_v2_router)
|
|
app.include_router(review_assist_router)
|
|
app.include_router(xingbang_unified_router)
|
|
app.include_router(risk_assess_router, prefix="/api/risk-assess")
|
|
app.include_router(xingbang_workspace_router)
|
|
app.include_router(risk_agent_config_router)
|
|
from api.report_agent import router as report_agent_router
|
|
app.include_router(report_agent_router)
|
|
from api.coordination_agent import router as coordination_agent_router
|
|
app.include_router(coordination_agent_router)
|
|
from api.collateral_agent import router as collateral_agent_router
|
|
app.include_router(collateral_agent_router)
|
|
from api.xingbang_projects import router as projects_router
|
|
app.include_router(projects_router)
|
|
from api.project_assistant import router as project_assistant_router
|
|
from api.operations_center import router as operations_router
|
|
app.include_router(operations_router)
|
|
app.include_router(project_assistant_router)
|
|
# S1 预审接口：必须注册在 project_assistant 之后（更具体的前缀先匹配；
|
|
# 其路径为 .../project-assistant/prescreen/...）——见 claude_use/T00/04 §1
|
|
from fastapi.exceptions import RequestValidationError
|
|
from api.prescreen import router as prescreen_router, validation_handler
|
|
|
|
# 只为 prescreen 路径输出统一错误信封；处理器内部对其它路径交回 FastAPI 默认实现，
|
|
# 因此既有端点的 {"detail": ...} 形状不变（有测试与冒烟锁定该隔离）。
|
|
app.add_exception_handler(RequestValidationError, validation_handler)
|
|
app.include_router(prescreen_router)
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"name": "mini OpenClaw", "version": "0.1.0", "status": "running"}
|
|
|
|
|
|
# 支持 `python app.py` 直接启动后端(FastAPI + uvicorn)
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8002)
|