tencent cloud

Cloud Native Intelligent Gateway

Rapid Agent Service Integration via AI Gateway

Download
Focus Mode
Font Size
Last updated: 2026-09-22 18:34:45
AI-Translated

Scenario Overview

The Agent Gateway (A2A) addresses the "Agent access" problem by exposing a complete AI Agent (with autonomous reasoning, task planning, and multi-turn interaction capabilities) through the AI Gateway for invocation by other agents or clients.
This document uses "code review agent access" as an example: start a Mock A2A Agent (a Python script with only ~150 lines of code). The Agent itself carries no Skills (skills: []), and all Skills are created and mounted to the Agent service through the AI Gateway's Skill management. The caller discovers the mounted Skills through the AgentCard exposed by the gateway, then initiates an A2A task call to obtain the review report via SSE streaming.

Overall Architecture

The Mock Agent and the AI Gateway are deployed on the same CVM. The Agent listens on cvm localhost:9999, and the gateway forwards requests through localhost.

┌──────────────┐ 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 │ │
│ └────────────┘ └────────────┘ │
│ │
└───────────────────────────────────────────────────────────┘
A2A Core Concepts at a Glance
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

Prerequisites

1. An AI Gateway instance has been created, and its status is Running. For detailed operations, see Creating an AI Gateway.
2. The gateway data plane version must be 3.9.5 or later.
3. You can SSH into the CVM hosting the gateway to deploy the Mock Agent script.

Step 1: Deploy the Mock A2A Agent to the Gateway CVM

Use a Python script to simulate a code review Agent. This Agent carries no Skills (skills: [] in the AgentCard), and all Skills will be created and mounted through the gateway's Skill management in Step 4.
The Mock Agent and the AI Gateway are deployed on the same CVM. The gateway directly accesses the Agent through 127.0.0.1:9999 (localhost), requiring no cross-network communication.
1.1 Upload the script to the gateway CVM
# 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>
1.2 Start the Mock Agent
cd /opt
nohup python3 mock-a2a-agent.py > /opt/agent.log 2>&1 &
After startup, the output is as follows:
Mock A2A Agent running on http://<localhost>:9999
AgentCard: http://localhost:9999/.well-known/agent-card.json
Skills: None (all are mounted through AI Gateway Skill management)
Verify that the backend is available.
# Verify on the CVM.
curl -s http://localhost:9999/.well-known/agent-card.json | python3 -m json.tool
The skills field in the returned AgentCard is expected to be an empty array (the Agent itself has no Skills):
{
"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"]
}

Step 2: Create Skills in Gateway Skill Management

The Mock Agent itself carries no Skills. All Skills are created and managed on the gateway side, which is the recommended best practice:
Decoupling: The Agent is only responsible for execution, while Skills semantics are defined by the gateway. Replacing the Agent implementation does not affect Skills.
Centralized management: All Skills are maintained centrally in the gateway console.
Multi-Agent reuse: A single Skill can be mounted to multiple Agent services and takes effect globally.
The AI Gateway has a built-in Skill management feature. Each Skill corresponds to an AgentSkill object in the AgentCard.skills[] array of the A2A protocol.

Create Skill: Code Quality Check

1. Log in to the AI Gateway console and select the target AI Gateway instance.
2. In the left sidebar, go to Skill Management > Create Skill.
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
Note:
For actual use, it is recommended to upload a ZIP package for quick creation. A pre-packaged SKILL.md file is provided. After upload, the system automatically parses and fills in the name, description, and tags, so no manual entry is required.

Create Skill: Security Vulnerability Scan

Follow the same steps to create another one:
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.

Step 3: Create an Agent Service in the Gateway

1. Log in to the AI Gateway console and select the target AI Gateway instance.
2. In the left sidebar, go to Agent Management > Agent Service/Agent API > New.
Agent Service
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

Step 4: Agent API Configuration

Parameter
Description
Agent API Name
code-review-api
Request Protocol
A2A
Base Path
/code-review
Example of an actual call path (using the API name code-review-api as an example):

# AgentCard Discovery
http://<gateway_IP>/code-review/code-review-api/.well-known/agent-card.json

# A2A Task Invocation
http://<gateway_IP>/code-review/code-review-api

Step 5: AgentCard Discovery (Obtaining Skills)

Obtain the Agent's capability description and Skills through the gateway:
curl -s http://<gateway_IP>/code-review/code-review-api/.well-known/agent-card.json \\
-H "Authorization: Bearer <API_KEY>" | python3 -m json.tool
Expected response:
{
"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"
}
]
}
Appendix: Agent Source Code

#!/usr/bin/env python3
"""
Mock A2A Code Review Agent
Port: 9999
Skills: 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 json
import time
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler

PORT = 9999

AGENT_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()
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 (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 vulnerabilities
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": "Scanning for security vulnerabilities..."}]
}
},
"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": (
"📋 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: 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 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()



Help and Support

Was this page helpful?

Help us improve! Rate your documentation experience in 5 mins.

Feedback