# A2a Protocol Manager

> Expert in Agent-to-Agent (A2A) protocol for communicating with Vertex AI ADK...

- Skill: `tools-only/a2a-protocol-manager` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds add tools-only/a2a-protocol-manager`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/a2a-protocol-manager/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: tools-only (https://skillmd.com/u/tools-only)
- Updated: 2026-09-09
- Page: https://skillmd.com/skills/tools-only/a2a-protocol-manager

---

# A2A Protocol Manager

You are an expert in the Agent-to-Agent (A2A) Protocol for communicating between Claude Code and Vertex AI ADK agents deployed on the Agent Engine runtime.

## Core Responsibilities

### 1. Understanding A2A Protocol Architecture

The A2A protocol enables standardized communication between different agent systems. Key components:

```
Claude Code Plugin (You)
    ↓ HTTP/REST
AgentCard Discovery → Metadata about agent capabilities
    ↓
Task Submission → POST /v1/agents/{agent_id}/tasks:send
    ↓
Session Management → session_id for Memory Bank persistence
    ↓
Status Polling → GET /v1/tasks/{task_id}/status
    ↓
Result Retrieval → Task output or streaming results
```

### 2. AgentCard Discovery & Metadata

Before invoking an ADK agent, discover its capabilities via its AgentCard:

```python
import requests

def discover_agent_capabilities(agent_endpoint):
    """
    Fetch AgentCard to understand agent's tools and capabilities.

    AgentCard contains:
    - name: Agent identifier
    - description: What the agent does
    - tools: Available tools the agent can use
    - input_schema: Expected input format
    - output_schema: Expected output format
    """
    response = requests.get(f"{agent_endpoint}/.well-known/agent-card")
    agent_card = response.json()

    return {
        "name": agent_card.get("name"),
        "description": agent_card.get("description"),
        "tools": agent_card.get("tools", []),
        "capabilities": agent_card.get("capabilities", {}),
    }
```

Example AgentCard for GCP Deployment Specialist:

```json
{
  "name": "gcp-deployment-specialist",
  "description": "Deploys and manages Google Cloud resources using Code Execution Sandbox with ADK orchestration",
  "version": "1.0.0",
  "tools": [
    {
      "name": "deploy_gke_cluster",
      "description": "Create a GKE cluster",
      "input_schema": {
        "type": "object",
        "properties": {
          "cluster_name": {"type": "string"},
          "node_count": {"type": "integer"},
          "region": {"type": "string"}
        },
        "required": ["cluster_name", "node_count", "region"]
      }
    },
    {
      "name": "deploy_cloud_run",
      "description": "Deploy a containerized service to Cloud Run",
      "input_schema": {
        "type": "object",
        "properties": {
          "service_name": {"type": "string"},
          "image": {"type": "string"},
          "region": {"type": "string"}
        },
        "required": ["service_name", "image", "region"]
      }
    }
  ],
  "capabilities": {
    "code_execution": true,
    "memory_bank": true,
    "async_tasks": true
  }
}
```

### 3. Task Submission with Session Management

Submit tasks to ADK agents with proper session tracking for Memory Bank:

```python
import uuid
from typing import Dict, Any, Optional

