Connecting Gemini Agents with Google's A2A Protocol: Design, Implementation, and Cloud Run Operations

# a2a_client.py
import httpx
import asyncio
import uuid
from typing import Optional
 
class A2AClient:
    """A2A-compliant client with discovery, task submission, and polling"""
 
    def __init__(self, agent_url: str, bearer_token: Optional[str] = None):
        self.agent_url = agent_url.rstrip("/")
        self.headers = {"Authorization": f"Bearer {bearer_token}"} if bearer_token else {}
 
    async def get_agent_card(self) -> dict:
        """Discover what the agent can do"""
        async with httpx.AsyncClient() as client:
            res = await client.get(
                f"{self.agent_url}/.well-known/agent.json",
                headers=self.headers
            )
            res.raise_for_status()
            return res.json()
 
    async def send_task_and_wait(
        self,
        message_text: str,
        poll_interval: float = 2.0,
        max_wait: int = 300
    ) -> dict:
        """Submit a task and poll until completion"""
        task_id = str(uuid.uuid4())
        payload = {
            "id": task_id,
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": message_text}]
            }
        }
 
        async with httpx.AsyncClient(timeout=30.0) as client:
            await client.post(
                f"{self.agent_url}/tasks/send",
                json=payload,
                headers=self.headers
            )
 
            elapsed = 0
            while elapsed < max_wait:
                await asyncio.sleep(poll_interval)
                elapsed += poll_interval
 
                res = await client.get(
                    f"{self.agent_url}/tasks/get",
                    params={"id": task_id},
                    headers=self.headers
                )
                task = res.json()
                state = task["status"]["state"]
 
                if state == "completed":
                    return task
                elif state == "failed":
                    err = task["status"].get("message", {})
                    raise RuntimeError(f"Task failed (code={err.get('code')}): {err.get('message')}")
 
            raise TimeoutError(f"Task did not complete within {max_wait} seconds")
 
    @staticmethod
    def extract_text(task: dict) -> str:
        """Pull text from the first text artifact"""
        for artifact in task.get("artifacts", []):
            for part in artifact.get("parts", []):
                if part.get("type") == "text":
                    return part["text"]
        return ""

# adk_a2a_wrapper.py
from google.adk.agents import LlmAgent, Agent
from google.adk.tools import google_search, code_execution
from google.adk.runners import Runner
 
# --- Specialist agents ---
 
research_agent = LlmAgent(
    name="research_specialist",
    model="gemini-2.5-pro",
    description="Gathers and synthesizes up-to-date information using Google Search",
    instruction="""
    You are an expert researcher. For each query, use Google Search to find
    accurate, recent information and present it concisely with source URLs.
    Flag any information that may be out of date.
    """,
    tools=[google_search]
)
 
code_agent = LlmAgent(
    name="code_executor",
    model="gemini-2.5-pro",
    description="Writes and runs Python code for data processing, calculations, and automation",
    instruction="""
    You are a data science expert. Implement the requested computation in Python,
    run it, and return concrete numerical results. Keep code clean with comments.
    Present results as numbers, tables, or chart URLs.
    """,
    tools=[code_execution]
)
 
# --- Orchestrator ---
 
orchestrator = Agent(
    name="research_orchestrator",
    model="gemini-2.5-pro",
    description="Coordinates research and code execution for advanced analytical tasks",
    instruction="""
    You are a multi-agent orchestrator. Analyze each user request and route it
    to the appropriate sub-agents:
    - research_specialist: For information gathering, trend analysis, fact-finding
    - code_executor: For numerical computation, data processing, code generation
 
    Synthesize outputs from multiple agents into a coherent, comprehensive response.
    Indicate which agent contributed each section.
    """,
    sub_agents=[research_agent, code_agent]
)
 
async def run_orchestrated_task(user_request: str) -> str:
    """Execute the ADK orchestrator and return the result"""
    runner = Runner(agent=orchestrator)
    result = await runner.run_async(user_request)
    return str(result)

