Gemini CLI × A2A プロトコル — リモートエージェント連携

単一のエージェントでは対応できない複雑なワークフローがあります。営業支援システムなら営業エージェント × 契約確認エージェント × レポート作成エージェントの連携が必要。研究支援なら文献検索エージェント × データ分析エージェント × 論文執筆エージェントの協働が不可欠です。

Gemini CLI v0.33.0（2026年3月）で導入された A2A（Agent-to-Agent）プロトコル は、HTTP認証とエージェントカード探索機能を追加し、複数のエージェントが安全に、スケーラブルに連携できる環境を実現しました。

以下では、A2A プロトコルの仕組みを押さえたうえで、実装から Cloud Run での本番デプロイまでを順に追っていきます。

A2A プロトコルの基本概念

A2A とは何か

A2A（Agent-to-Agent）は、エージェント同士が相互に通信するための標準プロトコルです。2023年にGoogleが Linux Foundation に寄贈したオープンスタンダードをベースとしており、2026年3月にGemini CLIで本格採用されましました。

A2Aの3つの特徴：

オープンスタンダード: Linux Foundation が管理。他のAIフレームワーク（LangChain、LlamaIndex等）との相互運用性を前提設計
HTTP ベース: REST API として実装可能。エージェントはサーバー型で動作可能
エージェントカード: メタデータベースの発見機構。エージェントの機能・入力形式・認証方式を定義

Gemini CLI での A2A 実装

Gemini CLI では、エージェント定義ファイル（.md形式）に YAML フロントマターでA2Aメタデータを埋め込みます。

例：営業支援エージェント定義

---
name: "Sales Support Agent"
description: "営業提案書の自動生成と客先リスク評価"
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...
(以下エージェント本文)

このファイルをプロジェクトの .gemini/agents/ に配置すると、Gemini CLI が自動的にA2Aエンドポイントを認識し、他のエージェントから呼び出し可能になります。

HTTP 認証とセキュリティ設計

認証方式の比較

API Key: 実装複雑度低 / セキュリティ中 / 本番利用性低 — 内部テスト向け
OAuth 2.0: 実装複雑度中 / セキュリティ高 / 本番利用性高 — 企業間連携向け
Service Account: 実装複雑度中 / セキュリティ高 / 本番利用性高 — Google Cloud内部向け
mTLS: 実装複雑度高 / セキュリティ最高 / 本番利用性高 — 金融・医療向け

API Key認証（シンプル実装）

---
name: "Internal Analytics Agent"
a2a:
  endpoint: "http://localhost:8000/agent"
  auth:
    type: "api_key"
    header: "X-Gemini-API-Key"
    secret: "${INTERNAL_AGENT_KEY}"  # 環境変数参照
---

環境変数設定：

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

OAuth 2.0認証（エンタープライズ向け）

---
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"
---

トークン取得フロー：

from google.auth.transport.requests import Request
from google.oauth2.service_account import Credentials
import requests
 
def get_oauth_token(client_id: str, client_secret: str, token_endpoint: str) -> str:
    """OAuth 2.0トークン取得"""
    payload = {
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "client_credentials",
        "scope": "agent:read agent:write"
    }
    response = requests.post(token_endpoint, json=payload)
    return response.json()["access_token"]
 
# エージェント呼び出し時に自動使用
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認証（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"
---

設定ファイル（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 認証（最高セキュリティ）

---
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}"
---

証明書生成（テスト環境）：

# CA証明書
openssl genrsa -out ca-key.pem 2048
openssl req -new -x509 -days 365 -key ca-key.pem -out ca-cert.pem
 
# クライアント証明書
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
 
# 検証
openssl verify -CAfile ca-cert.pem client-cert.pem

エージェント定義ファイルの完全実装

ステップ1：エージェント定義の作成

プロジェクト構造：

my-gemini-project/
├── .gemini/
│   ├── config.yaml          # Gemini CLI設定
│   └── agents/
│       ├── sales-support.md      # メインエージェント
│       ├── contract-reviewer.md   # サブエージェント
│       └── report-generator.md    # サブエージェント
└── agents/
    ├── server.py                  # A2Aサーバー実装
    └── requirements.txt

営業支援エージェント（sales-support.md）：

---
name: "Sales Support Agent"
slug: "sales-support"
version: "1.0.0"
description: "営業提案資料の自動生成とリスク評価を行う営業支援エージェント"
author: "Sales Team"
tags: ["sales", "proposal", "risk-assessment"]
 
