Gemini CLI × A2A Protocol — Remote Agent Integration

Complex workflows exceed single-agent capabilities. Sales pipelines require coordination between proposal agents, contract validation agents, and report generators. Research workflows demand collaboration across literature search, data analysis, and paper synthesis agents.

Gemini CLI v0.33.0 (March 2026) introduced the A2A (Agent-to-Agent) protocol with HTTP authentication and agent card discovery, enabling secure, scalable multi-agent collaboration.

What follows covers the A2A protocol's fundamentals, the implementation and security patterns that matter, and a production deployment on Cloud Run and Gemini Enterprise.

Understanding the A2A Protocol

What is A2A?

A2A (Agent-to-Agent) is an open standard for inter-agent communication. Google donated the specification to the Linux Foundation in 2023, and Gemini CLI adopted it fully in March 2026.

A2A's Core Characteristics:

Open standard: Managed by Linux Foundation. Designed for interoperability with other AI frameworks (LangChain, LlamaIndex, etc.)
HTTP-based: REST API implementation enables server-based agent deployment
Agent cards: Metadata-driven discovery mechanism defining capabilities, input schemas, and authentication methods

A2A in Gemini CLI

Gemini CLI embeds A2A metadata in agent definition files (Markdown with YAML frontmatter).

Example: Sales Support Agent Definition

---
name: "Sales Support Agent"
description: "Auto-generates sales proposals and performs risk assessment"
version: "1.0.0"
a2a:
  endpoint: "https://sales-agent.example.com/a2a"
  auth:
    type: "api_key"
    header: "X-API-Key"
  capabilities:
    - "proposal_generation"
    - "risk_assessment"
    - "contract_review"
  input_format:
    type: "json"
    schema:
      properties:
        client_name: { type: "string" }
        product_id: { type: "string" }
        budget_range: { type: "number" }
---
 
You are a sales support specialist...
(Agent instruction follows)

Placing this file in .gemini/agents/ enables Gemini CLI to recognize the A2A endpoint, making it callable from other agents.

HTTP Authentication and Security Design

Authentication Methods Comparison

API Key: Implementation Low / Security Medium / Production Low — Internal & testing use
OAuth 2.0: Implementation Medium / Security High / Production High — Enterprise partnerships
Service Account: Implementation Medium / Security High / Production High — Google Cloud internal
mTLS: Implementation High / Security Highest / Production High — Finance & healthcare

API Key Authentication (Simple)

---
name: "Internal Analytics Agent"
a2a:
  endpoint: "http://localhost:8000/agent"
  auth:
    type: "api_key"
    header: "X-Gemini-API-Key"
    secret: "${INTERNAL_AGENT_KEY}"
---

Environment setup:

export INTERNAL_AGENT_KEY="gemini-agent-key-xyz123..."

OAuth 2.0 Authentication (Enterprise)

---
name: "Enterprise Contract Agent"
a2a:
  endpoint: "https://contract-service.company.com/agent"
  auth:
    type: "oauth2"
    client_id: "${OAUTH_CLIENT_ID}"
    client_secret: "${OAUTH_CLIENT_SECRET}"
    token_endpoint: "https://auth.company.com/oauth/token"
    scopes:
      - "agent:read"
      - "agent:write"
---

Token acquisition flow:

import requests
from typing import Optional
 
def get_oauth_token(
    client_id: str,
    client_secret: str,
    token_endpoint: str,
    scopes: list = None
) -> str:
    """Obtain OAuth 2.0 access token"""
    if scopes is None:
        scopes = ["agent:read", "agent:write"]
 
    payload = {
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "client_credentials",
        "scope": " ".join(scopes)
    }
 
    response = requests.post(token_endpoint, json=payload)
    response.raise_for_status()
    return response.json()["access_token"]
 
# Token automatically used in agent invocations
token = get_oauth_token(
    client_id="YOUR_OAUTH_CLIENT_ID",
    client_secret="YOUR_OAUTH_CLIENT_SECRET",
    token_endpoint="https://auth.company.com/oauth/token"
)

Service Account Authentication (Google Cloud)

---
name: "GCP Data Agent"
a2a:
  endpoint: "https://data-agent-xxxxx-uc.a.run.app"
  auth:
    type: "service_account"
    service_account_file: "${SERVICE_ACCOUNT_JSON}"
    scopes:
      - "https://www.googleapis.com/auth/cloud-platform"
---

Configuration file (service-account.json):