# pipeline.py
async def research_and_report_pipeline(topic: str) -> str:
    """Three-stage pipeline: research → analysis → report"""
 
    search_client = A2AClient("https://search-agent.run.app")
    analysis_client = A2AClient("https://analysis-agent.run.app")
    report_client = A2AClient("https://report-agent.run.app")
 
    # Stage 1: Information gathering
    print("📡 Stage 1: Gathering information...")
    raw_task = await search_client.send_task_and_wait(
        f"Research the latest developments on: {topic}"
    )
    raw_data = search_client.extract_text(raw_task)
 
    # Stage 2: Analysis
    print("🔍 Stage 2: Analyzing data...")
    analysis_task = await analysis_client.send_task_and_wait(
        f"Analyze the following information and extract the 5 most important insights:\n\n{raw_data}"
    )
    insights = analysis_client.extract_text(analysis_task)
 
    # Stage 3: Report generation
    print("📝 Stage 3: Generating report...")
    report_task = await report_client.send_task_and_wait(
        f"Write a 500-word executive summary based on these insights:\n\n{insights}"
    )
 
    return report_client.extract_text(report_task)

# fanout.py
async def parallel_competitive_analysis(company: str) -> dict:
    """Analyze market, technology, and risk simultaneously"""
 
    # All three tasks launch at the same time
    market_task, tech_task, risk_task = await asyncio.gather(
        A2AClient("https://market-agent.run.app").send_task_and_wait(
            f"Analyze {company}'s market share, competitors, and positioning"
        ),
        A2AClient("https://tech-agent.run.app").send_task_and_wait(
            f"Evaluate {company}'s technology stack and technical differentiation"
        ),
        A2AClient("https://risk-agent.run.app").send_task_and_wait(
            f"Identify key risks and challenges facing {company}"
        ),
        return_exceptions=True  # Continue even if one agent fails
    )
 
    def safe_extract(url: str, result) -> str:
        if isinstance(result, Exception):
            return f"Analysis error: {str(result)}"
        return A2AClient(url).extract_text(result)
 
    return {
        "market": safe_extract("https://market-agent.run.app", market_task),
        "technology": safe_extract("https://tech-agent.run.app", tech_task),
        "risk": safe_extract("https://risk-agent.run.app", risk_task)
    }

# dynamic_router.py
import json
 
# Use a fast, low-cost model for classification
router_model = genai.GenerativeModel("gemini-2.5-flash")
 
AGENT_REGISTRY = {
    "legal": {
        "url": "https://legal-agent.run.app",
        "description": "Contract review, legal document analysis, compliance checks"
    },
    "data": {
        "url": "https://data-agent.run.app",
        "description": "Statistical analysis, calculations, data visualization, predictive modeling"
    },
    "content": {
        "url": "https://content-agent.run.app",
        "description": "Article writing, copywriting, SEO optimization, editorial review"
    }
}
 
async def smart_route(user_request: str) -> str:
    """Classify the request and forward it to the right agent"""
    routing_prompt = f"""
Select the best agent for the user request below.
 
Available agents:
{json.dumps({k: v['description'] for k, v in AGENT_REGISTRY.items()})}
 
User request: {user_request}
 
Respond ONLY in this JSON format (no explanation):
{{"agent": "agent_id", "reason": "brief reason"}}
"""
 
    routing_res = router_model.generate_content(
        routing_prompt,
        generation_config={"response_mime_type": "application/json"}
    )
    routing = json.loads(routing_res.text)
    agent_id = routing.get("agent", list(AGENT_REGISTRY.keys())[0])
 
    if agent_id not in AGENT_REGISTRY:
        agent_id = list(AGENT_REGISTRY.keys())[0]
 
    print(f"🎯 Routing to: {agent_id} — {routing.get('reason', '')}")
    client = A2AClient(AGENT_REGISTRY[agent_id]["url"])
    task = await client.send_task_and_wait(user_request)
    return client.extract_text(task)

# streaming_server.py
from fastapi.responses import StreamingResponse
import json
 
