diff --git a/Dockerfile b/Dockerfile index 0786558..3bc33a2 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,10 +2,17 @@ FROM python:3.11-slim WORKDIR /app +# 安装系统依赖 +RUN apt-get update && apt-get install -y \ + curl \ + && rm -rf /var/lib/apt/lists/* + # 复制应用代码 COPY requirements.txt . COPY app.py . COPY k8s_manager.py . +COPY database.py . +COPY agent_manager/ ./agent_manager/ # 安装依赖 RUN pip install --no-cache-dir -r requirements.txt diff --git a/agent_manager.db b/agent_manager.db index 99f88e4..54c5a59 100644 Binary files a/agent_manager.db and b/agent_manager.db differ diff --git a/agent_templates/agent_manager.db b/agent_templates/agent_manager.db index 77813d3..c43e224 100644 Binary files a/agent_templates/agent_manager.db and b/agent_templates/agent_manager.db differ diff --git a/app.py b/app.py index dc14582..bce5c42 100644 --- a/app.py +++ b/app.py @@ -198,8 +198,10 @@ 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): @@ -207,6 +209,7 @@ class AgentResponse(BaseModel): name: str namespace: str status: str + framework: Optional[str] = None created_at: Optional[str] = None template: Optional[str] = None service_port: Optional[int] = None @@ -240,6 +243,7 @@ class PodStatusResponse(BaseModel): 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 @@ -249,6 +253,8 @@ class PodStatusResponse(BaseModel): 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): @@ -279,15 +285,16 @@ async def root(): @app.post("/agents", response_model=AgentResponse) -async def create_agent(request: CreateAgentRequest): +async def create_agent(request: CreateAgentRequest, db: Session = Depends(get_db)): """ - 创建AI Agent Pod + 创建AI Agent Pod（在独立命名空间中，并创建Service和Ingress） Args: - request: 创建请求（name, template, config, user_id可选） + request: 创建请求（name, template, config, namespace可选, user_id可选） + db: 数据库会话 Returns: - 创建的Agent信息包括pod_id + 创建的Agent信息包括pod_id、service和ingress信息 """ try: logger.info(f"收到创建Agent请求: {request.name}, 模板: {request.template}") @@ -300,8 +307,18 @@ async def create_agent(request: CreateAgentRequest): 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())}") @@ -312,36 +329,180 @@ async def create_agent(request: CreateAgentRequest): 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() # 添加框架标签 - # 创建Pod - result = k8s_manager.create_pod( + # 步骤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 = k8s_manager.v1.read_namespaced_pod( + pod = temp_manager.v1.read_namespaced_pod( name=request.name, - namespace=NAMESPACE + 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": NAMESPACE, + "namespace": agent_namespace, + "framework": framework, "labels": pod.metadata.labels } - logger.info(f"✅ Agent创建成功，Pod ID: {result['pod_id']}, 用户: {user_id}") + + # 添加 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中创建成功 + return AgentResponse(**result) except Exception as e: @@ -350,23 +511,48 @@ async def create_agent(request: CreateAgentRequest): @app.delete("/agents/{agent_name}", response_model=MessageResponse) -async def delete_agent(agent_name: str): +async def delete_agent(agent_name: str, db: Session = Depends(get_db)): """ - 删除AI Agent Pod + 删除AI Agent（包括独立命名空间、LoadBalancer Service、DNS 记录和数据库记录） Args: agent_name: Agent名称 + db: 数据库会话 Returns: 删除结果 """ try: logger.info(f"收到删除Agent请求: {agent_name}") - result = k8s_manager.delete_pod(pod_name=agent_name) + + # 步骤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) 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: @@ -377,23 +563,68 @@ async def delete_agent(agent_name: str): @app.get("/agents/{agent_name}/status", response_model=PodStatusResponse) -async def get_agent_status(agent_name: str): +async def get_agent_status(agent_name: str, db: Session = Depends(get_db)): """ - 获取Agent状态 + 获取Agent详细状态（包括访问信息） Args: agent_name: Agent名称 + db: 数据库会话 Returns: - Agent状态信息 + Agent状态信息（包括Pod状态和访问信息） """ try: logger.info(f"获取Agent状态: {agent_name}") + + # 从Kubernetes获取Pod状态 result = k8s_manager.get_pod_status(pod_name=agent_name) if result.get("status") == "not_found": raise HTTPException(status_code=404, detail=result.get("message")) + # 从数据库获取Agent记录（包含访问信息） + try: + db_agent = db.query(Agent).filter(Agent.name == agent_name).first() + + 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} 在数据库中未找到，可能是在数据库启用前创建的") + + except Exception as db_error: + logger.error(f"从数据库读取访问信息失败: {str(db_error)}") + # 不抛出异常，继续返回Pod状态信息 + return PodStatusResponse(**result) except HTTPException: @@ -458,7 +689,7 @@ async def list_templates(): Returns: 模板列表及其配置信息 """ - valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent"] + valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a"] templates_info = [] for template in valid_templates: @@ -480,7 +711,7 @@ async def list_platform_templates(): 平台提供的Agent模板列表 """ # 平台 Agent 是预定义的标准模板 - platform_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "jina_search_agent", "azure_blob_agent"] + platform_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a"] templates_info = [] for template in platform_templates: @@ -530,7 +761,7 @@ async def get_template_info(template_name: str): Returns: 模板详细信息（端口、所需环境变量等） """ - valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent"] + valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a"] if template_name not in valid_templates: raise HTTPException( diff --git a/database.py b/database.py index ce26b1b..b366720 100644 --- a/database.py +++ b/database.py @@ -14,13 +14,10 @@ from sqlalchemy.orm import sessionmaker, relationship, Session import enum import os -# Database URL from environment or default to SQLite -DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./agent_manager.db") +# Database URL - PostgreSQL (hardcoded) +DATABASE_URL = "postgresql://taiji:By%40123456.@taijipda.postgres.database.azure.com:5432/taijiagnet" -engine = create_engine( - DATABASE_URL, - connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {} -) +engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() @@ -97,8 +94,8 @@ class Agent(Base): name = Column(String(100), unique=True, nullable=False, index=True) display_name = Column(String(200)) - # Template reference - template_id = Column(Integer, ForeignKey("templates.id"), nullable=False) + # Template reference (nullable for backward compatibility with non-template agents) + template_id = Column(Integer, ForeignKey("templates.id"), nullable=True) template = relationship("Template", back_populates="agents") # Ownership and organization @@ -149,6 +146,13 @@ class Agent(Base): service_url = Column(String(500)) namespace = Column(String(100), default="ai-agents") + # Access information (NEW) - LoadBalancer and DNS details + external_ip = Column(String(100), nullable=True) # LoadBalancer external IP + domain = Column(String(500), nullable=True) # Full DNS domain (e.g., agent-name.taijiagnet.com) + ip_url = Column(String(500), nullable=True) # HTTP URL via IP (http://x.x.x.x:80) + domain_url = Column(String(500), nullable=True) # HTTP URL via domain + recommended_url = Column(String(500), nullable=True) # Recommended access URL + # Metadata created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index d38d205..b5c9fd2 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -2,7 +2,7 @@ apiVersion: apps/v1 kind: Deployment metadata: name: agent-manager - namespace: default + namespace: agent-manager labels: app: agent-manager spec: @@ -69,13 +69,13 @@ apiVersion: v1 kind: Service metadata: name: agent-manager - namespace: default + namespace: agent-manager labels: app: agent-manager spec: - type: ClusterIP + type: LoadBalancer ports: - - port: 8000 + - port: 80 targetPort: 8000 protocol: TCP name: http diff --git a/k8s_manager.py b/k8s_manager.py index 10298aa..3d17178 100644 --- a/k8s_manager.py +++ b/k8s_manager.py @@ -6,6 +6,8 @@ from kubernetes.client.rest import ApiException from typing import Dict, List, Optional import logging import os +import requests +import time logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -14,6 +16,14 @@ logger = logging.getLogger(__name__) class K8sManager: """Kubernetes资源管理器""" + # Azure DNS 配置（从环境变量读取） + AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID", "263c3ff6-1be5-4141-8308-b188464fb297") + AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "c5ba26db-f180-425f-bac3-93708d853988") + AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "ydt8Q~DIOfKXDUOJkzfgaNVNh2d1VrbeIaYuKbpD") + AZURE_SUBSCRIPTION_ID = os.getenv("AZURE_SUBSCRIPTION_ID", "c6c47e4c-f5f4-49f8-b26f-7728862c17d6") + AZURE_RESOURCE_GROUP = os.getenv("AZURE_RESOURCE_GROUP", "taiji-ai-v0") + AZURE_DNS_ZONE = os.getenv("AZURE_DNS_ZONE", "taijiagnet.com") + def __init__(self, namespace: str = "ai-agents", kubeconfig_path: str = None): """ 初始化Kubernetes管理器 @@ -27,6 +37,7 @@ class K8sManager: self.v1 = client.CoreV1Api() self.apps_v1 = client.AppsV1Api() + self.networking_v1 = client.NetworkingV1Api() # 确保命名空间存在 self._ensure_namespace() @@ -73,8 +84,240 @@ class K8sManager: else: raise + def create_agent_namespace(self, agent_name: str, owner_id: str = None) -> str: + """为 Agent 创建独立的命名空间 + + Args: + agent_name: Agent 名称 + owner_id: 所有者 ID + + Returns: + 创建的命名空间名称 + """ + # 生成命名空间名称（使用 agent-{agent_name} 格式） + namespace_name = f"agent-{agent_name}" + + # 确保命名空间名称符合 DNS 标准（最多 63 个字符，只能包含小写字母、数字和连字符） + namespace_name = namespace_name[:63].lower().strip('-') + + try: + # 检查命名空间是否已存在 + self.v1.read_namespace(name=namespace_name) + logger.info(f"命名空间 {namespace_name} 已存在") + return namespace_name + except ApiException as e: + if e.status == 404: + # 创建命名空间 + labels = { + "managed-by": "agent-manager", + "agent-name": agent_name + } + if owner_id: + labels["owner-id"] = owner_id + + namespace_manifest = client.V1Namespace( + metadata=client.V1ObjectMeta( + name=namespace_name, + labels=labels + ) + ) + self.v1.create_namespace(body=namespace_manifest) + logger.info(f"✅ 创建命名空间 {namespace_name}") + + # 复制 ACR secret 到新命名空间 + self._copy_acr_secret_to_namespace(namespace_name) + + return namespace_name + else: + raise + + def _copy_acr_secret_to_namespace(self, target_namespace: str): + """复制 ACR secret 到目标命名空间 + + Args: + target_namespace: 目标命名空间 + """ + try: + # 从 agent-manager 命名空间读取 acr-secret + source_secret = self.v1.read_namespaced_secret( + name="acr-secret", + namespace="agent-manager" + ) + + # 创建新的 secret（去除自动生成的字段） + new_secret = client.V1Secret( + metadata=client.V1ObjectMeta( + name="acr-secret", + namespace=target_namespace + ), + data=source_secret.data, + type=source_secret.type + ) + + # 在目标命名空间创建 secret + self.v1.create_namespaced_secret( + namespace=target_namespace, + body=new_secret + ) + logger.info(f"✅ 已复制 ACR secret 到命名空间 {target_namespace}") + except ApiException as e: + if e.status == 404: + logger.warning(f"⚠️ 源 ACR secret 不存在，跳过复制") + elif e.status == 409: + logger.info(f"ACR secret 已存在于命名空间 {target_namespace}") + else: + logger.error(f"❌ 复制 ACR secret 失败: {e}") + + def create_service(self, service_name: str, namespace: str, pod_selector: Dict[str, str], + service_port: int, target_port: int) -> Dict: + """为 Agent Pod 创建 LoadBalancer Service + + Args: + service_name: Service 名称 + namespace: 命名空间 + pod_selector: Pod 选择器标签 + service_port: Service 端口 + target_port: Pod 目标端口 + + Returns: + 创建的 Service 信息（包含外网 IP） + """ + try: + service_manifest = client.V1Service( + metadata=client.V1ObjectMeta( + name=service_name, + namespace=namespace, + labels={"managed-by": "agent-manager"} + ), + spec=client.V1ServiceSpec( + selector=pod_selector, + ports=[ + client.V1ServicePort( + name="http", + protocol="TCP", + port=service_port, + target_port=target_port + ) + ], + type="LoadBalancer" + ) + ) + + response = self.v1.create_namespaced_service( + namespace=namespace, + body=service_manifest + ) + + logger.info(f"✅ LoadBalancer Service {service_name} 创建成功 (namespace: {namespace})") + + # 获取外网 IP（可能需要等待分配） + external_ip = None + if response.status.load_balancer.ingress: + lb_ingress = response.status.load_balancer.ingress[0] + external_ip = lb_ingress.ip or lb_ingress.hostname + + return { + "name": response.metadata.name, + "namespace": response.metadata.namespace, + "cluster_ip": response.spec.cluster_ip, + "port": service_port, + "external_ip": external_ip, + "type": "LoadBalancer" + } + except ApiException as e: + logger.error(f"创建 Service 失败: {e}") + raise Exception(f"创建 Service 失败: {e.reason}") + + def create_ingress(self, ingress_name: str, namespace: str, service_name: str, + service_port: int, host: str = None, path: str = "/") -> Dict: + """为 Agent Service 创建 Ingress + + Args: + ingress_name: Ingress 名称 + namespace: 命名空间 + service_name: 后端 Service 名称 + service_port: Service 端口 + host: 域名（可选，如果为 None 则使用默认） + path: 路径前缀 + + Returns: + 创建的 Ingress 信息 + """ + try: + # 构建路径规则 + http_ingress_path = client.V1HTTPIngressPath( + path=path, + path_type="Prefix", + backend=client.V1IngressBackend( + service=client.V1IngressServiceBackend( + name=service_name, + port=client.V1ServiceBackendPort(number=service_port) + ) + ) + ) + + # 构建规则 + ingress_rule = client.V1IngressRule( + http=client.V1HTTPIngressRuleValue(paths=[http_ingress_path]) + ) + + # 如果指定了 host，添加到规则中 + if host: + ingress_rule.host = host + + # 创建 Ingress manifest + ingress_manifest = client.V1Ingress( + metadata=client.V1ObjectMeta( + name=ingress_name, + namespace=namespace, + labels={"managed-by": "agent-manager"}, + annotations={ + "nginx.ingress.kubernetes.io/rewrite-target": "/", + "nginx.ingress.kubernetes.io/ssl-redirect": "false" + } + ), + spec=client.V1IngressSpec( + ingress_class_name="nginx", # 使用 nginx ingress controller + rules=[ingress_rule] + ) + ) + + response = self.networking_v1.create_namespaced_ingress( + namespace=namespace, + body=ingress_manifest + ) + + logger.info(f"✅ Ingress {ingress_name} 创建成功 (namespace: {namespace})") + + # 获取 Ingress IP/域名 + ingress_url = None + if response.status.load_balancer.ingress: + lb_ingress = response.status.load_balancer.ingress[0] + if lb_ingress.ip: + ingress_url = f"http://{lb_ingress.ip}{path}" + elif lb_ingress.hostname: + ingress_url = f"http://{lb_ingress.hostname}{path}" + + return { + "name": response.metadata.name, + "namespace": response.metadata.namespace, + "host": host, + "path": path, + "url": ingress_url, + "note": "Ingress URL 将在负载均衡器配置完成后可用" + } + except ApiException as e: + logger.error(f"创建 Ingress 失败: {e}") + raise Exception(f"创建 Ingress 失败: {e.reason}") + # 模板端口映射 TEMPLATE_PORTS = { + "echo_agent": 8000, + "chat_agent": 8000, + "code_agent": 8000, + "search_agent": 8000, + "mysql_agent": 8000, + "postgresql_agent": 8000, "jina_search_agent": 8080, "azure_blob_agent": 8080, "azure_blob_agent_mcp": 8080, @@ -397,6 +640,195 @@ class K8sManager: logger.error(f"删除Pod失败: {e}") raise Exception(f"删除Pod失败: {e.reason}") + def delete_agent_namespace(self, agent_name: str) -> Dict: + """删除Agent的命名空间及其所有资源 + + Args: + agent_name: Agent 名称 + + Returns: + 删除结果 + """ + namespace_name = f"agent-{agent_name}".lower().strip('-')[:63] + + try: + # 删除命名空间会自动删除其中的所有资源（Pod、Service、Ingress等） + self.v1.delete_namespace( + name=namespace_name, + body=client.V1DeleteOptions() + ) + logger.info(f"✅ 命名空间 {namespace_name} 及其所有资源删除成功") + return { + "status": "success", + "message": f"Agent {agent_name} 的命名空间 {namespace_name} 及所有相关资源已删除", + "namespace": namespace_name + } + except ApiException as e: + if e.status == 404: + return { + "status": "not_found", + "message": f"命名空间 {namespace_name} 不存在" + } + logger.error(f"删除命名空间失败: {e}") + raise Exception(f"删除命名空间失败: {e.reason}") + + def _get_azure_access_token(self) -> str: + """获取 Azure Access Token + + Returns: + Azure 访问令牌 + """ + try: + token_url = f"https://login.microsoftonline.com/{self.AZURE_TENANT_ID}/oauth2/v2.0/token" + data = { + "client_id": self.AZURE_CLIENT_ID, + "client_secret": self.AZURE_CLIENT_SECRET, + "grant_type": "client_credentials", + "scope": "https://management.azure.com/.default" + } + + response = requests.post(token_url, data=data, timeout=10) + response.raise_for_status() + + token = response.json().get('access_token') + if not token: + raise Exception("未能获取访问令牌") + + return token + except Exception as e: + logger.error(f"获取 Azure 访问令牌失败: {e}") + raise + + def create_dns_record(self, subdomain: str, ip_address: str, ttl: int = 300) -> Dict: + """为 IP 地址创建 Azure DNS A 记录 + + Args: + subdomain: 子域名（不包括主域名） + ip_address: 要指向的 IP 地址 + ttl: DNS TTL（秒），默认 300 + + Returns: + DNS 记录信息 + """ + try: + # 获取访问令牌 + token = self._get_azure_access_token() + + # 构建 DNS API URL + dns_url = ( + f"https://management.azure.com/subscriptions/{self.AZURE_SUBSCRIPTION_ID}/" + f"resourceGroups/{self.AZURE_RESOURCE_GROUP}/providers/Microsoft.Network/" + f"dnsZones/{self.AZURE_DNS_ZONE}/A/{subdomain}?api-version=2018-05-01" + ) + + # DNS 记录数据 + dns_data = { + "properties": { + "TTL": ttl, + "ARecords": [{"ipv4Address": ip_address}] + } + } + + # 创建 DNS 记录 + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json" + } + + response = requests.put(dns_url, json=dns_data, headers=headers, timeout=30) + response.raise_for_status() + + full_domain = f"{subdomain}.{self.AZURE_DNS_ZONE}" + logger.info(f"✅ DNS 记录创建成功: {full_domain} -> {ip_address}") + + return { + "subdomain": subdomain, + "domain": full_domain, + "ip_address": ip_address, + "ttl": ttl, + "status": "created" + } + except requests.exceptions.RequestException as e: + logger.error(f"创建 DNS 记录失败: {e}") + if hasattr(e.response, 'text'): + logger.error(f"错误详情: {e.response.text}") + raise Exception(f"创建 DNS 记录失败: {str(e)}") + + def delete_dns_record(self, subdomain: str) -> Dict: + """删除 Azure DNS A 记录 + + Args: + subdomain: 要删除的子域名 + + Returns: + 删除结果 + """ + try: + # 获取访问令牌 + token = self._get_azure_access_token() + + # 构建 DNS API URL + dns_url = ( + f"https://management.azure.com/subscriptions/{self.AZURE_SUBSCRIPTION_ID}/" + f"resourceGroups/{self.AZURE_RESOURCE_GROUP}/providers/Microsoft.Network/" + f"dnsZones/{self.AZURE_DNS_ZONE}/A/{subdomain}?api-version=2018-05-01" + ) + + headers = {"Authorization": f"Bearer {token}"} + + response = requests.delete(dns_url, headers=headers, timeout=30) + + if response.status_code == 404: + logger.info(f"DNS 记录 {subdomain} 不存在，跳过删除") + return {"status": "not_found", "subdomain": subdomain} + + response.raise_for_status() + + full_domain = f"{subdomain}.{self.AZURE_DNS_ZONE}" + logger.info(f"✅ DNS 记录删除成功: {full_domain}") + + return { + "subdomain": subdomain, + "domain": full_domain, + "status": "deleted" + } + except requests.exceptions.RequestException as e: + logger.error(f"删除 DNS 记录失败: {e}") + raise Exception(f"删除 DNS 记录失败: {str(e)}") + + def wait_for_loadbalancer_ip(self, service_name: str, namespace: str, + max_wait: int = 300, interval: int = 5) -> str: + """等待 LoadBalancer 分配外网 IP + + Args: + service_name: Service 名称 + namespace: 命名空间 + max_wait: 最大等待时间（秒） + interval: 检查间隔（秒） + + Returns: + 分配的外网 IP 地址 + """ + logger.info(f"等待 LoadBalancer IP 分配... (最多等待 {max_wait} 秒)") + + elapsed = 0 + while elapsed < max_wait: + try: + service = self.v1.read_namespaced_service(service_name, namespace) + if service.status.load_balancer.ingress: + ip = service.status.load_balancer.ingress[0].ip + if ip: + logger.info(f"✅ LoadBalancer IP 已分配: {ip}") + return ip + except Exception as e: + logger.warning(f"查询 Service 状态失败: {e}") + + time.sleep(interval) + elapsed += interval + logger.info(f"等待中... ({elapsed}/{max_wait} 秒)") + + raise Exception(f"LoadBalancer IP 分配超时（{max_wait} 秒）") + def get_pod_status(self, pod_name: str) -> Dict: """ 获取Pod状态 diff --git a/scripts/init_database.sh b/scripts/init_database.sh index ad05c68..d42d122 100644 --- a/scripts/init_database.sh +++ b/scripts/init_database.sh @@ -119,6 +119,41 @@ try: ) db.add(jina_template) + # Intelligent Search Agent (Custom with HTTP service) + search_template = Template( + name="search_agent", + display_name="Intelligent Search Agent", + description="智能搜索代理，提供多源搜索和答案生成功能", + agent_type=AgentType.CUSTOM, + agent_framework="langchain", + image="agnettaiji.azurecr.io/ai-agents/search-agent:v1.0", + port=8080, + env_requirements={ + "required": { + "LLM_BASE_URL": "LLM API基础URL", + "LLM_API_KEY": "LLM API密钥", + "SERPER_API_KEY": "Serper API密钥（Google搜索）", + "JINA_API_KEY": "Jina API密钥（内容提取和重排序）" + }, + "optional": { + "LLM_MODEL": "LLM模型名称，默认xchat52", + "MAX_ITERATIONS": "最大迭代次数，默认3", + "MAX_RESULTS_PER_QUERY": "每次搜索最大结果数，默认10", + "CONTENT_MAX_LENGTH": "内容最大长度，默认5000", + "LOG_LEVEL": "日志级别，默认INFO", + "TIMEOUT": "超时时间(秒)，默认30" + } + }, + cpu_request="500m", + cpu_limit="1000m", + memory_request="512Mi", + memory_limit="1Gi", + min_replicas=1, + max_replicas=3, + target_cpu_utilization=70 + ) + db.add(search_template) + # Echo Agent (Platform - simple example) echo_template = Template( name="echo_agent", @@ -139,7 +174,7 @@ try: db.add(echo_template) db.commit() - print("✓ Created 4 default templates") + print("✓ Created 5 default templates") # 创建默认配额 print("\nCreating default quotas...")