""" Agent 管理 API(管理员) Phase 2.3: AI PaaS Agent Management """ from fastapi import APIRouter, Depends, HTTPException, Query, Body from sqlalchemy.orm import Session from sqlalchemy import or_, func from pydantic import BaseModel from typing import Optional, List from ...database import get_db from ...models import MarketApp, EvolutionSkill, AIModel, User, ModelProvider from ...auth import get_current_admin from ...middleware.log_service import LogService import uuid from datetime import datetime router = APIRouter() # ============================================================ # Pydantic 模型 # ============================================================ class AgentConfig(BaseModel): system_prompt: Optional[str] = None temperature: Optional[float] = None max_tokens: Optional[int] = None model_id: Optional[str] = None skills: Optional[List[str]] = None # skill IDs tools: Optional[List[str]] = None # MCP server IDs memory_enabled: Optional[bool] = False reasoning_enabled: Optional[bool] = False class AgentCreate(BaseModel): name: str description: Optional[str] = None icon: Optional[str] = None price: float = 0.0 is_member_free: bool = False config: Optional[AgentConfig] = None class AgentUpdate(BaseModel): name: Optional[str] = None description: Optional[str] = None icon: Optional[str] = None price: Optional[float] = None is_member_free: Optional[bool] = None config: Optional[AgentConfig] = None status: Optional[str] = None # active/inactive tutorial: Optional[str] = None # ============================================================ # 辅助函数 # ============================================================ def agent_to_dict(a: MarketApp) -> dict: """将 MarketApp (agent) 转换为字典""" # 解析 skill_content 中的配置 import json config = {} if a.skill_content: try: config = json.loads(a.skill_content) except Exception: pass return { "id": a.id, "name": a.name, "type": a.type, "source": a.source, "publisher_id": a.publisher_id, "publisher_name": a.publisher_name, "description": a.description, "icon": a.icon, "tutorial": a.tutorial, "price": float(a.price) if a.price else 0, "is_member_free": a.is_member_free, "join_promotion": a.join_promotion, "status": a.status, "rating_avg": float(a.rating_avg) if a.rating_avg else 0, "review_count": a.review_count, "install_count": a.install_count, "agent_endpoint": a.agent_endpoint, "skill_content": a.skill_content, "config": config, "orchestration": a.orchestration, "created_at": a.created_at.isoformat() if a.created_at else None, "updated_at": a.updated_at.isoformat() if a.updated_at else None, } # ============================================================ # 端点 # ============================================================ @router.get("/", summary="获取 Agent 列表") async def list_agents( keyword: str = Query(None, description="搜索关键词"), status: str = Query(None, description="状态: pending/published/removed"), source: str = Query(None, description="来源: official/user"), page: int = Query(1, ge=1), page_size: int = Query(15, ge=1, le=100), admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取所有 Agent 列表(分页、筛选)""" query = db.query(MarketApp).filter(MarketApp.type == "agent") if keyword: query = query.filter( or_( MarketApp.name.ilike(f"%{keyword}%"), MarketApp.description.ilike(f"%{keyword}%"), MarketApp.publisher_name.ilike(f"%{keyword}%"), ) ) if status: query = query.filter(MarketApp.status == status) if source: query = query.filter(MarketApp.source == source) total = query.count() agents = query.order_by(MarketApp.created_at.desc()).offset( (page - 1) * page_size ).limit(page_size).all() return { "code": 0, "data": { "total": total, "page": page, "page_size": page_size, "items": [agent_to_dict(a) for a in agents], } } @router.get("/stats/overview", summary="Agent 统计概览") async def get_agent_stats( admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取 Agent 统计数据""" base = db.query(MarketApp).filter(MarketApp.type == "agent") total = base.count() published = base.filter(MarketApp.status == "published").count() pending = base.filter(MarketApp.status == "pending").count() official = base.filter(MarketApp.source == "official").count() user_created = base.filter(MarketApp.source == "user").count() # 安装量统计 install_base = db.query(func.sum(MarketApp.install_count)).filter( MarketApp.type == "agent" ).scalar() or 0 # 按类型分布 by_source = { "official": official, "user": user_created, } # 评分统计 avg_rating = db.query(func.avg(MarketApp.rating_avg)).filter( MarketApp.type == "agent", MarketApp.rating_avg > 0, ).scalar() or 0 return { "code": 0, "data": { "total_agents": total, "published_agents": published, "pending_agents": pending, "official_agents": official, "user_agents": user_created, "total_installs": int(install_base), "average_rating": round(float(avg_rating), 2), "by_source": by_source, } } @router.get("/{agent_id}", summary="获取 Agent 详情") async def get_agent( agent_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取单个 Agent 的详细信息""" agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent", ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") result = agent_to_dict(agent) # 获取关联的 Skills import json config = json.loads(agent.skill_content or "{}") skill_ids = config.get("skills", []) if skill_ids: skills = db.query(EvolutionSkill).filter( EvolutionSkill.id.in_(skill_ids) ).all() result["linked_skills"] = [ { "id": s.id, "skill_name": s.skill_name, "state": s.state, "version": s.version, } for s in skills ] else: result["linked_skills"] = [] # 获取可用模型列表 models = db.query(AIModel, ModelProvider).join( ModelProvider, AIModel.provider_id == ModelProvider.id ).filter(AIModel.is_enabled == True).all() result["available_models"] = [ {"id": m.id, "name": m.name, "provider": p.name} for m, p in models ] return {"code": 0, "data": result} @router.post("/", summary="创建 Agent") async def create_agent( req: AgentCreate, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """创建新的 Agent""" import json # 检查名称是否已存在 existing = db.query(MarketApp).filter( MarketApp.name == req.name, MarketApp.type == "agent", ).first() if existing: raise HTTPException(status_code=400, detail="Agent 名称已存在") # 构建 skill_content config = {} if req.config: cfg = req.config config = { "system_prompt": cfg.system_prompt or "", "temperature": cfg.temperature or 0.7, "max_tokens": cfg.max_tokens or 2048, "model_id": cfg.model_id, "skills": cfg.skills or [], "tools": cfg.tools or [], "memory_enabled": cfg.memory_enabled or False, "reasoning_enabled": cfg.reasoning_enabled or False, } agent = MarketApp( id=str(uuid.uuid4()), name=req.name, type="agent", source="official", publisher_id=admin_user.id, publisher_name=admin_user.username, description=req.description or "", icon=req.icon or "", price=0.0, # 管理员导入/安装/开发的 Agent 默认免费 is_member_free=req.is_member_free, skill_content=json.dumps(config, ensure_ascii=False), status="published", created_at=datetime.now(), updated_at=datetime.now(), ) db.add(agent) db.commit() db.refresh(agent) LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="create_agent", target_type="agent", target_id=agent.id, detail=f"创建 Agent: {agent.name}", ) return {"code": 0, "message": "Agent 创建成功", "data": agent_to_dict(agent)} @router.put("/{agent_id}", summary="更新 Agent") async def update_agent( agent_id: str, req: AgentUpdate, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """更新 Agent 配置""" import json agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent", ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") # 更新基本字段 if req.name is not None: # 检查新名称是否重复 dup = db.query(MarketApp).filter( MarketApp.name == req.name, MarketApp.type == "agent", MarketApp.id != agent_id, ).first() if dup: raise HTTPException(status_code=400, detail="Agent 名称已存在") agent.name = req.name if req.description is not None: agent.description = req.description if req.icon is not None: agent.icon = req.icon if req.price is not None: agent.price = req.price if req.is_member_free is not None: agent.is_member_free = req.is_member_free if req.status is not None: agent.status = req.status if req.tutorial is not None: agent.tutorial = req.tutorial # 更新配置 if req.config is not None: cfg = req.config # 合并现有配置 existing_config = {} if agent.skill_content: try: existing_config = json.loads(agent.skill_content) except Exception: pass new_config = { "system_prompt": cfg.system_prompt if cfg.system_prompt is not None else existing_config.get("system_prompt", ""), "temperature": cfg.temperature if cfg.temperature is not None else existing_config.get("temperature", 0.7), "max_tokens": cfg.max_tokens if cfg.max_tokens is not None else existing_config.get("max_tokens", 2048), "model_id": cfg.model_id if cfg.model_id is not None else existing_config.get("model_id"), "skills": cfg.skills if cfg.skills is not None else existing_config.get("skills", []), "tools": cfg.tools if cfg.tools is not None else existing_config.get("tools", []), "memory_enabled": cfg.memory_enabled if cfg.memory_enabled is not None else existing_config.get("memory_enabled", False), "reasoning_enabled": cfg.reasoning_enabled if cfg.reasoning_enabled is not None else existing_config.get("reasoning_enabled", False), } agent.skill_content = json.dumps(new_config, ensure_ascii=False) agent.updated_at = datetime.now() db.commit() db.refresh(agent) LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="update_agent", target_type="agent", target_id=agent_id, detail=f"更新 Agent: {agent.name}", ) return {"code": 0, "message": "Agent 已更新", "data": agent_to_dict(agent)} @router.delete("/{agent_id}", summary="删除 Agent") async def delete_agent( agent_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """删除 Agent""" agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent", ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") agent_name = agent.name db.delete(agent) db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="delete_agent", target_type="agent", target_id=agent_id, detail=f"删除 Agent: {agent_name}", ) return {"code": 0, "message": "Agent 已删除"} @router.put("/{agent_id}/toggle", summary="启用/禁用 Agent") async def toggle_agent( agent_id: str, is_active: bool = Query(..., description="是否启用"), admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """启用或禁用 Agent""" agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent", ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") agent.status = "published" if is_active else "pending" agent.updated_at = datetime.now() db.commit() LogService.log_audit( admin_id=admin_user.id, admin_name=admin_user.username, action="toggle_agent", target_type="agent", target_id=agent_id, detail=f"{'启用' if is_active else '禁用'} Agent: {agent.name}", ) return { "code": 0, "message": f"Agent 已{'启用' if is_active else '禁用'}", "data": agent_to_dict(agent), } @router.get("/{agent_id}/reviews", summary="获取 Agent 评价") async def get_agent_reviews( agent_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取 Agent 的用户评价""" from ...models import AppReview, User agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent", ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") reviews = db.query(AppReview).filter( AppReview.app_id == agent_id ).order_by(AppReview.created_at.desc()).limit(50).all() return { "code": 0, "data": [ { "id": r.id, "user_id": r.user_id, "rating": r.rating, "content": r.content, "created_at": r.created_at.isoformat() if r.created_at else None, } for r in reviews ] } @router.get("/available/skills", summary="获取可关联的 Skills") async def get_available_skills( keyword: str = Query(None), state: str = Query(None, description="技能状态筛选"), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取可关联到 Agent 的 Skills 列表""" query = db.query(EvolutionSkill) if keyword: query = query.filter( or_( EvolutionSkill.skill_name.ilike(f"%{keyword}%"), EvolutionSkill.description.ilike(f"%{keyword}%"), ) ) if state: query = query.filter(EvolutionSkill.state == state) total = query.count() skills = query.order_by(EvolutionSkill.use_count.desc()).offset( (page - 1) * page_size ).limit(page_size).all() return { "code": 0, "data": { "total": total, "items": [ { "id": s.id, "skill_name": s.skill_name, "description": s.description, "state": s.state, "version": s.version, "use_count": s.use_count, "category": s.category, "tags": s.tags, } for s in skills ] } } @router.get("/available/models", summary="获取可用模型列表") async def get_available_models( admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): """获取可用于 Agent 的模型列表""" models = db.query(AIModel, ModelProvider).join( ModelProvider, AIModel.provider_id == ModelProvider.id ).filter( AIModel.is_enabled == True ).order_by(AIModel.name).all() return { "code": 0, "data": [ { "id": m.id, "name": m.name, "provider": p.name, "model_type": p.protocol, "input_price": float(m.price_per_million_tokens) if m.price_per_million_tokens else 0, "output_price": float(m.price_per_million_tokens) if m.price_per_million_tokens else 0, } for m, p in models ] } # ============================================================ # 3.1 tool_policy 沙箱:Agent 工具/MCP 权限策略 # ============================================================ class ToolPolicyModel(BaseModel): tool_policy: Optional[dict] = None # null = 清除策略(恢复默认允许全部) @router.get("/{agent_id}/tool-policy", summary="获取 Agent 工具策略") async def get_agent_tool_policy( agent_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent" ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") return {"agent_id": agent_id, "tool_policy": agent.tool_policy} @router.put("/{agent_id}/tool-policy", summary="设置 Agent 工具策略") async def set_agent_tool_policy( agent_id: str, body: ToolPolicyModel, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent" ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") tp = body.tool_policy if tp is not None: if not isinstance(tp, dict): raise HTTPException(status_code=400, detail="tool_policy 必须为对象") # 轻量结构校验 if "allowed_tools" in tp and tp["allowed_tools"] is not None: at = tp["allowed_tools"] if not isinstance(at, dict): raise HTTPException(status_code=400, detail="allowed_tools 必须为对象") for k in ("skill_ids", "mcp_ids"): if k in at and at[k] is not None and not isinstance(at[k], list): raise HTTPException(status_code=400, detail=f"allowed_tools.{k} 必须为数组") for k in ("max_calls_per_conversation", "per_round_cap"): if k in tp and tp[k] is not None and not isinstance(tp[k], int): raise HTTPException(status_code=400, detail=f"{k} 必须为整数") agent.tool_policy = tp db.commit() return {"agent_id": agent_id, "tool_policy": agent.tool_policy, "updated": True} # 3.3 多 Agent 编排图(图引擎) # ============================================================ class OrchestrationModel(BaseModel): orchestration: Optional[dict] = None # null = 清除编排图 @router.get("/{agent_id}/orchestration", summary="获取 Agent 编排图") async def get_agent_orchestration( agent_id: str, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent" ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") return {"agent_id": agent_id, "orchestration": agent.orchestration} @router.put("/{agent_id}/orchestration", summary="设置 Agent 编排图") async def set_agent_orchestration( agent_id: str, body: OrchestrationModel, admin_user=Depends(get_current_admin), db: Session = Depends(get_db), ): agent = db.query(MarketApp).filter( MarketApp.id == agent_id, MarketApp.type == "agent" ).first() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") og = body.orchestration if og is not None: if not isinstance(og, dict): raise HTTPException(status_code=400, detail="orchestration 必须为对象") try: from app.core.orchestrator import validate_graph validate_graph(og) except Exception as e: raise HTTPException(status_code=400, detail=f"编排图非法: {str(e)}") agent.orchestration = og db.commit() return {"agent_id": agent_id, "orchestration": agent.orchestration, "updated": True} # ============================================================ # Phase 4.2: YAML workflow import/export/validate/templates # ============================================================ import yaml as _yaml import os as _os from app.core.yaml_workflow import validate_yaml, yaml_to_orchestration, orchestration_to_yaml from app.core.orchestrator import validate_graph _YTMPL = _os.path.join(_os.path.dirname(_os.path.dirname(_os.path.dirname(__file__))), "workflow_templates") class YAMLBody(BaseModel): yaml_text: str @router.post("/orchestration/yaml/validate", summary="校验 YAML 工作流") async def yaml_validate(req: YAMLBody, admin_user=Depends(get_current_admin)): try: data = _yaml.safe_load(req.yaml_text) except _yaml.YAMLError as e: return {"ok": False, "errors": [f"YAML parse: {str(e)}"], "warnings": []} r = validate_yaml(data) if r["ok"]: try: validate_graph(yaml_to_orchestration(data)) except Exception as e: r["ok"] = False; r["errors"].append(f"Graph fail: {str(e)}") return r @router.post("/{agent_id}/orchestration/yaml/import", summary="YAML 导入编排") async def yaml_import(agent_id: str, req: YAMLBody, admin_user=Depends(get_current_admin), db=Depends(get_db)): ag = db.query(MarketApp).filter(MarketApp.id == agent_id, MarketApp.type == "agent").first() if not ag: raise HTTPException(status_code=404, detail="Agent not found") try: data = _yaml.safe_load(req.yaml_text) except _yaml.YAMLError as e: raise HTTPException(status_code=400, detail=f"YAML parse: {str(e)}") v = validate_yaml(data) if not v["ok"]: raise HTTPException(status_code=400, detail="YAML invalid: " + "; ".join(v["errors"])) orch = yaml_to_orchestration(data) try: validate_graph(orch) except Exception as e: raise HTTPException(status_code=400, detail=f"Graph invalid: {str(e)}") ag.orchestration = orch ag.updated_at = datetime.now() db.commit() return {"agent_id": agent_id, "orchestration": ag.orchestration, "imported": True} @router.post("/{agent_id}/orchestration/yaml/export", summary="编排导出 YAML") async def yaml_export(agent_id: str, admin_user=Depends(get_current_admin), db=Depends(get_db)): ag = db.query(MarketApp).filter(MarketApp.id == agent_id, MarketApp.type == "agent").first() if not ag: raise HTTPException(status_code=404, detail="Agent not found") if not ag.orchestration: raise HTTPException(status_code=400, detail="No orchestration") yaml_str = _yaml.dump(orchestration_to_yaml(ag.orchestration), allow_unicode=True, default_flow_style=False, sort_keys=False) return {"agent_id": agent_id, "yaml": yaml_str, "version": "1.0"} @router.get("/orchestration/yaml/templates", summary="内置模板") async def yaml_templates(admin_user=Depends(get_current_admin)): tmpls = [] if _os.path.isdir(_YTMPL): for fp in sorted(_os.listdir(_YTMPL)): if fp.endswith((".yaml", ".yml")): try: with open(_os.path.join(_YTMPL, fp), encoding="utf-8") as f: txt = f.read() d = _yaml.safe_load(txt) or {} tmpls.append({"filename": fp, "name": d.get("name", fp), "description": d.get("description", ""), "category": d.get("category", "general"), "yaml": txt}) except Exception: tmpls.append({"filename": fp, "name": fp, "error": True}) return {"templates": tmpls, "count": len(tmpls)} # ============================================================ # Phase 4.3: ACA 协议抽象 — protocols / tools / health # ============================================================ from app.core.aca import registry as _aca_registry, init_aca_registry # 懒初始化:首次调用 ACA 端点时确保 registry 已填充 _aca_initialized = False def _ensure_aca(): global _aca_initialized if not _aca_initialized: init_aca_registry() _aca_initialized = True @router.get("/aca/protocols", summary="ACA 已注册协议") async def aca_protocols(admin_user=Depends(get_current_admin)): _ensure_aca() return [ {"name": a.name, "version": a.version, "capabilities": a.capabilities} for a in _aca_registry.list_all() ] @router.get("/aca/protocols/{name}/tools", summary="协议工具列表") async def aca_protocol_tools(name: str, admin_user=Depends(get_current_admin)): _ensure_aca() adapter = _aca_registry.get(name) if not adapter: raise HTTPException(status_code=404, detail=f"协议 {name} 未注册") tools = await adapter.list_tools() return {"protocol": name, "tools": tools, "count": len(tools)} @router.post("/aca/protocols/{name}/health", summary="协议健康检查") async def aca_protocol_health(name: str, admin_user=Depends(get_current_admin)): _ensure_aca() adapter = _aca_registry.get(name) if not adapter: raise HTTPException(status_code=404, detail=f"协议 {name} 未注册") ok = await adapter.health_check() return {"protocol": name, "healthy": ok}