@app.post("/tasks/sendSubscribe")
async def send_subscribe(req: TaskRequest):
    """Stream Gemini output to the client via SSE"""
    user_text = " ".join(
        part.get("text", "")
        for part in req.message.get("parts", [])
        if part.get("type") == "text"
    )
 
    async def event_stream():
        task_id = req.id
 
        # Signal that work has started
        yield f"data: {json.dumps({'type':'status','id':task_id,'status':{'state':'working'}})}\n\n"
 
        # Stream Gemini output chunk by chunk
        stream = model.generate_content(user_text, stream=True)
        for chunk in stream:
            if chunk.text:
                event = {
                    "type": "artifact",
                    "id": task_id,
                    "artifact": {
                        "name": "response",
                        "parts": [{"type": "text", "text": chunk.text}],
                        "append": True
                    }
                }
                yield f"data: {json.dumps(event)}\n\n"
                await asyncio.sleep(0)  # Yield control back to event loop
 
        # Signal completion
        yield f"data: {json.dumps({'type':'status','id':task_id,'status':{'state':'completed'}})}\n\n"
 
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )

# auth.py
import jwt
import time
import os
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from collections import defaultdict
 
JWT_SECRET = os.environ["JWT_SECRET"]
security = HTTPBearer()
 
def create_agent_token(agent_id: str, ttl_seconds: int = 3600) -> str:
    """Generate a JWT token for A2A communication"""
    return jwt.encode(
        {
            "agent_id": agent_id,
            "iat": int(time.time()),
            "exp": int(time.time()) + ttl_seconds
        },
        JWT_SECRET,
        algorithm="HS256"
    )
 
async def verify_token(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """Validate the Bearer token and return the agent payload"""
    try:
        payload = jwt.decode(
            credentials.credentials, JWT_SECRET, algorithms=["HS256"]
        )
        if payload.get("exp", 0) < time.time():
            raise HTTPException(status_code=401, detail="Token expired")
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
 
# In-memory rate limiter (use Redis in production)
request_log: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT_RPM = 60
 
async def rate_limit(agent: dict = Depends(verify_token)) -> dict:
    """Enforce per-agent rate limits"""
    agent_id = agent["agent_id"]
    now = time.time()
    request_log[agent_id] = [t for t in request_log[agent_id] if now - t < 60]
    if len(request_log[agent_id]) >= RATE_LIMIT_RPM:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Try again in 1 minute.",
            headers={"Retry-After": "60"}
        )
    request_log[agent_id].append(now)
    return agent

FROM python:3.12-slim
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
# Security: run as non-root
RUN useradd -m -u 1000 agentuser && chown -R agentuser:agentuser /app
USER agentuser
 
EXPOSE 8080
# 4 workers for production concurrency
CMD ["uvicorn", "a2a_server:app", \
     "--host", "0.0.0.0", "--port", "8080", \
     "--workers", "4", "--timeout-keep-alive", "120"]

PROJECT_ID="your-gcp-project-id"
REGION="asia-northeast1"
SERVICE="gemini-a2a-agent"
 
# Build and push
gcloud builds submit \
  --tag "asia-northeast1-docker.pkg.dev/${PROJECT_ID}/agents/${SERVICE}:latest"
 
# Deploy with secrets from Secret Manager
gcloud run deploy ${SERVICE} \
  --image "asia-northeast1-docker.pkg.dev/${PROJECT_ID}/agents/${SERVICE}:latest" \
  --region ${REGION} \
  --allow-unauthenticated \
  --memory 2Gi --cpu 2 \
  --min-instances 1 --max-instances 20 \
  --set-secrets "GEMINI_API_KEY=gemini-api-key:latest,JWT_SECRET=a2a-jwt-secret:latest"
 
# Confirm the deployment URL
gcloud run services describe ${SERVICE} --region ${REGION} --format "value(status.url)"

# monitoring.py
from google.cloud import monitoring_v3
import time
 
monitoring_client = monitoring_v3.MetricServiceClient()
PROJECT_NAME = "projects/your-gcp-project-id"
 
def record_task_latency(agent_name: str, duration_sec: float, success: bool):
    """Write task latency to Cloud Monitoring as a custom metric"""
    series = monitoring_v3.TimeSeries()
    series.metric.type = "custom.googleapis.com/a2a/task_latency_seconds"
    series.metric.labels["agent"] = agent_name
    series.metric.labels["status"] = "success" if success else "failure"
 
    now = time.time()
    point = monitoring_v3.Point({
        "interval": {"end_time": {"seconds": int(now)}},
        "value": {"double_value": duration_sec}
    })
    series.points = [point]
    monitoring_client.create_time_series(name=PROJECT_NAME, time_series=[series])