# A2Aメタデータ
a2a:
  # エンドポイント設定
  endpoint: "https://sales-agent.example.com/a2a"
 
  # HTTP認証
  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:
    - name: "proposal_generation"
      description: "顧客に合わせた提案書を自動生成"
      input: { type: "object", properties: { client_name: { type: "string" }, budget: { type: "number" } } }
 
    - name: "risk_assessment"
      description: "契約リスク評価を実施"
      input: { type: "object", properties: { client_name: { type: "string" }, contract_amount: { type: "number" } } }
 
  # サブエージェント指定
  subagents:
    - slug: "contract-reviewer"
      role: "contract_validation"
    - slug: "report-generator"
      role: "report_creation"
 
  # キャッシング設定
  caching:
    enabled: true
    ttl_seconds: 3600
 
  # レート制限
  rate_limiting:
    requests_per_minute: 100
    burst_size: 10
 
# Gemini プロンプト
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 for complete pipeline
 
  You have access to the following subagents:
  - contract-reviewer: Validates contracts for legal compliance
  - report-generator: Creates comprehensive reports
 
  Always prioritize accuracy and compliance. For high-value deals (>$100K),
  escalate to senior management review.
---
 
I will help with sales proposals and risk assessment...

ステップ2：サブエージェントの定義

契約レビュー用エージェント（contract-reviewer.md）：

---
name: "Contract Reviewer Agent"
slug: "contract-reviewer"
version: "1.0.0"
description: "契約書の法的リスク評価と条項検証"
a2a:
  endpoint: "https://contract-service.example.com/a2a"
  auth:
    type: "service_account"
    service_account_file: "${SERVICE_ACCOUNT_JSON}"
  capabilities:
    - name: "review_contract"
      description: "契約書のリスク評価"
      input: { type: "object", properties: { contract_text: { type: "string" } } }
    - name: "flag_problematic_clauses"
      description: "問題のある条項を指摘"
      input: { type: "object", properties: { contract_text: { type: "string" } } }
---
 
You are a legal expert specializing in enterprise contracts...

ステップ3：Gemini CLI 設定

.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:
    host: "0.0.0.0"
    port: 8000
    ssl:
      enabled: true
      cert_file: "certs/server.crt"
      key_file: "certs/server.key"
 
  # エージェントカード探索
  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サーバーの実装

Python（FastAPI）での実装

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
 
app = FastAPI(title="Sales Agent Server")
logger = logging.getLogger(__name__)
 
# OAuth トークン検証
SECRET_KEY = "YOUR_SECRET_KEY"
 
class AgentRequest(BaseModel):
    """A2A リクエストモデル"""
    agent_id: str
    action: str
    params: Dict[str, Any]
    context: Optional[Dict[str, Any]] = None
 
class AgentResponse(BaseModel):
    """A2A レスポンスモデル"""
    status: str
    result: Optional[Dict[str, Any]]
    error: Optional[str] = None
 
