diff --git a/backend/server/main.py b/backend/server/main.py index cf6a26b1..1af66310 100644 --- a/backend/server/main.py +++ b/backend/server/main.py @@ -23,6 +23,7 @@ from starlette.middleware.base import BaseHTTPMiddleware from server.routers import router from server.routers.ws_chat_router import ws_chat from yuxi.channels.adapters.slack.http_handler import slack_webhook +from yuxi.channels.adapters.nostr.profile_api import profile_router from server.utils.lifespan import lifespan from server.utils.auth_middleware import is_public_path from server.utils.common_utils import setup_logging @@ -46,6 +47,8 @@ app.include_router(router, prefix="/api") app.include_router(ws_chat) # Slack HTTP Webhook (单独注册以处理 URL verification challenge) app.include_router(slack_webhook) +# Nostr Profile API +app.include_router(profile_router) # CORS 设置 app.add_middleware( diff --git a/backend/server/routers/channels_router.py b/backend/server/routers/channels_router.py index d694d106..562d7509 100644 --- a/backend/server/routers/channels_router.py +++ b/backend/server/routers/channels_router.py @@ -6,8 +6,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from yuxi.channels.manager import channel_manager -from yuxi.channels.registry import BUILTIN_ADAPTERS +from yuxi.channels.manager import get_channel_manager from yuxi.storage.postgres.models_business import User from yuxi.storage.postgres.models_channels import ChannelThreadMapping, ChannelUserMapping from yuxi.utils.logging_config import logger @@ -23,7 +22,7 @@ def _make_response(data: Any = None, code: int = 0, message: str = "ok") -> dict async def _check_rate_limit(user_id: str, key: str, max_req: int, window: int) -> None: rate_key = f"{key}:{user_id}" - allowed = await channel_manager.check_rate_limit(rate_key, max_req, window) + allowed = await get_channel_manager().check_rate_limit(rate_key, max_req, window) if not allowed: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, @@ -33,7 +32,7 @@ async def _check_rate_limit(user_id: str, key: str, max_req: int, window: int) - @channels.get("/status") async def list_channels_status(current_user: User = Depends(get_admin_user)): - status_data = await channel_manager.get_channel_status() + status_data = await get_channel_manager().get_channel_status() return _make_response(data=status_data) @@ -64,6 +63,28 @@ async def list_channels_for_action( return _make_response(data=result) +@channels.post("/register") +async def register_channel( + body: dict[str, Any], + current_user: User = Depends(get_admin_user), +): + channel_type = body.get("channel_type") + if not channel_type: + raise HTTPException(status_code=400, detail="channel_type is required") + + config = body.get("config", {}) + manager = get_channel_manager() + + try: + await manager.register_channel(channel_type, config) + return _make_response( + data={"channel_id": channel_type, "registered": True}, + message="渠道注册成功", + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + @channels.get("/{channel_id}/actions") async def get_channel_actions( channel_id: str, @@ -92,10 +113,7 @@ async def get_channel_actions( return _make_response( data={ "channel_id": channel_id, - "actions": [ - {"action": a, "status": "supported", "reason": "", "impl": ""} - for a in supported - ], + "actions": [{"action": a, "status": "supported", "reason": "", "impl": ""} for a in supported], } ) @@ -105,7 +123,7 @@ async def get_channel_status( channel_id: str, current_user: User = Depends(get_admin_user), ): - status_data = await channel_manager.get_channel_status(channel_id) + status_data = await get_channel_manager().get_channel_status(channel_id) if status_data.get("status") == "not_found": raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") return _make_response(data=status_data) @@ -116,26 +134,26 @@ async def start_channel( channel_id: str, current_user: User = Depends(get_admin_user), ): - await _check_rate_limit(str(current_user.id), "channel_start", 1, 1) + await _check_rate_limit(str(current_user.id), "channel_start", 3, 3) - registered_ids = set(channel_manager._registry.list_channels()) | set(BUILTIN_ADAPTERS.keys()) - if channel_id not in registered_ids: - return _make_response(code=-1, message=f"Channel '{channel_id}' is not registered") + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") - if channel_id in channel_manager._adapters: + if manager.is_running(channel_id): return _make_response(code=409, message=f"Channel '{channel_id}' is already running") try: - await channel_manager.start_channel(channel_id) + await manager.start_channel(channel_id) return _make_response( data={"channel_id": channel_id, "status": "connecting"}, message="Channel start initiated", ) except ValueError as e: - return _make_response(code=-1, message=str(e)) + raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Failed to start channel {channel_id}: {e}") - return _make_response(code=-1, message=f"Failed to start channel: {e}") + raise HTTPException(status_code=500, detail=f"Failed to start channel: {e}") @channels.post("/{channel_id}/stop") @@ -143,12 +161,13 @@ async def stop_channel( channel_id: str, current_user: User = Depends(get_admin_user), ): - await _check_rate_limit(str(current_user.id), "channel_stop", 1, 1) + await _check_rate_limit(str(current_user.id), "channel_stop", 3, 3) - if channel_id not in channel_manager._adapters: + manager = get_channel_manager() + if not manager.is_running(channel_id): return _make_response(code=409, message=f"Channel '{channel_id}' is not running") - await channel_manager.stop_channel(channel_id) + await manager.stop_channel(channel_id) return _make_response( data={"channel_id": channel_id, "status": "disconnected"}, message="Channel stopped", @@ -160,21 +179,21 @@ async def restart_channel( channel_id: str, current_user: User = Depends(get_admin_user), ): - await _check_rate_limit(str(current_user.id), "channel_restart", 1, 1) + await _check_rate_limit(str(current_user.id), "channel_restart", 3, 3) - registered_ids = set(channel_manager._registry.list_channels()) | set(BUILTIN_ADAPTERS.keys()) - if channel_id not in registered_ids: - return _make_response(code=-1, message=f"Channel '{channel_id}' is not registered") + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") try: - await channel_manager.restart_channel(channel_id) + await manager.restart_channel(channel_id) return _make_response( data={"channel_id": channel_id, "status": "connecting"}, message="Channel restart initiated", ) except Exception as e: logger.error(f"Failed to restart channel {channel_id}: {e}") - return _make_response(code=-1, message=f"Failed to restart channel: {e}") + raise HTTPException(status_code=500, detail=f"Failed to restart channel: {e}") @channels.put("/{channel_id}/config") @@ -183,18 +202,18 @@ async def update_channel_config( config_updates: dict[str, Any], current_user: User = Depends(get_admin_user), ): - registered_ids = set(channel_manager._registry.list_channels()) | set(BUILTIN_ADAPTERS.keys()) - if channel_id not in registered_ids: - return _make_response(code=-1, message=f"Channel '{channel_id}' is not registered") + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") try: - result = await channel_manager.update_channel_config(channel_id, config_updates) + result = await manager.update_channel_config(channel_id, config_updates) return _make_response( data=result, message="Configuration updated. Restart channel to apply changes.", ) except Exception as e: - return _make_response(code=-1, message=str(e)) + raise HTTPException(status_code=400, detail=str(e)) @channels.post("/{channel_id}/test") @@ -202,11 +221,11 @@ async def test_channel( channel_id: str, current_user: User = Depends(get_admin_user), ): - registered_ids = set(channel_manager._registry.list_channels()) | set(BUILTIN_ADAPTERS.keys()) - if channel_id not in registered_ids: - return _make_response(code=-1, message=f"Channel '{channel_id}' is not registered") + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") - result = await channel_manager.test_channel(channel_id) + result = await manager.test_channel(channel_id) return _make_response(data=result) @@ -230,9 +249,9 @@ async def get_channel_stats( period_map = {"1d": 1, "7d": 7, "30d": 30} days = period_map.get(period, 7) - registered_ids = set(channel_manager._registry.list_channels()) | set(BUILTIN_ADAPTERS.keys()) - if channel_id not in registered_ids: - return _make_response(code=-1, message=f"Channel '{channel_id}' is not registered") + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") now = utc_now_naive() since = now - timedelta(days=days) @@ -299,14 +318,19 @@ async def get_channel_stats( daily_trend: list[dict] = [] daily_success_rate: list[dict] = [] if granularity == "day": - daily_query = select( - func.date(ChannelMsgRecord.created_at).label("date"), - func.count().label("count"), - func.avg(ChannelMsgRecord.response_time_ms).label("avg_ms"), - ).where( - ChannelMsgRecord.channel_id == channel_id, - ChannelMsgRecord.created_at >= since, - ).group_by(func.date(ChannelMsgRecord.created_at)).order_by(func.date(ChannelMsgRecord.created_at)) + daily_query = ( + select( + func.date(ChannelMsgRecord.created_at).label("date"), + func.count().label("count"), + func.avg(ChannelMsgRecord.response_time_ms).label("avg_ms"), + ) + .where( + ChannelMsgRecord.channel_id == channel_id, + ChannelMsgRecord.created_at >= since, + ) + .group_by(func.date(ChannelMsgRecord.created_at)) + .order_by(func.date(ChannelMsgRecord.created_at)) + ) daily_result = await db.execute(daily_query) daily_trend = [ @@ -314,19 +338,24 @@ async def get_channel_stats( for r in daily_result.all() ] - success_rate_query = select( - func.date(ChannelMsgRecord.created_at).label("date"), - func.count().label("total"), - func.sum( - func.cast( - (ChannelMsgRecord.status == "success").cast(func.Integer), - func.Integer, - ) - ).label("success"), - ).where( - ChannelMsgRecord.channel_id == channel_id, - ChannelMsgRecord.created_at >= since, - ).group_by(func.date(ChannelMsgRecord.created_at)).order_by(func.date(ChannelMsgRecord.created_at)) + success_rate_query = ( + select( + func.date(ChannelMsgRecord.created_at).label("date"), + func.count().label("total"), + func.sum( + func.cast( + (ChannelMsgRecord.status == "success").cast(func.Integer), + func.Integer, + ) + ).label("success"), + ) + .where( + ChannelMsgRecord.channel_id == channel_id, + ChannelMsgRecord.created_at >= since, + ) + .group_by(func.date(ChannelMsgRecord.created_at)) + .order_by(func.date(ChannelMsgRecord.created_at)) + ) success_rate_result = await db.execute(success_rate_query) daily_success_rate = [ @@ -386,10 +415,13 @@ async def get_channel_user_mapping( raise HTTPException(status_code=404, detail="User mapping not found") thread_result = await db.execute( - select(ChannelThreadMapping).where( + select(ChannelThreadMapping) + .where( ChannelThreadMapping.channel_id == channel_id, ChannelThreadMapping.internal_user_id == mapping.internal_user_id, - ).order_by(ChannelThreadMapping.last_active_at.desc()).limit(10) + ) + .order_by(ChannelThreadMapping.last_active_at.desc()) + .limit(10) ) recent_chats = [ { @@ -529,25 +561,37 @@ async def export_channel_messages( output = StringIO() writer = csv.writer(output) - writer.writerow([ - "id", "created_at", "sender_user_id", "content_preview", "content_type", - "status", "response_time_ms", "reply_content_preview", "error_message", - "chat_id", "chat_type" - ]) + writer.writerow( + [ + "id", + "created_at", + "sender_user_id", + "content_preview", + "content_type", + "status", + "response_time_ms", + "reply_content_preview", + "error_message", + "chat_id", + "chat_type", + ] + ) for r in rows: - writer.writerow([ - r.id, - r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else "", - r.sender_user_id, - r.content_preview, - r.content_type, - r.status, - r.response_time_ms or "", - r.reply_content_preview or "", - r.error_message or "", - r.chat_id, - r.chat_type, - ]) + writer.writerow( + [ + r.id, + r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else "", + r.sender_user_id, + r.content_preview, + r.content_type, + r.status, + r.response_time_ms or "", + r.reply_content_preview or "", + r.error_message or "", + r.chat_id, + r.chat_type, + ] + ) output.seek(0) return StreamingResponse( @@ -565,9 +609,7 @@ async def get_channel_policy( ): from yuxi.storage.postgres.models_channels import ChannelPolicyConfig - result = await db.execute( - select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id) - ) + result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) policy = result.scalar_one_or_none() if not policy: return _make_response( @@ -595,15 +637,18 @@ async def update_channel_policy( from yuxi.storage.postgres.models_channels import ChannelPolicyConfig from yuxi.utils.datetime_utils import utc_now_naive - result = await db.execute( - select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id) - ) + result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) policy = result.scalar_one_or_none() if policy: for key in [ - "group_chat_mode", "whitelist_ids", "welcome_message", - "work_hours_start", "work_hours_end", "timezone_offset", "off_hours_reply", + "group_chat_mode", + "whitelist_ids", + "welcome_message", + "work_hours_start", + "work_hours_end", + "timezone_offset", + "off_hours_reply", ]: if key in policy_updates: setattr(policy, key, policy_updates[key]) @@ -611,10 +656,20 @@ async def update_channel_policy( else: policy = ChannelPolicyConfig( channel_id=channel_id, - **{k: v for k, v in policy_updates.items() if k in [ - "group_chat_mode", "whitelist_ids", "welcome_message", - "work_hours_start", "work_hours_end", "timezone_offset", "off_hours_reply", - ]} + **{ + k: v + for k, v in policy_updates.items() + if k + in [ + "group_chat_mode", + "whitelist_ids", + "welcome_message", + "work_hours_start", + "work_hours_end", + "timezone_offset", + "off_hours_reply", + ] + }, ) db.add(policy) @@ -649,12 +704,11 @@ async def update_channel_routing( ): from yuxi.storage.postgres.models_channels import ChannelRoutingRule - await db.execute( - select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id) + existing = ( + (await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id))) + .scalars() + .all() ) - existing = (await db.execute( - select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id) - )).scalars().all() for rule in existing: await db.delete(rule) @@ -662,7 +716,7 @@ async def update_channel_routing( rule = ChannelRoutingRule( channel_id=channel_id, command=item.get("command", ""), - agent_config_id=item.get("agent_config_id", 0), + agent_config_id=str(item.get("agent_config_id", "0")), priority=item.get("priority", len(routing_updates) - i), ) db.add(rule) @@ -696,32 +750,38 @@ async def get_channel_config_schema( elif isinstance(value, (int, float)): field_type = "number" is_credential = any(k in key for k in ("token", "secret", "key", "app_id")) - fields.append({ - "key": key, - "label": key.replace("_", " ").title(), - "type": field_type, - "default": value, - "required": "token" in key or "secret" in key, - "group": "credentials" if is_credential else "general", - }) + fields.append( + { + "key": key, + "label": key.replace("_", " ").title(), + "type": field_type, + "default": value, + "required": "token" in key or "secret" in key, + "group": "credentials" if is_credential else "general", + } + ) - fields.append({ - "key": "agent_config_id", - "label": "默认 Agent", - "type": "agent_select", - "required": False, - "group": "routing", - }) + fields.append( + { + "key": "agent_config_id", + "label": "默认 Agent", + "type": "agent_select", + "required": False, + "group": "routing", + } + ) - return _make_response(data={ - "channel_type": channel_id, - "fields": fields, - "groups": [ - {"key": "credentials", "label": "认证凭据"}, - {"key": "general", "label": "通用设置"}, - {"key": "routing", "label": "路由配置"}, - ], - }) + return _make_response( + data={ + "channel_type": channel_id, + "fields": fields, + "groups": [ + {"key": "credentials", "label": "认证凭据"}, + {"key": "general", "label": "通用设置"}, + {"key": "routing", "label": "路由配置"}, + ], + } + ) return _make_response(data={"channel_type": channel_id, "fields": [], "groups": []}) @@ -734,10 +794,13 @@ async def get_channel_chat_mapping( db: AsyncSession = Depends(get_db), ): result = await db.execute( - select(ChannelThreadMapping).where( + select(ChannelThreadMapping) + .where( ChannelThreadMapping.channel_id == channel_id, ChannelThreadMapping.channel_chat_id == chat_id, - ).order_by(ChannelThreadMapping.last_active_at.desc()).limit(1) + ) + .order_by(ChannelThreadMapping.last_active_at.desc()) + .limit(1) ) mapping = result.scalar_one_or_none() if not mapping: @@ -753,3 +816,16 @@ async def get_channel_chat_mapping( "last_active_at": mapping.last_active_at.strftime("%Y-%m-%dT%H:%M:%SZ") if mapping.last_active_at else None, } ) + + +@channels.delete("/{channel_id}") +async def unregister_channel( + channel_id: str, + current_user: User = Depends(get_admin_user), +): + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found") + + await manager.unregister_channel(channel_id) + return _make_response(data={"channel_id": channel_id, "unregistered": True}, message="渠道已注销") diff --git a/backend/server/routers/dashboard_router.py b/backend/server/routers/dashboard_router.py index 2018084e..1c7fb2aa 100644 --- a/backend/server/routers/dashboard_router.py +++ b/backend/server/routers/dashboard_router.py @@ -647,6 +647,7 @@ class FeedbackListItem(BaseModel): avatar: str | None rating: str reason: str | None + channel: str created_at: str message_content: str conversation_title: str | None @@ -701,6 +702,7 @@ async def get_all_feedbacks( "avatar": user.avatar if user else None, "rating": feedback.rating, "reason": feedback.reason, + "channel": feedback.channel, "created_at": feedback.created_at.isoformat(), "message_content": message.content, "conversation_title": conversation.title, diff --git a/backend/server/routers/webhook_router.py b/backend/server/routers/webhook_router.py index 3070d00a..05c5a885 100644 --- a/backend/server/routers/webhook_router.py +++ b/backend/server/routers/webhook_router.py @@ -3,9 +3,9 @@ from __future__ import annotations import asyncio from fastapi import APIRouter, Request, status -from fastapi.responses import JSONResponse +from fastapi.responses import JSONResponse, Response -from yuxi.channels.manager import channel_manager +from yuxi.channels.manager import get_channel_manager from yuxi.channels.registry import BUILTIN_ADAPTERS from yuxi.utils.logging_config import logger @@ -14,7 +14,7 @@ webhook = APIRouter(prefix="/webhook", tags=["webhook"]) @webhook.post("/feishu") async def feishu_webhook(request: Request): - adapter = channel_manager._adapters.get("feishu") + adapter = get_channel_manager()._adapters.get("feishu") if not adapter: return JSONResponse( @@ -58,9 +58,95 @@ async def feishu_webhook(request: Request): ) +@webhook.api_route("/wechat", methods=["GET", "POST"]) +async def wechat_webhook(request: Request): + adapter = get_channel_manager()._adapters.get("wechat") + + if not adapter: + return JSONResponse( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + content={"code": 503, "data": None, "message": "WeChat channel is not running"}, + ) + + if request.method == "GET": + return await _handle_wechat_url_verification(adapter, request) + + body = await request.body() + headers = dict(request.headers) + source_ip = request.client.host if request.client else "unknown" + + max_body_bytes = getattr(adapter, "_MAX_WEBHOOK_BODY_BYTES", 256 * 1024) + if len(body) > max_body_bytes: + logger.warning(f"[WeChatWebhook] Body too large: {len(body)} > {max_body_bytes}") + return JSONResponse( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + content={"code": 413, "message": "Request body too large"}, + ) + + try: + if not await adapter.verify_webhook_signature(headers, body): + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content={"code": 403, "data": None, "message": "Invalid webhook signature"}, + ) + except NotImplementedError: + pass + + try: + parsed_dict = adapter._parse_webhook_body(body) + passive_reply = await adapter.try_passive_reply(parsed_dict) + message = adapter.normalize_inbound(parsed_dict) + except Exception as e: + logger.error(f"[WeChatWebhook] Failed to parse message: {e}") + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content={"code": 400, "data": None, "message": f"Failed to parse message: {e}"}, + ) + + asyncio.create_task(adapter._handle_message(message)) + if passive_reply is not None: + return Response(content=passive_reply, media_type="application/xml") + return JSONResponse(content={"code": 0, "data": None, "message": "ok"}) + + +async def _handle_wechat_url_verification(adapter, request: Request): + from yuxi.channels.adapters.wechat.mp.crypto import verify_url_echostr + from yuxi.channels.adapters.wechat.wecom.crypto import verify_url_signature + + params = dict(request.query_params) + signature = params.get("signature", params.get("msg_signature", "")) + timestamp = params.get("timestamp", "") + nonce = params.get("nonce", "") + echostr = params.get("echostr", "") + token = adapter.config.get("token", "") + + if not token or not echostr: + logger.warning("[WeChatWebhook] URL verification missing token or echostr") + return JSONResponse( + status_code=status.HTTP_400_BAD_REQUEST, + content={"code": 400, "message": "Missing token or echostr"}, + ) + + mode = getattr(adapter, "_mode", "personal") + if mode == "wecom": + ok, result = verify_url_signature(token, timestamp, nonce, echostr, signature) + else: + ok, result = verify_url_echostr(token, timestamp, nonce, echostr, signature) + + if not ok: + logger.warning("[WeChatWebhook] URL verification signature mismatch") + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content={"code": 403, "message": "Signature verification failed"}, + ) + + logger.info(f"[WeChatWebhook] URL verification succeeded for mode={mode}") + return JSONResponse(content=result) + + @webhook.post("/{channel_type}") async def webhook_dispatch(channel_type: str, request: Request): - adapter = channel_manager._adapters.get(channel_type) + adapter = get_channel_manager()._adapters.get(channel_type) adapter_cls = BUILTIN_ADAPTERS.get(channel_type) if not adapter: @@ -124,7 +210,7 @@ async def webhook_dispatch(channel_type: str, request: Request): return JSONResponse(content={"code": 0, "data": None, "message": "ok"}) else: message = adapter.normalize_inbound(body_data) - except asyncio.TimeoutError: + except TimeoutError: logger.error(f"Webhook processing timed out for {channel_type}") return JSONResponse( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, diff --git a/backend/server/routers/ws_chat_router.py b/backend/server/routers/ws_chat_router.py index 9616795c..e09edc7d 100644 --- a/backend/server/routers/ws_chat_router.py +++ b/backend/server/routers/ws_chat_router.py @@ -8,7 +8,7 @@ import uuid from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query from jose import JWTError -from yuxi.channels.manager import channel_manager +from yuxi.channels.manager import get_channel_manager from yuxi.channels.models import ( ChannelIdentity, ChannelMessage, @@ -53,8 +53,17 @@ class ConnectionRegistry: def count(self, user_id: str) -> int: return len(self._connections.get(user_id, [])) + async def broadcast_all(self, message: dict) -> None: + for ws_list in list(self._connections.values()): + for ws in ws_list: + try: + await ws.send_json(message) + except Exception: + pass + connections = ConnectionRegistry() +get_channel_manager().set_ws_broadcast(connections.broadcast_all) @ws_chat.websocket("/ws/chat/{client_id}") @@ -171,7 +180,7 @@ async def websocket_chat(ws: WebSocket, client_id: str, token: str = Query(None) }) try: - await channel_manager.dispatch_inbound(message) + await get_channel_manager().dispatch_inbound(message) except Exception as e: logger.error(f"WS message processing error: {e}") await ws.send_json({ diff --git a/backend/server/utils/lifespan.py b/backend/server/utils/lifespan.py index de0950b0..32f7f76c 100644 --- a/backend/server/utils/lifespan.py +++ b/backend/server/utils/lifespan.py @@ -13,7 +13,7 @@ from yuxi.storage.postgres.manager import pg_manager from yuxi.knowledge import knowledge_base from yuxi.utils import logger from yuxi.agents.backends.sandbox import init_sandbox_provider, shutdown_sandbox_provider -from yuxi.channels.manager import channel_manager +from yuxi.channels.manager import get_channel_manager from yuxi import get_version @@ -90,8 +90,8 @@ async def lifespan(app: FastAPI): # 多渠道网关初始化 try: - await channel_manager.initialize() - app.state.channel_manager = channel_manager + await get_channel_manager().initialize() + app.state.channel_manager = get_channel_manager() except Exception as e: logger.error(f"Failed to initialize channel manager during startup: {e}")