{
  "type": "service_account",
  "project_id": "my-gemini-project",
  "private_key_id": "key123...",
  "private_key": "-----BEGIN PRIVATE KEY-----\n...",
  "client_email": "agent@my-gemini-project.iam.gserviceaccount.com",
  "client_id": "123456789",
  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
  "token_uri": "https://oauth2.googleapis.com/token",
  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs"
}

mTLS Authentication (Maximum Security)

---
name: "Secure Medical Agent"
a2a:
  endpoint: "https://medical-agent.hospital.com:8443"
  auth:
    type: "mtls"
    client_cert: "${CLIENT_CERT_PATH}"
    client_key: "${CLIENT_KEY_PATH}"
    ca_cert: "${CA_CERT_PATH}"
---

Certificate generation (testing):

# CA certificate
openssl genrsa -out ca-key.pem 2048
openssl req -new -x509 -days 365 -key ca-key.pem -out ca-cert.pem
 
# Client certificate
openssl genrsa -out client-key.pem 2048
openssl req -new -key client-key.pem -out client.csr
openssl x509 -req -days 365 -in client.csr \
  -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial \
  -out client-cert.pem
 
# Verify
openssl verify -CAfile ca-cert.pem client-cert.pem

Complete Agent Definition Implementation

Step 1: Creating Agent Definition

Project structure:

my-gemini-project/
├── .gemini/
│   ├── config.yaml
│   └── agents/
│       ├── sales-support.md
│       ├── contract-reviewer.md
│       └── report-generator.md
└── agents/
    ├── server.py
    └── requirements.txt

Sales support agent (sales-support.md):

---
name: "Sales Support Agent"
slug: "sales-support"
version: "1.0.0"
description: "Auto-generates sales proposals and performs risk evaluation"
author: "Sales Team"
tags: ["sales", "proposal", "risk-assessment"]
 
# A2A Metadata
a2a:
  # Endpoint configuration
  endpoint: "https://sales-agent.example.com/a2a"
 
  # HTTP authentication
  auth:
    type: "oauth2"
    client_id: "${OAUTH_CLIENT_ID}"
    client_secret: "${OAUTH_CLIENT_SECRET}"
    token_endpoint: "https://auth.example.com/oauth/token"
    scopes:
      - "agent:sales:read"
      - "agent:contract:read"
 
  # Capabilities list
  capabilities:
    - name: "proposal_generation"
      description: "Generate customized proposals"
      input:
        type: "object"
        properties:
          client_name: { type: "string" }
          budget: { type: "number" }
 
    - name: "risk_assessment"
      description: "Conduct contract risk evaluation"
      input:
        type: "object"
        properties:
          client_name: { type: "string" }
          contract_amount: { type: "number" }
 
  # Subagent specification
  subagents:
    - slug: "contract-reviewer"
      role: "contract_validation"
    - slug: "report-generator"
      role: "report_creation"
 
  # Caching configuration
  caching:
    enabled: true
    ttl_seconds: 3600
 
  # Rate limiting
  rate_limiting:
    requests_per_minute: 100
    burst_size: 10
 
# Gemini Prompt Configuration
model: "gemini-2.5-pro"
temperature: 0.7
max_tokens: 4000
 
system_prompt: |
  You are an expert sales support agent with deep knowledge of enterprise contracts
  and risk assessment.
 
  Your responsibilities:
  1. Generate customized sales proposals based on client needs and budget
  2. Assess contract risks and identify potential issues
  3. Coordinate with contract reviewers and report generators
 
  Available subagents:
  - contract-reviewer: Validates contracts for legal compliance
  - report-generator: Creates comprehensive reports
 
  Always prioritize accuracy and compliance. For deals >$100K,
  escalate to senior management review.
---
 
I will help with sales proposals and risk assessment...

Step 2: Defining Subagents

Contract review agent (contract-reviewer.md):

---
name: "Contract Reviewer Agent"
slug: "contract-reviewer"
version: "1.0.0"
description: "Legal risk assessment and clause validation"
a2a:
  endpoint: "https://contract-service.example.com/a2a"
  auth:
    type: "service_account"
    service_account_file: "${SERVICE_ACCOUNT_JSON}"
  capabilities:
    - name: "review_contract"
      description: "Assess contract risk"
      input:
        type: "object"
        properties:
          contract_text: { type: "string" }
    - name: "flag_problematic_clauses"
      description: "Identify problematic clauses"
      input:
        type: "object"
        properties:
          contract_text: { type: "string" }