async def verify_token(authorization: str = Header(...)) -> Dict:
    """OAuth トークン検証"""
    try:
        scheme, token = authorization.split(" ")
        if scheme.lower() != "bearer":
            raise HTTPException(status_code=401, detail="Invalid auth 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 エージェント実行エンドポイント
 
    他のエージェントから呼び出される
    """
    logger.info(f"Agent {agent_id} executing action: {request.action}")
 
    try:
        # エージェント ID の検証
        if agent_id != "sales-support":
            raise ValueError(f"Unknown agent: {agent_id}")
 
        # アクション別の処理
        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":
            # contract-reviewer エージェントを呼び出し
            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:
    """
    サブエージェント呼び出し（A2A プロトコル）
    """
    import aiohttp
 
    # サブエージェント URL 取得（エージェントレジストリから）
    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:
    """提案書生成ロジック"""
    return {
        "proposal_id": f"PROP-{client_name}-001",
        "title": f"{client_name} 向けカスタムソリューション提案",
        "budget": budget,
        "timeline": "30 days",
        "components": ["Feature A", "Support Package", "Training"]
    }
 
def get_service_token() -> str:
    """Service Account トークン取得"""
    # Google Cloud credentials から OAuth トークンを取得
    # 実装は省略
    return "service-token-xyz"
 
@app.get("/a2a/agents/{agent_id}/card")
async def get_agent_card(agent_id: str) -> Dict:
    """
    エージェントカード取得（discovery用）
 
    他のエージェントが機能を検出する際に使用
    """
    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"
        }
    }
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

環境設定

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

エージェント呼び出しの実装

Gemini CLI での直接呼び出し

# ローカルエージェント実行
gemini agent run sales-support --prompt "Create a proposal for TechCorp with $500K budget"
 
# A2A サブエージェント自動呼び出し
gemini agent run sales-support --prompt "Review contract for ABC Corp" \
  --subagent-enabled
 
# JSON 入力での呼び出し
gemini agent run sales-support << 'EOF'
{
  "action": "proposal_generation",
  "client_name": "Enterprise Inc",
  "budget": 250000
}
EOF

Python クライアント実装

from gemini_sdk import GeminiAgent
import json
 
class SalesOrchestrator:
    """営業ワークフロー オーケストレーション"""
 
    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:
        """営業機会を処理（A2A フロー）"""
 
        # Step 1: 提案書生成
        proposal = await self.agent.call(
            action="proposal_generation",
            params={
                "client_name": opportunity["client_name"],
                "budget": opportunity["budget"],
                "industry": opportunity["industry"]
            }
        )
 
        # Step 2: 契約レビュー（contract-reviewer サブエージェント呼び出し）
        review = await self.agent.call(
            action="risk_assessment",
            params={
                "client_name": opportunity["client_name"],
                "contract_amount": opportunity["budget"]
            }
        )
 
        # Step 3: 統合レポート生成
        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"
        }
 
# 使用例
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))

本番デプロイ（Cloud Run + Gemini Enterprise）

Cloud Run へのデプロイ

Dockerfile:

FROM python:3.11-slim
 
WORKDIR /app
 
# 依存関係インストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# コード配置
COPY agents/ ./agents/
COPY certs/ ./certs/
 
# ポート設定
EXPOSE 8000
 
# 起動
CMD ["python", "agents/server.py"]

デプロイスクリプト:

#!/bin/bash
 
PROJECT_ID="my-gemini-project"
SERVICE_NAME="sales-agent"
REGION="us-central1"
 
# イメージビルド
docker build -t gcr.io/${PROJECT_ID}/${SERVICE_NAME}:latest .
 
# Registry にプッシュ
docker push gcr.io/${PROJECT_ID}/${SERVICE_NAME}:latest
 
# 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
 
# URL 確認
gcloud run services describe ${SERVICE_NAME} --region ${REGION}

Gemini Enterprise での統合

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

トラブルシューティングと監視

一般的なエラーと解決方法

エラー 1：認証失敗

Error: A2A authentication failed: Invalid token

解決策：

# トークン有効期限確認
import jwt
 
def check_token_validity(token):
    try:
        decoded = jwt.decode(token, options={"verify_signature": False})
        expiry = decoded.get("exp")
        print(f"Token expires at: {expiry}")
    except jwt.DecodeError as e:
        print(f"Invalid token: {e}")

エラー 2：サブエージェント発見失敗

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

解決策：

# エージェントカード確認
curl -H "Authorization: Bearer $TOKEN" \
  https://agent-registry.example.com/agents/contract-reviewer
 
# または Gemini CLI で確認
gemini agent list --a2a-enabled

エラー 3：レート制限超過

Error: Rate limit exceeded: 100 requests/minute

解決策：リクエストをバッチ処理または非同期化：

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
async def batch_process(requests, max_concurrent=10):
    """レート制限内での並列処理"""
    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])

監視とロギング

import logging
from pythonjsonlogger import jsonlogger
 
# JSON ロギング設定
logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)
 
# メトリクス記録
def log_agent_execution(agent_id, action, duration_ms, success):
    logger.info("agent_execution", extra={
        "agent_id": agent_id,
        "action": action,
        "duration_ms": duration_ms,
        "success": success,
        "timestamp": datetime.utcnow().isoformat()
    })

ベストプラクティス

セキュリティ

認証方式の選択: 内部用は API Key、企業間連携は OAuth 2.0、Google Cloud内部は Service Account
トークンローテーション: 定期的（90日ごと）にトークンを更新
機密情報: .env ファイルまたは Secrets Manager で管理

パフォーマンス

キャッシング: 頻繁に呼ばれるサブエージェント結果をキャッシュ
タイムアウト設定: A2A リクエストには必ずタイムアウト設定（デフォルト30秒）
非同期処理: async/await で複数エージェント呼び出しを並列化

運用

ヘルスチェック: .../health エンドポイント実装
ログ集約: Cloud Logging または ELK で全エージェントログを一元管理
定期更新: A2A プロトコル仕様の最新版を月次確認

全体を振り返って

Gemini CLI の A2A プロトコルにより、エージェント連携が本格的に可能になりましました。

この記事で学んだこと：

A2Aプロトコルの仕様: オープンスタンダード、HTTP ベース、エージェントカード
HTTP認証実装: API Key から mTLS までの4パターン
サーバー実装: FastAPI での A2A エンドポイント構築
本番デプロイ: Cloud Run + Gemini Enterprise 統合

次のステップ：

Gemini CLI 入門ガイドで基礎固め
Gemini CLI 2026年3月アップデート — Plan Mode・サブエージェント・サンドボックスで最新機能確認
Google ADK vs LangChain — AIエージェントフレームワーク比較で他フレームワークとの使い分け検討

A2A プロトコルで複雑なエージェント連携を安全に実装し、企業システムの知能化を加速させましょう。

参考資料

Gemini CLI 公式ドキュメント
A2A プロトコル仕様（Linux Foundation）