Google A2A プロトコル × Gemini API でエージェントを相互接続する — 設計から Cloud Run 運用まで

# a2a_client.py
import httpx
import asyncio
import uuid
from typing import Optional
 
class A2AClient:
    """A2A プロトコル準拠のクライアントライブラリ"""
 
    def __init__(self, agent_url: str, bearer_token: Optional[str] = None):
        self.agent_url = agent_url.rstrip("/")
        self.headers = {}
        if bearer_token:
            self.headers["Authorization"] = f"Bearer {bearer_token}"
 
    async def get_agent_card(self) -> dict:
        """エージェントの能力を確認する"""
        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:
        """タスクを送信し、完了まで待機してから結果を返す"""
        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
 
                status_res = await client.get(
                    f"{self.agent_url}/tasks/get",
                    params={"id": task_id},
                    headers=self.headers
                )
                task = status_res.json()
                state = task["status"]["state"]
 
                if state == "completed":
                    return task
                elif state == "failed":
                    error = task["status"].get("message", {})
                    raise RuntimeError(f"タスク失敗 (code={error.get('code')}): {error.get('message')}")
 
            raise TimeoutError(f"タスクが {max_wait} 秒以内に完了しませんでした")
 
    @staticmethod
    def extract_text(task: dict) -> str:
        """アーティファクトからテキストを取得するユーティリティ"""
        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
 
# --- 専門エージェント定義 ---
 
research_agent = LlmAgent(
    name="research_specialist",
    model="gemini-2.5-pro",
    description="Google 検索で最新情報を収集・整理する専門エージェント",
    instruction="""
    あなたは情報収集の専門家です。
    ユーザーの質問に対し、Google 検索で最新の正確な情報を収集し、
    出典 URL を明示したうえで 500 文字以内にまとめてください。
    情報が古い可能性がある場合はその旨を明記してください。
    """,
    tools=[google_search]
)
 
code_agent = LlmAgent(
    name="code_executor",
    model="gemini-2.5-pro",
    description="Python コードを生成・実行し、計算や数値処理を担当するエージェント",
    instruction="""
    あなたはデータサイエンスとプログラミングの専門家です。
    要求された計算や処理を Python で実装し、実際に実行して数値結果を返してください。
    コードはシンプルに保ち、各処理にコメントを付けてください。
    結果は数値・表・グラフの URL などで具体的に示してください。
    """,
    tools=[code_execution]
)
 
# --- オーケストレーター定義 ---
 
orchestrator = Agent(
    name="research_orchestrator",
    model="gemini-2.5-pro",
    description="調査・コード実行を組み合わせた高度な分析を行うオーケストレーター",
    instruction="""
    あなたはマルチエージェントシステムのオーケストレーターです。
    ユーザーのタスクを分析し、以下のサブエージェントを適切に活用してください：
    - research_specialist: 情報収集・最新動向の調査
    - code_executor: 数値計算・データ処理・コード生成
 
    複数のエージェントの結果を統合し、一貫性のある包括的な回答を返してください。
    各エージェントの担当箇所を明示すると、読者の理解が深まります。
    """,
    sub_agents=[research_agent, code_agent]
)
 
async def run_orchestrated_task(user_request: str) -> str:
    """ADK オーケストレーターを実行し、結果を返す"""
    runner = Runner(agent=orchestrator)
    # Runner の実行（ADK バージョンにより API が異なる場合があります）
    result = await runner.run_async(user_request)
    return str(result)

# pipeline.py
async def research_then_report_pipeline(topic: str) -> str:
    """調査 → 分析 → レポート生成の 3 段パイプライン"""
 
    search_client = A2AClient("https://search-agent.run.app")
    analysis_client = A2AClient("https://analysis-agent.run.app")
    report_client = A2AClient("https://report-agent.run.app")
 
    # Step 1: 情報収集（検索エージェント）
    print("📡 Step 1: 情報収集中...")
    raw_task = await search_client.send_task_and_wait(
        f"「{topic}」について最新動向を詳しく調べてください"
    )
    raw_data = search_client.extract_text(raw_task)
 
    # Step 2: データ分析（分析エージェント）
    print("🔍 Step 2: 分析中...")
    analysis_task = await analysis_client.send_task_and_wait(
        f"以下の情報を分析し、重要なポイントを 5 つ抽出してください:\n\n{raw_data}"
    )
    insights = analysis_client.extract_text(analysis_task)
 
    # Step 3: レポート生成（レポートエージェント）
    print("📝 Step 3: レポート生成中...")
    report_task = await report_client.send_task_and_wait(
        f"次の洞察をもとに、1,000 文字のエグゼクティブサマリーを作成してください:\n\n{insights}"
    )
 
    return report_client.extract_text(report_task)

# fanout.py
async def parallel_competitive_analysis(company: str) -> dict:
    """市場・技術・リスクを並列分析する"""
 
    # 3 エージェントの send_task を同時実行（直列比で約 3 倍速）
    market_task, tech_task, risk_task = await asyncio.gather(
        A2AClient("https://market-agent.run.app").send_task_and_wait(
            f"{company} の市場シェアと競合状況を分析してください"
        ),
        A2AClient("https://tech-agent.run.app").send_task_and_wait(
            f"{company} の技術スタックと技術的強みを評価してください"
        ),
        A2AClient("https://risk-agent.run.app").send_task_and_wait(
            f"{company} のリスク要因と課題を列挙してください"
        ),
        return_exceptions=True  # 一部失敗でも残りを継続
    )
 
    def safe_extract(client_url: str, task_or_exc) -> str:
        if isinstance(task_or_exc, Exception):
            return f"分析エラー: {str(task_or_exc)}"
        return A2AClient(client_url).extract_text(task_or_exc)
 
    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
 
