一、场景概述
这篇以“代码审查智能体接入” 为例：启动一个 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
# 从本地电脑 上传到网关所在的 CVMscp outputs/mock-a2a-agent.py root@<网关CVM的IP>:/opt/# SSH 登录网关 CVMssh root@<网关CVM的IP>
1.2 启动 Mock Agent
cd /optnohup python3 mock-a2a-agent.py > /opt/agent.log 2>&1 &
启动后输出：
Mock A2A Agent running on http://<localhost>:9999AgentCard: http://localhost:9999/.well-known/agent-card.jsonSkills: 无（全部由 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端口: 9999Skills: 无（全部由 AI 网关 Skill 管理挂载）端点: GET /.well-known/agent-card.json → AgentCard（skills=[]） POST / → A2A tasks/sendSubscribe（SSE 流式）"""import jsonimport timeimport sysfrom http.server import HTTPServer, BaseHTTPRequestHandlerPORT = 9999AGENT_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()