云原生智能网关通过 AI 网关实现 Agent 服务快速接入

场景概述
Agent 网关（A2A） 解决的是“智能体接入”问题——将一个完整的 AI Agent（具备自主推理、任务规划、多轮交互能力）通过 AI 网关暴露给其他 Agent 或客户端调用。
本文以“代码审查智能体接入” 为例：启动一个 Mock A2A Agent（Python 脚本，仅 ~150 行代码），Agent 本身不携带 Skills（skills: []），所有 Skills 通过 AI 网关的 Skill 管理 创建并挂载到 Agent 服务。调用方通过网关暴露的 AgentCard 发现挂载后的 Skills，再发起 A2A 任务调用，以 SSE 流式方式获取审查报告。
整体架构
Mock Agent 与 AI 网关部署在同一台 CVM 上，Agent 监听 cvm localhost:9999，网关通过 localhost 转发请求。
﻿
┌──────────────┐     AgentCard 发现      ┌───────────────────────────────────────────────────────────┐
│  调用方       │ ◄─────────────────────►│           AI 网关所在 CVM                                  │
│  WorkBuddy   │    message/send (SSE)   │                                                           │
│  Cursor      │                         │  ┌────────────┐  localhost:9999 ┌────────────┐           │
│  其他 Agent  │                         │  │  AI 网关    │ ◄──────────────►│ Mock Agent │           │       
└──────────────┘                         │  │  Skill 管理 │   AgentCard/RPC │ skills: [] │          │      
                                         │  │  2条 Skill  │                 │ port: 9999 │          │
                                         │  └────────────┘                  └────────────┘          │
                                         │                                                           │ 
                                         └───────────────────────────────────────────────────────────┘
A2A 核心概念速览
概念
说明
AgentCard
A2A 的服务发现机制，描述 Agent 的能力和 Skills
Skill
Agent 对外声明的能力单元，每个 Skill 有独立的 ID 和描述
message/send
A2A 的任务调用方法，支持 SSE 流式响应
tasks/status
任务状态查询，支持 submitted → working → completed 生命周期
前提条件
1. 已创建 AI 网关实例，实例状态为运行中。具体操作请参见 新建 AI 网关。
2. 网关数据面版本 ≥ 3.9.5。
3. 能 SSH 登录网关所在的 CVM（用于部署 Mock Agent 脚本）。
步骤一：部署 Mock A2A Agent 到网关 CVM
使用 Python 脚本模拟一个代码审查 Agent。此 Agent 不携带任何 Skills（AgentCard 中 skills: []），所有 Skills 将在步骤四通过网关 Skill 管理创建并挂载。
Mock Agent 与 AI 网关部署在同一台 CVM 上，网关通过 127.0.0.1:9999（localhost）直接访问 Agent，无需跨网络。
1.1 将脚本上传到网关 CVM
# 从本地电脑 上传到网关所在的 CVM
scp outputs/mock-a2a-agent.py root@<网关CVM的IP>:/opt/
﻿
# SSH 登录网关 CVM
ssh root@<网关CVM的IP>
1.2 启动 Mock Agent
cd /opt
nohup python3 mock-a2a-agent.py > /opt/agent.log 2>&1 &
启动后输出：
Mock A2A Agent running on http://<localhost>:9999
AgentCard: http://localhost:9999/.well-known/agent-card.json
Skills: 无（全部由 AI 网关 Skill 管理挂载）
验证后端可用
# 在 CVM 上验证
curl -s http://localhost:9999/.well-known/agent-card.json | python3 -m json.tool
预期返回的 AgentCard 中 skills 为空数组（Agent 本身无 Skills）：
{
  "name": "code-review-agent",
  "description": "代码审查智能体 — Skills 全部由 AI 网关 Skill 管理挂载，Agent 本身不携带 Skills",
  "version": "1.0.0",
  "capabilities": { "streaming": true },
  "skills": [],
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text"]
}
步骤二：在网关 Skill 管理中创建 Skills
Mock Agent 本身不携带任何 Skills。所有 Skills 全部在网关侧创建和管理，这是推荐的最佳实践：
解耦：Agent 只负责执行，Skills 语义由网关定义，换 Agent 实现不影响 Skills
集中管理：所有 Skills 在网关控制台统一维护
多 Agent 复用：同一条 Skill 可挂载到多个 Agent 服务，全局生效
AI 网关内置 Skill 管理功能，每条 Skill 对应 A2A 协议 AgentCard.skills[] 中的一个 AgentSkill 对象。
创建 Skill：代码质量检查
1. 登录 AI 网关控制台，选择目标 AI 网关实例。
2. 左侧导航 > Skill 管理 > 新建 Skill。
参数
值
说明
Skill 名称
code-review
展示名称，对外可见
描述
基于 SonarQube 规则检查代码规范、复杂度、重复率。适用于 Java/Go/Python/JS 项目
写入 AgentCard skills[].description，LLM 据此判断是否匹配
Tags
代码审查 代码质量 代码规范
写入 AgentCard skills[].tags，支持按标签搜索
Examples
帮我检查这个项目的代码规范是否符合团队标准
使用示例，帮助 LLM 理解何时触发
说明：
实际使用过程中推荐：上传 ZIP 包快速创建。已提供预制的 SKILL.md 打包文件，上传后系统自动解析名称、描述、标签并回填，无需手动填写。
创建 Skill：安全漏洞扫描
同样步骤再创建一条：
参数
值
Skill 名称
security-scan
描述
扫描 OWASP Top 10 漏洞、CVE 依赖项风险、硬编码密钥、不安全加密算法
Tags
安全扫描 漏洞检测 依赖检查
Examples
帮我扫描这段代码有没有 SQL 注入风险
步骤三：在网关创建 Agent 服务
1. 登录 AI 网关控制台，选择目标 AI 网关实例。
2. 左侧导航 > Agent 管理 > Agent 服务/Agent API > 新建。
Agent 服务
参数
填写说明
服务名称
code-review-agent
服务平台
A2A
服务地址
http://127.0.0.1:9999（Agent 与网关在同一台 CVM，用 localhost）
协议版本
1.0.0
协议版本
1.0
协议绑定类型
JSON-RPC
支持能力
勾选 流式输出
挂载 skill
code-review、security-scan
超时时间
120000 毫秒（Mock 有延时模拟）
重试次数
3
描述
代码审查智能体 — Mock 演示版
步骤四：Agent API 配置
参数
填写说明
Agent API 名称
code-review-api
请求协议
A2A
 Base Path