router_model = genai.GenerativeModel("gemini-2.5-flash")  # 低コスト・高速なモデルで分類
 
AGENT_REGISTRY = {
    "legal": {
        "url": "https://legal-agent.run.app",
        "description": "契約書レビュー・法的文書の分析・コンプライアンスチェック"
    },
    "data": {
        "url": "https://data-agent.run.app",
        "description": "データ分析・統計計算・グラフ生成・予測モデル"
    },
    "content": {
        "url": "https://content-agent.run.app",
        "description": "記事執筆・コピーライティング・SEO 最適化"
    }
}
 
async def smart_route(user_request: str) -> str:
    """リクエストを分類し、最適なエージェントに転送する"""
    routing_prompt = f"""
ユーザーのリクエストに最適なエージェントを 1 つ選んでください。
 
エージェント一覧:
{json.dumps({k: v['description'] for k, v in AGENT_REGISTRY.items()}, ensure_ascii=False)}
 
リクエスト: {user_request}
 
次の JSON 形式のみで回答（説明不要）:
{{"agent": "agent_id", "reason": "選択理由（15文字以内）"}}
"""
 
    routing_res = router_model.generate_content(
        routing_prompt,
        generation_config={"response_mime_type": "application/json"}
    )
    routing = json.loads(routing_res.text)
    agent_id = routing["agent"]
 
    if agent_id not in AGENT_REGISTRY:
        # フォールバック: 最初のエージェントを使用
        agent_id = list(AGENT_REGISTRY.keys())[0]
 
    print(f"🎯 ルーティング → {agent_id}（理由: {routing['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):
    """Gemini のストリーミング出力を 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
 
        # working イベント配信
        yield f"data: {json.dumps({'type': 'status', 'id': task_id, 'status': {'state': 'working'}})}\n\n"
 
        # Gemini ストリーミング生成
        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, ensure_ascii=False)}\n\n"
                await asyncio.sleep(0)  # イベントループに制御を返す
 
        # completed イベント配信
        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
 
JWT_SECRET = os.environ["JWT_SECRET"]
security = HTTPBearer()
 
def create_agent_token(agent_id: str, ttl_sec: int = 3600) -> str:
    """A2A 通信用 JWT トークンを生成する"""
    return jwt.encode(
        {"agent_id": agent_id, "iat": int(time.time()), "exp": int(time.time()) + ttl_sec},
        JWT_SECRET,
        algorithm="HS256"
    )
 
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> dict:
    """Bearer トークンを検証し、エージェント情報を返す"""
    try:
        payload = jwt.decode(credentials.credentials, JWT_SECRET, algorithms=["HS256"])
        if payload.get("exp", 0) < time.time():
            raise HTTPException(status_code=401, detail="トークンの有効期限切れ")
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="無効なトークン")
 
# レート制限（メモリベース、本番では Redis を推奨）
from collections import defaultdict
request_log: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT_RPM = 60  # 1 分あたりの最大リクエスト数
 
async def rate_limit(agent: dict = Depends(verify_token)) -> dict:
    """エージェント ID ごとにレート制限を適用する"""
    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="レート制限超過。1 分後に再試行してください。")
    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 . .
 
# セキュリティ: 非 root ユーザーで実行
RUN useradd -m -u 1000 agentuser && chown -R agentuser:agentuser /app
USER agentuser
 
EXPOSE 8080
# 4 ワーカーで並列処理（本番推奨）
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"
 
# ビルド & プッシュ
gcloud builds submit \
  --tag "asia-northeast1-docker.pkg.dev/${PROJECT_ID}/agents/${SERVICE}:latest"
 
# デプロイ（シークレットは 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"
 
# デプロイ URL 確認
gcloud run services describe ${SERVICE} --region ${REGION} --format "value(status.url)"
# → https://gemini-a2a-agent-xxxxxxxxxx-an.a.run.app

# 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):
    """タスク処理時間をカスタムメトリクスとして記録する"""
    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": "技術記事を指定文字数に要約するエージェント",
    "url": "https://summarizer-agent.example.run.app",
    "version": "2.1.0",  # skills を変更したら必ず上げる
    "capabilities": {"streaming": True, "pushNotifications": False},
    "skills": [
        {
            "id": "summarize",
            "name": "summarize",
            "description": "本文を要約する",
            "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 を返し、変更時は確実に再取得させる
    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,
            # 5分は許容し、それ以上は握らせない
            "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} のカードが古すぎます: {version} < {required}. "
            "キャッシュを疑ってください"
        )

# 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)
 
 
# 本番では Firestore / Redis などの共有ストアに置く
# （Cloud Run は複数インスタンスに分散するため、プロセス内 dict では守れない）
_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]:
    """(結果, 再利用したか) を返す。"""
    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:
            # 先行リクエストが処理中。409 で呼び出し側に待たせる
            raise HTTPException(status_code=409, detail="task in progress")
 
    try:
        result = await work()
    except Exception:
        async with _LOCK:
            _STORE.pop(key, None)  # 失敗は残さず、再送を許す
        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 ヘッダが必要です")
 
    async def _work() -> dict:
        # ここで実際の Gemini 呼び出しを行う
        return await handle_task(payload)
 
    result, reused = await run_idempotent(idempotency_key, _work)
    return {"result": result, "reused": reused}