""" FastAPI Web服务 - AI Agent管理服务 支持平台Agent和自定义Agent两种类型 """ from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel, Field from typing import Dict, List, Optional from sqlalchemy.orm import Session from datetime import datetime import logging from k8s_manager import K8sManager from database import ( get_db, Template, Agent, Quota, AgentMetric, AgentType, AgentStatus, parse_resource_string ) from template_manager import template_manager from tool_generator_api import router as tool_generator_router import os # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 创建FastAPI应用 app = FastAPI( title="AI Agent Manager", description="Kubernetes AI Agent管理服务，支持动态工具生成和 CI/CD 自动构建", version="2.0.0" ) # 注册动态工具生成 Router app.include_router(tool_generator_router) # 初始化K8s管理器 NAMESPACE = os.getenv("NAMESPACE", "ai-agents") KUBECONFIG_PATH = os.getenv("KUBECONFIG_PATH", None) # 可选：指定kubeconfig路径 k8s_manager = K8sManager(namespace=NAMESPACE, kubeconfig_path=KUBECONFIG_PATH) # ==================== 请求/响应模型 ==================== # Template Management Models class CreateTemplateRequest(BaseModel): """创建模板请求""" name: str = Field(..., min_length=1, max_length=100) display_name: str description: Optional[str] = None agent_type: str = Field(..., description="platform or custom") agent_framework: str = Field(default="langchain", description="langchain, mcp, or a2a") image: str port: Optional[int] = None env_requirements: Optional[Dict] = Field(default_factory=dict) tools_config: Optional[Dict] = Field(default_factory=dict, description="Tools configuration JSON") default_model_provider: Optional[str] = Field(None, description="Default model provider") default_model_name: Optional[str] = Field(None, description="Default model name") cpu_request: Optional[str] = None cpu_limit: Optional[str] = None memory_request: Optional[str] = None memory_limit: Optional[str] = None min_replicas: int = 1 max_replicas: int = 3 target_cpu_utilization: int = 80 class UpdateTemplateRequest(BaseModel): """更新模板请求""" display_name: Optional[str] = None description: Optional[str] = None image: Optional[str] = None port: Optional[int] = None env_requirements: Optional[Dict] = None cpu_request: Optional[str] = None cpu_limit: Optional[str] = None memory_request: Optional[str] = None memory_limit: Optional[str] = None min_replicas: Optional[int] = None max_replicas: Optional[int] = None target_cpu_utilization: Optional[int] = None is_active: Optional[bool] = None class TemplateResponse(BaseModel): """模板响应""" id: int name: str display_name: str description: Optional[str] agent_type: str image: str port: Optional[int] env_requirements: Dict cpu_request: Optional[str] cpu_limit: Optional[str] memory_request: Optional[str] memory_limit: Optional[str] min_replicas: int max_replicas: int target_cpu_utilization: int is_active: bool created_at: datetime class Config: from_attributes = True # Platform Agent Models class CreatePlatformAgentRequest(BaseModel): """创建平台Agent请求""" name: str = Field(..., min_length=1, max_length=63) template_name: str owner_id: str channel_id: Optional[str] = None tenant_id: Optional[str] = None namespace: Optional[str] = Field(default="ai-agents", description="Kubernetes namespace") query_params: Optional[Dict] = Field(default_factory=dict) # NEW: Framework-specific configurations agent_framework: Optional[str] = Field(None, description="Override template framework") tools_config: Optional[Dict] = Field(default_factory=dict, description="Tools configuration") tool_endpoint: Optional[str] = Field(None, description="External tool endpoint") tool_api_key: Optional[str] = Field(None, description="Tool API key") model_provider: Optional[str] = Field(None, description="Model provider") model_name: Optional[str] = Field(None, description="Model name") model_endpoint: Optional[str] = Field(None, description="Model endpoint") model_api_key: Optional[str] = Field(None, description="Model API key") storage_connection_string: Optional[str] = Field(None, description="Storage connection string") storage_account_name: Optional[str] = Field(None, description="Storage account name") # Custom Agent Models class ScalingConfig(BaseModel): """弹性伸缩配置""" min_replicas: int = Field(1, ge=0) max_replicas: int = Field(3, ge=1) target_cpu_utilization: int = Field(80, ge=1, le=100) class CreateCustomAgentRequest(BaseModel): """创建自定义Agent请求""" name: str = Field(..., min_length=1, max_length=63) template_name: str owner_id: str channel_id: Optional[str] = None tenant_id: Optional[str] = None namespace: Optional[str] = Field(default="ai-agents", description="Kubernetes namespace") environment_vars: Dict[str, str] # NEW: Framework-specific configurations agent_framework: Optional[str] = Field(None, description="Override template framework") tools_config: Optional[Dict] = Field(default_factory=dict, description="Tools configuration") tool_endpoint: Optional[str] = Field(None, description="External tool endpoint") tool_api_key: Optional[str] = Field(None, description="Tool API key") model_provider: Optional[str] = Field(None, description="Model provider") model_name: Optional[str] = Field(None, description="Model name") model_endpoint: Optional[str] = Field(None, description="Model endpoint") model_api_key: Optional[str] = Field(None, description="Model API key") storage_connection_string: Optional[str] = Field(None, description="Storage connection string") storage_account_name: Optional[str] = Field(None, description="Storage account name") # Resource configuration cpu_request: Optional[str] = None cpu_limit: Optional[str] = None memory_request: Optional[str] = None memory_limit: Optional[str] = None scaling_config: Optional[ScalingConfig] = None class UpdateAgentEnvRequest(BaseModel): """更新Agent环境变量请求""" environment_vars: Dict[str, str] class UpdateScalingRequest(BaseModel): """更新伸缩配置请求""" min_replicas: Optional[int] = None max_replicas: Optional[int] = None target_cpu_utilization: Optional[int] = None # Unified Agent Response class AgentResponseNew(BaseModel): """Agent响应（新）""" id: int name: str display_name: Optional[str] template_name: str agent_type: str status: str owner_id: str channel_id: Optional[str] tenant_id: Optional[str] service_url: Optional[str] current_replicas: int min_replicas: int max_replicas: int created_at: datetime last_accessed_at: Optional[datetime] class Config: from_attributes = True # Legacy Models (for backward compatibility) class CreateAgentRequest(BaseModel): """创建Agent请求（旧版）""" name: str = Field(..., description="Agent名称", min_length=1, max_length=63) template: str = Field(..., description="模板类型") framework: Optional[str] = Field(default="API", description="Agent框架类型: MCP, A2A, API") config: Dict = Field(default_factory=dict, description="配置信息") env: Optional[Dict[str, str]] = Field(default_factory=dict, description="环境变量") namespace: Optional[str] = Field(default=None, description="Kubernetes命名空间，默认使用环境变量NAMESPACE的值") class AgentResponse(BaseModel): """Agent响应""" name: str namespace: str status: str framework: Optional[str] = None created_at: Optional[str] = None template: Optional[str] = None service_port: Optional[int] = None access_info: Optional[Dict] = None pod_id: Optional[str] = None pod_ip: Optional[str] = None host_ip: Optional[str] = None node_name: Optional[str] = None owner_info: Optional[Dict] = None class ResourceUsage(BaseModel): """资源使用情况""" cpu: Optional[str] = None memory: Optional[str] = None available: Optional[bool] = None reason: Optional[str] = None class ResourceInfo(BaseModel): """资源信息（配额和使用情况）""" requests: Optional[Dict] = None limits: Optional[Dict] = None usage: Optional[ResourceUsage] = None class PodStatusResponse(BaseModel): """Pod状态响应""" name: str namespace: str status: str health_status: Optional[str] = None # 新增：健康状态 (healthy, unhealthy, degraded) template: Optional[str] = None framework: Optional[str] = None # Agent框架类型 (MCP, A2A, API) created_at: Optional[str] = None node: Optional[str] = None pod_ip: Optional[str] = None containers: Optional[List[Dict]] = None # 新增：容器详细信息 resources: Optional[ResourceInfo] = None service_port: Optional[int] = None access_url: Optional[str] = None endpoints: Optional[Dict] = None conditions: Optional[List[Dict]] = None # 访问信息（从数据库读取） access_info: Optional[Dict] = None # 包含 external_ip, domain, URLs 等 class PodMetricsResponse(BaseModel): """Pod资源使用响应""" name: str namespace: Optional[str] = None requests: Dict limits: Dict usage: Optional[Dict] = None # 实时使用情况（需要 metrics-server） timestamp: Optional[str] = None # metrics 时间戳 metrics_available: Optional[bool] = None # metrics-server 是否可用 class MessageResponse(BaseModel): """通用消息响应""" status: str message: str @app.get("/") async def root(): """健康检查""" return { "service": "AI Agent Manager", "status": "running", "namespace": NAMESPACE } @app.post("/agents", response_model=AgentResponse) async def create_agent(request: CreateAgentRequest, db: Session = Depends(get_db)): """ 创建AI Agent Pod（在独立命名空间中，并创建Service和Ingress） Args: request: 创建请求（name, template, config, namespace可选, user_id可选） db: 数据库会话 Returns: 创建的Agent信息包括pod_id、service和ingress信息 """ try: logger.info(f"收到创建Agent请求: {request.name}, 模板: {request.template}") # 验证模板类型（从数据库动态获取） valid_templates = template_manager.get_template_names() if request.template not in valid_templates: raise HTTPException( status_code=400, detail=f"无效的模板类型。支持的模板: {', '.join(valid_templates)}" ) # 验证 framework 类型 valid_frameworks = ["MCP", "A2A", "API"] framework = (request.framework or "API").upper() if framework not in valid_frameworks: raise HTTPException( status_code=400, detail=f"无效的框架类型。支持的框架: {', '.join(valid_frameworks)}" ) # 合并环境变量到config config_data = request.config.copy() config_data["agent_framework"] = framework # 添加框架类型到配置 if request.env: config_data["env"] = request.env logger.info(f"环境变量: {list(request.env.keys())}") # 添加 user_id 标签 user_id = config_data.get("user_id", "default") if "labels" not in config_data: config_data["labels"] = {} config_data["labels"]["user-id"] = user_id config_data["labels"]["managed-by"] = "agent-manager" config_data["labels"]["app"] = request.name config_data["labels"]["framework"] = framework.lower() # 添加框架标签 # 步骤1: 为每个Agent创建独立的命名空间 agent_namespace = k8s_manager.create_agent_namespace( agent_name=request.name, owner_id=user_id ) logger.info(f"✅ Agent {request.name} 将部署在独立命名空间: {agent_namespace}") # 步骤2: 创建Pod（在独立命名空间中） temp_manager = K8sManager(namespace=agent_namespace, kubeconfig_path=KUBECONFIG_PATH) result = temp_manager.create_pod( pod_name=request.name, template=request.template, config_data=config_data ) # 步骤3: 获取服务端口 service_port = k8s_manager.TEMPLATE_PORTS.get(request.template) # 步骤4: 创建 LoadBalancer Service（AKS 会自动分配外网 IP） service_info = None dns_info = None if service_port: try: import time time.sleep(2) # 等待Pod启动 # 创建 LoadBalancer Service service_info = temp_manager.create_service( service_name=f"{request.name}-service", namespace=agent_namespace, pod_selector={"app": request.name}, service_port=80, # 外部访问端口 target_port=service_port # Pod内部端口 ) logger.info(f"✅ LoadBalancer Service 创建成功: {service_info['name']}") # 步骤5: 等待 LoadBalancer IP 分配并创建 DNS try: logger.info("等待 LoadBalancer 外网 IP 分配...") external_ip = temp_manager.wait_for_loadbalancer_ip( service_name=f"{request.name}-service", namespace=agent_namespace, max_wait=120, # 最多等待2分钟 interval=5 ) service_info["external_ip"] = external_ip logger.info(f"✅ LoadBalancer 外网 IP: {external_ip}") # 步骤6: 自动创建 Azure DNS 记录 try: logger.info(f"创建 DNS 记录: {request.name}.taijiagnet.com") dns_info = temp_manager.create_dns_record( subdomain=request.name, ip_address=external_ip ) logger.info(f"✅ DNS 记录: {dns_info['domain']} -> {external_ip}") except Exception as e: logger.warning(f"DNS 记录创建失败: {str(e)}") dns_info = {"status": "failed", "error": str(e)} except Exception as e: logger.warning(f"等待 LoadBalancer IP 超时: {str(e)}") logger.info(" 外网 IP 将在后台继续分配") except Exception as e: logger.warning(f"创建 LoadBalancer Service 失败: {str(e)}") # 获取 Pod 详细信息（包括 pod_id） try: import time time.sleep(1) # 等待 Pod 创建完成 pod = temp_manager.v1.read_namespaced_pod( name=request.name, namespace=agent_namespace ) result["pod_id"] = pod.metadata.uid result["pod_ip"] = pod.status.pod_ip result["host_ip"] = pod.status.host_ip result["node_name"] = pod.spec.node_name result["namespace"] = agent_namespace # 更新为实际使用的命名空间 result["framework"] = framework # 添加框架类型到响应 result["owner_info"] = { "user_id": user_id, "agent_name": request.name, "namespace": agent_namespace, "framework": framework, "labels": pod.metadata.labels } # 添加 LoadBalancer Service 信息到响应 if service_info: result["service_info"] = service_info # 构建访问信息 external_ip = service_info.get("external_ip") if external_ip: # 有外网 IP result["access_info"] = { "external_ip": external_ip, "ip_url": f"http://{external_ip}:80", "service_url": f"http://{service_info['cluster_ip']}:80", "pod_url": f"http://{pod.status.pod_ip}:{service_port}" if pod.status.pod_ip and service_port else None } # 添加 DNS 信息 if dns_info and dns_info.get("status") == "created": result["dns_info"] = dns_info result["access_info"]["domain"] = dns_info["domain"] result["access_info"]["domain_url"] = f"http://{dns_info['domain']}" result["access_info"]["recommended"] = f"http://{dns_info['domain']}" logger.info(f" - 推荐访问: http://{dns_info['domain']}") else: result["access_info"]["recommended"] = f"http://{external_ip}:80" logger.info(f" - 外网访问: http://{external_ip}:80") else: # IP 还在分配中 result["access_info"] = { "status": "pending", "external_ip": None, "note": "LoadBalancer IP 正在分配中，请稍后查询" } logger.info(f" - 外网 IP 正在分配中") logger.info(f"✅ Agent创建成功！") logger.info(f" - Pod ID: {result['pod_id']}") logger.info(f" - Namespace: {agent_namespace}") logger.info(f" - User: {user_id}") except Exception as e: logger.warning(f"获取Pod详细信息失败: {str(e)}") # 保存到数据库 try: # 提取访问信息 access_info = result.get("access_info", {}) external_ip = access_info.get("external_ip") domain = access_info.get("domain") ip_url = access_info.get("ip_url") domain_url = access_info.get("domain_url") recommended_url = access_info.get("recommended", domain_url or ip_url) # 创建Agent记录 db_agent = Agent( name=request.name, display_name=request.name, owner_id=user_id, agent_type=AgentType.PLATFORM, # 默认为平台类型 status=AgentStatus.RUNNING, agent_framework=framework.lower(), namespace=agent_namespace, service_name=service_info.get("name") if service_info else None, external_ip=external_ip, domain=domain, ip_url=ip_url, domain_url=domain_url, recommended_url=recommended_url, min_replicas=1, max_replicas=1, current_replicas=1 ) db.add(db_agent) db.commit() db.refresh(db_agent) logger.info(f"✅ Agent信息已保存到数据库: {db_agent.id}") except Exception as e: logger.error(f"保存Agent到数据库失败: {str(e)}") # 不抛出异常，因为Agent已经在K8s中创建成功 # 确保 result 包含所有必需字段 if "name" not in result: result["name"] = request.name if "namespace" not in result: result["namespace"] = agent_namespace if "status" not in result: result["status"] = "Pending" try: return AgentResponse(**result) except Exception as validation_error: logger.error(f"AgentResponse 验证失败: {validation_error}") logger.error(f"result 数据: {result}") raise except Exception as e: import traceback error_msg = str(e) if str(e) else repr(e) logger.error(f"创建Agent失败: {error_msg}") logger.error(f"异常详情: {traceback.format_exc()}") raise HTTPException(status_code=500, detail=error_msg) @app.delete("/agents/{agent_name}", response_model=MessageResponse) async def delete_agent(agent_name: str, db: Session = Depends(get_db)): """ 删除AI Agent（包括独立命名空间、LoadBalancer Service、DNS 记录和数据库记录） Args: agent_name: Agent名称 db: 数据库会话 Returns: 删除结果 """ try: logger.info(f"收到删除Agent请求: {agent_name}") # 保护机制：防止删除 agent-manager 命名空间 computed_namespace = f"agent-{agent_name}".lower().strip('-')[:63] if computed_namespace == "agent-manager": logger.error(f"❌ 禁止删除 agent-manager 命名空间！agent_name={agent_name}, computed_namespace={computed_namespace}") raise HTTPException( status_code=400, detail=f"禁止删除 agent-manager 命名空间。这是系统保护命名空间，不能被删除。" ) # region agent log try: import json, time with open("/home/taiji/tools/agent-manager/.cursor/debug.log", "a") as _f: _f.write(json.dumps({ "sessionId": "debug-session", "runId": "pre-fix", "hypothesisId": "H1", "location": "app.py:delete_agent:entry", "message": "delete_agent called", "data": { "agent_name": agent_name, "computed_namespace": f"agent-{agent_name}"[:63].lower().strip('-'), "manager_namespace": "agent-manager" }, "timestamp": int(time.time() * 1000) }) + "\n") except Exception: pass # endregion # 步骤1: 删除 DNS 记录 try: dns_result = k8s_manager.delete_dns_record(subdomain=agent_name) if dns_result.get("status") == "deleted": logger.info(f"✅ DNS 记录已删除: {dns_result.get('domain')}") except Exception as e: logger.warning(f"删除 DNS 记录失败（可忽略）: {str(e)}") # 步骤2: 删除 Agent 的独立命名空间（会自动删除Pod、Service等所有资源） result = k8s_manager.delete_agent_namespace(agent_name=agent_name) # region agent log try: import json, time with open("/home/taiji/tools/agent-manager/.cursor/debug.log", "a") as _f: _f.write(json.dumps({ "sessionId": "debug-session", "runId": "pre-fix", "hypothesisId": "H2", "location": "app.py:delete_agent:after_delete_namespace", "message": "delete_agent_namespace result", "data": { "agent_name": agent_name, "result_status": result.get("status"), "result_namespace": result.get("namespace") }, "timestamp": int(time.time() * 1000) }) + "\n") except Exception: pass # endregion if result.get("status") == "not_found": raise HTTPException(status_code=404, detail=result.get("message")) # 步骤3: 从数据库删除Agent记录 try: db_agent = db.query(Agent).filter(Agent.name == agent_name).first() if db_agent: db.delete(db_agent) db.commit() logger.info(f"✅ Agent数据库记录已删除: {agent_name}") else: logger.warning(f"⚠️ Agent {agent_name} 在数据库中未找到") except Exception as db_error: logger.error(f"删除数据库记录失败: {str(db_error)}") # 不抛出异常，因为K8s资源已经删除 logger.info(f"✅ Agent {agent_name} 及其所有资源删除成功") return MessageResponse(**result) except HTTPException: raise except Exception as e: logger.error(f"删除Agent失败: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/agents/{agent_name}/status", response_model=PodStatusResponse) async def get_agent_status(agent_name: str, db: Session = Depends(get_db)): """ 获取Agent详细状态（包括访问信息） Args: agent_name: Agent名称 db: 数据库会话 Returns: Agent状态信息（包括Pod状态和访问信息） """ try: logger.info(f"获取Agent状态: {agent_name}") # 先从数据库获取Agent的namespace db_agent = db.query(Agent).filter(Agent.name == agent_name).first() # 确定Agent所在的namespace agent_namespace = None if db_agent and db_agent.namespace: agent_namespace = db_agent.namespace else: # 如果数据库中没有，尝试查找以 agent-{agent_name} 开头的命名空间 try: namespaces = k8s_manager.v1.list_namespace( label_selector=f"agent-name={agent_name}" ) if namespaces.items: agent_namespace = namespaces.items[0].metadata.name else: # 尝试常见的命名空间格式 for ns_pattern in [f"agent-{agent_name}", f"agent-test-{agent_name}"]: try: k8s_manager.v1.read_namespace(name=ns_pattern) agent_namespace = ns_pattern break except: continue except Exception as e: logger.warning(f"查找命名空间失败: {e}") if not agent_namespace: raise HTTPException(status_code=404, detail=f"Agent {agent_name} 的命名空间未找到") # 使用正确的namespace获取Pod状态 temp_manager = K8sManager(namespace=agent_namespace, kubeconfig_path=KUBECONFIG_PATH) result = temp_manager.get_pod_status(pod_name=agent_name) if result.get("status") == "not_found": raise HTTPException(status_code=404, detail=result.get("message")) # 添加数据库中的信息 if db_agent: # 添加框架类型 result["framework"] = db_agent.agent_framework.upper() if db_agent.agent_framework else "API" # 添加访问信息 access_info = {} if db_agent.external_ip: access_info["external_ip"] = db_agent.external_ip if db_agent.ip_url: access_info["ip_url"] = db_agent.ip_url if db_agent.domain: access_info["domain"] = db_agent.domain if db_agent.domain_url: access_info["domain_url"] = db_agent.domain_url if db_agent.recommended_url: access_info["recommended_url"] = db_agent.recommended_url if db_agent.service_name: access_info["service_name"] = db_agent.service_name # 只有当有访问信息时才添加 if access_info: result["access_info"] = access_info logger.info(f"✅ 已添加访问信息: domain={db_agent.domain}, ip={db_agent.external_ip}") else: logger.warning(f"⚠️ Agent {agent_name} 在数据库中没有访问信息") else: logger.warning(f"⚠️ Agent {agent_name} 在数据库中未找到") return PodStatusResponse(**result) except HTTPException: raise except Exception as e: logger.error(f"获取Agent状态失败: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/agents/{agent_name}/metrics", response_model=PodMetricsResponse) async def get_agent_metrics(agent_name: str, db: Session = Depends(get_db)): """ 获取Agent资源使用情况 Args: agent_name: Agent名称 db: 数据库会话 Returns: Agent资源使用信息 """ try: logger.info(f"获取Agent资源信息: {agent_name}") # 先从数据库获取Agent的namespace db_agent = db.query(Agent).filter(Agent.name == agent_name).first() # 确定Agent所在的namespace agent_namespace = None if db_agent and db_agent.namespace: agent_namespace = db_agent.namespace else: # 如果数据库中没有，尝试查找以 agent-{agent_name} 开头的命名空间 try: namespaces = k8s_manager.v1.list_namespace( label_selector=f"agent-name={agent_name}" ) if namespaces.items: agent_namespace = namespaces.items[0].metadata.name else: # 尝试常见的命名空间格式 for ns_pattern in [f"agent-{agent_name}", f"agent-test-{agent_name}"]: try: k8s_manager.v1.read_namespace(name=ns_pattern) agent_namespace = ns_pattern break except: continue except Exception as e: logger.warning(f"查找命名空间失败: {e}") if not agent_namespace: raise HTTPException(status_code=404, detail=f"Agent {agent_name} 的命名空间未找到") # 使用正确的namespace获取Pod指标 temp_manager = K8sManager(namespace=agent_namespace, kubeconfig_path=KUBECONFIG_PATH) result = temp_manager.get_pod_metrics(pod_name=agent_name) return PodMetricsResponse(**result) except HTTPException: raise except Exception as e: logger.error(f"获取Agent资源信息失败: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/agents") async def list_agents(template: Optional[str] = None, db: Session = Depends(get_db)): """ 列出所有Agent（跨所有命名空间） Args: template: 模板类型过滤（可选） db: 数据库会话 Returns: Agent列表 """ try: logger.info(f"列出Agents, 模板过滤: {template}") all_agents = [] # 方法1: 从数据库获取Agent列表（推荐） try: query = db.query(Agent) db_agents = query.all() for db_agent in db_agents: # 尝试从K8s获取Pod状态 pod_status = "Unknown" pod_ip = None try: if db_agent.namespace: temp_manager = K8sManager(namespace=db_agent.namespace, kubeconfig_path=KUBECONFIG_PATH) pod = temp_manager.v1.read_namespaced_pod( name=db_agent.name, namespace=db_agent.namespace ) pod_status = pod.status.phase pod_ip = pod.status.pod_ip except Exception: pod_status = "NotFound" agent_info = { "name": db_agent.name, "namespace": db_agent.namespace, "status": pod_status, "template": db_agent.agent_framework or "unknown", "created_at": db_agent.created_at.isoformat() if db_agent.created_at else None, "pod_ip": pod_ip, "external_ip": db_agent.external_ip, "domain": db_agent.domain, "service_url": db_agent.recommended_url } # 应用模板过滤 if template and agent_info.get("template") != template: continue all_agents.append(agent_info) logger.info(f"从数据库获取到 {len(all_agents)} 个Agent") except Exception as db_error: logger.warning(f"从数据库获取Agent列表失败: {db_error}") # 方法2: 遍历所有以 agent- 开头的命名空间（作为补充） try: namespaces = k8s_manager.v1.list_namespace() agent_namespaces = [ ns.metadata.name for ns in namespaces.items if ns.metadata.name.startswith("agent-") ] # 已经从数据库获取的Agent名称 known_agents = {a["name"] for a in all_agents} for ns_name in agent_namespaces: try: temp_manager = K8sManager(namespace=ns_name, kubeconfig_path=KUBECONFIG_PATH) label_selector = "managed-by=agent-manager" if template: label_selector += f",template={template}" pods = temp_manager.v1.list_namespaced_pod( namespace=ns_name, label_selector=label_selector ) for pod in pods.items: if pod.metadata.name not in known_agents: agent_info = { "name": pod.metadata.name, "namespace": ns_name, "status": pod.status.phase, "template": pod.metadata.labels.get("template", "unknown"), "created_at": pod.metadata.creation_timestamp.isoformat() if pod.metadata.creation_timestamp else None, "pod_ip": pod.status.pod_ip } all_agents.append(agent_info) except Exception as e: logger.debug(f"命名空间 {ns_name} 查询失败: {e}") except Exception as ns_error: logger.warning(f"遍历命名空间失败: {ns_error}") return {"agents": all_agents, "count": len(all_agents)} except Exception as e: logger.error(f"列出Agents失败: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/templates") async def list_templates(): """ 列出所有可用的Agent模板及其所需参数 Returns: 模板列表及其配置信息（从数据库动态加载） """ templates = template_manager.get_all_templates() return { "templates": templates, "count": len(templates) } @app.get("/templates/platform") async def list_platform_templates(): """ 获取平台 Agent 镜像列表 Returns: 平台提供的Agent模板列表（从数据库动态加载） """ templates = template_manager.get_all_templates() platform_templates = [t for t in templates if t.get("agent_type") == "platform"] for t in platform_templates: t["type"] = "platform" return { "templates": platform_templates, "count": len(platform_templates), "type": "platform" } @app.get("/templates/custom") async def list_custom_templates(): """ 获取自定义 Agent 镜像列表 Returns: 用户自定义的Agent模板列表（从数据库动态加载） """ templates = template_manager.get_all_templates() custom_templates = [t for t in templates if t.get("agent_type") == "custom"] for t in custom_templates: t["type"] = "custom" return { "templates": custom_templates, "count": len(custom_templates), "type": "custom" } @app.get("/templates/{template_name}") async def get_template_info_endpoint(template_name: str): """ 获取指定模板的详细信息 Args: template_name: 模板名称 Returns: 模板详细信息（端口、所需环境变量等） """ template = template_manager.get_template(template_name) if not template: valid_templates = template_manager.get_template_names() raise HTTPException( status_code=404, detail=f"模板 {template_name} 不存在。可用模板: {', '.join(valid_templates)}" ) return template # ==================== 模板管理 CRUD API ==================== @app.post("/templates/create") async def create_template(request: CreateTemplateRequest, db: Session = Depends(get_db)): """ 创建新的 Agent 模板 通过 API 动态添加模板，无需修改代码或重新部署 Args: request: 模板创建请求 Returns: 创建的模板信息 Example: POST /templates/create { "name": "my_custom_agent", "display_name": "My Custom Agent", "description": "自定义 Agent 描述", "image": "agnettaiji.azurecr.io/ai-agents/my-agent:latest", "port": 8000, "agent_framework": "api", "env_requirements": { "required": {"MY_API_KEY": "API 密钥"} } } """ try: template = template_manager.create_template( name=request.name, image=request.image, display_name=request.display_name, description=request.description, port=request.port or 8000, agent_framework=request.agent_framework or "api", env_requirements=request.env_requirements, tools_config=request.tools_config, cpu_request=request.cpu_request, cpu_limit=request.cpu_limit, memory_request=request.memory_request, memory_limit=request.memory_limit, created_by="api", ) logger.info(f"✅ 模板创建成功: {request.name}") return { "status": "success", "message": f"模板 {request.name} 创建成功", "template": template } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"创建模板失败: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.put("/templates/{template_name}") async def update_template(template_name: str, request: UpdateTemplateRequest, db: Session = Depends(get_db)): """ 更新现有模板 Args: template_name: 模板名称 request: 更新请求（只更新提供的字段） Returns: 更新后的模板信息 """ try: # 检查模板是否存在 if not template_manager.template_exists(template_name): raise HTTPException( status_code=404, detail=f"模板 {template_name} 不存在" ) template = template_manager.update_template( name=template_name, image=request.image, display_name=request.display_name, description=request.description, port=request.port, env_requirements=request.env_requirements, cpu_request=request.cpu_request, cpu_limit=request.cpu_limit, memory_request=request.memory_request, memory_limit=request.memory_limit, is_active=request.is_active, ) logger.info(f"✅ 模板更新成功: {template_name}") return { "status": "success", "message": f"模板 {template_name} 更新成功", "template": template } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"更新模板失败: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.delete("/templates/{template_name}") async def delete_template(template_name: str, force: bool = False, db: Session = Depends(get_db)): """ 删除模板 Args: template_name: 模板名称 force: 是否强制删除（默认软删除/禁用） Returns: 删除结果 """ try: # 检查模板是否存在 if not template_manager.template_exists(template_name): raise HTTPException( status_code=404, detail=f"模板 {template_name} 不存在" ) template_manager.delete_template(template_name, force=force) action = "永久删除" if force else "禁用" logger.info(f"✅ 模板{action}成功: {template_name}") return { "status": "success", "message": f"模板 {template_name} 已{action}" } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"删除模板失败: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/templates/search/{keyword}") async def search_templates(keyword: str): """ 搜索模板 Args: keyword: 搜索关键字（匹配名称、显示名、描述） Returns: 匹配的模板列表 """ templates = template_manager.search_templates(keyword) return { "templates": templates, "count": len(templates), "keyword": keyword } @app.post("/templates/refresh-cache") async def refresh_template_cache(): """ 刷新模板缓存 手动刷新内存中的模板缓存，立即生效新的模板配置 """ template_manager.invalidate_cache() templates = template_manager.get_all_templates() return { "status": "success", "message": "模板缓存已刷新", "count": len(templates) } if __name__ == "__main__": import uvicorn host = os.getenv("SERVICE_HOST", "0.0.0.0") port = int(os.getenv("SERVICE_PORT", "8000")) logger.info(f"启动AI Agent Manager服务: {host}:{port}") uvicorn.run(app, host=host, port=port)