/code-review
实际调用路径示例（以 API 名称 code-review-api 为例）：
﻿
# AgentCard 发现
http://<网关IP>/code-review/code-review-api/.well-known/agent-card.json
﻿
# A2A 任务调用
http://<网关IP>/code-review/code-review-api
步骤五：AgentCard 发现（获取 Skills）
通过网关获取 Agent 的能力描述和 Skills：
curl -s http://<网关IP>/code-review/code-review-api/.well-known/agent-card.json \\
  -H "Authorization: Bearer <API_KEY>" | python3 -m json.tool
预期响应：
{
    "name": "code-review-api",
    "description": "代码审查智能体 — Mock 演示版",
    "version": "1.0.0",
    "capabilities": {
        "streaming": true,
        "pushNotifications": false
    },
    "skills": [
        {
            "id": "skill-c8af22d30d67bff5",
            "name": "security-scan",
            "description": "扫描 OWASP Top 10 漏洞、CVE 依赖项风险、硬编码密钥、不安全加密算法",
            "tags": ["安全扫描", "漏洞检测", "依赖检查"],
            "examples": ["帮我扫描这段代码有没有 SQL 注入风险"]
        },
        {
            "id": "skill-fa0f9b5e2518286b",
            "name": "code-review",
            "description": "基于 SonarQube 规则检查代码规范、复杂度、重复率。适用于 Java/Go/Python/JS 项目",
            "tags": ["代码审查", "代码质量", "代码规范"],
            "examples": ["帮我检查这个项目的代码规范是否符合团队标准"]
        }
    ],
    "defaultInputModes": ["text/plain", "application/json"],
    "defaultOutputModes": ["text/plain", "application/json"],
    "supportedInterfaces": [
        {
            "protocolVersion": "1.0",
            "protocolBinding": "JSONRPC",
            "url": "http://<网关IP>/code-review/code-review-api"
        }
    ]
}
附录：Agent 源码
﻿
#!/usr/bin/env python3
"""
Mock A2A Code Review Agent
端口: 9999
Skills: 无（全部由 AI 网关 Skill 管理挂载）
﻿
端点:
  GET  /.well-known/agent-card.json  → AgentCard（skills=[]）
  POST /                             → A2A tasks/sendSubscribe（SSE 流式）
"""
﻿
import json
import time
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
﻿
PORT = 9999
﻿
AGENT_CARD = {
    "name": "code-review-agent",
    "description": "代码审查智能体 — Skills 全部由 AI 网关 Skill 管理挂载，Agent 本身不携带 Skills",
    "version": "1.0.0",
    "capabilities": {"streaming": True},
    "skills": [],
    "defaultInputModes": ["text"],
    "defaultOutputModes": ["text"],
    "url": f"http://localhost:{PORT}"
}
﻿
﻿
class A2AHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/.well-known/agent-card.json":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
            self.wfile.write(json.dumps(AGENT_CARD, ensure_ascii=False).encode())
        else:
            self.send_response(404)
            self.end_headers()
﻿
    def do_POST(self):
        if self.path != "/":
            self.send_response(404)
            self.end_headers()
            return
﻿
        content_len = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_len)
        request = json.loads(body)
﻿
        method = request.get("method", "")
        req_id = request.get("id", 1)
﻿
        if method == "tasks/sendSubscribe":
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
﻿
            task_id = f"task-{int(time.time())}"
﻿
            # Stage 1: submitted
            self._sse_event("status", {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "type": "TaskStatusUpdateEvent",
                    "taskId": task_id,
                    "status": {"state": "submitted"},
                    "final": False
                }
            })
