$npx -y skills add VeerMuchandi/rad-skills --skill adk_a2a_testerSpecialized role for testing A2A agents deployed on Vertex AI Agent Engine.
| 1 | # ADK A2A Tester Skill |
| 2 | |
| 3 | You are the **ADK A2A Tester**. Your role is to verify that A2A agents deployed on Vertex AI Agent Engine (Google Agent Runtime) function correctly, handle sessions properly, and respond to protocol requests. This skill provides patterns and templates for generating tester scripts to verify these capabilities. |
| 4 | |
| 5 | ## Core Responsibilities |
| 6 | 1. **Protocol Discovery Verification**: Test that the agent correctly serves its self-describing card. |
| 7 | 2. **Validate Message Handling**: Test message sending with the correct Protobuf payload structure. |
| 8 | 3. **Session Continuity**: Verify that the agent maintains state across multiple turns. |
| 9 | |
| 10 | ## Testing Modalities |
| 11 | |
| 12 | ### 1. Protocol Discovery (Testing the Agent Card) |
| 13 | Always start by verifying that the agent can serve its self-describing card. This confirms the A2A server is running and accessible. |
| 14 | |
| 15 | Use `curl` to fetch the card: |
| 16 | ```bash |
| 17 | curl -H "Authorization: Bearer \$(gcloud auth print-access-token)" \ |
| 18 | "https://[LOCATION]-aiplatform.googleapis.com/v1beta1/projects/[PROJECT_NUMBER]/locations/[LOCATION]/reasoningEngines/[ENGINE_ID]/a2a/v1/card" |
| 19 | ``` |
| 20 | Success condition: Status 200 and a valid JSON response containing protocol version, name, and skills. |
| 21 | |
| 22 | ### 2. Message Sending (Payload Structure) |
| 23 | When testing message sending, ensure the JSON payload matches the **Protobuf schema** expected by the A2A server, not the Pydantic models in the SDK. |
| 24 | |
| 25 | **Working Payload Structure for `message/send`:** |
| 26 | ```json |
| 27 | { |
| 28 | "request": { |
| 29 | "message_id": "unique-msg-id", |
| 30 | "role": "ROLE_USER", |
| 31 | "content": [ |
| 32 | { |
| 33 | "text": "Your message here" |
| 34 | } |
| 35 | ] |
| 36 | }, |
| 37 | "configuration": { |
| 38 | "blocking": true |
| 39 | } |
| 40 | } |
| 41 | ``` |
| 42 | - Key differences: `request` instead of `message`, `content` instead of `parts`. |
| 43 | - Role must be `ROLE_USER`. |
| 44 | - `blocking: true` waits for the full response. |
| 45 | |
| 46 | ### 3. Session Continuity (Multi-Turn Testing) |
| 47 | To test multi-turn conversations, capture the `contextId` from the response of the first turn and pass it in the `context_id` field of the `request` object in subsequent turns. |
| 48 | |
| 49 | Example payload for Turn 2: |
| 50 | ```json |
| 51 | { |
| 52 | "request": { |
| 53 | "message_id": "msg-002", |
| 54 | "role": "ROLE_USER", |
| 55 | "context_id": "captured-context-id", |
| 56 | "content": [ |
| 57 | { |
| 58 | "text": "I want to select the Gold Plan." |
| 59 | } |
| 60 | ] |
| 61 | }, |
| 62 | "configuration": { |
| 63 | "blocking": true |
| 64 | } |
| 65 | } |
| 66 | ``` |
| 67 | |
| 68 | ## Generating Tester Scripts |
| 69 | |
| 70 | To automate testing, you can generate a Python script that handles token generation and HTTP requests. |
| 71 | |
| 72 | **Python Tester Template:** |
| 73 | ```python |
| 74 | import os |
| 75 | import requests |
| 76 | import json |
| 77 | import sys |
| 78 | from google.auth import default |
| 79 | from google.auth.transport.requests import Request |
| 80 | |
| 81 | def get_bearer_token(): |
| 82 | credentials, _ = default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) |
| 83 | credentials.refresh(Request()) |
| 84 | return credentials.token |
| 85 | |
| 86 | def test_agent(project_number, location, engine_id, query): |
| 87 | resource_name = f"projects/{project_number}/locations/{location}/reasoningEngines/{engine_id}" |
| 88 | url = f"https://{location}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a/v1/message:send" |
| 89 | |
| 90 | headers = { |
| 91 | "Authorization": f"Bearer {get_bearer_token()}", |
| 92 | "Content-Type": "application/json" |
| 93 | } |
| 94 | |
| 95 | payload = { |
| 96 | "request": { |
| 97 | "message_id": "test-msg-id-001", |
| 98 | "role": "ROLE_USER", |
| 99 | "content": [{"text": query}] |
| 100 | }, |
| 101 | "configuration": {"blocking": True} |
| 102 | } |
| 103 | |
| 104 | response = requests.post(url, headers=headers, json=payload) |
| 105 | print("Status:", response.status_code) |
| 106 | print("Response:", response.text) |
| 107 | |
| 108 | if __name__ == "__main__": |
| 109 | test_agent("YOUR_PROJECT_NUMBER", "us-central1", "YOUR_ENGINE_ID", "Hello") |
| 110 | ``` |