┌──────────────┐ AgentCard discovery ┌───────────────────────────────────────────────────────────┐│ Caller │ ◄─────────────────────►│ CVM hosting the AI Gateway ││ WorkBuddy │ message/send (SSE) │ ││ Cursor │ │ ┌────────────┐ localhost:9999 ┌────────────┐ ││ Other Agents │ │ │ AI Gateway │ ◄──────────────►│ Mock Agent │ │└──────────────┘ │ │ Skill Management │ AgentCard/RPC │ skills: [] │ ││ │ 2 Skills │ │ port: 9999 │ ││ └────────────┘ └────────────┘ ││ │└───────────────────────────────────────────────────────────┘
Concept | Description |
AgentCard | Service discovery mechanism of A2A that describes the capabilities and Skills of an Agent |
Skill | Capability unit declared by an Agent. Each Skill has an independent ID and description. |
message/send | Task invocation method of A2A that supports SSE streaming responses |
tasks/status | Task status query that supports the submitted → working → completed lifecycle |
# Upload from the local computer to the CVM hosting the gateway.scp outputs/mock-a2a-agent.py root@<gateway_CVM_IP>:/opt/# SSH into the gateway CVM.ssh root@<gateway_CVM_IP>
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: None (all are mounted through AI Gateway Skill management)
# Verify on the CVM.curl -s http://localhost:9999/.well-known/agent-card.json | python3 -m json.tool
{"name": "code-review-agent","description": "Code review agent — All Skills are mounted through AI Gateway Skill management, and the Agent itself carries no Skills","version": "1.0.0","capabilities": { "streaming": true },"skills": [],"defaultInputModes": ["text"],"defaultOutputModes": ["text"]}
Parameter | Value | Description |
Skill name | code-review | Display name, visible to external users |
Description | Checks code standards, complexity, and duplication based on SonarQube rules. Applies to Java/Go/Python/JS projects. | Write to AgentCard skills[].description. The LLM determines whether it matches based on this information. |
Tags | Code review, code quality, code standards | Write to AgentCard skills[].tags. Supports searching by tag. |
Examples | Check whether this project's code standards comply with the team standards. | Usage examples that help the LLM understand when to trigger |
Parameter | Value |
Skill name | security-scan |
Description | Scan OWASP Top 10 vulnerabilities, CVE dependency risks, hardcoded keys, and insecure encryption algorithms. |
Tags | Security scanning, vulnerability detection, and dependency checking |
Examples | Scan this code for SQL injection risks. |
Parameter | Description |
Service Name | code-review-agent |
Service Platform | A2A |
Service Address | http://127.0.0.1:9999 (Agent and gateway are on the same CVM, use localhost) |
Protocol Version | 1.0.0 |
Protocol Version | 1.0 |
Protocol Binding Type | JSON-RPC |
Supported Capability | Select streaming output. |
Mount skill. | code-review,security-scan |
Timeout Time | 120000 ms (Mock has delay simulation) |
Number of retries | 3 |
Description | Code review agent Mock demo version |
Parameter | Description |
Agent API Name | code-review-api |
Request Protocol | A2A |
Base Path | /code-review |
# AgentCard Discoveryhttp://<gateway_IP>/code-review/code-review-api/.well-known/agent-card.json# A2A Task Invocationhttp://<gateway_IP>/code-review/code-review-api
curl -s http://<gateway_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": "Code review agent — Mock demo version","version": "1.0.0","capabilities": {"streaming": true,"pushNotifications": false},"skills": [{"id": "skill-c8af22d30d67bff5","name": "security-scan","description": "Scans for OWASP Top 10 vulnerabilities, CVE dependency risks, hardcoded secrets, and insecure encryption algorithms","tags": ["Security scanning", "Vulnerability detection", "Dependency checking"],"examples": ["Scan this code for SQL injection risks"]},{"id": "skill-fa0f9b5e2518286b","name": "code-review","description": "Checks code standards, complexity, and duplication based on SonarQube rules. Applicable to Java/Go/Python/JS projects","tags": ["Code review", "Code quality", "Coding standards"],"examples": ["Check whether this project's coding standards comply with the team standards"]}],"defaultInputModes": ["text/plain", "application/json"],"defaultOutputModes": ["text/plain", "application/json"],"supportedInterfaces": [{"protocolVersion": "1.0","protocolBinding": "JSONRPC","url": "http://<gateway_IP>/code-review/code-review-api"}]}
#!/usr/bin/env python3"""Mock A2A Code Review AgentPort: 9999Skills: None (all are mounted through AI Gateway Skill management)Endpoint:GET /.well-known/agent-card.json → AgentCard(skills=[])POST / → A2A tasks/sendSubscribe (SSE streaming)"""import jsonimport timeimport sysfrom http.server import HTTPServer, BaseHTTPRequestHandlerPORT = 9999AGENT_CARD = {"name": "code-review-agent","description": "Code review agent — All Skills are mounted and managed by AI Gateway Skill, and the Agent itself carries no 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()returncontent_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: submittedself._sse_event("status", {"jsonrpc": "2.0","id": req_id,"result": {"type": "TaskStatusUpdateEvent","taskId": task_id,"status": {"state": "submitted"},"final": False}})# Stage 2: working (Analyzing code)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": "Analyzing code..."}]}},"final": False}})time.sleep(0.5)# Scan for security vulnerabilitiesself._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": "Scanning for security vulnerabilities..."}]}},"final": False}})time.sleep(0.5)# Stage 3: artifactself._sse_event("artifact", {"jsonrpc": "2.0","id": req_id,"result": {"type": "TaskArtifactUpdateEvent","taskId": task_id,"artifact": {"parts": [{"type": "text","text": ("📋 Code Review Report\\n""══════════════════\\n""✅ Code quality check passed\\n"" - Naming convention: Compliant with PEP8 standards\\n"" - Cyclomatic complexity: Average 3.2 (Good)\\n"" - Code duplication rate: 2.1% (Excellent)\\n\\n""⚠️ Security issues (Low risk)\\n"" - Line 42: A hardcoded temporary debug key is used. Remove it.\\n"" - Line 78: input() lacks input validation. Add allowlist filtering.\\n\\n""💡 Suggestions for improvement\\n"" - Consider extracting configuration items to a .env file\\n"" - Add unit tests for critical functions.\\n")}]},"final": False}})# Stage 4: completedself._sse_event("status", {"jsonrpc": "2.0","id": req_id,"result": {"type": "TaskStatusUpdateEvent","taskId": task_id,"status": {"state": "completed"},"final": True}})# Stream complete (final: true marked as ended)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: None (all are mounted through AI Gateway Skill management)")print(f"Press Ctrl+C to stop")try:server.serve_forever()except KeyboardInterrupt:print("\\nStopped")server.server_close()
Was this page helpful?
You can also Contact sales or Submit a Ticket for help.
Help us improve! Rate your documentation experience in 5 mins.
Feedback