﻿
            # Stage 2: working (分析代码)
            self._sse_event("status", {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "type": "TaskStatusUpdateEvent",
                    "taskId": task_id,
                    "status": {
                        "state": "working",
                        "message": {
                            "role": "agent",
                            "parts": [{"type": "text", "text": "正在分析代码..."}]
                        }
                    },
                    "final": False
                }
            })
            time.sleep(0.5)
﻿
            # 扫描安全漏洞
            self._sse_event("status", {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "type": "TaskStatusUpdateEvent",
                    "taskId": task_id,
                    "status": {
                        "state": "working",
                        "message": {
                            "role": "agent",
                            "parts": [{"type": "text", "text": "正在扫描安全漏洞..."}]
                        }
                    },
                    "final": False
                }
            })
            time.sleep(0.5)
﻿
            # Stage 3: artifact
            self._sse_event("artifact", {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "type": "TaskArtifactUpdateEvent",
                    "taskId": task_id,
                    "artifact": {
                        "parts": [{
                            "type": "text",
                            "text": (
                                "📋 代码审查报告\\n"
                                "══════════════════\\n"
                                "✅ 代码质量检查通过\\n"
                                "   - 命名规范: 符合 PEP8 标准\\n"
                                "   - 圈复杂度: 平均 3.2（良好）\\n"
                                "   - 代码重复率: 2.1%（优秀）\\n\\n"
                                "⚠️ 安全问题（低风险）\\n"
                                "   - 第 42 行: 使用了硬编码的临时调试密钥，建议移除\\n"
                                "   - 第 78 行: input() 未做输入校验，建议添加白名单过滤\\n\\n"
                                "💡 改进建议\\n"
                                "   - 考虑将配置项提取到 .env 文件\\n"
                                "   - 建议为关键函数添加单元测试"
                            )
                        }]
                    },
                    "final": False
                }
            })
﻿
            # Stage 4: completed
            self._sse_event("status", {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "type": "TaskStatusUpdateEvent",
                    "taskId": task_id,
                    "status": {"state": "completed"},
                    "final": True
                }
            })
﻿
            # Stream complete (final: true 已标记结束)
            self.wfile.flush()
﻿
        else:
            self.send_response(400)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {"code": -32601, "message": f"Method not found: {method}"}
            }, ensure_ascii=False).encode())
﻿
    def _sse_event(self, event_type, data):
        payload = json.dumps(data, ensure_ascii=False)
        self.wfile.write(f"event: {event_type}\\ndata: {payload}\\n\\n".encode())
        self.wfile.flush()
﻿
    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization")
        self.end_headers()
﻿
    def log_message(self, format, *args):
        print(f"[A2A] {args[0]}")
﻿
﻿
if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", PORT), A2AHandler)
    print(f"Mock A2A Agent running on http://0.0.0.0:{PORT}")
    print(f"AgentCard: http://localhost:{PORT}/.well-known/agent-card.json")
    print(f"Skills: 无（全部由 AI 网关 Skill 管理挂载）")
    print(f"按 Ctrl+C 停止")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\\n已停止")
        server.server_close()
﻿
﻿

概念	说明
AgentCard	A2A 的服务发现机制，描述 Agent 的能力和 Skills
Skill	Agent 对外声明的能力单元，每个 Skill 有独立的 ID 和描述
message/send	A2A 的任务调用方法，支持 SSE 流式响应
tasks/status	任务状态查询，支持 submitted → working → completed 生命周期

参数	值	说明
Skill 名称	code-review	展示名称，对外可见
描述	基于 SonarQube 规则检查代码规范、复杂度、重复率。适用于 Java/Go/Python/JS 项目	写入 AgentCard skills[].description，LLM 据此判断是否匹配
Tags	代码审查代码质量代码规范	写入 AgentCard skills[].tags，支持按标签搜索
Examples	帮我检查这个项目的代码规范是否符合团队标准	使用示例，帮助 LLM 理解何时触发

参数	值
Skill 名称	security-scan
描述	扫描 OWASP Top 10 漏洞、CVE 依赖项风险、硬编码密钥、不安全加密算法
Tags	安全扫描漏洞检测依赖检查
Examples	帮我扫描这段代码有没有 SQL 注入风险

参数	填写说明
服务名称	code-review-agent
服务平台	A2A
服务地址	http://127.0.0.1:9999（Agent 与网关在同一台 CVM，用 localhost）
协议版本	1.0.0
协议版本	1.0
协议绑定类型	JSON-RPC
支持能力	勾选流式输出
挂载 skill	code-review、security-scan
超时时间	120000 毫秒（Mock 有延时模拟）
重试次数	3
描述	代码审查智能体 — Mock 演示版

参数	填写说明
Agent API 名称	code-review-api
请求协议	A2A
Base Path	/code-review

通过 AI 网关实现 Agent 服务快速接入

本页目录：

场景概述

整体架构

前提条件

步骤一：部署 Mock A2A Agent 到网关 CVM

步骤二：在网关 Skill 管理中创建 Skills

创建 Skill：代码质量检查

创建 Skill：安全漏洞扫描

步骤三：在网关创建 Agent 服务

步骤四：Agent API 配置

步骤五：AgentCard 发现（获取 Skills）