---
 
You are a legal expert specializing in enterprise contracts...

Step 3: Gemini CLI Configuration

.gemini/config.yaml:

project_name: "Sales Automation Pipeline"
version: "1.0.0"
 
agents:
  default: "sales-support"
  directory: ".gemini/agents"
 
  discovery:
    enabled: true
    auto_register: true
 
a2a:
  enabled: true
 
  # HTTP server configuration
  server:
    host: "0.0.0.0"
    port: 8000
    ssl:
      enabled: true
      cert_file: "certs/server.crt"
      key_file: "certs/server.key"
 
  # Agent card discovery
  discovery:
    enabled: true
    endpoints:
      - "https://agent-registry.example.com/agents"
      - "http://localhost:8000/agents"
 
api_keys:
  gemini: "${GEMINI_API_KEY}"
 
logging:
  level: "info"
  format: "json"

A2A Server Implementation

Python (FastAPI) Implementation

from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional, Dict, Any
import jwt
import json
import logging
import aiohttp
 
app = FastAPI(title="Sales Agent Server")
logger = logging.getLogger(__name__)
 
# OAuth token validation
SECRET_KEY = "YOUR_SECRET_KEY"
 
class AgentRequest(BaseModel):
    """A2A request model"""
    agent_id: str
    action: str
    params: Dict[str, Any]
    context: Optional[Dict[str, Any]] = None
 
class AgentResponse(BaseModel):
    """A2A response model"""
    status: str
    result: Optional[Dict[str, Any]]
    error: Optional[str] = None
 
async def verify_token(authorization: str = Header(...)) -> Dict:
    """OAuth token verification"""
    try:
        scheme, token = authorization.split(" ")
        if scheme.lower() != "bearer":
            raise HTTPException(status_code=401, detail="Invalid scheme")
 
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
 
@app.post("/a2a/agents/{agent_id}/execute")
async def execute_agent_action(
    agent_id: str,
    request: AgentRequest,
    token: Dict = Depends(verify_token)
) -> AgentResponse:
    """
    A2A agent execution endpoint
 
    Callable by other agents
    """
    logger.info(f"Agent {agent_id} executing action: {request.action}")
 
    try:
        # Validate agent ID
        if agent_id != "sales-support":
            raise ValueError(f"Unknown agent: {agent_id}")
 
        # Action routing
        if request.action == "proposal_generation":
            result = generate_proposal(
                client_name=request.params.get("client_name"),
                budget=request.params.get("budget")
            )
        elif request.action == "risk_assessment":
            # Invoke contract-reviewer subagent
            result = await call_subagent(
                subagent="contract-reviewer",
                action="review_contract",
                params=request.params
            )
        else:
            raise ValueError(f"Unknown action: {request.action}")
 
        return AgentResponse(
            status="success",
            result=result
        )
 
    except Exception as e:
        logger.error(f"Agent execution failed: {e}")
        return AgentResponse(
            status="error",
            result=None,
            error=str(e)
        )
 
async def call_subagent(subagent: str, action: str, params: Dict) -> Dict:
    """
    Invoke subagent via A2A protocol
    """
    # Retrieve subagent URL from agent registry
    subagent_url = f"https://contract-service.example.com/a2a/agents/{subagent}/execute"
 
    headers = {
        "Authorization": f"Bearer {get_service_token()}",
        "Content-Type": "application/json"
    }
 
    payload = {
        "agent_id": subagent,
        "action": action,
        "params": params
    }
 
    async with aiohttp.ClientSession() as session:
        async with session.post(subagent_url, json=payload, headers=headers) as resp:
            if resp.status != 200:
                raise Exception(f"Subagent failed: {await resp.text()}")
            return await resp.json()
 
def generate_proposal(client_name: str, budget: float) -> Dict:
    """Proposal generation logic"""
    return {
        "proposal_id": f"PROP-{client_name}-001",
        "title": f"Custom Solution for {client_name}",
        "budget": budget,
        "timeline": "30 days",
        "components": ["Feature A", "Support Package", "Training"]
    }
 
def get_service_token() -> str:
    """Retrieve Service Account token"""
    # Obtain OAuth token from Google Cloud credentials
    return "service-token-xyz"
 