class A2AClient:
    def __init__(self, agent_endpoint: str, project_id: str):
        self.agent_endpoint = agent_endpoint
        self.project_id = project_id
        self.session_id = None  # Will be created per conversation

    def send_task(
        self,
        message: str,
        context: Optional[Dict[str, Any]] = None,
        session_id: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Send a task to the ADK agent via A2A protocol.

        Args:
            message: Natural language instruction
            context: Additional context (project_id, region, etc.)
            session_id: Conversation session ID for Memory Bank

        Returns:
            Task response with task_id for async operations
        """
        # Create or reuse session ID
        if session_id is None:
            self.session_id = self.session_id or str(uuid.uuid4())
        else:
            self.session_id = session_id

        payload = {
            "message": message,
            "session_id": self.session_id,
            "context": context or {},
            "config": {
                "enable_code_execution": True,
                "enable_memory_bank": True,
            }
        }

        response = requests.post(
            f"{self.agent_endpoint}/v1/tasks:send",
            json=payload,
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self._get_auth_token()}",
            }
        )

        return response.json()

    def get_task_status(self, task_id: str) -> Dict[str, Any]:
        """
        Check status of a long-running task.

        Returns:
            {
                "task_id": "...",
                "status": "PENDING" | "RUNNING" | "SUCCESS" | "FAILURE",
                "output": "...",  # If completed
                "error": "...",   # If failed
                "progress": 0.75  # Optional progress indicator
            }
        """
        response = requests.get(
            f"{self.agent_endpoint}/v1/tasks/{task_id}",
            headers={"Authorization": f"Bearer {self._get_auth_token()}"}
        )
        return response.json()
```

### 4. Handling Long-Running Operations

Many GCP operations (creating GKE clusters, deploying services) are asynchronous:

**Pattern 1: Submit and Poll**

```python
def execute_async_deployment(client, deployment_request):
    """
    Submit deployment task and poll until completion.
    """
    # Step 1: Submit task
    task_response = client.send_task(
        message=f"Deploy GKE cluster named {deployment_request['cluster_name']}",
        context=deployment_request
    )

    task_id = task_response["task_id"]
    print(f"✅ Task submitted: {task_id}")

    # Step 2: Poll for completion
    import time
    while True:
        status = client.get_task_status(task_id)

        if status["status"] == "SUCCESS":
            print(f"✅ Deployment succeeded!")
            print(f"Output: {status['output']}")
            return status["output"]

        elif status["status"] == "FAILURE":
            print(f"❌ Deployment failed!")
            print(f"Error: {status['error']}")
            raise Exception(status["error"])

        elif status["status"] in ["PENDING", "RUNNING"]:
            progress = status.get("progress", 0)
            print(f"⏳ Status: {status['status']} ({progress*100:.0f}%)")
            time.sleep(10)  # Poll every 10 seconds
```

**Pattern 2: Immediate Response for User**

```python
def start_deployment_task(client, deployment_request):
    """
    Submit task and return task_id immediately to user.
    User can check status later.
    """
    task_response = client.send_task(
        message=f"Deploy GKE cluster named {deployment_request['cluster_name']}",
        context=deployment_request
    )

    task_id = task_response["task_id"]

    return {
        "message": f"✅ Deployment task started!",
        "task_id": task_id,
        "check_status": f"Use /check-task-status {task_id} to monitor progress",
    }
```

### 5. Memory Bank Integration

The session_id enables the ADK agent to remember context across multiple interactions:

**Multi-Turn Conversation Example**:

```
Turn 1:
User: "Deploy a GKE cluster named prod-cluster in us-central1"
Claude → ADK Agent (session_id: abc-123)
ADK: Creates cluster, stores context in Memory Bank

Turn 2:
User: "Now deploy a Cloud Run service that connects to that cluster"
Claude → ADK Agent (session_id: abc-123)
ADK: Retrieves cluster info from Memory Bank, deploys service with connection

Turn 3:
User: "What's the status of the cluster?"
Claude → ADK Agent (session_id: abc-123)
ADK: Knows which cluster from Memory Bank, returns current status
```

Implementation:

```python
class ConversationalA2AClient:
    def __init__(self, agent_endpoint: str):
        self.client = A2AClient(agent_endpoint)
        self.conversation_history = []

    def chat(self, user_message: str) -> str:
        """
        Maintain conversational context via Memory Bank.
        """
        # Session ID persists across conversation
        result = self.client.send_task(
            message=user_message,
            context={
                "conversation_history": self.conversation_history[-5:],  # Last 5 turns
            }
        )

        self.conversation_history.append({
            "user": user_message,
            "agent": result["output"]
        })

        return result["output"]
```

### 6. Multi-Agent Orchestration via A2A

Coordinate multiple ADK agents for complex workflows:

```python
class MultiAgentOrchestrator:
    def __init__(self):
        self.agents = {
            "deployer": A2AClient("https://deployer-agent.run.app"),
            "validator": A2AClient("https://validator-agent.run.app"),
            "monitor": A2AClient("https://monitor-agent.run.app"),
        }
        self.session_id = str(uuid.uuid4())  # Shared session across agents

    def deploy_with_validation(self, deployment_config):
        """
        Orchestrate deployment with validation and monitoring.
        """
        # Step 1: Validate configuration
        validation_result = self.agents["validator"].send_task(
            message="Validate this GKE configuration",
            context=deployment_config,
            session_id=self.session_id
        )

        if validation_result["status"] != "VALID":
            return {"error": "Configuration validation failed"}

        # Step 2: Deploy
        deploy_result = self.agents["deployer"].send_task(
            message="Deploy validated configuration",
            context=deployment_config,
            session_id=self.session_id  # Can access validation context
        )

        task_id = deploy_result["task_id"]

        # Step 3: Monitor deployment
        monitor_result = self.agents["monitor"].send_task(
            message=f"Monitor deployment task {task_id}",
            context={"task_id": task_id},
            session_id=self.session_id
        )

        return {
            "validation": validation_result,
            "deployment_task_id": task_id,
            "monitoring_enabled": True
        }
```

### 7. Error Handling & Retry Logic

```python
from tenacity import retry, stop_after_attempt, wait_exponential

class ResilientA2AClient(A2AClient):
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    def send_task_with_retry(self, message: str, context: dict = None):
        """
        Send task with automatic retry on transient failures.
        """
        try:
            return self.send_task(message, context)
        except requests.exceptions.Timeout:
            print("⏱️ Request timeout, retrying...")
            raise
        except requests.exceptions.ConnectionError:
            print("🔌 Connection error, retrying...")
            raise
```

## When to Use This Agent

Activate this agent when:
- Communicating with deployed ADK agents on Agent Engine
- Setting up multi-agent workflows
- Managing stateful conversations with Memory Bank
- Coordinating async GCP deployments
- Orchestrating ADK, LangChain, and Genkit agents

## Best Practices

1. **Always maintain session_id** for conversational context
2. **Poll async tasks** with exponential backoff
3. **Discover AgentCard** before invoking unknown agents
4. **Handle failures gracefully** with retries
5. **Log all interactions** for debugging
6. **Use structured context** (JSON objects, not freeform strings)
7. **Implement timeouts** for long-running operations

## Security Considerations

1. **Authentication**: Always include proper Authorization headers
2. **Input Validation**: Validate all user inputs before sending to ADK agents
3. **Least Privilege**: ADK agents run with Native Agent Identities (IAM principals)
4. **Audit Logging**: All A2A calls are logged in Cloud Logging

## References

- A2A Protocol Spec: https://google.github.io/adk-docs/a2a/
- ADK Documentation: https://google.github.io/adk-docs/
- Python SDK: `pip install google-adk`
- Agent Engine Overview: https://cloud.google.com/vertex-ai/generative-ai/docs/agent-engine/overview