diff --git a/backend/server/main.py b/backend/server/main.py index d3b9e7dd..20b5c37e 100644 --- a/backend/server/main.py +++ b/backend/server/main.py @@ -21,6 +21,9 @@ from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from server.routers import router +from server.routers.gateway_ws_router import router as gateway_ws_router +from server.routers.webhook_router import router as webhook_router +from yuxi.channel.gateway.sse import router as gateway_sse_router from server.utils.lifespan import lifespan from server.utils.auth_middleware import is_public_path from server.utils.common_utils import setup_logging @@ -40,6 +43,9 @@ _attempt_lock = asyncio.Lock() app = FastAPI(lifespan=lifespan) # 所有业务接口统一挂载到 /api，具体分组在 server.routers 中集中注册。 app.include_router(router, prefix="/api") +app.include_router(gateway_sse_router) +app.include_router(gateway_ws_router) +app.include_router(webhook_router) # CORS 设置 app.add_middleware( diff --git a/backend/server/routers/__init__.py b/backend/server/routers/__init__.py index fefc6ce8..61bb5ed9 100644 --- a/backend/server/routers/__init__.py +++ b/backend/server/routers/__init__.py @@ -2,10 +2,14 @@ import os from fastapi import APIRouter +from server.routers.auth_apikey_router import apikey_router +from server.routers.auth_dept_router import department from server.routers.auth_router import auth +from server.routers.channel_router import channel, gateway_router from server.routers.chat_router import chat from server.routers.dashboard_router import dashboard -from server.routers.auth_dept_router import department +from server.routers.filesystem_router import filesystem_router +from server.routers.gateway_sse_router import sse_polling_router from server.routers.mcp_router import mcp from server.routers.model_provider_router import model_providers from server.routers.skill_router import skills @@ -13,8 +17,6 @@ from server.routers.subagent_router import subagents_router from server.routers.system_router import system from server.routers.system_task_router import tasks from server.routers.tool_router import tools -from server.routers.auth_apikey_router import apikey_router -from server.routers.filesystem_router import filesystem_router from server.routers.workspace_router import workspace _LITE_MODE = os.environ.get("LITE_MODE", "").lower() in ("true", "1") @@ -23,6 +25,9 @@ router = APIRouter() # 基础系统接口：健康检查、配置、认证与聊天主链路。 router.include_router(system) # /api/system/* 系统状态与全局配置 +router.include_router(channel) # /api/system/channel/* 渠道网关管理 +router.include_router(gateway_router) # /api/system/gateway/* 渠道网关健康检查 +router.include_router(sse_polling_router) # /api/poll SSE 流轮询降级 router.include_router(auth) # /api/auth/* 登录与用户信息 router.include_router(chat) # /api/chat/* 对话、消息流、运行态 @@ -41,9 +46,9 @@ router.include_router(workspace) # /api/workspace/* 用户个人工作区 if not _LITE_MODE: from server.routers.graph_router import graph - from server.routers.knowledge_router import knowledge from server.routers.knowledge_eval_router import evaluation from server.routers.knowledge_mindmap_router import mindmap + from server.routers.knowledge_router import knowledge # 知识库与图谱能力依赖较重，LITE 模式下跳过这组接口。 router.include_router(knowledge) # /api/knowledge/* 知识库管理与检索 diff --git a/backend/server/routers/auth_router.py b/backend/server/routers/auth_router.py index 980638b2..5ea696e9 100644 --- a/backend/server/routers/auth_router.py +++ b/backend/server/routers/auth_router.py @@ -50,6 +50,7 @@ class Token(BaseModel): role: str department_id: int | None = None department_name: str | None = None + source: str | None = None class UserCreate(BaseModel): @@ -58,6 +59,7 @@ class UserCreate(BaseModel): role: str = "user" phone_number: str | None = None department_id: int | None = None + source: str | None = None class UserUpdate(BaseModel): @@ -67,6 +69,7 @@ class UserUpdate(BaseModel): phone_number: str | None = None avatar: str | None = None department_id: int | None = None + source: str | None = None class UserProfileUpdate(BaseModel): @@ -83,6 +86,7 @@ class UserResponse(BaseModel): role: str department_id: int | None = None department_name: str | None = None # 部门名称 + source: str | None = None created_at: str last_login: str | None = None @@ -124,6 +128,7 @@ class OIDCLoginResponse(BaseModel): role: str department_id: int | None = None department_name: str | None = None + source: str | None = None # ============================================================================= @@ -236,6 +241,7 @@ async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends( "role": user.role, "department_id": user.department_id, "department_name": department_name, + "source": user.source, } @@ -319,6 +325,7 @@ async def initialize_admin(admin_data: InitializeAdmin, db: AsyncSession = Depen "phone_number": new_admin.phone_number, "avatar": new_admin.avatar, "role": new_admin.role, + "source": new_admin.source, } @@ -492,6 +499,7 @@ async def create_user( "password_hash": hashed_password, "role": user_data.role, "department_id": department_id, + "source": user_data.source or "local", } ) @@ -623,6 +631,10 @@ async def update_user( user.avatar = user_data.avatar update_details.append(f"头像: {user_data.avatar or '已清空'}") + if user_data.source is not None: + user.source = user_data.source + update_details.append(f"来源: {user_data.source}") + # 部门修改权限控制（只有超级管理员可以修改用户部门） if user_data.department_id is not None and user_data.department_id != user.department_id: if current_user.role != "superadmin": @@ -857,6 +869,7 @@ async def impersonate_user( "role": target_user.role, "department_id": target_user.department_id, "department_name": department_name, + "source": target_user.source, } diff --git a/backend/server/routers/channel_router.py b/backend/server/routers/channel_router.py new file mode 100644 index 00000000..e0f2f5d5 --- /dev/null +++ b/backend/server/routers/channel_router.py @@ -0,0 +1,660 @@ +from pydantic import BaseModel, Field + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from yuxi.channel.doctor import doctor_runner +from yuxi.channel.protocols import DirectoryProtocol, DirectorySearchParams +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.services.channel_service import ( + approve_channel_pairing, + create_channel_binding, + create_channel_config, + create_thread_mapping, + create_user_mapping, + delete_channel_binding, + delete_channel_config, + delete_channel_pairing_record, + delete_thread_mapping, + delete_user_mapping, + find_thread_mapping_by_channel_chat, + find_thread_mapping_by_thread_id, + find_user_mapping_by_channel, + find_user_mappings_by_internal_user, + generate_agent_pairing_code, + get_channel_binding, + get_channel_config, + get_channel_pairing_record, + get_gateway_health, + get_gateway_healthz, + get_hook_status, + get_thread_mapping, + get_user_mapping, + install_hook, + list_channel_bindings, + list_channel_bindings_by_type, + list_channel_configs, + list_channel_pairing_records, + list_user_mappings, + reject_channel_pairing, + reload_hooks, + resolve_channel_thread, + resolve_channel_user, + uninstall_hook, + update_channel_binding, + update_channel_config, +) +from yuxi.storage.postgres.models_business import User + +from server.utils.auth_middleware import get_admin_user + +channel = APIRouter(prefix="/system/channel", tags=["channel"]) +gateway_router = APIRouter(prefix="/system/gateway", tags=["gateway"]) + + +class RepairRequest(BaseModel): + step_id: str = Field(..., min_length=1) + account_id: str = "default" + + +class ChannelConfigCreate(BaseModel): + channel_type: str + name: str + config: dict | None = None + + +class ChannelConfigUpdate(BaseModel): + channel_type: str | None = None + name: str | None = None + config: dict | None = None + enabled: bool | None = None + + +class ChannelBindingCreate(BaseModel): + channel_type: str + channel_config_id: str | None = None + account_id: str | None = None + peer_kind: str | None = None + peer_id: str | None = None + agent_config_id: int + priority: int = 0 + dm_scope: str = "per-channel-peer" + guild_id: str | None = None + team_id: str | None = None + roles: list[str] | None = None + session_dm_scope: str | None = None + + +class ChannelBindingUpdate(BaseModel): + channel_type: str | None = None + channel_config_id: str | None = None + account_id: str | None = None + peer_kind: str | None = None + peer_id: str | None = None + agent_config_id: int | None = None + priority: int | None = None + dm_scope: str | None = None + guild_id: str | None = None + team_id: str | None = None + roles: list[str] | None = None + session_dm_scope: str | None = None + + +class ChannelUserMappingCreate(BaseModel): + channel_id: str + channel_user_id: str + internal_user_id: str + + +class ChannelThreadMappingCreate(BaseModel): + channel_id: str + channel_chat_id: str + internal_user_id: str + thread_id: str + agent_id: str | None = None + + +# ============================================================================= +# === 渠道配置 CRUD === +# ============================================================================= + + +@channel.get("/channels") +async def list_channels( + channel_type: str | None = Query(None, description="按渠道类型过滤"), + enabled: bool | None = Query(None, description="按启用状态过滤"), + current_user: User = Depends(get_admin_user), +): + return await list_channel_configs(channel_type=channel_type, enabled=enabled) + + +@channel.get("/channels/{config_id}") +async def get_channel(config_id: str, current_user: User = Depends(get_admin_user)): + config = await get_channel_config(config_id) + if not config: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道配置不存在") + return config + + +@channel.post("/channels", status_code=status.HTTP_201_CREATED) +async def create_channel( + data: ChannelConfigCreate, + current_user: User = Depends(get_admin_user), +): + return await create_channel_config( + { + "channel_type": data.channel_type, + "name": data.name, + "config": data.config or {}, + } + ) + + +@channel.put("/channels/{config_id}") +async def update_channel( + config_id: str, + data: ChannelConfigUpdate, + current_user: User = Depends(get_admin_user), +): + update_data = {k: v for k, v in data.model_dump().items() if v is not None} + config = await update_channel_config(config_id, update_data) + if not config: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道配置不存在") + return config + + +@channel.delete("/channels/{config_id}", status_code=status.HTTP_200_OK) +async def delete_channel(config_id: str, current_user: User = Depends(get_admin_user)): + deleted = await delete_channel_config(config_id) + if not deleted: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道配置不存在") + return {"success": True, "message": "渠道配置已删除"} + + +# ============================================================================= +# === 渠道绑定 CRUD === +# ============================================================================= + + +@channel.get("/bindings") +async def list_bindings( + agent_config_id: int | None = Query(None, description="按 AgentConfig ID 过滤"), + channel_type: str | None = Query(None, description="按渠道类型过滤"), + current_user: User = Depends(get_admin_user), +): + if channel_type: + return await list_channel_bindings_by_type(channel_type) + return await list_channel_bindings(agent_config_id=agent_config_id) + + +@channel.get("/bindings/{binding_id}") +async def get_binding(binding_id: str, current_user: User = Depends(get_admin_user)): + binding = await get_channel_binding(binding_id) + if not binding: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道绑定不存在") + return binding + + +@channel.post("/bindings", status_code=status.HTTP_201_CREATED) +async def create_binding( + data: ChannelBindingCreate, + current_user: User = Depends(get_admin_user), +): + return await create_channel_binding(data.model_dump()) + + +@channel.put("/bindings/{binding_id}") +async def update_binding( + binding_id: str, + data: ChannelBindingUpdate, + current_user: User = Depends(get_admin_user), +): + update_data = {k: v for k, v in data.model_dump().items() if v is not None} + binding = await update_channel_binding(binding_id, update_data) + if not binding: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道绑定不存在") + return binding + + +@channel.delete("/bindings/{binding_id}", status_code=status.HTTP_200_OK) +async def delete_binding(binding_id: str, current_user: User = Depends(get_admin_user)): + deleted = await delete_channel_binding(binding_id) + if not deleted: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道绑定不存在") + return {"success": True, "message": "渠道绑定已删除"} + + +# ============================================================================= +# === 配对管理 === +# ============================================================================= + + +@channel.get("/pairing") +async def list_pairing( + status_filter: str | None = Query(None, alias="status"), + current_user: User = Depends(get_admin_user), +): + return await list_channel_pairing_records(status=status_filter) + + +@channel.get("/pairing/agent/{agent_config_id}/code") +async def get_agent_pairing_code( + agent_config_id: int, + current_user: User = Depends(get_admin_user), +): + try: + return await generate_agent_pairing_code(agent_config_id) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +@channel.post("/pairing/{record_id}/approve") +async def approve_pairing( + record_id: str, + current_user: User = Depends(get_admin_user), +): + record = await approve_channel_pairing(record_id) + if not record: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="配对记录不存在") + return record + + +@channel.post("/pairing/{record_id}/reject") +async def reject_pairing( + record_id: str, + current_user: User = Depends(get_admin_user), +): + record = await reject_channel_pairing(record_id) + if not record: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="配对记录不存在") + return record + + +@channel.get("/pairing/{record_id}") +async def get_pairing( + record_id: str, + current_user: User = Depends(get_admin_user), +): + record = await get_channel_pairing_record(record_id) + if not record: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="配对记录不存在") + return record + + +@channel.delete("/pairing/{record_id}", status_code=status.HTTP_200_OK) +async def delete_pairing( + record_id: str, + current_user: User = Depends(get_admin_user), +): + deleted = await delete_channel_pairing_record(record_id) + if not deleted: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="配对记录不存在") + return {"success": True, "message": "配对记录已删除"} + + +# ============================================================================= +# === Gateway 健康检查 === +# ============================================================================= + + +@gateway_router.get("/health") +async def gateway_health(): + return await get_gateway_health() + + +@gateway_router.get("/healthz") +async def gateway_healthz(): + return await get_gateway_healthz() + + +@gateway_router.get("/health/ready") +async def gateway_readiness(): + from yuxi.channel.runtime.manager import gateway + + trace = gateway.get_startup_trace() + if trace is None: + return { + "ready": False, + "state": "not_ready", + "reason": "Gateway not started", + } + + success = bool(trace.get("completed_at")) and not trace.get("failed_at") + ready_state = trace.get("ready_state", "unknown") + + return { + "ready": success and ready_state in ("ready", "degraded"), + "state": ready_state, + "trace": trace, + } + + +@gateway_router.get("/health/boot") +async def gateway_boot_status(): + from yuxi.channel.runtime.manager import gateway + + return gateway.get_boot_status() + + +@gateway_router.get("/health/channels") +async def gateway_channels_health(): + from yuxi.channel.monitoring.readiness import readiness_checker + + result = await readiness_checker.check() + return { + "status": "healthy" if result.ready else "degraded", + "timestamp": result.checked_at, + "channels": [ + { + "key": c.key, + "healthy": c.healthy, + "reason": c.reason, + "detail": c.detail, + } + for c in result.channels + ], + "failing": result.failing, + "uptime_ms": result.uptime_ms, + } + + +@gateway_router.get("/health/channels/metrics") +async def gateway_channels_health_metrics(): + from yuxi.channel.monitoring.health_monitor import channel_health_monitor + + return channel_health_monitor.metrics.to_dict() + + +@gateway_router.get("/hooks") +async def gateway_hook_status(): + return await get_hook_status() + + +class InstallHookRequest(BaseModel): + hook_id: str + hook_md_content: str + target: str = "workspace" + + +@gateway_router.post("/hooks/install", status_code=status.HTTP_201_CREATED) +async def gateway_install_hook(req: InstallHookRequest, _user=Depends(get_admin_user)): + return await install_hook(req.hook_id, req.hook_md_content, req.target) + + +@gateway_router.delete("/hooks/{source}/{hook_id}") +async def gateway_uninstall_hook(source: str, hook_id: str, _user=Depends(get_admin_user)): + result = await uninstall_hook(source, hook_id) + if not result.get("success"): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=result.get("error")) + return result + + +@gateway_router.post("/hooks/reload") +async def gateway_reload_hooks(_user=Depends(get_admin_user)): + return await reload_hooks() + + +# ============================================================================= +# === 渠道诊断 === +# ============================================================================= + + +async def _resolve_config(channel_type: str, account_id: str) -> dict: + plugin = ChannelPluginRegistry.get(channel_type) + if plugin is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"渠道 {channel_type} 未注册") + return await plugin.resolve_account(account_id) + + +@channel.get("/{channel_type}/diagnose") +async def diagnose_channel( + channel_type: str, + account_id: str = Query("default"), + current_user: User = Depends(get_admin_user), +): + config = await _resolve_config(channel_type, account_id) + result = await doctor_runner.diagnose(channel_type, account_id, config) + return result.to_dict() + + +@channel.post("/{channel_type}/diagnose/repair") +async def repair_channel( + channel_type: str, + body: RepairRequest, + current_user: User = Depends(get_admin_user), +): + config = await _resolve_config(channel_type, body.account_id) + result = await doctor_runner.repair(channel_type, body.step_id, config) + return { + "step_id": result.step_id, + "success": result.success, + "message": result.message, + } + + +# ============================================================================= +# === Directory 目录查询 === +# ============================================================================= + + +async def _resolve_plugin_direct(config_id: str): + config = await get_channel_config(config_id) + if not config: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="渠道配置不存在") + channel_type = config["channel_type"] + plugin = ChannelPluginRegistry.get(channel_type) + if not plugin or not isinstance(plugin, DirectoryProtocol): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"渠道 {channel_type} 不支持目录查询", + ) + return plugin, config + + +def _plugin_config(config: dict) -> dict: + cfg = config.get("config") + return cfg if isinstance(cfg, dict) else {} + + +@channel.get("/{config_id}/directory/peers") +async def list_directory_peers( + config_id: str, + query: str = Query(default=""), + limit: int = Query(default=20, ge=1, le=200), + offset: int = Query(default=0, ge=0), + current_user: User = Depends(get_admin_user), +): + plugin, config = await _resolve_plugin_direct(config_id) + plugin_cfg = _plugin_config(config) + + if query: + params = DirectorySearchParams(query=query, kind="user", limit=limit, offset=offset) + result = await plugin.search_peers(plugin_cfg, params) + return {"items": [item.__dict__ for item in result.items], "total": result.total, "has_more": result.has_more} + + peers = await plugin.list_peers(plugin_cfg) + return {"items": [p.__dict__ for p in peers], "total": len(peers)} + + +@channel.get("/{config_id}/directory/groups") +async def list_directory_groups( + config_id: str, + query: str = Query(default=""), + limit: int = Query(default=20, ge=1, le=200), + offset: int = Query(default=0, ge=0), + current_user: User = Depends(get_admin_user), +): + plugin, config = await _resolve_plugin_direct(config_id) + plugin_cfg = _plugin_config(config) + + if query: + params = DirectorySearchParams(query=query, kind="group", limit=limit, offset=offset) + result = await plugin.search_groups(plugin_cfg, params) + return {"items": [item.__dict__ for item in result.items], "total": result.total, "has_more": result.has_more} + + groups = await plugin.list_groups(plugin_cfg) + return {"items": [g.__dict__ for g in groups], "total": len(groups)} + + +@channel.get("/{config_id}/directory/peers/{peer_id}") +async def get_directory_peer( + config_id: str, + peer_id: str, + current_user: User = Depends(get_admin_user), +): + plugin, config = await _resolve_plugin_direct(config_id) + peer = await plugin.get_peer(_plugin_config(config), peer_id) + if peer is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") + return peer.__dict__ + + +@channel.get("/{config_id}/directory/groups/{group_id}") +async def get_directory_group( + config_id: str, + group_id: str, + current_user: User = Depends(get_admin_user), +): + plugin, config = await _resolve_plugin_direct(config_id) + group = await plugin.get_group(_plugin_config(config), group_id) + if group is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="群组不存在") + return group.__dict__ + + +# ============================================================================= +# === 渠道用户映射 CRUD === +# ============================================================================= + + +@channel.get("/user-mappings/{mapping_id}") +async def get_user_mapping_endpoint(mapping_id: int, current_user: User = Depends(get_admin_user)): + mapping = await get_user_mapping(mapping_id) + if not mapping: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户映射不存在") + return mapping + + +@channel.get("/user-mappings") +async def find_user_mapping( + channel_id: str | None = Query(default=None), + channel_user_id: str | None = Query(default=None), + internal_user_id: str | None = Query(default=None), + current_user: User = Depends(get_admin_user), +): + if channel_id and channel_user_id: + mapping = await find_user_mapping_by_channel(channel_id, channel_user_id) + return {"items": [mapping] if mapping else [], "total": 1 if mapping else 0} + if internal_user_id: + mappings = await find_user_mappings_by_internal_user(internal_user_id) + return {"items": mappings, "total": len(mappings)} + mappings = await list_user_mappings() + return {"items": mappings, "total": len(mappings)} + + +@channel.post("/user-mappings", status_code=status.HTTP_201_CREATED) +async def create_user_mapping_endpoint( + data: ChannelUserMappingCreate, + current_user: User = Depends(get_admin_user), +): + return await create_user_mapping(data.model_dump()) + + +@channel.post("/user-mappings/resolve") +async def resolve_user_mapping( + channel_id: str = Query(...), + channel_user_id: str = Query(...), + current_user: User = Depends(get_admin_user), +): + return await resolve_channel_user(channel_id, channel_user_id) + + +@channel.delete("/user-mappings/{mapping_id}", status_code=status.HTTP_200_OK) +async def delete_user_mapping_endpoint(mapping_id: int, current_user: User = Depends(get_admin_user)): + deleted = await delete_user_mapping(mapping_id) + if not deleted: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户映射不存在") + return {"success": True, "message": "用户映射已删除"} + + +# ============================================================================= +# === 渠道会话线程映射 CRUD === +# ============================================================================= + + +@channel.get("/thread-mappings/{mapping_id}") +async def get_thread_mapping_endpoint(mapping_id: int, current_user: User = Depends(get_admin_user)): + mapping = await get_thread_mapping(mapping_id) + if not mapping: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="线程映射不存在") + return mapping + + +@channel.get("/thread-mappings") +async def find_thread_mapping( + channel_id: str | None = Query(default=None), + channel_chat_id: str | None = Query(default=None), + internal_user_id: str | None = Query(default=None), + thread_id: str | None = Query(default=None), + current_user: User = Depends(get_admin_user), +): + if channel_id and channel_chat_id and internal_user_id: + mapping = await find_thread_mapping_by_channel_chat( + channel_id, + channel_chat_id, + internal_user_id, + ) + return {"items": [mapping] if mapping else [], "total": 1 if mapping else 0} + if thread_id: + mapping = await find_thread_mapping_by_thread_id(thread_id) + return {"items": [mapping] if mapping else [], "total": 1 if mapping else 0} + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="请提供 channel_id+channel_chat_id+internal_user_id 或 thread_id", + ) + + +@channel.post("/thread-mappings", status_code=status.HTTP_201_CREATED) +async def create_thread_mapping_endpoint( + data: ChannelThreadMappingCreate, + current_user: User = Depends(get_admin_user), +): + return await create_thread_mapping(data.model_dump()) + + +@channel.post("/thread-mappings/resolve") +async def resolve_thread_mapping( + channel_id: str = Query(...), + channel_chat_id: str = Query(...), + internal_user_id: str = Query(...), + agent_id: str | None = Query(default=None), + current_user: User = Depends(get_admin_user), +): + return await resolve_channel_thread(channel_id, channel_chat_id, internal_user_id, agent_id) + + +@channel.delete("/thread-mappings/{mapping_id}", status_code=status.HTTP_200_OK) +async def delete_thread_mapping_endpoint(mapping_id: int, current_user: User = Depends(get_admin_user)): + deleted = await delete_thread_mapping(mapping_id) + if not deleted: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="线程映射不存在") + return {"success": True, "message": "线程映射已删除"} + + +@channel.get("/ui-schemas") +async def list_channel_ui_schemas(current_user: User = Depends(get_admin_user)): + from yuxi.channel.ui.schema import list_all_ui_metadata + + schemas = [m.to_dict() for m in list_all_ui_metadata()] + return {"schemas": schemas, "total": len(schemas)} + + +@channel.get("/ui-schemas/{channel_type}") +async def get_channel_ui_schema(channel_type: str, current_user: User = Depends(get_admin_user)): + from yuxi.channel.ui.schema import get_channel_ui_metadata + + metadata = get_channel_ui_metadata(channel_type) + if metadata is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"渠道类型 '{channel_type}' 的 UI Schema 不存在", + ) + return metadata.to_dict() diff --git a/backend/server/routers/gateway_sse_router.py b/backend/server/routers/gateway_sse_router.py new file mode 100644 index 00000000..13fc1eae --- /dev/null +++ b/backend/server/routers/gateway_sse_router.py @@ -0,0 +1,22 @@ +import logging + +from fastapi import APIRouter, Depends, Query + +from server.utils.auth_middleware import get_admin_user +from yuxi.channel.gateway.polling import polling_fallback +from yuxi.storage.postgres.models_business import User + +logger = logging.getLogger(__name__) + +sse_polling_router = APIRouter(prefix="/poll", tags=["sse-polling"]) + + +@sse_polling_router.get("") +async def sse_poll( + session_id: str = Query(..., description="轮询会话 ID"), + current_user: User = Depends(get_admin_user), +): + events = await polling_fallback.poll(session_id) + if not events: + return {"status": "empty", "data": None} + return {"status": "ok", "data": events} diff --git a/backend/server/routers/gateway_ws_router.py b/backend/server/routers/gateway_ws_router.py new file mode 100644 index 00000000..2f89a525 --- /dev/null +++ b/backend/server/routers/gateway_ws_router.py @@ -0,0 +1,12 @@ +from fastapi import APIRouter, Query, WebSocket +from yuxi.channel.gateway.server import gateway_ws_server + +router = APIRouter(tags=["Gateway WebSocket"]) + + +@router.websocket("/ws/gateway") +async def gateway_ws_endpoint( + ws: WebSocket, + token: str | None = Query(None), +): + await gateway_ws_server.handle_connection(ws, token) diff --git a/backend/server/routers/webhook_router.py b/backend/server/routers/webhook_router.py new file mode 100644 index 00000000..e1e50df8 --- /dev/null +++ b/backend/server/routers/webhook_router.py @@ -0,0 +1,56 @@ +import logging + +from fastapi import APIRouter, Request, Response + +from yuxi.channel.gateway.routes import webhook_registry + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/webhook", tags=["webhook"]) + + +def _extract_signature(headers: dict[str, str]) -> str | None: + sig = headers.get("x-hub-signature-256") or headers.get("x-signature") + if sig and sig.startswith("sha256="): + sig = sig[7:] + return sig + + +@router.api_route( + "/{channel_type}", + methods=["POST", "PUT", "PATCH"], + summary="接收渠道 Webhook 回调", +) +async def webhook_receive( + channel_type: str, + request: Request, +): + body = await request.body() + method = request.method + content_type = request.headers.get("content-type") + headers = {k.lower(): v for k, v in request.headers.items()} + + guard_result, handler_result = await webhook_registry.dispatch_guarded( + channel_type=channel_type, + method=method, + content_type=content_type, + body=body, + signature=_extract_signature(headers), + extra_headers=headers, + ) + + if guard_result is not None: + logger.warning( + "Webhook guard blocked: channel=%s method=%s status=%d error=%s", + channel_type, + method, + guard_result.http_status, + guard_result.error_code, + ) + return Response( + content=guard_result.error_message or "", + status_code=guard_result.http_status, + media_type="text/plain", + ) + + return handler_result or {"status": "ok"} diff --git a/backend/server/utils/lifespan.py b/backend/server/utils/lifespan.py index 10380f25..1b486081 100644 --- a/backend/server/utils/lifespan.py +++ b/backend/server/utils/lifespan.py @@ -80,6 +80,13 @@ async def lifespan(app: FastAPI): except Exception as e: logger.error(f"Failed to initialize sandbox provider during startup: {e}") + try: + from yuxi.channel.gateway.rpc_handlers import register_all_handlers + register_all_handlers() + logger.info("Gateway RPC handlers registered (%d methods)", 42) + except Exception as e: + logger.error(f"Failed to register gateway RPC handlers: {e}") + # ========================================================= # 2. 核心修复：在这里执行一次 setup()，建完表就拉倒 # ========================================================= @@ -87,6 +94,19 @@ async def lifespan(app: FastAPI): await checkpointer.setup() print("LangGraph Checkpoint tables verified/created!") + try: + from yuxi.channel.runtime.manager import gateway + + await gateway.boot() + trace = gateway.get_startup_trace() + if not trace.get("success"): + logger.error("ChannelGateway boot failed: %s", trace.get("error")) + else: + logger.info("ChannelGateway ready, %d channels, %.2fs", + len(gateway.list_channels()), trace.get("duration_seconds", 0)) + except Exception as e: + logger.error("ChannelGateway boot error: %s", e) + await tasker.start() logger.info(f""" @@ -102,6 +122,13 @@ async def lifespan(app: FastAPI): logger.info("Yuxi backend startup complete") yield await tasker.shutdown() + + try: + from yuxi.channel.gateway.server import gateway_ws_server + await gateway_ws_server.shutdown() + except Exception as e: + logger.error(f"Gateway WS shutdown error: {e}") + shutdown_sandbox_provider() await close_queue_clients() await pg_manager.close()