@app.get("/a2a/agents/{agent_id}/card")
async def get_agent_card(agent_id: str) -> Dict:
    """
    Retrieve agent card for discovery
 
    Used by other agents to discover capabilities
    """
    return {
        "id": agent_id,
        "name": "Sales Support Agent",
        "version": "1.0.0",
        "endpoint": "https://sales-agent.example.com/a2a",
        "capabilities": [
            {
                "name": "proposal_generation",
                "description": "Generate proposals"
            },
            {
                "name": "risk_assessment",
                "description": "Assess contract risks"
            }
        ],
        "auth": {
            "type": "oauth2",
            "token_endpoint": "https://auth.example.com/oauth/token"
        }
    }
 
@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "version": "1.0.0"}
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Environment Setup

requirements.txt:

fastapi==0.104.0
uvicorn[standard]==0.24.0
pydantic==2.5.0
pyjwt==2.8.1
aiohttp==3.9.1
google-auth==2.25.2
google-cloud-secret-manager==2.16.4

Agent Invocation Patterns

Direct Gemini CLI Invocation

# Run local agent
gemini agent run sales-support --prompt "Create a proposal for TechCorp with $500K budget"
 
# Enable A2A subagent invocation
gemini agent run sales-support --prompt "Review contract for ABC Corp" \
  --subagent-enabled
 
# JSON-based invocation
gemini agent run sales-support << 'EOF'
{
  "action": "proposal_generation",
  "client_name": "Enterprise Inc",
  "budget": 250000
}
EOF

Python Client Implementation

from gemini_sdk import GeminiAgent
import json
import asyncio
 
class SalesOrchestrator:
    """Sales workflow orchestration"""
 
    def __init__(self, api_key: str):
        self.agent = GeminiAgent(
            agent_id="sales-support",
            api_key=api_key,
            enable_subagents=True
        )
 
    async def process_sales_opportunity(self, opportunity: Dict) -> Dict:
        """Process sales opportunity (A2A workflow)"""
 
        # Step 1: Generate proposal
        proposal = await self.agent.call(
            action="proposal_generation",
            params={
                "client_name": opportunity["client_name"],
                "budget": opportunity["budget"],
                "industry": opportunity["industry"]
            }
        )
 
        # Step 2: Contract review (contract-reviewer subagent)
        review = await self.agent.call(
            action="risk_assessment",
            params={
                "client_name": opportunity["client_name"],
                "contract_amount": opportunity["budget"]
            }
        )
 
        # Step 3: Generate comprehensive report
        report = await self.agent.call(
            action="generate_report",
            params={
                "proposal": proposal,
                "review": review,
                "client": opportunity["client_name"]
            }
        )
 
        return {
            "proposal": proposal,
            "risk_review": review,
            "final_report": report,
            "status": "ready_for_approval"
        }
 
# Usage example
async def main():
    orchestrator = SalesOrchestrator(api_key="YOUR_GEMINI_API_KEY")
 
    opportunity = {
        "client_name": "Global Tech Solutions",
        "budget": 750000,
        "industry": "Financial Services",
        "timeline": "Q2 2026"
    }
 
    result = await orchestrator.process_sales_opportunity(opportunity)
    print(json.dumps(result, indent=2))
 
if __name__ == "__main__":
    asyncio.run(main())

Production Deployment (Cloud Run + Gemini Enterprise)

Deploying to Cloud Run

Dockerfile:

FROM python:3.11-slim
 
WORKDIR /app
 
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# Copy application code
COPY agents/ ./agents/
COPY certs/ ./certs/
 
# Port configuration
EXPOSE 8000
 
# Start service
CMD ["python", "agents/server.py"]

Deployment script:

#!/bin/bash
 
PROJECT_ID="my-gemini-project"
SERVICE_NAME="sales-agent"
REGION="us-central1"
 
# Build image
docker build -t gcr.io/${PROJECT_ID}/${SERVICE_NAME}:latest .
 
# Push to registry
docker push gcr.io/${PROJECT_ID}/${SERVICE_NAME}:latest
 
# Deploy to Cloud Run
gcloud run deploy ${SERVICE_NAME} \
  --image gcr.io/${PROJECT_ID}/${SERVICE_NAME}:latest \
  --region ${REGION} \
  --platform managed \
  --allow-unauthenticated \
  --set-env-vars GEMINI_API_KEY=${GEMINI_API_KEY},\
OAUTH_CLIENT_ID=${OAUTH_CLIENT_ID},\
OAUTH_CLIENT_SECRET=${OAUTH_CLIENT_SECRET} \
  --memory 2Gi \
  --cpu 2 \
  --timeout 3600
 