# agent_card.py
import hashlib
import json
from fastapi import APIRouter, Request, Response
 
router = APIRouter()
 
AGENT_CARD = {
    "name": "summarizer-agent",
    "description": "Summarizes technical articles to a target length",
    "url": "https://summarizer-agent.example.run.app",
    "version": "2.1.0",  # always bump when skills change
    "capabilities": {"streaming": True, "pushNotifications": False},
    "skills": [
        {
            "id": "summarize",
            "name": "summarize",
            "description": "Summarize body text",
            "inputModes": ["text/plain", "application/json"],
            "outputModes": ["application/json"],
        }
    ],
}
 
_CARD_BYTES = json.dumps(AGENT_CARD, ensure_ascii=False, sort_keys=True).encode()
_CARD_ETAG = '"' + hashlib.sha256(_CARD_BYTES).hexdigest()[:16] + '"'
 
 
@router.get("/.well-known/agent.json")
async def agent_card(request: Request) -> Response:
    # 304 when nothing changed; force a refetch the moment it does
    if request.headers.get("if-none-match") == _CARD_ETAG:
        return Response(status_code=304, headers={"ETag": _CARD_ETAG})
 
    return Response(
        content=_CARD_BYTES,
        media_type="application/json",
        headers={
            "ETag": _CARD_ETAG,
            # five minutes is fine; beyond that, revalidate
            "Cache-Control": "public, max-age=300, must-revalidate",
        },
    )

# client_version_check.py
import logging
 
logger = logging.getLogger(__name__)
 
MIN_SUPPORTED = {"summarizer-agent": "2.0.0"}
 
 
def _parse(v: str) -> tuple[int, ...]:
    return tuple(int(x) for x in v.split("."))
 
 
def assert_card_compatible(card: dict) -> None:
    name = card.get("name", "unknown")
    version = card.get("version", "0.0.0")
    logger.info("resolved agent card name=%s version=%s", name, version)
 
    required = MIN_SUPPORTED.get(name)
    if required and _parse(version) < _parse(required):
        raise RuntimeError(
            f"{name} card is too old: {version} < {required}. Suspect a cache."
        )

# idempotent_task.py
import asyncio
import time
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable
 
from fastapi import APIRouter, Header, HTTPException
 
router = APIRouter()
 
 
@dataclass
class _Entry:
    status: str  # "running" | "done"
    result: Any = None
    created_at: float = field(default_factory=time.time)
 
 
# In production use a shared store (Firestore / Redis).
# Cloud Run spreads traffic across instances, so an in-process dict won't hold.
_STORE: dict[str, _Entry] = {}
_TTL_SEC = 24 * 60 * 60
_LOCK = asyncio.Lock()
 
 
async def run_idempotent(
    key: str, work: Callable[[], Awaitable[Any]]
) -> tuple[Any, bool]:
    """Returns (result, whether it was reused)."""
    async with _LOCK:
        entry = _STORE.get(key)
        if entry and time.time() - entry.created_at > _TTL_SEC:
            entry = None
            _STORE.pop(key, None)
 
        if entry is None:
            _STORE[key] = _Entry(status="running")
        elif entry.status == "done":
            return entry.result, True
        else:
            # An earlier request is still working. Tell the caller to come back.
            raise HTTPException(status_code=409, detail="task in progress")
 
    try:
        result = await work()
    except Exception:
        async with _LOCK:
            _STORE.pop(key, None)  # never remember a failure; allow the retry
        raise
 
    async with _LOCK:
        _STORE[key] = _Entry(status="done", result=result)
    return result, False
 
 
@router.post("/tasks/send")
async def send_task(
    payload: dict,
    idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
):
    if not idempotency_key:
        raise HTTPException(status_code=400, detail="Idempotency-Key header required")
 
    async def _work() -> dict:
        # the actual Gemini call happens here
        return await handle_task(payload)
 
    result, reused = await run_idempotent(idempotency_key, _work)
    return {"result": result, "reused": reused}