$npx -y skills add VeerMuchandi/rad-skills --skill a2ui_remote_testerSpecialized role for testing A2UI agents deployed on Vertex AI Agent Engine using a local proxy server.
| 1 | # A2UI Remote Tester Skill |
| 2 | |
| 3 | You are the **A2UI Remote Tester**. Your primary responsibility is to verify that A2UI agents deployed on Vertex AI Agent Engine (Google Agent Runtime) function correctly, handle sessions properly, and respond to protocol requests using an interactive mock client. |
| 4 | |
| 5 | This skill provides patterns and templates for setting up a remote testing harness, which involves running a local proxy server that forwards A2A requests to the deployed agent endpoint, while serving the interactive mock client locally. |
| 6 | |
| 7 | ## Core Responsibilities |
| 8 | 1. **Proxy Server Setup**: Create a local server (e.g., using FastAPI) that forwards requests to the remote Agent Engine instance. |
| 9 | 2. **A2A Protocol Bridge**: Handle JSON-RPC 2.0 requests from the client and translate them to the format expected by the Agent Engine API. |
| 10 | 3. **UI Rendering Verification**: Use the mock client to verify component rendering and interaction flow with the live remote agent. |
| 11 | |
| 12 | ## Implementation Guide |
| 13 | |
| 14 | ### Step 1: Create a Remote Tester Sub-folder |
| 15 | Create a folder named `remote_tester` directly under the agent's root directory. Copy the `index.html` file from your `local_tester` folder into this new folder, and update the heading in it to `<h1>A2UI Remote Tester</h1>` and the title to `<title>A2UI Remote Tester</title>` to distinguish it from the local tester. |
| 16 | |
| 17 | ### Step 2: Implement the Proxy Server |
| 18 | Create a Python script (e.g., `proxy_server.py`) inside the `remote_tester` folder. |
| 19 | |
| 20 | Here is the complete template for `proxy_server.py`: |
| 21 | |
| 22 | ```python |
| 23 | import os |
| 24 | from fastapi import FastAPI, Request |
| 25 | from fastapi.responses import FileResponse |
| 26 | from fastapi.middleware.cors import CORSMiddleware |
| 27 | import uvicorn |
| 28 | import json |
| 29 | import logging |
| 30 | import requests |
| 31 | import time |
| 32 | from google.auth import default |
| 33 | from google.auth.transport.requests import Request as AuthRequest |
| 34 | |
| 35 | app = FastAPI() |
| 36 | |
| 37 | app.add_middleware( |
| 38 | CORSMiddleware, |
| 39 | allow_origins=["*"], |
| 40 | allow_credentials=True, |
| 41 | allow_methods=["*"], |
| 42 | allow_headers=["*"], |
| 43 | ) |
| 44 | |
| 45 | logger = logging.getLogger(__name__) |
| 46 | logging.basicConfig(level=logging.INFO) |
| 47 | |
| 48 | # Configure these for your deployed agent |
| 49 | PROJECT_ID = "your-project-id" |
| 50 | LOCATION = "us-central1" |
| 51 | ENGINE_ID = "your-engine-id" |
| 52 | |
| 53 | def get_bearer_token(): |
| 54 | credentials, _ = default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) |
| 55 | credentials.refresh(AuthRequest()) |
| 56 | return credentials.token |
| 57 | |
| 58 | @app.get("/.well-known/agent-card.json") |
| 59 | async def get_agent_card(): |
| 60 | token = get_bearer_token() |
| 61 | url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{ENGINE_ID}/a2a/v1/card" |
| 62 | headers = {"Authorization": f"Bearer {token}"} |
| 63 | res = requests.get(url, headers=headers) |
| 64 | return res.json() |
| 65 | |
| 66 | @app.get("/") |
| 67 | async def get_index(): |
| 68 | return FileResponse(os.path.join(os.path.dirname(os.path.abspath(__file__)), "index.html")) |
| 69 | |
| 70 | @app.post("/jsonrpc") |
| 71 | async def handle_jsonrpc(request: Request): |
| 72 | body = await request.json() |
| 73 | logger.info(f"Received JSON-RPC request: {body}") |
| 74 | |
| 75 | if body.get("jsonrpc") != "2.0": |
| 76 | return {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": body.get("id")} |
| 77 | |
| 78 | method = body.get("method") |
| 79 | params = body.get("params", {}) |
| 80 | request_id = body.get("id") |
| 81 | |
| 82 | if method == "message/send": |
| 83 | message = params.get("message", {}) |
| 84 | |
| 85 | token = get_bearer_token() |
| 86 | url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{ENGINE_ID}/a2a/v1/message:send" |
| 87 | headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} |
| 88 | |
| 89 | # Translate client payload to standard A2A Message format |
| 90 | a2a_parts = [] |
| 91 | if "text" in message and message["text"]: |
| 92 | a2a_parts.append({"text": message["text"]}) |
| 93 | if "parts" in message: |
| 94 | for part in message["parts"]: |
| 95 | a2a_part = {} |
| 96 | if "data" in part: |
| 97 | a2a_part["data"] = {"data": part["data"]} |
| 98 | if "metadata" in part: |
| 99 | a2a_part["metadata"] = part["metadata"] |
| 100 | if "text" in part: |
| 101 | a2a_part["text"] = part["text"] |
| 102 | a2a_parts.append(a2a_part) |
| 103 | |
| 104 | a2a_message = { |
| 105 | "role": "ROLE_USER", |
| 106 | "content": a2a_parts |
| 107 | } |
| 108 | |
| 109 | payload = {"message": a2a_message} |
| 110 | |
| 111 | logger.info(f"Forwarding to remote agent: {url}") |
| 112 | res = requests.post(url, headers=headers, json=payload) |
| 113 | |
| 114 | if res.status_code != 200: |
| 115 | return {"jsonrpc": "2.0", "error": {"code": res.status_code, "message": res.text}, "id": request_id} |
| 116 | |
| 117 | remote_response = res.json() |