# Retrieve service URL
gcloud run services describe ${SERVICE_NAME} --region ${REGION} --format='value(status.url)'

Gemini Enterprise Integration

gemini-enterprise.yaml:

project_id: "my-gemini-project"
enterprise_enabled: true
 
agents:
  sales-support:
    name: "Sales Support Agent"
    deployment:
      platform: "cloud_run"
      url: "https://sales-agent-xxxxx-uc.a.run.app"
      region: "us-central1"
      min_replicas: 2
      max_replicas: 10
 
    monitoring:
      enabled: true
      metrics:
        - "request_latency"
        - "error_rate"
        - "token_usage"
 
    scaling:
      metric: "cpu"
      target_cpu_percent: 70
 
    backup:
      enabled: true
      frequency: "daily"
 
a2a_network:
  discovery:
    enabled: true
    registry_url: "https://agent-registry.example.com"
 
  mesh:
    mtls:
      enabled: true
      cert_rotation_days: 90
 
  rate_limiting:
    global: 10000
    per_agent: 1000

Troubleshooting and Monitoring

Common Errors and Solutions

Error 1: Authentication Failure

Error: A2A authentication failed: Invalid token

Solution:

import jwt
from datetime import datetime
 
def check_token_validity(token):
    try:
        decoded = jwt.decode(token, options={"verify_signature": False})
        expiry = decoded.get("exp")
        expiry_time = datetime.utcfromtimestamp(expiry)
        print(f"Token expires at: {expiry_time}")
    except jwt.DecodeError as e:
        print(f"Invalid token: {e}")

Error 2: Subagent Discovery Failure

Error: Subagent 'contract-reviewer' not found in registry

Solution:

# Verify agent card
curl -H "Authorization: Bearer $TOKEN" \
  https://agent-registry.example.com/agents/contract-reviewer
 
# List available A2A agents
gemini agent list --a2a-enabled

Error 3: Rate Limit Exceeded

Error: Rate limit exceeded: 100 requests/minute

Solution: Batch requests or implement async processing:

import asyncio
from typing import List
 
async def batch_process(requests: List[dict], max_concurrent: int = 10):
    """Process requests within rate limits"""
    semaphore = asyncio.Semaphore(max_concurrent)
 
    async def process_one(request):
        async with semaphore:
            return await agent.call(request)
 
    return await asyncio.gather(*[process_one(r) for r in requests])

Monitoring and Logging

import logging
from pythonjsonlogger import jsonlogger
from datetime import datetime
 
# JSON logging setup
logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)
 
# Metric recording
def log_agent_execution(agent_id: str, action: str, duration_ms: float, success: bool):
    logger.info("agent_execution", extra={
        "agent_id": agent_id,
        "action": action,
        "duration_ms": duration_ms,
        "success": success,
        "timestamp": datetime.utcnow().isoformat()
    })

Best Practices

Security

Auth selection: API Key for internal use, OAuth 2.0 for partnerships, Service Account for Google Cloud
Token rotation: Update tokens regularly (every 90 days)
Secret management: Store credentials in .env or Secrets Manager

Performance

Caching: Cache frequently-called subagent results
Timeouts: Always set A2A request timeouts (default: 30 seconds)
Async operations: Use async/await for concurrent agent invocations

Operations

Health checks: Implement .../health endpoint
Log aggregation: Centralize all agent logs in Cloud Logging or ELK
Regular updates: Check A2A protocol specifications monthly

Looking back

Gemini CLI's A2A protocol enables enterprise-grade multi-agent collaboration.

Key Takeaways:

A2A Protocol Fundamentals: Open standard, HTTP-based, agent card discovery
Authentication Implementation: Four approaches from simple API Key to mTLS
Server Architecture: FastAPI endpoints for A2A communication
Production Deployment: Cloud Run integration with Gemini Enterprise

Next Steps:

Review Getting Started with Gemini CLI for fundamentals
Study Gemini CLI March 2026 Update — Plan Mode, Subagents, Sandboxing for latest features
Explore Google ADK vs LangChain — AI Agent Framework Comparison for framework selection

Build robust multi-agent systems with A2A protocol and unlock new possibilities in AI-driven enterprise automation.

References

Gemini CLI official documentation
A2A protocol specification (Linux Foundation)
Cloud Run deployment guide (Google Cloud)