diff --git a/.env.example b/.env.example index 13104a9..70e5b23 100644 --- a/.env.example +++ b/.env.example @@ -33,3 +33,8 @@ RAPIDAPI_HOST=rapidapi.com REDIS_URL=redis://redis:6379 NATS_URL=nats://nats:4222 DATABASE_URL=postgresql://taiji_user:taiji_pass@postgres:5432/taiji_db + +# ========== AI Agent Manager 配置 ========== +# Kubernetes Agent Pod 管理服务 +AGENT_MANAGER_URL=http://localhost:8000 +AGENT_K8S_NAMESPACE=ai-agents diff --git a/API_DOCUMENTATION.md b/API_DOCUMENTATION.md new file mode 100644 index 0000000..bf59ec9 --- /dev/null +++ b/API_DOCUMENTATION.md @@ -0,0 +1,666 @@ +# AI Agent Manager API 文档 + +## 概述 + +AI Agent Manager 是一个基于 FastAPI 构建的 Kubernetes AI Agent 管理服务。该服务提供 RESTful API 接口，用于在 Kubernetes 集群中创建、删除、查询和管理 AI Agent Pod。 + +### 基本信息 + +| 项目 | 值 | +|------|-----| +| 服务名称 | AI Agent Manager | +| 版本 | 1.0.0 | +| 基础URL | `http://:` | +| 默认端口 | 8000 | +| 内容类型 | `application/json` | + +### 环境变量配置 + +| 变量名 | 默认值 | 说明 | +|--------|--------|------| +| `NAMESPACE` | `ai-agents` | AI Agent 部署的 Kubernetes 命名空间 | +| `KUBECONFIG_PATH` | `None` | kubeconfig 文件路径（可选） | +| `SERVICE_HOST` | `0.0.0.0` | 服务监听地址 | +| `SERVICE_PORT` | `8000` | 服务监听端口 | + +--- + +## API 端点 + +### 1. 健康检查 + +检查服务运行状态。 + +**请求** + +``` +GET / +``` + +**响应** + +```json +{ + "service": "AI Agent Manager", + "status": "running", + "namespace": "ai-agents" +} +``` + +**cURL 示例** + +```bash +curl -X GET "http://localhost:8000/" +``` + +--- + +### 2. 列出所有模板 + +获取所有可用的 Agent 模板及其所需参数。 + +**请求** + +``` +GET /templates +``` + +**成功响应 (200)** + +```json +{ + "templates": [ + { + "template": "echo_agent", + "port": null, + "env_info": {} + }, + { + "template": "jina_search_agent", + "port": 8080, + "env_info": { + "required": { + "JINA_API_KEY": "Jina API密钥，从 https://jina.ai/ 获取" + }, + "optional": { + "SERVICE_PORT": "HTTP服务端口，默认8080", + "SERVICE_HOST": "HTTP服务监听地址，默认0.0.0.0" + } + } + }, + { + "template": "mysql_agent", + "port": null, + "env_info": { + "required": { + "MYSQL_HOST": "MySQL数据库主机地址", + "MYSQL_USER": "MySQL用户名", + "MYSQL_PASSWORD": "MySQL密码", + "MYSQL_DATABASE": "MySQL数据库名", + "OPENAI_API_KEY": "OpenAI API密钥" + }, + "optional": { + "MYSQL_PORT": "MySQL端口，默认3306" + } + } + } + ], + "count": 7 +} +``` + +**cURL 示例** + +```bash +curl -X GET "http://localhost:8000/templates" +``` + +--- + +### 3. 获取模板详情 + +获取指定模板的详细信息，包括所需环境变量。 + +**请求** + +``` +GET /templates/{template_name} +``` + +**路径参数** + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `template_name` | string | 是 | 模板名称 | + +**成功响应 (200)** + +```json +{ + "template": "jina_search_agent", + "port": 8080, + "env_info": { + "required": { + "JINA_API_KEY": "Jina API密钥，从 https://jina.ai/ 获取" + }, + "optional": { + "SERVICE_PORT": "HTTP服务端口，默认8080", + "SERVICE_HOST": "HTTP服务监听地址，默认0.0.0.0" + } + } +} +``` + +**错误响应** + +| 状态码 | 说明 | +|--------|------| +| 404 | 模板不存在 | + +**cURL 示例** + +```bash +curl -X GET "http://localhost:8000/templates/jina_search_agent" +``` + +--- + +### 4. 创建 Agent + +创建一个新的 AI Agent Pod。 + +**请求** + +``` +POST /agents +``` + +**请求头** + +``` +Content-Type: application/json +``` + +**请求体参数** + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `name` | string | 是 | Agent 名称，长度 1-63 字符 | +| `template` | string | 是 | 模板类型（见模板列表） | +| `config` | object | 否 | 配置信息（见下表） | + +**config 配置参数** + +| 参数 | 类型 | 默认值 | 说明 | +|------|------|--------|------| +| `replicas` | integer | 1 | 副本数量 | +| `cpu_request` | string | `100m` | CPU 请求量 | +| `cpu_limit` | string | `500m` | CPU 限制量 | +| `memory_request` | string | `128Mi` | 内存请求量 | +| `memory_limit` | string | `512Mi` | 内存限制量 | +| `env` | object | `{}` | 自定义环境变量 | + +**请求体示例（带环境变量）** + +```json +{ + "name": "my-jina-agent", + "template": "jina_search_agent", + "config": { + "cpu_request": "100m", + "cpu_limit": "500m", + "memory_request": "128Mi", + "memory_limit": "512Mi", + "env": { + "JINA_API_KEY": "your-jina-api-key" + } + } +} +``` + +**成功响应 (200) - HTTP服务类型Agent** + +```json +{ + "name": "my-jina-agent", + "namespace": "ai-agents", + "status": "Pending", + "created_at": "2025-12-31T01:00:00.000000+00:00", + "template": "jina_search_agent", + "service_port": 8080, + "access_info": { + "note": "Pod IP将在Pod运行后可用，请通过 /agents/{name}/status 获取", + "port": 8080, + "endpoints": { + "root": "http://:8080/", + "health": "http://:8080/health" + } + } +} +``` + +**成功响应 (200) - 普通Agent** + +```json +{ + "name": "my-echo-agent", + "namespace": "ai-agents", + "status": "Pending", + "created_at": "2025-12-31T01:00:00.000000+00:00", + "template": "echo_agent" +} +``` + +**错误响应** + +| 状态码 | 说明 | +|--------|------| +| 400 | 无效的模板类型 | +| 500 | 服务器内部错误 | + +**cURL 示例** + +```bash +# 创建 Jina Search Agent（带环境变量） +curl -X POST "http://localhost:8000/agents" \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-jina-agent", + "template": "jina_search_agent", + "config": { + "env": { + "JINA_API_KEY": "your-jina-api-key" + } + } + }' + +# 创建 MySQL Agent +curl -X POST "http://localhost:8000/agents" \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-mysql-agent", + "template": "mysql_agent", + "config": { + "env": { + "MYSQL_HOST": "mysql.example.com", + "MYSQL_USER": "root", + "MYSQL_PASSWORD": "password", + "MYSQL_DATABASE": "mydb", + "OPENAI_API_KEY": "sk-xxx" + } + } + }' +``` + +--- + +### 5. 删除 Agent + +删除指定的 AI Agent Pod。 + +**请求** + +``` +DELETE /agents/{agent_name} +``` + +**路径参数** + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `agent_name` | string | 是 | 要删除的 Agent 名称 | + +**成功响应 (200)** + +```json +{ + "status": "success", + "message": "Pod my-jina-agent 已删除" +} +``` + +**cURL 示例** + +```bash +curl -X DELETE "http://localhost:8000/agents/my-jina-agent" +``` + +--- + +### 6. 获取 Agent 状态 + +获取指定 Agent 的详细状态信息，包括访问URL。 + +**请求** + +``` +GET /agents/{agent_name}/status +``` + +**路径参数** + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `agent_name` | string | 是 | Agent 名称 | + +**成功响应 (200) - HTTP服务类型Agent（运行中）** + +```json +{ + "name": "my-jina-agent", + "namespace": "ai-agents", + "status": "Running", + "template": "jina_search_agent", + "created_at": "2025-12-31T01:00:00.000000+00:00", + "node": "aks-nodepool1-12345678-vmss000000", + "pod_ip": "10.244.0.15", + "service_port": 8080, + "access_url": "http://10.244.0.15:8080", + "endpoints": { + "root": "http://10.244.0.15:8080/", + "health": "http://10.244.0.15:8080/health" + }, + "conditions": [ + { + "type": "Ready", + "status": "True", + "reason": null + } + ] +} +``` + +**状态值说明** + +| 状态 | 说明 | +|------|------| +| `Pending` | Pod 已被接受，但容器尚未创建 | +| `Running` | Pod 已绑定到节点，所有容器已创建 | +| `Succeeded` | Pod 中所有容器已成功终止 | +| `Failed` | Pod 中所有容器已终止，至少一个容器失败 | +| `Unknown` | 无法获取 Pod 状态 | + +**cURL 示例** + +```bash +curl -X GET "http://localhost:8000/agents/my-jina-agent/status" +``` + +--- + +### 7. 获取 Agent 资源使用情况 + +获取指定 Agent 的 CPU 和内存资源配置信息。 + +**请求** + +``` +GET /agents/{agent_name}/metrics +``` + +**成功响应 (200)** + +```json +{ + "name": "my-jina-agent", + "requests": { + "cpu": "100m", + "memory": "128Mi" + }, + "limits": { + "cpu": "500m", + "memory": "512Mi" + } +} +``` + +**cURL 示例** + +```bash +curl -X GET "http://localhost:8000/agents/my-jina-agent/metrics" +``` + +--- + +### 8. 列出所有 Agent + +获取所有 AI Agent 的列表，支持按模板类型过滤。 + +**请求** + +``` +GET /agents +``` + +**查询参数** + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `template` | string | 否 | 按模板类型过滤 | + +**成功响应 (200)** + +```json +{ + "agents": [ + { + "name": "my-jina-agent", + "status": "Running", + "template": "jina_search_agent", + "created_at": "2025-12-31T01:00:00.000000+00:00", + "pod_ip": "10.244.0.15" + } + ], + "count": 1 +} +``` + +**cURL 示例** + +```bash +# 列出所有 Agent +curl -X GET "http://localhost:8000/agents" + +# 按模板类型过滤 +curl -X GET "http://localhost:8000/agents?template=jina_search_agent" +``` + +--- + +## 模板类型说明 + +### echo_agent + +简单的回显 Agent，用于测试和演示。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/echo-agent:latest` +- **用途**: 测试、演示、健康检查 +- **所需环境变量**: 无 + +### chat_agent + +聊天对话 Agent，支持对话交互。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/chat-agent:latest` +- **用途**: 对话系统、客服机器人 +- **所需环境变量**: 无 + +### code_agent + +代码生成 Agent，支持代码相关任务。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/code-agent:latest` +- **用途**: 代码生成、代码审查、代码解释 +- **所需环境变量**: 无 + +### search_agent + +搜索 Agent。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/search-agent:latest` +- **用途**: 搜索功能 +- **所需环境变量**: 无 + +### jina_search_agent + +使用 Jina Reader API 获取网站内容的 HTTP 服务 Agent。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/jina-search-agent:latest` +- **服务端口**: 8080 +- **用途**: 网站内容抓取、网页转文本 +- **所需环境变量**: + | 变量名 | 必填 | 说明 | + |--------|------|------| + | `JINA_API_KEY` | 是 | Jina API密钥 | + | `SERVICE_PORT` | 否 | HTTP服务端口，默认8080 | + | `SERVICE_HOST` | 否 | HTTP服务监听地址，默认0.0.0.0 | + +- **Agent API端点**: + - `GET /` - 服务信息和所需参数 + - `GET /health` - 健康检查 + - `POST /search` - 搜索网站内容 + - `GET /fetch?url=` - 快速获取网站内容 + +### mysql_agent + +MySQL 数据库查询 Agent，使用 LangChain 实现。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/mysql-agent:latest` +- **用途**: MySQL 数据库自然语言查询 +- **所需环境变量**: + | 变量名 | 必填 | 说明 | + |--------|------|------| + | `MYSQL_HOST` | 是 | MySQL数据库主机地址 | + | `MYSQL_USER` | 是 | MySQL用户名 | + | `MYSQL_PASSWORD` | 是 | MySQL密码 | + | `MYSQL_DATABASE` | 是 | MySQL数据库名 | + | `OPENAI_API_KEY` | 是 | OpenAI API密钥 | + | `MYSQL_PORT` | 否 | MySQL端口，默认3306 | + +### postgresql_agent + +PostgreSQL 数据库查询 Agent，使用 LangChain 实现。 + +- **镜像**: `agnettaiji.azurecr.io/ai-agents/postgresql-agent:latest` +- **用途**: PostgreSQL 数据库自然语言查询 +- **所需环境变量**: + | 变量名 | 必填 | 说明 | + |--------|------|------| + | `POSTGRES_HOST` | 是 | PostgreSQL数据库主机地址 | + | `POSTGRES_USER` | 是 | PostgreSQL用户名 | + | `POSTGRES_PASSWORD` | 是 | PostgreSQL密码 | + | `POSTGRES_DATABASE` | 是 | PostgreSQL数据库名 | + | `OPENAI_API_KEY` | 是 | OpenAI API密钥 | + | `POSTGRES_PORT` | 否 | PostgreSQL端口，默认5432 | + +--- + +## 完整使用流程示例 + +### 场景：创建并使用 Jina Search Agent + +```python +import requests +import time + +BASE_URL = "http://localhost:8000" + +# 1. 查看模板所需参数 +print("=== 查看模板信息 ===") +response = requests.get(f"{BASE_URL}/templates/jina_search_agent") +template_info = response.json() +print(f"所需环境变量: {template_info['env_info']}") + +# 2. 创建 Agent +print("\n=== 创建 Agent ===") +response = requests.post( + f"{BASE_URL}/agents", + json={ + "name": "my-jina-agent", + "template": "jina_search_agent", + "config": { + "env": { + "JINA_API_KEY": "your-jina-api-key" + } + } + } +) +create_result = response.json() +print(f"创建结果: {create_result}") + +# 3. 等待 Agent 运行 +print("\n=== 等待 Agent 启动 ===") +for i in range(30): + response = requests.get(f"{BASE_URL}/agents/my-jina-agent/status") + status = response.json() + if status.get("status") == "Running": + print(f"Agent 已启动!") + print(f"访问地址: {status.get('access_url')}") + break + print(f"当前状态: {status.get('status')}, 等待中...") + time.sleep(2) + +# 4. 调用 Agent 服务 +if status.get("access_url"): + agent_url = status["access_url"] + + # 查看 Agent 信息 + print("\n=== Agent 服务信息 ===") + response = requests.get(f"{agent_url}/") + print(response.json()) + + # 搜索网站内容 + print("\n=== 搜索网站内容 ===") + response = requests.post( + f"{agent_url}/search", + json={"url": "https://www.example.com"} + ) + print(response.json()) + +# 5. 删除 Agent +print("\n=== 删除 Agent ===") +response = requests.delete(f"{BASE_URL}/agents/my-jina-agent") +print(response.json()) +``` + +--- + +## 错误处理 + +### 通用错误响应格式 + +```json +{ + "detail": "错误描述信息" +} +``` + +### HTTP 状态码 + +| 状态码 | 说明 | +|--------|------| +| 200 | 请求成功 | +| 400 | 请求参数错误 | +| 404 | 资源不存在 | +| 500 | 服务器内部错误 | + +--- + +## OpenAPI/Swagger 文档 + +FastAPI 自动生成交互式 API 文档： + +- **Swagger UI**: `http://localhost:8000/docs` +- **ReDoc**: `http://localhost:8000/redoc` +- **OpenAPI JSON**: `http://localhost:8000/openapi.json` + +--- + +## 注意事项 + +1. **Agent 名称规范**: 名称必须符合 Kubernetes 命名规范（小写字母、数字、连字符，长度 1-63） +2. **资源配额**: 请根据实际需求配置 CPU 和内存，避免资源浪费或不足 +3. **命名空间**: 所有 Agent 默认部署在 `ai-agents` 命名空间 +4. **镜像拉取**: 需要配置 ACR 密钥（`acr-secret`）才能拉取私有镜像 +5. **网络访问**: 服务默认监听所有网络接口（0.0.0.0），生产环境请注意安全配置 +6. **环境变量安全**: 敏感信息（如 API 密钥）应通过安全方式传递，避免在日志中暴露 diff --git a/docker-compose.yml b/docker-compose.yml index 58a5013..311f9af 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -84,6 +84,9 @@ services: - REDIS_URL=${REDIS_URL} - NATS_URL=nats://nats:4222 - LITELLM_URL=http://litellm-gateway:4000 + - AGENT_MANAGER_URL=${AGENT_MANAGER_URL:-http://host.docker.internal:8000} + extra_hosts: + - "host.docker.internal:host-gateway" volumes: - ./services/mcp-server:/app - ./logs:/app/logs diff --git a/plans/agent-api-integration-plan.md b/plans/agent-api-integration-plan.md new file mode 100644 index 0000000..94afc79 --- /dev/null +++ b/plans/agent-api-integration-plan.md @@ -0,0 +1,191 @@ +# AI Agent Manager API 集成计划 + +## 概述 + +将项目中的 Agent 管理功能与新的 AI Agent Manager API 集成，实现真正的 Kubernetes Pod 部署和资源管控。 + +## 当前状态分析 + +### 现有实现 +- **Agent 数据模型** ([`models.py`](../services/mcp-server/models.py:122))：数据库级别的 Agent 记录 +- **Agent 路由** ([`agents.py`](../services/mcp-server/app/routes/agents.py:1))：CRUD 操作，无真正 K8s 部署 +- **资源管控** ([`resource_control.py`](../services/mcp-server/app/resource_control.py:1))：配额检查和速率限制 +- **部署功能** ([`user.py:deploy_agent`](../services/mcp-server/app/routes/user.py:327))：仅模拟，未实际调用 K8s + +### 新 API 能力 +- 真正的 Kubernetes Pod 创建/删除/查询 +- 模板系统（echo_agent, jina_search_agent, mysql_agent 等） +- 资源配置（cpu_request, cpu_limit, memory_request, memory_limit） +- Pod 状态监控和访问信息 + +## 架构设计 + +```mermaid +flowchart TB + subgraph Frontend[前端] + UI[用户界面] + end + + subgraph MCPServer[MCP Server] + AgentRoutes[Agent 路由] + ResourceControl[资源管控] + AgentClient[Agent Manager 客户端] + DB[(PostgreSQL)] + end + + subgraph K8sCluster[Kubernetes 集群] + AgentManager[AI Agent Manager API] + AgentPods[Agent Pods] + end + + UI --> AgentRoutes + AgentRoutes --> ResourceControl + ResourceControl --> AgentClient + AgentClient --> AgentManager + AgentManager --> AgentPods + AgentRoutes --> DB +``` + +## 实施任务清单 + +### 阶段一：创建 Agent Manager 客户端 + +- [x] **1.1** 创建 `services/mcp-server/app/agent_manager_client.py` + - 封装对 AI Agent Manager API 的 HTTP 调用 + - 支持健康检查、模板列表、创建/删除/查询 Agent + - 使用 httpx 异步客户端 + - 配置通过环境变量 `AGENT_MANAGER_URL` 设置 + +- [x] **1.2** 创建请求/响应模型（集成到 `schemas.py`） + - K8sResourceConfig（cpu_request, cpu_limit, memory_request, memory_limit, env） + - AgentStatusResponse（Pod 状态、IP、端点信息） + - AgentMetricsResponse（资源使用情况） + - TemplateInfo, TemplateListResponse + +### 阶段二：修改 Agent 路由 + +- [x] **2.1** 更新 `services/mcp-server/app/routes/agents.py` + - 修改 `create_agent` 函数： + - 保留资源管控检查 + - 调用 Agent Manager API 创建 Pod + - 同步更新数据库记录（添加 pod_name, pod_ip, template 等字段） + - 新增 `delete_agent` 函数：调用 API 删除 Pod + - 新增 `get_agent_status` 函数：获取 Pod 实时状态 + - 新增 `get_agent_metrics` 函数：获取资源使用情况 + - 新增 `list_templates` 和 `get_template` 函数 + +- [x] **2.2** 更新 Agent 数据模型 `services/mcp-server/models.py` + - 添加字段：`pod_name`, `pod_ip`, `template`, `service_port` + - 添加字段：`cpu_request`, `cpu_limit`, `memory_request`, `memory_limit` + - 添加字段：`k8s_status`（Pending, Running, Failed 等） + - 添加字段：`k8s_namespace`, `access_url`, `endpoints`, `env_config`, `pod_created_at` + +- [x] **2.3** 创建数据库迁移脚本 + - `services/mcp-server/migrations/004_add_k8s_agent_fields.sql` + +### 阶段三：更新资源管控 + +- [ ] **3.1** 更新 `services/mcp-server/app/resource_control.py` + - 添加 Agent 资源配额检查（用户可创建的 Agent 数量限制） + - 添加 CPU/内存总量限制检查 + - 集成 Agent Manager 的 metrics API 获取实际资源使用 + +- [x] **3.2** 更新 `services/mcp-server/app/routes/admin.py` + - 修改 `update_agent_config` 函数： + - 支持更新 cpu_request, cpu_limit, memory_request, memory_limit + - 注意：已运行的 Pod 需要重新创建才能更新资源配置 + +### 阶段四：更新用户侧功能 + +- [x] **4.1** 更新 `services/mcp-server/app/routes/user.py` + - 修改 `deploy_agent` 函数：调用 Agent Manager API + - 修改 `generate_tool` 函数：支持自定义资源配置（待完成） + +- [x] **4.2** 更新 `services/mcp-server/app/schemas.py` + - 更新 `UpdateAgentConfigRequest`：添加 K8s 资源配置字段 + +### 阶段五：模板管理 + +- [x] **5.1** 模板路由已集成到 `services/mcp-server/app/routes/agents.py` + - `GET /agents/templates`：获取可用模板列表 + - `GET /agents/templates/{name}`：获取模板详情和所需环境变量 + +- [ ] **5.2** 更新路由注册 `services/mcp-server/app/routes/__init__.py`（无需修改，已自动包含） + +### 阶段六：资源监控集成 + +- [ ] **6.1** 更新 `services/mcp-server/app/routes/resource_monitoring.py` + - 集成 Agent Manager 的 `/agents/{name}/metrics` API + - 提供实时 Pod 资源使用数据 + +- [ ] **6.2** 更新 `services/mcp-server/app/routes/monitoring.py` + - 添加 Agent Pod 健康状态监控 + - 添加 Agent Manager 服务健康检查 + +### 阶段七：配置和部署 + +- [x] **7.1** 更新环境变量配置 + - `.env.example`：添加 `AGENT_MANAGER_URL` + - `services/mcp-server/config.py`：添加配置项 + +- [ ] **7.2** 更新 Kubernetes 部署配置 + - `k8s/mcp-server.yaml`：添加环境变量 + - `k8s/configmap.yaml`：添加 Agent Manager URL 配置 + +- [ ] **7.3** 更新 Docker Compose 配置 + - `docker-compose.yml`：添加 Agent Manager 服务依赖 + +## API 映射关系 + +| MCP Server 功能 | AI Agent Manager API | 说明 | +|----------------|---------------------|------| +| 创建 Agent | `POST /agents` | 创建 K8s Pod | +| 删除 Agent | `DELETE /agents/{name}` | 删除 K8s Pod | +| 获取 Agent 状态 | `GET /agents/{name}/status` | 获取 Pod 状态和 IP | +| 获取资源使用 | `GET /agents/{name}/metrics` | 获取 CPU/内存配置 | +| 列出所有 Agent | `GET /agents` | 列出所有 Pod | +| 获取模板列表 | `GET /templates` | 获取可用模板 | +| 获取模板详情 | `GET /templates/{name}` | 获取模板所需参数 | + +## 资源配置映射 + +| 用户配置 | API 参数 | 默认值 | +|---------|---------|-------| +| CPU 请求 | `cpu_request` | `100m` | +| CPU 限制 | `cpu_limit` | `500m` | +| 内存请求 | `memory_request` | `128Mi` | +| 内存限制 | `memory_limit` | `512Mi` | + +## 需要新增的接口 + +根据业务需求，建议新增以下接口： + +1. **Agent 日志查询** + - `GET /agents/{name}/logs` - 获取 Pod 日志 + - 需要 Agent Manager API 支持 + +2. **Agent 重启** + - `POST /agents/{name}/restart` - 重启 Pod + - 可通过删除后重新创建实现 + +3. **Agent 扩缩容** + - `PUT /agents/{name}/scale` - 调整副本数 + - 需要 Agent Manager API 支持 replicas 参数 + +4. **批量操作** + - `POST /agents/batch/create` - 批量创建 + - `DELETE /agents/batch` - 批量删除 + +## 注意事项 + +1. **数据一致性**：数据库记录和 K8s Pod 状态需要保持同步 +2. **错误处理**：API 调用失败时需要回滚数据库操作 +3. **权限控制**：保留现有的资源管控和权限检查 +4. **向后兼容**：保留现有 API 接口格式，扩展返回字段 + +## 测试计划 + +- [ ] 单元测试：Agent Manager 客户端 +- [ ] 集成测试：创建/删除/查询 Agent 流程 +- [ ] 端到端测试：前端到 K8s Pod 完整流程 +- [ ] 性能测试：并发创建 Agent 场景 diff --git a/services/mcp-server/app/agent_manager_client.py b/services/mcp-server/app/agent_manager_client.py new file mode 100644 index 0000000..8337443 --- /dev/null +++ b/services/mcp-server/app/agent_manager_client.py @@ -0,0 +1,385 @@ +""" +AI Agent Manager API 客户端 +封装对 AI Agent Manager 服务的 HTTP 调用 +""" + +import os +import httpx +import structlog +from typing import Dict, List, Optional, Any +from dataclasses import dataclass, field +from enum import Enum + +logger = structlog.get_logger(__name__) + +# 从环境变量获取 Agent Manager URL +AGENT_MANAGER_URL = os.getenv("AGENT_MANAGER_URL", "http://localhost:8000") + + +class AgentStatus(str, Enum): + """Agent Pod 状态""" + PENDING = "Pending" + RUNNING = "Running" + SUCCEEDED = "Succeeded" + FAILED = "Failed" + UNKNOWN = "Unknown" + + +@dataclass +class AgentConfig: + """Agent 资源配置""" + replicas: int = 1 + cpu_request: str = "100m" + cpu_limit: str = "500m" + memory_request: str = "128Mi" + memory_limit: str = "512Mi" + env: Dict[str, str] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "replicas": self.replicas, + "cpu_request": self.cpu_request, + "cpu_limit": self.cpu_limit, + "memory_request": self.memory_request, + "memory_limit": self.memory_limit, + "env": self.env + } + + +@dataclass +class TemplateInfo: + """模板信息""" + template: str + port: Optional[int] + env_info: Dict[str, Any] + + +@dataclass +class AgentCreateResult: + """Agent 创建结果""" + name: str + namespace: str + status: str + created_at: str + template: str + service_port: Optional[int] = None + access_info: Optional[Dict[str, Any]] = None + + +@dataclass +class AgentStatusResult: + """Agent 状态结果""" + name: str + namespace: str + status: str + template: str + created_at: str + node: Optional[str] = None + pod_ip: Optional[str] = None + service_port: Optional[int] = None + access_url: Optional[str] = None + endpoints: Optional[Dict[str, str]] = None + conditions: Optional[List[Dict[str, Any]]] = None + + +@dataclass +class AgentMetricsResult: + """Agent 资源使用结果""" + name: str + requests: Dict[str, str] + limits: Dict[str, str] + + +class AgentManagerError(Exception): + """Agent Manager API 错误""" + def __init__(self, message: str, status_code: int = 500, detail: Any = None): + self.message = message + self.status_code = status_code + self.detail = detail + super().__init__(message) + + +class AgentManagerClient: + """AI Agent Manager API 客户端""" + + def __init__(self, base_url: Optional[str] = None, timeout: float = 30.0): + """ + 初始化客户端 + + Args: + base_url: API 基础 URL，默认从环境变量获取 + timeout: 请求超时时间（秒） + """ + self.base_url = (base_url or AGENT_MANAGER_URL).rstrip("/") + self.timeout = timeout + self._client: Optional[httpx.AsyncClient] = None + + async def _get_client(self) -> httpx.AsyncClient: + """获取或创建 HTTP 客户端""" + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient( + base_url=self.base_url, + timeout=self.timeout, + headers={"Content-Type": "application/json"} + ) + return self._client + + async def close(self): + """关闭客户端""" + if self._client and not self._client.is_closed: + await self._client.aclose() + self._client = None + + async def _request( + self, + method: str, + path: str, + json: Optional[Dict] = None, + params: Optional[Dict] = None + ) -> Dict[str, Any]: + """ + 发送 HTTP 请求 + + Args: + method: HTTP 方法 + path: API 路径 + json: 请求体 + params: 查询参数 + + Returns: + 响应 JSON + + Raises: + AgentManagerError: API 调用失败 + """ + client = await self._get_client() + + try: + response = await client.request( + method=method, + url=path, + json=json, + params=params + ) + + if response.status_code >= 400: + try: + detail = response.json() + except Exception: + detail = response.text + + logger.error( + "agent_manager_api_error", + method=method, + path=path, + status_code=response.status_code, + detail=detail + ) + + raise AgentManagerError( + message=f"API 调用失败: {response.status_code}", + status_code=response.status_code, + detail=detail + ) + + return response.json() + + except httpx.RequestError as e: + logger.error( + "agent_manager_connection_error", + method=method, + path=path, + error=str(e) + ) + raise AgentManagerError( + message=f"连接 Agent Manager 失败: {str(e)}", + status_code=503, + detail={"error": "connection_error", "message": str(e)} + ) + + # ==================== 健康检查 ==================== + + async def health_check(self) -> Dict[str, Any]: + """ + 检查 Agent Manager 服务状态 + + Returns: + 服务状态信息 + """ + return await self._request("GET", "/") + + # ==================== 模板管理 ==================== + + async def list_templates(self) -> List[TemplateInfo]: + """ + 获取所有可用模板 + + Returns: + 模板列表 + """ + data = await self._request("GET", "/templates") + templates = [] + for t in data.get("templates", []): + templates.append(TemplateInfo( + template=t["template"], + port=t.get("port"), + env_info=t.get("env_info", {}) + )) + return templates + + async def get_template(self, template_name: str) -> TemplateInfo: + """ + 获取模板详情 + + Args: + template_name: 模板名称 + + Returns: + 模板信息 + """ + data = await self._request("GET", f"/templates/{template_name}") + return TemplateInfo( + template=data["template"], + port=data.get("port"), + env_info=data.get("env_info", {}) + ) + + # ==================== Agent 管理 ==================== + + async def create_agent( + self, + name: str, + template: str, + config: Optional[AgentConfig] = None + ) -> AgentCreateResult: + """ + 创建 Agent Pod + + Args: + name: Agent 名称（1-63 字符，小写字母、数字、连字符） + template: 模板类型 + config: 资源配置 + + Returns: + 创建结果 + """ + payload = { + "name": name, + "template": template + } + + if config: + payload["config"] = config.to_dict() + + logger.info( + "creating_agent", + name=name, + template=template, + config=config.to_dict() if config else None + ) + + data = await self._request("POST", "/agents", json=payload) + + return AgentCreateResult( + name=data["name"], + namespace=data["namespace"], + status=data["status"], + created_at=data["created_at"], + template=data["template"], + service_port=data.get("service_port"), + access_info=data.get("access_info") + ) + + async def delete_agent(self, agent_name: str) -> Dict[str, Any]: + """ + 删除 Agent Pod + + Args: + agent_name: Agent 名称 + + Returns: + 删除结果 + """ + logger.info("deleting_agent", name=agent_name) + return await self._request("DELETE", f"/agents/{agent_name}") + + async def get_agent_status(self, agent_name: str) -> AgentStatusResult: + """ + 获取 Agent 状态 + + Args: + agent_name: Agent 名称 + + Returns: + Agent 状态信息 + """ + data = await self._request("GET", f"/agents/{agent_name}/status") + + return AgentStatusResult( + name=data["name"], + namespace=data["namespace"], + status=data["status"], + template=data["template"], + created_at=data["created_at"], + node=data.get("node"), + pod_ip=data.get("pod_ip"), + service_port=data.get("service_port"), + access_url=data.get("access_url"), + endpoints=data.get("endpoints"), + conditions=data.get("conditions") + ) + + async def get_agent_metrics(self, agent_name: str) -> AgentMetricsResult: + """ + 获取 Agent 资源使用情况 + + Args: + agent_name: Agent 名称 + + Returns: + 资源使用信息 + """ + data = await self._request("GET", f"/agents/{agent_name}/metrics") + + return AgentMetricsResult( + name=data["name"], + requests=data.get("requests", {}), + limits=data.get("limits", {}) + ) + + async def list_agents(self, template: Optional[str] = None) -> List[Dict[str, Any]]: + """ + 列出所有 Agent + + Args: + template: 按模板类型过滤（可选） + + Returns: + Agent 列表 + """ + params = {} + if template: + params["template"] = template + + data = await self._request("GET", "/agents", params=params) + return data.get("agents", []) + + +# 全局客户端实例 +_agent_manager_client: Optional[AgentManagerClient] = None + + +def get_agent_manager_client() -> AgentManagerClient: + """获取全局 Agent Manager 客户端实例""" + global _agent_manager_client + if _agent_manager_client is None: + _agent_manager_client = AgentManagerClient() + return _agent_manager_client + + +async def close_agent_manager_client(): + """关闭全局客户端""" + global _agent_manager_client + if _agent_manager_client: + await _agent_manager_client.close() + _agent_manager_client = None diff --git a/services/mcp-server/app/auth.py b/services/mcp-server/app/auth.py index 9c50dac..b0a7cc9 100644 --- a/services/mcp-server/app/auth.py +++ b/services/mcp-server/app/auth.py @@ -106,8 +106,16 @@ async def require_auth( "/api/channel/auth/login", "/api/admin/auth/login", "/api/providers/auth/login", + "/agents/templates", # 模板列表公开访问 } - if path in allow_paths or not path.startswith("/api"): + # 允许公开路径和非 API/agents 路径 + if path in allow_paths: + return {} + # 模板详情也公开访问 + if path.startswith("/agents/templates/"): + return {} + # 非 API 且非 agents 路径不需要认证 + if not path.startswith("/api") and not path.startswith("/agents"): return {} api_key_header = request.headers.get("X-API-Key") @@ -137,6 +145,7 @@ async def require_auth( async def authenticate_request(request: Request, db: AsyncSession) -> Optional[Dict[str, Any]]: """Authenticate a request without FastAPI dependency injection (middleware use).""" + path = request.url.path allow_paths = { "/health", "/metrics", @@ -146,8 +155,16 @@ async def authenticate_request(request: Request, db: AsyncSession) -> Optional[D "/api/channel/auth/login", "/api/admin/auth/login", "/api/providers/auth/login", + "/agents/templates", # 模板列表公开访问 } - if request.url.path in allow_paths or not request.url.path.startswith("/api"): + # 允许公开路径 + if path in allow_paths: + return {} + # 模板详情也公开访问 + if path.startswith("/agents/templates/"): + return {} + # 非 API 且非 agents 路径不需要认证 + if not path.startswith("/api") and not path.startswith("/agents"): return {} api_key_header = request.headers.get("X-API-Key") diff --git a/services/mcp-server/app/routes/admin.py b/services/mcp-server/app/routes/admin.py index 5c4aaf9..92bba90 100644 --- a/services/mcp-server/app/routes/admin.py +++ b/services/mcp-server/app/routes/admin.py @@ -892,6 +892,12 @@ async def update_agent_config( ): """ 更新Agent资源配置（super_admin 和 billing_admin 可用） + + 支持两种格式： + 1. 旧格式（数值）：cpu=2.0, memory=4.0 + 2. K8s格式（字符串）：cpu_request="100m", memory_limit="512Mi" + + 注意：如果 Agent 有关联的 K8s Pod，更新资源配置需要重新创建 Pod 才能生效。 """ _verify_write_permission(principal) @@ -907,7 +913,7 @@ async def update_agent_config( detail="Agent资源不存在" ) - # 更新配置 + # 更新旧格式配置 if req.cpu is not None: agent.cpu = req.cpu if req.memory is not None: @@ -915,18 +921,41 @@ async def update_agent_config( if req.maxInstances is not None: agent.max_instances = req.maxInstances + # 更新 K8s 格式配置 + if req.cpu_request is not None: + agent.cpu_request = req.cpu_request + if req.cpu_limit is not None: + agent.cpu_limit = req.cpu_limit + if req.memory_request is not None: + agent.memory_request = req.memory_request + if req.memory_limit is not None: + agent.memory_limit = req.memory_limit + await db.commit() await db.refresh(agent) + # 构建响应数据 + response_data = { + "id": str(agent.id), + "name": agent.name, + "cpu": float(agent.cpu) if agent.cpu else None, + "memory": float(agent.memory) if agent.memory else None, + "maxInstances": agent.max_instances, + # K8s 格式 + "cpu_request": agent.cpu_request, + "cpu_limit": agent.cpu_limit, + "memory_request": agent.memory_request, + "memory_limit": agent.memory_limit, + } + + # 如果有 Pod，提示需要重新创建 + message = "Agent资源配置更新成功" + if agent.pod_name: + message += "。注意：已运行的 Pod 需要重新创建才能应用新的资源配置。" + return SuccessResponse( - data={ - "id": str(agent.id), - "name": agent.name, - "cpu": float(agent.cpu), - "memory": float(agent.memory), - "maxInstances": agent.max_instances, - }, - message="Agent资源配置更新成功" + data=response_data, + message=message ) diff --git a/services/mcp-server/app/routes/agents.py b/services/mcp-server/app/routes/agents.py index 7a1d9cb..971e873 100644 --- a/services/mcp-server/app/routes/agents.py +++ b/services/mcp-server/app/routes/agents.py @@ -1,4 +1,4 @@ -"""Agent CRUD and execution endpoints.""" +"""Agent CRUD and execution endpoints with Kubernetes integration.""" from __future__ import annotations @@ -8,15 +8,20 @@ import uuid import structlog from datetime import datetime from decimal import Decimal -from typing import List, Optional +from typing import List, Optional, Dict, Any -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from database import get_db from models import Agent, User, Billing, Execution, Balance, Session -from schemas import AgentCard, AgentCreateRequest, ExecutionResult, MCPRequest, SessionCreate, SessionResponse +from schemas import ( + AgentCard, AgentCreateRequest, ExecutionResult, MCPRequest, + SessionCreate, SessionResponse, + AgentStatusResponse, AgentMetricsResponse, TemplateInfo, TemplateListResponse, + K8sResourceConfig +) from ..metrics import ( agents_queries_total, agents_registered_total, @@ -27,12 +32,18 @@ from ..state import get_state from ..utils import record_tool_metrics from ..auth import get_current_user from ..resource_control import enforce_resource_control, resource_controller +from ..agent_manager_client import ( + get_agent_manager_client, + AgentConfig, + AgentManagerError, +) logger = structlog.get_logger(__name__) router = APIRouter(prefix="/agents", tags=["agents"]) def _build_agent_card(agent: Agent) -> AgentCard: + """构建 Agent 卡片信息""" return AgentCard( id=agent.id, name=agent.name, @@ -41,18 +52,32 @@ def _build_agent_card(agent: Agent) -> AgentCard: goal=agent.goal, tools=agent.tools or [], capabilities=agent.capabilities or [], - endpoints={ + endpoints=agent.endpoints or { "mcp": f"mcp://localhost:8002/agents/{agent.id}", "http": f"http://localhost:8002/agents/{agent.id}", "websocket": f"ws://localhost:8002/agents/{agent.id}/ws", }, status=agent.status, version=agent.version, + # K8s 相关信息 + template=agent.template, + pod_name=agent.pod_name, + pod_ip=agent.pod_ip, + k8s_status=agent.k8s_status, + service_port=agent.service_port, + access_url=agent.access_url, + # 资源配置 + cpu_request=agent.cpu_request, + cpu_limit=agent.cpu_limit, + memory_request=agent.memory_request, + memory_limit=agent.memory_limit, + # 统计信息 total_executions=agent.total_executions, success_rate=agent.success_rate, avg_execution_time=agent.avg_execution_time, created_at=agent.created_at, updated_at=agent.updated_at, + pod_created_at=agent.pod_created_at, ) @@ -66,20 +91,73 @@ async def _get_balance(db: AsyncSession, user_id: uuid.UUID) -> Balance: return balance +# ==================== 模板管理 ==================== + +@router.get("/templates", response_model=TemplateListResponse) +async def list_templates() -> TemplateListResponse: + """获取所有可用的 Agent 模板""" + try: + client = get_agent_manager_client() + templates = await client.list_templates() + + return TemplateListResponse( + templates=[ + TemplateInfo( + template=t.template, + port=t.port, + env_info=t.env_info + ) + for t in templates + ], + count=len(templates) + ) + except AgentManagerError as e: + logger.error("获取模板列表失败", error=str(e)) + raise HTTPException(status_code=e.status_code, detail=e.detail) + except Exception as e: + logger.error("获取模板列表失败", error=str(e)) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.get("/templates/{template_name}", response_model=TemplateInfo) +async def get_template(template_name: str) -> TemplateInfo: + """获取模板详情""" + try: + client = get_agent_manager_client() + template = await client.get_template(template_name) + + return TemplateInfo( + template=template.template, + port=template.port, + env_info=template.env_info + ) + except AgentManagerError as e: + logger.error("获取模板详情失败", template=template_name, error=str(e)) + raise HTTPException(status_code=e.status_code, detail=e.detail) + except Exception as e: + logger.error("获取模板详情失败", template=template_name, error=str(e)) + raise HTTPException(status_code=500, detail=str(e)) + + +# ==================== Agent CRUD ==================== + @router.post("", response_model=AgentCard) async def create_agent( request: AgentCreateRequest, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ) -> AgentCard: - """Create a new agent for the authenticated user.""" + """ + 创建新的 Agent。 + + 如果提供了 template，将在 Kubernetes 中创建对应的 Pod。 + """ state = get_state() try: user_id = uuid.UUID(current_user["user_id"]) # ========== 资源管控检查 ========== - # 创建Agent前验证账户状态（不扣费） await enforce_resource_control( user_id=str(user_id), resource_type="agent", @@ -89,6 +167,10 @@ async def create_agent( ) # ================================== + # 准备资源配置 + resource_config = request.resource_config or K8sResourceConfig() + + # 创建数据库记录 agent = Agent( name=request.name, description=request.description, @@ -98,18 +180,86 @@ async def create_agent( config=request.config, capabilities=request.capabilities, owner_id=user_id, + # K8s 相关字段 + template=request.template, + cpu_request=resource_config.cpu_request, + cpu_limit=resource_config.cpu_limit, + memory_request=resource_config.memory_request, + memory_limit=resource_config.memory_limit, + env_config=resource_config.env, + k8s_status="Unknown", ) db.add(agent) + await db.flush() # 获取 agent.id + + # 如果指定了模板，创建 K8s Pod + if request.template: + try: + client = get_agent_manager_client() + + # 构建 Pod 名称（使用 agent ID 确保唯一性） + pod_name = f"{request.name}-{str(agent.id)[:8]}" + + # 创建 Agent 配置 + agent_config = AgentConfig( + replicas=resource_config.replicas, + cpu_request=resource_config.cpu_request, + cpu_limit=resource_config.cpu_limit, + memory_request=resource_config.memory_request, + memory_limit=resource_config.memory_limit, + env=resource_config.env + ) + + # 调用 Agent Manager API 创建 Pod + result = await client.create_agent( + name=pod_name, + template=request.template, + config=agent_config + ) + + # 更新数据库记录 + agent.pod_name = result.name + agent.k8s_namespace = result.namespace + agent.k8s_status = result.status + agent.service_port = result.service_port + agent.pod_created_at = datetime.fromisoformat(result.created_at.replace("Z", "+00:00")) + + if result.access_info: + agent.endpoints = result.access_info.get("endpoints", {}) + + logger.info( + "K8s Pod 创建成功", + agent_id=str(agent.id), + pod_name=result.name, + status=result.status + ) + + except AgentManagerError as e: + # K8s 创建失败，回滚数据库 + await db.rollback() + logger.error("创建 K8s Pod 失败", error=str(e)) + raise HTTPException( + status_code=e.status_code, + detail={ + "error": "k8s_creation_failed", + "message": f"创建 Kubernetes Pod 失败: {e.message}", + "detail": e.detail + } + ) + await db.commit() await db.refresh(agent) card = _build_agent_card(agent) + + # 缓存到 Redis redis_client = state.redis_client if redis_client: await redis_client.setex( f"agent:{agent.id}", 3600, json.dumps(card.model_dump(mode="json"), ensure_ascii=False) ) + # 发布事件到 NATS nats_client = state.nats_client if nats_client: await nats_client.publish( @@ -118,14 +268,19 @@ async def create_agent( { "agent_id": str(agent.id), "name": agent.name, + "template": agent.template, + "pod_name": agent.pod_name, "timestamp": datetime.utcnow().isoformat(), } ).encode(), ) agents_registered_total.labels(status="success").inc() - logger.info("Agent创建成功", agent_id=str(agent.id)) + logger.info("Agent创建成功", agent_id=str(agent.id), template=agent.template) return card + + except HTTPException: + raise except Exception as exc: await db.rollback() agents_registered_total.labels(status="error").inc() @@ -135,17 +290,26 @@ async def create_agent( @router.get("", response_model=List[AgentCard]) async def list_agents( - skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db) + skip: int = 0, + limit: int = 100, + template: Optional[str] = Query(None, description="按模板类型过滤"), + db: AsyncSession = Depends(get_db) ) -> List[AgentCard]: - """Return paginated agent cards.""" + """返回分页的 Agent 列表""" state = get_state() redis_client = state.redis_client try: agents_queries_total.labels(operation="list").inc() - result = await db.execute( - select(Agent).order_by(Agent.created_at.desc()).offset(skip).limit(limit) - ) + + query = select(Agent).order_by(Agent.created_at.desc()) + + # 按模板过滤 + if template: + query = query.where(Agent.template == template) + + query = query.offset(skip).limit(limit) + result = await db.execute(query) agent_rows = result.scalars().all() cards: List[AgentCard] = [] @@ -164,7 +328,7 @@ async def list_agents( @router.get("/{agent_id}", response_model=AgentCard) async def get_agent(agent_id: str, db: AsyncSession = Depends(get_db)) -> AgentCard: - """Fetch a specific agent, using Redis cache when available.""" + """获取指定 Agent 的详细信息""" state = get_state() redis_client = state.redis_client @@ -196,6 +360,210 @@ async def get_agent(agent_id: str, db: AsyncSession = Depends(get_db)) -> AgentC raise HTTPException(status_code=500, detail=str(exc)) from exc +@router.delete("/{agent_id}") +async def delete_agent( + agent_id: str, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +) -> Dict[str, Any]: + """ + 删除 Agent。 + + 如果 Agent 有关联的 K8s Pod，也会一并删除。 + """ + try: + agent_uuid = uuid.UUID(agent_id) + user_id = uuid.UUID(current_user["user_id"]) + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid agent ID") from exc + + agent = await db.get(Agent, agent_uuid) + if not agent: + raise HTTPException(status_code=404, detail="Agent not found") + + # 检查权限 + if agent.owner_id != user_id and current_user.get("role") != "super_admin": + raise HTTPException(status_code=403, detail="Access denied") + + # 如果有 Pod，先删除 K8s Pod + if agent.pod_name: + try: + client = get_agent_manager_client() + await client.delete_agent(agent.pod_name) + logger.info("K8s Pod 删除成功", pod_name=agent.pod_name) + except AgentManagerError as e: + # 如果 Pod 不存在（404），继续删除数据库记录 + if e.status_code != 404: + logger.error("删除 K8s Pod 失败", pod_name=agent.pod_name, error=str(e)) + raise HTTPException( + status_code=e.status_code, + detail={ + "error": "k8s_deletion_failed", + "message": f"删除 Kubernetes Pod 失败: {e.message}", + "detail": e.detail + } + ) + + # 删除数据库记录 + await db.delete(agent) + await db.commit() + + # 清除缓存 + state = get_state() + redis_client = state.redis_client + if redis_client: + await redis_client.delete(f"agent:{agent_id}") + + # 发布事件 + nats_client = state.nats_client + if nats_client: + await nats_client.publish( + "agent.deleted", + json.dumps( + { + "agent_id": agent_id, + "name": agent.name, + "pod_name": agent.pod_name, + "timestamp": datetime.utcnow().isoformat(), + } + ).encode(), + ) + + logger.info("Agent删除成功", agent_id=agent_id) + return {"status": "success", "message": f"Agent {agent.name} 已删除"} + + +# ==================== Agent 状态和监控 ==================== + +@router.get("/{agent_id}/status", response_model=AgentStatusResponse) +async def get_agent_status( + agent_id: str, + db: AsyncSession = Depends(get_db), +) -> AgentStatusResponse: + """ + 获取 Agent 的实时状态。 + + 如果 Agent 有关联的 K8s Pod，会从 Agent Manager 获取最新状态。 + """ + try: + agent_uuid = uuid.UUID(agent_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid agent ID") from exc + + agent = await db.get(Agent, agent_uuid) + if not agent: + raise HTTPException(status_code=404, detail="Agent not found") + + # 如果有 Pod，获取实时状态 + if agent.pod_name: + try: + client = get_agent_manager_client() + status = await client.get_agent_status(agent.pod_name) + + # 更新数据库中的状态 + agent.k8s_status = status.status + agent.pod_ip = status.pod_ip + agent.access_url = status.access_url + if status.endpoints: + agent.endpoints = status.endpoints + + await db.commit() + + return AgentStatusResponse( + id=agent.id, + name=agent.name, + status=agent.status, + k8s_status=status.status, + pod_name=status.name, + pod_ip=status.pod_ip, + node=status.node, + service_port=status.service_port, + access_url=status.access_url, + endpoints=status.endpoints or {}, + cpu_request=agent.cpu_request, + cpu_limit=agent.cpu_limit, + memory_request=agent.memory_request, + memory_limit=agent.memory_limit, + created_at=agent.created_at, + pod_created_at=agent.pod_created_at, + conditions=status.conditions, + ) + + except AgentManagerError as e: + logger.warning("获取 Pod 状态失败", pod_name=agent.pod_name, error=str(e)) + # 返回数据库中的状态 + + # 返回数据库中的状态 + return AgentStatusResponse( + id=agent.id, + name=agent.name, + status=agent.status, + k8s_status=agent.k8s_status or "Unknown", + pod_name=agent.pod_name, + pod_ip=agent.pod_ip, + service_port=agent.service_port, + access_url=agent.access_url, + endpoints=agent.endpoints or {}, + cpu_request=agent.cpu_request, + cpu_limit=agent.cpu_limit, + memory_request=agent.memory_request, + memory_limit=agent.memory_limit, + created_at=agent.created_at, + pod_created_at=agent.pod_created_at, + ) + + +@router.get("/{agent_id}/metrics", response_model=AgentMetricsResponse) +async def get_agent_metrics( + agent_id: str, + db: AsyncSession = Depends(get_db), +) -> AgentMetricsResponse: + """ + 获取 Agent 的资源使用情况。 + """ + try: + agent_uuid = uuid.UUID(agent_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid agent ID") from exc + + agent = await db.get(Agent, agent_uuid) + if not agent: + raise HTTPException(status_code=404, detail="Agent not found") + + # 如果有 Pod，获取实时资源使用 + if agent.pod_name: + try: + client = get_agent_manager_client() + metrics = await client.get_agent_metrics(agent.pod_name) + + return AgentMetricsResponse( + id=agent.id, + name=agent.name, + requests=metrics.requests, + limits=metrics.limits, + ) + + except AgentManagerError as e: + logger.warning("获取 Pod 资源使用失败", pod_name=agent.pod_name, error=str(e)) + + # 返回数据库中的配置 + return AgentMetricsResponse( + id=agent.id, + name=agent.name, + requests={ + "cpu": agent.cpu_request or "100m", + "memory": agent.memory_request or "128Mi", + }, + limits={ + "cpu": agent.cpu_limit or "500m", + "memory": agent.memory_limit or "512Mi", + }, + ) + + +# ==================== Agent 执行 ==================== + + @router.post("/{agent_id}/execute", response_model=ExecutionResult) async def execute_agent( agent_id: str, diff --git a/services/mcp-server/app/routes/user.py b/services/mcp-server/app/routes/user.py index 20a442d..5c960dc 100644 --- a/services/mcp-server/app/routes/user.py +++ b/services/mcp-server/app/routes/user.py @@ -330,8 +330,12 @@ async def deploy_agent( db: AsyncSession = Depends(get_db) ): """ - 部署Agent + 部署Agent到Kubernetes + + 如果Agent已有模板配置，将创建对应的K8s Pod。 """ + from app.agent_manager_client import get_agent_manager_client, AgentConfig, AgentManagerError + user_id = principal.get("user_id") # 查询Agent @@ -346,19 +350,94 @@ async def deploy_agent( detail="Agent不存在" ) - # 这里简化处理，实际应该创建部署记录并调用K8s API - deployment_config = { - "agentId": req.agentId, - "instances": req.instances, - "model": req.model, - "gateway": req.gateway, - "userId": user_id, - } + # 检查是否已部署 + if agent.pod_name and agent.k8s_status == "Running": + return SuccessResponse( + data={ + "agentId": req.agentId, + "podName": agent.pod_name, + "podIp": agent.pod_ip, + "status": agent.k8s_status, + "accessUrl": agent.access_url, + }, + message=f"Agent {agent.name} 已在运行中" + ) - return SuccessResponse( - data=deployment_config, - message=f"Agent {agent.name} 部署成功" - ) + # 如果有模板，创建K8s Pod + if agent.template: + try: + client = get_agent_manager_client() + + # 构建 Pod 名称 + pod_name = f"{agent.name}-{str(agent.id)[:8]}" + + # 创建 Agent 配置 + agent_config = AgentConfig( + replicas=req.instances, + cpu_request=agent.cpu_request or "100m", + cpu_limit=agent.cpu_limit or "500m", + memory_request=agent.memory_request or "128Mi", + memory_limit=agent.memory_limit or "512Mi", + env=agent.env_config or {} + ) + + # 调用 Agent Manager API 创建 Pod + result = await client.create_agent( + name=pod_name, + template=agent.template, + config=agent_config + ) + + # 更新数据库记录 + agent.pod_name = result.name + agent.k8s_namespace = result.namespace + agent.k8s_status = result.status + agent.service_port = result.service_port + + if result.access_info: + agent.endpoints = result.access_info.get("endpoints", {}) + + await db.commit() + await db.refresh(agent) + + return SuccessResponse( + data={ + "agentId": req.agentId, + "podName": result.name, + "namespace": result.namespace, + "status": result.status, + "servicePort": result.service_port, + "instances": req.instances, + "model": req.model, + "gateway": req.gateway, + }, + message=f"Agent {agent.name} 部署成功" + ) + + except AgentManagerError as e: + raise HTTPException( + status_code=e.status_code, + detail={ + "error": "deployment_failed", + "message": f"部署失败: {e.message}", + "detail": e.detail + } + ) + else: + # 没有模板，返回配置信息（兼容旧逻辑） + deployment_config = { + "agentId": req.agentId, + "instances": req.instances, + "model": req.model, + "gateway": req.gateway, + "userId": user_id, + "note": "Agent没有配置模板，无法创建K8s Pod" + } + + return SuccessResponse( + data=deployment_config, + message=f"Agent {agent.name} 配置已保存（未创建Pod）" + ) # ============= 编排中心 ============= diff --git a/services/mcp-server/app/schemas.py b/services/mcp-server/app/schemas.py index 72253f5..d9ce757 100644 --- a/services/mcp-server/app/schemas.py +++ b/services/mcp-server/app/schemas.py @@ -491,10 +491,22 @@ class UpdateChannelRequest(BaseModel): class UpdateAgentConfigRequest(BaseModel): - """更新Agent资源配置请求""" - cpu: Optional[float] = Field(None, ge=0.1, le=64) - memory: Optional[float] = Field(None, ge=0.5, le=256) - maxInstances: Optional[int] = Field(None, ge=1, le=1000) + """更新Agent资源配置请求 + + 支持两种格式： + 1. 旧格式（数值）：cpu=2.0, memory=4.0 + 2. K8s格式（字符串）：cpu_request="100m", memory_limit="512Mi" + """ + # 旧格式（保留兼容） + cpu: Optional[float] = Field(None, ge=0.1, le=64, description="CPU 核数（旧格式）") + memory: Optional[float] = Field(None, ge=0.5, le=256, description="内存 GB（旧格式）") + maxInstances: Optional[int] = Field(None, ge=1, le=1000, description="最大实例数") + + # K8s 格式（新格式） + cpu_request: Optional[str] = Field(None, description="CPU 请求量（K8s 格式，如 100m, 1）") + cpu_limit: Optional[str] = Field(None, description="CPU 限制量（K8s 格式，如 500m, 2）") + memory_request: Optional[str] = Field(None, description="内存请求量（K8s 格式，如 128Mi, 1Gi）") + memory_limit: Optional[str] = Field(None, description="内存限制量（K8s 格式，如 512Mi, 2Gi）") class ChannelInfo(BaseModel): diff --git a/services/mcp-server/config.py b/services/mcp-server/config.py index db02dad..3c7d66a 100644 --- a/services/mcp-server/config.py +++ b/services/mcp-server/config.py @@ -57,6 +57,15 @@ class Settings(BaseSettings): agent_memory_limit: str = "512MB" agent_cpu_limit: float = 1.0 # CPU核数 + # AI Agent Manager API 设置（K8s Pod 管理） + agent_manager_url: str = os.getenv("AGENT_MANAGER_URL", "http://localhost:8000") + agent_manager_timeout: float = 30.0 # 秒 + agent_default_cpu_request: str = "100m" + agent_default_cpu_limit: str = "500m" + agent_default_memory_request: str = "128Mi" + agent_default_memory_limit: str = "512Mi" + agent_k8s_namespace: str = os.getenv("AGENT_K8S_NAMESPACE", "ai-agents") + # 工具设置 max_tools_per_agent: int = 50 tool_execution_timeout: int = 60 # 秒 diff --git a/services/mcp-server/migrations/004_add_k8s_agent_fields.sql b/services/mcp-server/migrations/004_add_k8s_agent_fields.sql new file mode 100644 index 0000000..bb9fd61 --- /dev/null +++ b/services/mcp-server/migrations/004_add_k8s_agent_fields.sql @@ -0,0 +1,49 @@ +-- 迁移脚本：添加 Kubernetes Agent 相关字段 +-- 版本：004 +-- 日期：2024-12-31 +-- 描述：为 Agent 表添加 Kubernetes Pod 管理所需的字段 + +-- 添加 Pod 相关字段 +ALTER TABLE agents ADD COLUMN IF NOT EXISTS pod_name VARCHAR(100); +ALTER TABLE agents ADD COLUMN IF NOT EXISTS pod_ip VARCHAR(45); +ALTER TABLE agents ADD COLUMN IF NOT EXISTS template VARCHAR(100); +ALTER TABLE agents ADD COLUMN IF NOT EXISTS service_port INTEGER; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS k8s_namespace VARCHAR(100) DEFAULT 'ai-agents'; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS k8s_status VARCHAR(20) DEFAULT 'Unknown'; + +-- 添加资源配置字段（K8s 格式） +ALTER TABLE agents ADD COLUMN IF NOT EXISTS cpu_request VARCHAR(20) DEFAULT '100m'; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS cpu_limit VARCHAR(20) DEFAULT '500m'; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS memory_request VARCHAR(20) DEFAULT '128Mi'; +ALTER TABLE agents ADD COLUMN IF NOT EXISTS memory_limit VARCHAR(20) DEFAULT '512Mi'; + +-- 添加环境变量配置 +ALTER TABLE agents ADD COLUMN IF NOT EXISTS env_config JSONB DEFAULT '{}'; + +-- 添加访问信息 +ALTER TABLE agents ADD COLUMN IF NOT EXISTS access_url VARCHAR(500); +ALTER TABLE agents ADD COLUMN IF NOT EXISTS endpoints JSONB DEFAULT '{}'; + +-- 添加 Pod 创建时间（K8s 返回的时间） +ALTER TABLE agents ADD COLUMN IF NOT EXISTS pod_created_at TIMESTAMP; + +-- 添加索引 +CREATE INDEX IF NOT EXISTS idx_agent_pod_name ON agents(pod_name); +CREATE INDEX IF NOT EXISTS idx_agent_template ON agents(template); +CREATE INDEX IF NOT EXISTS idx_agent_k8s_status ON agents(k8s_status); + +-- 添加注释 +COMMENT ON COLUMN agents.pod_name IS 'Kubernetes Pod 名称'; +COMMENT ON COLUMN agents.pod_ip IS 'Pod IP 地址'; +COMMENT ON COLUMN agents.template IS 'Agent 模板类型（如 echo_agent, jina_search_agent）'; +COMMENT ON COLUMN agents.service_port IS '服务端口（HTTP 服务类型 Agent）'; +COMMENT ON COLUMN agents.k8s_namespace IS 'Kubernetes 命名空间'; +COMMENT ON COLUMN agents.k8s_status IS 'Pod 状态（Pending, Running, Succeeded, Failed, Unknown）'; +COMMENT ON COLUMN agents.cpu_request IS 'CPU 请求量（K8s 格式，如 100m）'; +COMMENT ON COLUMN agents.cpu_limit IS 'CPU 限制量（K8s 格式，如 500m）'; +COMMENT ON COLUMN agents.memory_request IS '内存请求量（K8s 格式，如 128Mi）'; +COMMENT ON COLUMN agents.memory_limit IS '内存限制量（K8s 格式，如 512Mi）'; +COMMENT ON COLUMN agents.env_config IS '环境变量配置（JSON 格式）'; +COMMENT ON COLUMN agents.access_url IS 'Agent 访问 URL'; +COMMENT ON COLUMN agents.endpoints IS 'Agent 端点信息（JSON 格式）'; +COMMENT ON COLUMN agents.pod_created_at IS 'Pod 创建时间（K8s 返回）'; diff --git a/services/mcp-server/models.py b/services/mcp-server/models.py index 30818ea..28857f6 100644 --- a/services/mcp-server/models.py +++ b/services/mcp-server/models.py @@ -135,11 +135,29 @@ class Agent(BaseModel, Base): tools = Column(JSON, default=list) # Agent授权使用的工具列表 capabilities = Column(JSON, default=list) # Agent的能力列表 - # 资源配置 + # 资源配置（旧格式，保留兼容） cpu = Column(sa.Numeric(5, 2), nullable=False, default=2) memory = Column(sa.Numeric(5, 2), nullable=False, default=4) # GB max_instances = Column(Integer, default=100) + # Kubernetes 资源配置（新格式） + cpu_request = Column(String(20), default="100m") # K8s CPU 请求量 + cpu_limit = Column(String(20), default="500m") # K8s CPU 限制量 + memory_request = Column(String(20), default="128Mi") # K8s 内存请求量 + memory_limit = Column(String(20), default="512Mi") # K8s 内存限制量 + env_config = Column(JSON, default=dict) # 环境变量配置 + + # Kubernetes Pod 信息 + pod_name = Column(String(100)) # Pod 名称 + pod_ip = Column(String(45)) # Pod IP 地址 + template = Column(String(100)) # Agent 模板类型 + service_port = Column(Integer) # 服务端口 + k8s_namespace = Column(String(100), default="ai-agents") # K8s 命名空间 + k8s_status = Column(String(20), default="Unknown") # Pod 状态 + access_url = Column(String(500)) # 访问 URL + endpoints = Column(JSON, default=dict) # 端点信息 + pod_created_at = Column(DateTime) # Pod 创建时间 + # 状态信息 status = Column(String(20), default="active") # active, inactive, error, available, unavailable version = Column(String(20), default="1.0.0") @@ -164,6 +182,9 @@ class Agent(BaseModel, Base): Index("idx_agent_type", type), Index("idx_agent_owner", owner_id), Index("idx_agent_status", status), + Index("idx_agent_pod_name", pod_name), + Index("idx_agent_template", template), + Index("idx_agent_k8s_status", k8s_status), ) diff --git a/services/mcp-server/schemas.py b/services/mcp-server/schemas.py index a8ac6ba..b8b4c2f 100644 --- a/services/mcp-server/schemas.py +++ b/services/mcp-server/schemas.py @@ -195,19 +195,42 @@ class ToolResponse(BaseSchema): # ========== Agent相关 ========== +class K8sResourceConfig(BaseModel): + """Kubernetes 资源配置""" + cpu_request: str = Field(default="100m", description="CPU 请求量（如 100m, 1）") + cpu_limit: str = Field(default="500m", description="CPU 限制量（如 500m, 2）") + memory_request: str = Field(default="128Mi", description="内存请求量（如 128Mi, 1Gi）") + memory_limit: str = Field(default="512Mi", description="内存限制量（如 512Mi, 2Gi）") + replicas: int = Field(default=1, ge=1, le=10, description="副本数量") + env: Dict[str, str] = Field(default_factory=dict, description="环境变量") + + class AgentCreateRequest(BaseModel): """创建Agent请求""" - name: str + name: str = Field(..., min_length=1, max_length=63, description="Agent 名称（小写字母、数字、连字符）") description: Optional[str] = None role: str = "general-purpose agent" goal: str = "Handle generic MCP tasks and routing" + # 模板配置（用于 K8s 部署） + template: Optional[str] = Field(None, description="Agent 模板类型（如 echo_agent, jina_search_agent）") + tools: List[str] = [] # 工具名称列表 config: Dict[str, Any] = {} capabilities: List[str] = [] + # K8s 资源配置 + resource_config: Optional[K8sResourceConfig] = Field(None, description="Kubernetes 资源配置") + owner_id: Optional[Union[uuid.UUID, str]] = None + @validator('name') + def validate_name(cls, v): + """验证名称符合 K8s 命名规范""" + import re + if not re.match(r'^[a-z0-9]([-a-z0-9]*[a-z0-9])?$', v): + raise ValueError('名称必须是小写字母、数字和连字符，且以字母或数字开头和结尾') + return v class AgentUpdateRequest(BaseModel): @@ -219,6 +242,9 @@ class AgentUpdateRequest(BaseModel): tools: Optional[List[str]] = None config: Optional[Dict[str, Any]] = None capabilities: Optional[List[str]] = None + + # K8s 资源配置更新 + resource_config: Optional[K8sResourceConfig] = None class AgentCard(BaseSchema): @@ -239,6 +265,20 @@ class AgentCard(BaseSchema): status: str = "active" version: str = "1.0.0" + # K8s 相关信息 + template: Optional[str] = None + pod_name: Optional[str] = None + pod_ip: Optional[str] = None + k8s_status: Optional[str] = None + service_port: Optional[int] = None + access_url: Optional[str] = None + + # 资源配置 + cpu_request: Optional[str] = None + cpu_limit: Optional[str] = None + memory_request: Optional[str] = None + memory_limit: Optional[str] = None + # 统计信息 total_executions: int = 0 success_rate: float = 0.0 @@ -247,6 +287,7 @@ class AgentCard(BaseSchema): # 时间信息 created_at: datetime updated_at: datetime + pod_created_at: Optional[datetime] = None @validator('id', pre=True) def validate_id(cls, v): @@ -259,6 +300,65 @@ class AgentCard(BaseSchema): return uuid.UUID(v_str) +class AgentStatusResponse(BaseSchema): + """Agent 状态响应""" + id: uuid.UUID + name: str + status: str # active, inactive, error + k8s_status: str # Pending, Running, Succeeded, Failed, Unknown + + # Pod 信息 + pod_name: Optional[str] = None + pod_ip: Optional[str] = None + node: Optional[str] = None + + # 访问信息 + service_port: Optional[int] = None + access_url: Optional[str] = None + endpoints: Dict[str, str] = {} + + # 资源配置 + cpu_request: Optional[str] = None + cpu_limit: Optional[str] = None + memory_request: Optional[str] = None + memory_limit: Optional[str] = None + + # 时间信息 + created_at: datetime + pod_created_at: Optional[datetime] = None + + # 条件信息 + conditions: Optional[List[Dict[str, Any]]] = None + + +class AgentMetricsResponse(BaseSchema): + """Agent 资源使用响应""" + id: uuid.UUID + name: str + + # 资源请求 + requests: Dict[str, str] = {} # {"cpu": "100m", "memory": "128Mi"} + + # 资源限制 + limits: Dict[str, str] = {} # {"cpu": "500m", "memory": "512Mi"} + + # 实际使用（如果可用） + usage: Optional[Dict[str, str]] = None + + +class TemplateInfo(BaseModel): + """模板信息""" + template: str + port: Optional[int] = None + env_info: Dict[str, Any] = {} + + +class TemplateListResponse(BaseModel): + """模板列表响应""" + templates: List[TemplateInfo] + count: int + + class AgentExecution(BaseModel): """Agent执行请求""" method: str