$npx -y skills add VeerMuchandi/rad-skills --skill a2ui_local_testerSpecialized role for setting up and performing local testing of A2UI agents using the A2A / JSON-RPC protocol.
| 1 | # A2UI Local Tester Skill |
| 2 | |
| 3 | You are the **A2UI Local Tester**. Your primary responsibility is to set up a local testing harness for A2UI agents, allowing developers to verify component rendering and interaction flow without deploying to Vertex AI Agent Engine or Gemini Enterprise. |
| 4 | |
| 5 | ## Core Responsibilities |
| 6 | 1. **Local Server Setup**: Create a local server (e.g., using FastAPI) that mimics the A2A protocol. |
| 7 | 2. **A2A Protocol Emulation**: Handle JSON-RPC 2.0 requests and bridge them to the agent's executor. |
| 8 | 3. **Isolation**: Ensure all local testing code is kept in a separate sub-folder under the agent and is excluded from deployment. |
| 9 | |
| 10 | ## Mandatory Workflow |
| 11 | 1. **Isolation**: Implement the A2UI agent in a separate folder from the reference agent (e.g., `[agent_name]_a2ui`). |
| 12 | 2. **Local First**: Always verify the agent using this local tester skill before suggesting or attempting deployment. |
| 13 | 3. **Explicit Permission**: Deploy to runtime environments only when explicitly instructed by the user after successful local testing. |
| 14 | |
| 15 | ## Implementation Guide (Option 2: A2A / JSON-RPC) |
| 16 | |
| 17 | If the agent is built as an A2A / JSON-RPC service (e.g., using `AdkAgentToA2AExecutor`), follow these steps to set up local testing: |
| 18 | |
| 19 | ### Step 1: Create a Local Tester Sub-folder |
| 20 | Create a folder named `local_tester` (or similar) directly under the agent's root directory. |
| 21 | Ensure this folder is added to the project's `.gitignore` to prevent it from being included in deployment packages. |
| 22 | |
| 23 | ### Step 2: Implement the Local Server |
| 24 | Create a Python script (e.g., `server.py`) inside the `local_tester` folder. This script should use FastAPI to expose the endpoints and serve the mock client. |
| 25 | |
| 26 | Here is the complete template for `server.py`: |
| 27 | |
| 28 | ```python |
| 29 | import os |
| 30 | import sys |
| 31 | from fastapi import FastAPI, Request |
| 32 | from fastapi.responses import FileResponse |
| 33 | from fastapi.middleware.cors import CORSMiddleware |
| 34 | import uvicorn |
| 35 | import json |
| 36 | import logging |
| 37 | from dotenv import load_dotenv |
| 38 | |
| 39 | # Add parent directory to path to import agent and tools |
| 40 | parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| 41 | sys.path.append(parent_dir) |
| 42 | |
| 43 | # Load environment variables from .env file in parent directory |
| 44 | load_dotenv(os.path.join(parent_dir, '.env')) |
| 45 | |
| 46 | import agent |
| 47 | from google.adk import runners |
| 48 | from google.adk.sessions import in_memory_session_service |
| 49 | from google.adk.artifacts import in_memory_artifact_service |
| 50 | from google.adk.memory import in_memory_memory_service |
| 51 | from google.genai import types as genai_types |
| 52 | |
| 53 | app = FastAPI() |
| 54 | |
| 55 | # Enable CORS for local testing |
| 56 | app.add_middleware( |
| 57 | CORSMiddleware, |
| 58 | allow_origins=["*"], |
| 59 | allow_credentials=True, |
| 60 | allow_methods=["*"], |
| 61 | allow_headers=["*"], |
| 62 | ) |
| 63 | |
| 64 | logger = logging.getLogger(__name__) |
| 65 | logging.basicConfig(level=logging.INFO) |
| 66 | |
| 67 | # Initialize ADK Runner |
| 68 | adk_agent = agent.root_agent |
| 69 | runner = runners.Runner( |
| 70 | app_name=adk_agent.name, |
| 71 | agent=adk_agent, |
| 72 | session_service=in_memory_session_service.InMemorySessionService(), |
| 73 | artifact_service=in_memory_artifact_service.InMemoryArtifactService(), |
| 74 | memory_service=in_memory_memory_service.InMemoryMemoryService(), |
| 75 | ) |
| 76 | |
| 77 | @app.get("/.well-known/agent-card.json") |
| 78 | async def get_agent_card(): |
| 79 | return { |
| 80 | "capabilities": { |
| 81 | "streaming": False, |
| 82 | "extensions": [{"uri": "https://a2ui.org/a2a-extension/a2ui/v0.8", "required": False}] |
| 83 | }, |
| 84 | "name": adk_agent.name, |
| 85 | "url": "/jsonrpc", |
| 86 | "version": "1.0.0" |
| 87 | } |
| 88 | |
| 89 | @app.get("/") |
| 90 | async def get_index(): |
| 91 | return FileResponse(os.path.join(os.path.dirname(os.path.abspath(__file__)), "index.html")) |
| 92 | |
| 93 | @app.post("/jsonrpc") |
| 94 | async def handle_jsonrpc(request: Request): |
| 95 | body = await request.json() |
| 96 | logger.info(f"Received JSON-RPC request: {body}") |
| 97 | |
| 98 | if body.get("jsonrpc") != "2.0": |
| 99 | return {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": body.get("id")} |
| 100 | |
| 101 | method = body.get("method") |
| 102 | params = body.get("params", {}) |
| 103 | request_id = body.get("id") |
| 104 | |
| 105 | if method == "message/send": |
| 106 | message = params.get("message", {}) |
| 107 | query = message.get("text", "") |
| 108 | parts = message.get("parts", []) |
| 109 | session_id = params.get("session_id", "local_session") |
| 110 | |
| 111 | # Extract userAction from DataPart |
| 112 | user_action = None |
| 113 | for part in parts: |
| 114 | if part.get("metadata", {}).get("mimeType") == "application/json+a2ui": |
| 115 | data = part.get("data") |
| 116 | if isinstance(data, str): |
| 117 | try: |
| 118 | data = json.loads(data) |
| 119 | except: |
| 120 | pass |
| 121 | if isinstance(data, dict) and 'userAction' in data: |
| 122 | user_action = data['userAction'] |
| 123 | break |
| 124 | |
| 125 | # Get session |
| 126 | session |