G-gen の奥田です。当記事では、Claude Agent SDK で実装したエージェントを Cloud Run にデプロイし、Gemini Enterprise app に A2A エージェントとして直接登録して呼び出す手順を解説します。 当記事について Gemini Enterprise app へのエージェント登録 当記事で行うこと 前提条件 手順 1. サービスアカウントと IAM の設定 2. エージェントの実装（Claude Agent SDK） 3. A2A サーバーの実装（a2a-sdk） 4. Cloud Run へのデプロイ 5. Gemini Enterprise app への登録 動作確認 サービスへの疎通確認 Agent Card と A2A エンドポイントの確認 Gemini Enterprise app からの実行 注意点 Agent Card の url はサービスの URL と一致させる a2a-sdk は 0.3 系に固定する MIME タイプは text/plain にする 当記事について Gemini Enterprise app へのエージェント登録 Gemini Enterprise app（通称 Gemini Enterprise）には、自社で開発したエージェントを登録して、ユーザーに使用させることができます。登録方法の 1 つが、Agent2Agent（以下、A2A）プロトコルによる登録です。エージェント側が A2A の Agent Card を公開していれば、Google Cloud コンソールまたは REST API から Gemini Enterprise app に登録できます。 Gemini Enterprise app の概要は以下の記事を参照してください。 blog.g-gen.co.jp A2A による登録では、エージェントの実行基盤は自分で用意します。Gemini Enterprise app が持つのは Agent Card の登録情報だけで、実行時には Agent Card の url を呼び出します。当記事では、その実行基盤として Cloud Run を使用します。 A2A エージェントをいったん Agent Registry に登録し、そこから Gemini Enterprise app に取り込むこともできます。しかし当記事では Agent Registry を経由せず、Gemini Enterprise app に直接登録します。 参考 : A2A エージェントを登録して管理する 当記事で行うこと Google が提供する AI エージェント開発用フレームワークである Agent Development Kit（ADK）で開発したエージェントを、Gemini Enterprise app に登録できることはよく知られています。 しかし Gemini Enterprise app には、ADK 以外で実装したエージェントも登録可能です。 当記事では、Anthropic が提供する Claude Agent SDK で開発したエージェントを、Gemini Enterprise app に登録する方法を検証します。Claude Agent SDK は、Claude Code のエージェントループとツール実行の仕組みをライブラリとして使用できるようにしたものです。LLM の呼び出し先は Agent Platform（旧称 Vertex AI）上の Claude Haiku 4.5 です。 構成は以下のとおりです。 [エンドユーザー] ↓ [Gemini Enterprise app] ↓ A2A（直接登録） [Cloud Run サービス] Python 3.12 / FastAPI / claude-agent-sdk / a2a-sdk ├─ Claude Haiku 4.5（Agent Platform） ├─ BigQuery リモート MCP サーバー → BigQuery のテーブル └─ 外部 API（気象データ） エージェントは、BigQuery のテーブルを BigQuery リモート MCP サーバー経由で参照し、外部 API から取得したデータと突き合わせて回答します。外部 API の例として、Open-Meteo の過去気象データ API を使用します。非商用の使用では API キーが不要と案内されています。 参考 : Claude Code Python Agent SDK 参考 : BigQuery リモート MCP サーバーを使用する 参考 : Historical Weather API 前提条件 当記事の手順を実行するには、以下が必要です。 Gemini Enterprise app が作成済みであること Cloud Run とエージェントを動かす Google Cloud プロジェクト（当記事では my-project ）で、Claude Haiku 4.5 が Model Garden で有効化されていること gcloud CLI と python3 が使用できること。 Claude モデルは、Google Cloud プロジェクトごとに Model Garden で有効化する必要があります。有効化されていないプロジェクトから呼び出すと 404 エラーが返ります。 参考 : Claude モデルで予測をリクエストする 参考 : Claude Code on Google Cloud's Agent Platform - Troubleshooting 当記事で付与する IAM ロールは以下のとおりです。 付与先 ロール 用途 エージェントのサービスアカウント Agent Platform ユーザー（ roles/aiplatform.user ） Claude モデルの呼び出し エージェントのサービスアカウント MCP ツールユーザー（ roles/mcp.toolUser ）、BigQuery ジョブユーザー（ roles/bigquery.jobUser ） BigQuery リモート MCP サーバーの呼び出し エージェントのサービスアカウント BigQuery データ閲覧者（ roles/bigquery.dataViewer ）。テーブル単位で付与 対象テーブルの参照 エージェントのサービスアカウント Cloud Run 起動元（ roles/run.invoker ）。サービス単位で付与 動作確認での ID トークンによる呼び出し Discovery Engine サービスエージェント Cloud Run 起動元（ roles/run.invoker ） Gemini Enterprise app からの呼び出し 作業者のユーザーアカウント サービス アカウント トークン作成者（ roles/iam.serviceAccountTokenCreator ） 動作確認用の ID トークンの発行 手順 1. サービスアカウントと IAM の設定 以降のコマンドで使う値を環境変数に代入しておきます。BigQuery のテーブルは Cloud Run と別のプロジェクトに置くこともできるため、 BQ_PROJECT_ID を分けています。 export PROJECT_ID =my-project export BQ_PROJECT_ID =my-project export REGION =us-central1 export SERVICE_NAME =analysis-agent export SERVICE_ACCOUNT =analysis-agent-sa@ ${PROJECT_ID} .iam.gserviceaccount.com export BQ_DATASET =sales_data export BQ_TABLE =sales API を有効化し、エージェント用のサービスアカウントを作成してロールを付与します。BigQuery データ閲覧者はデータセットではなくテーブルに対して付与し、エージェントが参照できる範囲を対象テーブルだけに絞ります。 gcloud services enable aiplatform.googleapis.com bigquery.googleapis.com \ discoveryengine.googleapis.com run.googleapis.com \ artifactregistry.googleapis.com cloudbuild.googleapis.com \ --project =" ${PROJECT_ID} " gcloud iam service-accounts create analysis-agent-sa --project =" ${PROJECT_ID} " gcloud projects add-iam-policy-binding " ${PROJECT_ID} " \ --member =" serviceAccount: ${SERVICE_ACCOUNT} " \ --role =" roles/aiplatform.user " for role in roles/bigquery.jobUser roles/mcp.toolUser; do gcloud projects add-iam-policy-binding " ${BQ_PROJECT_ID} " \ --member =" serviceAccount: ${SERVICE_ACCOUNT} " \ --role =" ${role} " done bq add-iam-policy-binding \ --project_id =" ${BQ_PROJECT_ID} " \ --member =" serviceAccount: ${SERVICE_ACCOUNT} " \ --role =" roles/bigquery.dataViewer " \ " ${BQ_PROJECT_ID} : ${BQ_DATASET} . ${BQ_TABLE} " 2. エージェントの実装（Claude Agent SDK） まず依存パッケージです。a2a-sdk は 0.3 系に固定します。理由は「注意点」で説明します。 claude-agent-sdk>=0.2.140 fastapi>=0.115.0 uvicorn[standard]>=0.32.0 google-auth>=2.35.0 httpx>=0.28.0 a2a-sdk[http-server]>=0.3.26,<1.0 システムプロンプトでは対象テーブルを固定し、集計は SQL 側で行うように指示します。 import os class Config : BQ_PROJECT_ID = os.environ[ "BQ_PROJECT_ID" ] BQ_DATASET = os.environ.get( "BQ_DATASET" , "sales_data" ) BQ_TABLE = os.environ.get( "BQ_TABLE" , "sales" ) BQ_MCP_URL = "https://bigquery.googleapis.com/mcp" MAX_TURNS = int (os.environ.get( "MAX_TURNS" , "20" )) AGENT_CWD = "/tmp/agent" AGENT_CONFIG_DIR = "/tmp/agent-config" AGENT_BASE_URL = os.environ.get( "AGENT_BASE_URL" , "http://localhost:8080" ).rstrip( "/" ) EXTERNAL_API_URL = "https://archive-api.open-meteo.com/v1/archive" @ classmethod def table_fqn (cls) -> str : return f "{cls.BQ_PROJECT_ID}.{cls.BQ_DATASET}.{cls.BQ_TABLE}" @ classmethod def system_prompt (cls) -> str : return f """あなたは BigQuery 上のデータを分析するアナリストです。 ## 対象テーブル `{cls.table_fqn()}` 1 行が 1 件の購入です。主な列は prefecture（都道府県）、sales_date（購入日）、sales_amount（金額）です。 ## ルール - 使えるツールは BigQuery MCP（mcp__bigquery__*）と外部データ（mcp__external__*）だけです - スキーマが不明なときは get_table_info で確認してから SQL を書いてください - 集計は SQL 側で行い、全行を取得しないでください - 回答には根拠にした SQL を添えてください - 外部データと社内データを突き合わせるときは、どちらがどの数値かを明示してください """ 外部 API を呼ぶツールは、Claude Agent SDK の tool デコレータと create_sdk_mcp_server で定義します。このサーバーはアプリケーションのプロセス内で動くため、別プロセスの起動は不要です。ツールの戻り値は content の配列で、失敗時は is_error を True にすると Claude が失敗として扱います。 import json import httpx from claude_agent_sdk import ToolAnnotations, create_sdk_mcp_server, tool from config import Config @ tool ( "get_daily_weather" , "指定した緯度経度・期間の日別平均気温（摂氏）を外部 API から取得する。日付は YYYY-MM-DD 形式。" , { "latitude" : float , "longitude" : float , "start_date" : str , "end_date" : str }, annotations=ToolAnnotations(readOnlyHint= True ), ) async def get_daily_weather (args: dict ) -> dict : params = { "latitude" : args[ "latitude" ], "longitude" : args[ "longitude" ], "start_date" : args[ "start_date" ], "end_date" : args[ "end_date" ], "daily" : "temperature_2m_mean" , "timezone" : "Asia/Tokyo" , } async with httpx.AsyncClient(timeout= 20 ) as client: response = await client.get(Config.EXTERNAL_API_URL, params=params) if response.status_code != 200 : return { "content" : [{ "type" : "text" , "text" : f "外部 API エラー: HTTP {response.status_code}" }], "is_error" : True , } daily = response.json().get( "daily" , {}) return { "content" : [{ "type" : "text" , "text" : json.dumps(daily, ensure_ascii= False )}]} external_server = create_sdk_mcp_server( name= "external" , version= "1.0.0" , tools=[get_daily_weather] ) 参考 : Give Claude custom tools 次に、エージェント本体のソースコードを解説します。 ClaudeAgentOptions の mcp_servers に、BigQuery リモート MCP サーバー（HTTP 型）と上記のプロセス内サーバーの 2 つを渡します。BigQuery リモート MCP サーバーは OAuth 2.0 のアクセストークンで認証するため、Application Default Credentials から取得したトークンを Authorization ヘッダーに載せます。トークンは失効するので、リクエストのたびに取得し直します。 allowed_tools には mcp__<サーバー名>__* の形で 2 つのサーバーのツールを指定します。 setting_sources=[] は、コンテナ内の設定ファイルを読み込まないための指定です。 import os from collections.abc import AsyncIterator import google.auth import google.auth.transport.requests from claude_agent_sdk import AssistantMessage, ClaudeAgentOptions, ResultMessage, query import external_tools from config import Config def access_token () -> str : credentials, _ = google.auth.default( scopes=[ "https://www.googleapis.com/auth/cloud-platform" ] ) credentials.refresh(google.auth.transport.requests.Request()) return credentials.token def build_options () -> ClaudeAgentOptions: os.makedirs(Config.AGENT_CWD, exist_ok= True ) os.makedirs(Config.AGENT_CONFIG_DIR, exist_ok= True ) return ClaudeAgentOptions( mcp_servers={ "bigquery" : { "type" : "http" , "url" : Config.BQ_MCP_URL, "headers" : { "Authorization" : f "Bearer {access_token()}" }, }, "external" : external_tools.external_server, }, allowed_tools=[ "mcp__bigquery__*" , "mcp__external__*" ], setting_sources=[], env={ "CLAUDE_CONFIG_DIR" : Config.AGENT_CONFIG_DIR}, cwd=Config.AGENT_CWD, max_turns=Config.MAX_TURNS, system_prompt=Config.system_prompt(), ) async def stream (prompt: str ) -> AsyncIterator[ dict ]: async for message in query(prompt=prompt, options=build_options()): if isinstance (message, AssistantMessage): for block in message.content: name = getattr (block, "name" , None ) if name: yield { "kind" : "tool" , "name" : name} elif isinstance (message, ResultMessage): yield { "kind" : "result" , "result" : message.result, "is_error" : message.is_error, } Claude の呼び出し先を Agent Platform にする設定は、コードではなく環境変数で行います。 CLAUDE_CODE_USE_VERTEX=1 、 CLOUD_ML_REGION 、 ANTHROPIC_VERTEX_PROJECT_ID 、 ANTHROPIC_MODEL の 4 つで、手順 4 で Cloud Run の環境変数として設定します。 参考 : Connect to external tools with MCP 参考 : Claude Code on Google Cloud's Agent Platform 3. A2A サーバーの実装（a2a-sdk） A2A のレイヤは a2a-sdk で実装します。Agent Card は、Gemini Enterprise app の公式ドキュメントのサンプルと同じフィールド構成にします。 url にはこのサービス自身の URL を入れます。 AgentExecutor を継承したクラスで、A2A のリクエストを手順 2 の stream() に橋渡しします。ツール呼び出しのたびに working 状態の更新を送り、回答を artifact として追加して completed にします。 Task の状態は completed か failed のどちらかに 1 回だけ遷移できます。結果を受け取ったらループを抜け、 except では未遷移のときだけ failed を呼びます。Claude 側がエラーを返したときは is_error で判定し、 failed として返します。 import logging from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.apps import A2AFastAPIApplication from a2a.server.events import EventQueue from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.tasks import InMemoryTaskStore, TaskUpdater from a2a.types import AgentCard, Part, TaskState, TextPart from a2a.utils import AGENT_CARD_WELL_KNOWN_PATH from fastapi import FastAPI import agent from config import Config log = logging.getLogger(__name__) def _text (value: str ) -> list [Part]: return [Part(root=TextPart(text=value))] def build_card () -> AgentCard: return AgentCard.model_validate({ "protocolVersion" : "0.3" , "name" : "業務データ分析エージェント" , "description" : "BigQuery 上の業務データと外部データを自然言語で分析するエージェント" , "url" : Config.AGENT_BASE_URL, "version" : "0.1.0" , "capabilities" : { "streaming" : True }, "defaultInputModes" : [ "text/plain" ], "defaultOutputModes" : [ "text/plain" ], "skills" : [ { "id" : "bigquery-analysis" , "name" : "BigQuery 分析" , "description" : "対象テーブルに対する集計・傾向分析を自然言語で受け付ける" , "tags" : [ "bigquery" , "analytics" ], }, { "id" : "external-join" , "name" : "外部データとの突き合わせ" , "description" : "社内データの集計結果を外部 API のデータと掛け合わせる" , "tags" : [ "external" , "join" ], }, ], }) class AnalysisAgentExecutor (AgentExecutor): async def execute (self, context: RequestContext, event_queue: EventQueue) -> None : updater = TaskUpdater(event_queue, context.task_id, context.context_id) if context.current_task is None : await updater.submit() await updater.start_work() done = False try : async for event in agent.stream(context.get_user_input()): if event[ "kind" ] == "tool" : await updater.update_status( TaskState.working, updater.new_agent_message(_text(f "実行中: {event['name']}" )), ) elif event[ "kind" ] == "result" : done = True if event[ "is_error" ]: message = event[ "result" ] or "エージェントがエラーを返しました" await updater.failed(updater.new_agent_message(_text(message))) else : await updater.add_artifact( _text(event[ "result" ] or "" ), name= "answer" ) await updater.complete() break except Exception as e: log.exception( "エージェントの実行に失敗" ) if not done: await updater.failed( updater.new_agent_message(_text(f "実行に失敗しました: {e}" )) ) async def cancel (self, context: RequestContext, event_queue: EventQueue) -> None : updater = TaskUpdater(event_queue, context.task_id, context.context_id) await updater.cancel(updater.new_agent_message(_text( "キャンセルされました" ))) def mount (app: FastAPI) -> AgentCard: card = build_card() handler = DefaultRequestHandler( agent_executor=AnalysisAgentExecutor(), task_store=InMemoryTaskStore(), ) A2AFastAPIApplication(agent_card=card, http_handler=handler).add_routes_to_app( app, agent_card_url=AGENT_CARD_WELL_KNOWN_PATH, rpc_url= "/" ) return card FastAPI のエントリーポイントでは、疎通確認用の /health を定義したうえで、A2A のルートを追加します。Agent Card は A2A の仕様どおり /.well-known/agent-card.json で配信され、JSON-RPC のエンドポイントは / です。 import logging from fastapi import FastAPI import a2a_app logging.basicConfig(level=logging.INFO, format = "%(message)s" ) app = FastAPI(title= "analysis-agent" ) @ app.get ( "/health" ) async def health () -> dict : return { "status" : "ok" } AGENT_CARD = a2a_app.mount(app) 参考 : A2A Protocol Specification (v0.3.0) 参考 : a2a-sdk - PyPI 4. Cloud Run へのデプロイ Dockerfile です。Claude Agent SDK は Claude Code の CLI をサブプロセスとして起動し、CLI がセッション情報を HOME 配下に書き込みます。Cloud Run のコンテナでは書き込み可能な /tmp を HOME にします。 FROM python:3.12-slim ENV HOME=/tmp \ PYTHONUNBUFFERED=1 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD exec uvicorn main:app --host 0 . 0 . 0 . 0 --port ${PORT :- 8080 } 環境変数は YAML ファイルで渡します。 ANTHROPIC_MODEL の値に @ が含まれるなど、区切り文字と衝突しやすい値があるため、 --set-env-vars ではなく --env-vars-file を使用します。 AGENT_BASE_URL は、この時点ではまだサービスの URL がわからないので仮の値にしておきます。 CLAUDE_CODE_USE_VERTEX : '1' CLOUD_ML_REGION : 'global' ANTHROPIC_VERTEX_PROJECT_ID : 'my-project' ANTHROPIC_MODEL : 'claude-haiku-4-5@20251001' BQ_PROJECT_ID : 'my-project' BQ_DATASET : 'sales_data' BQ_TABLE : 'sales' MAX_TURNS : '20' AGENT_BASE_URL : 'http://localhost:8080' 当記事では、デプロイを 2 回実行します。1 回目でサービスの URL を確定させ、その URL を AGENT_BASE_URL に書き戻してから 2 回目をデプロイします。Agent Card の url が実際のサービスの URL と一致していないと、Gemini Enterprise app からの呼び出しが認証で失敗するためです。詳細は「注意点」を参照してください。 まずは 1 回目のデプロイです。Cloud Run へソースコードを直接デプロイします。Gemini Enterprise app 以外の任意の主体から呼びだされることがないよう、認証を必須（ --no-allow-unauthenticated ）にします。デプロイ後、確定したサービスの URL を SERVICE_URL に取得します。 gcloud run deploy " ${SERVICE_NAME} " \ --source = . \ --project =" ${PROJECT_ID} " \ --region =" ${REGION} " \ --service-account =" ${SERVICE_ACCOUNT} " \ --no-allow-unauthenticated \ --memory = 2Gi \ --cpu = 2 \ --concurrency = 1 \ --timeout = 3600s \ --env-vars-file = env.yaml SERVICE_URL = $( gcloud run services describe " ${SERVICE_NAME} " \ --project =" ${PROJECT_ID} " --region =" ${REGION} " --format =' value(status.url) ' ) echo " ${SERVICE_URL} " 次に、2 回目のデプロイです。取得した SERVICE_URL を env.yaml の AGENT_BASE_URL に書き戻し、1 回目とまったく同じコマンドでもう一度デプロイします。これで Agent Card の url が、実際のサービスの URL と一致します。 sed -i " s|^AGENT_BASE_URL:.*|AGENT_BASE_URL: ' ${SERVICE_URL} '| " env.yaml gcloud run deploy " ${SERVICE_NAME} " \ --source = . \ --project =" ${PROJECT_ID} " \ --region =" ${REGION} " \ --service-account =" ${SERVICE_ACCOUNT} " \ --no-allow-unauthenticated \ --memory = 2Gi \ --cpu = 2 \ --concurrency = 1 \ --timeout = 3600s \ --env-vars-file = env.yaml 参考 : gcloud run deploy 参考 : gcloud topic gcloudignore 5. Gemini Enterprise app への登録 Gemini Enterprise app は、Discovery Engine のサービスエージェント（ service-プロジェクト番号@gcp-sa-discoveryengine.iam.gserviceaccount.com ）としてエージェントを呼び出します。認証必須の Cloud Run サービスを呼べるように、このサービスエージェントに Cloud Run 起動元を付与します。 PROJECT_NUMBER = $( gcloud projects describe " ${PROJECT_ID} " --format =' value(projectNumber) ' ) gcloud run services add-iam-policy-binding " ${SERVICE_NAME} " \ --project =" ${PROJECT_ID} " --region =" ${REGION} " \ --member =" serviceAccount:service- ${PROJECT_NUMBER} @gcp-sa-discoveryengine.iam.gserviceaccount.com " \ --role =" roles/run.invoker " gcloud run services add-iam-policy-binding " ${SERVICE_NAME} " \ --project =" ${PROJECT_ID} " --region =" ${REGION} " \ --member =" serviceAccount: ${SERVICE_ACCOUNT} " \ --role =" roles/run.invoker " 2 つ目は動作確認のためのものです。このあと、エージェントのサービスアカウントの権限を借用して ID トークンを発行し、そのトークンでサービスを呼び出します。サービスを実行するサービスアカウントであることと、そのサービスを呼び出せることは別なので、このアカウントにも Cloud Run 起動元を付与します。付与されていないと、呼び出しは 403 と insufficient_scope を返します。 登録には、デプロイ済みのサービスから取得した Agent Card の JSON をそのまま使用します。Cloud Run の認証を通すために、サービスアカウントの権限を借用して、オーディエンスをサービスの URL にした ID トークンを発行します。そのために、作業者のアカウントにサービス アカウント トークン作成者を付与しておきます。 gcloud iam service-accounts add-iam-policy-binding " ${SERVICE_ACCOUNT} " \ --project =" ${PROJECT_ID} " \ --member =" user: $( gcloud config get-value account ) " \ --role =" roles/iam.serviceAccountTokenCreator " TOKEN = $( gcloud auth print-identity-token \ --impersonate-service-account =" ${SERVICE_ACCOUNT} " --audiences =" ${SERVICE_URL} " ) CARD = $( curl -sS -H " Authorization: Bearer ${TOKEN} " " ${SERVICE_URL} /.well-known/agent-card.json " ) Gemini Enterprise app の登録 API に POST します。 a2aAgentDefinition.jsonAgentCard には、Agent Card の JSON を文字列として埋め込みます。当記事の Gemini Enterprise app はロケーションが global なので、エンドポイントのホスト名にロケーションの接頭辞は付きません。 export APP_ID =my-gemini-enterprise-app ENDPOINT = " https://discoveryengine.googleapis.com/v1alpha/projects/ ${PROJECT_ID} /locations/global/collections/default_collection/engines/ ${APP_ID} /assistants/default_assistant/agents " BODY = $( printf ' %s ' " ${CARD} " | python3 -c ' import json, sys print(json.dumps({ "displayName": "業務データ分析エージェント", "description": "BigQuery の業務データと外部データを分析するエージェント", "a2aAgentDefinition": {"jsonAgentCard": sys.stdin.read()}, }, ensure_ascii=False)) ' ) curl -sS -X POST \ -H " Authorization: Bearer $( gcloud auth print-access-token ) " \ -H " X-Goog-User-Project: ${PROJECT_ID} " \ -H " Content-Type: application/json " \ " ${ENDPOINT} " -d " ${BODY} " 登録されたエージェントは、同じエンドポイントへの GET で一覧できます。Google Cloud コンソールの Gemini Enterprise app の管理画面でも確認できます。 参考 : A2A エージェントを登録して管理する 参考 : サービス間認証 動作確認 サービスへの疎通確認 まず、Cloud Run のサービスにリクエストが届いているかを確認します。認証必須のサービスなので、手順 5 で発行した TOKEN を付けて /health を呼びます。 curl -sS -H " Authorization: Bearer ${TOKEN} " " ${SERVICE_URL} /health " {"status":"ok"} が返れば、認証を通ってコンテナまでリクエストが届き、FastAPI が応答しています。ここから先で失敗した場合は、原因をエージェント本体か A2A の実装に絞り込めます。 403 が返る場合は IAM の設定を、応答がない場合はコンテナの起動を疑います。 参考 : サービスにコンテナのヘルスチェックを構成する Agent Card と A2A エンドポイントの確認 Gemini Enterprise app を通さずに、A2A のエンドポイントを直接呼んで確認します。手順 5 で発行した ID トークンをそのまま使い、JSON-RPC の message/send を送ります。 curl -sS -X POST " ${SERVICE_URL} / " \ -H " Authorization: Bearer ${TOKEN} " \ -H " Content-Type: application/json " \ -d ' { "jsonrpc": "2.0", "id": "1", "method": "message/send", "params": { "message": { "role": "user", "parts": [{"kind": "text", "text": "このテーブルは全部で何行ありますか"}], "messageId": "msg-1" } } } ' レスポンスの status.state が completed になり、 artifacts に回答のテキストが入っていれば、A2A サーバーとエージェント本体は正常です。 Gemini Enterprise app からの実行 Gemini Enterprise app の画面で、登録したエージェントを選んで日本語で質問します。当記事では以下の 2 つを試しました。 質問 呼ばれたツール（ ToolSearch を除く） このテーブルは全部で何行ありますか BigQuery MCP のツールのみ 2023年の東京都の月別売上合計を出して、同じ期間の東京の月平均気温と突き合わせて get_table_info → execute_sql_readonly → get_daily_weather → execute_sql_readonly 2 つ目の質問では、1 回の応答の中で BigQuery のツールと外部 API のツールが両方呼ばれ、月別の売上と気温を対応づけた表と分析コメントが返りました。 ツール呼び出しの先頭には ToolSearch が入ります。Claude Agent SDK のツール検索は既定で有効と記載されており、Agent Platform 上の Claude Haiku 4.5 以降のモデルが対象です。ツールの数が少ない構成では先にすべて読み込むほうが速いと記載されているため、環境変数 ENABLE_TOOL_SEARCH に false を指定すると、この 1 往復を省けます。 参考 : Scale to many tools with tool search 注意点 Agent Card の url はサービスの URL と一致させる Cloud Run の ID トークンは、オーディエンスを受信側サービスの URL にする必要があります。Gemini Enterprise app は Agent Card の url を宛先として呼び出すため、 url が実際のサービスの URL と違うと認証に失敗します。 Cloud Run は、すべてのサービスにハッシュを含む非決定論的 URL を割り当てます。サービス名の長さが許せば、決定論的 URL も追加で割り当てられます。1 つのサービスに 2 つの URL があり、 gcloud run services describe の表示では決定論的 URL が優先されると記載されています。一方 --format='value(status.url)' は status.url の値をそのまま返すため、両者は一致しないことがあります。URL は手で組み立てず、コマンドでの出力をそのまま使います。 gcloud run services describe SERVICE --format ' value(status.url) ' 手順 4 でデプロイを 2 回実行しているのは、1 回目で確定したサービスの URL を env.yaml の AGENT_BASE_URL に書き戻し、Agent Card の url に反映させるためです。 参考 : サービス間認証 参考 : HTTPS リクエストで呼び出す 参考 : A2A エージェントを登録して管理する a2a-sdk は 0.3 系に固定する 2026年9月現在、Gemini Enterprise app は「A2A v0.3 のストリーミング機構をサポートしている」と案内されています。また a2a-sdk は 1.0 で A2AFastAPIApplication などのラッパークラスが削除され、サーバーの組み立て方が変わりました。 当記事のコードは 0.3 系を前提にしているため、バージョンの上限も含めて固定します。 参考 : a2a-python v1.0 Migration Guide MIME タイプは text/plain にする Agent Card の defaultInputModes と defaultOutputModes は、公式ドキュメントのサンプルと同じ text/plain にします。検証中に text と書いた Agent Card は、Gemini Enterprise app への登録で拒否されました。 奥田 梨紗 (記事一覧) クラウドソリューション部データインテリジェンス課 Google Cloudの可能性に惹かれ、2024年4月G-genにジョイン。 Google Cloud Partner Top Engineer 2025&2026 Follow @risa_hochiminh

目次 はじめに DevinConとは 自己紹介 WINTICKETとDevin活用 Agent設計技 ...

G-gen の河野です。当記事では、Google が開発した最適化問題を解くためのオープンソースライブラリ「 OR-Tools 」を使用して、製品の生産計画を作成します。 OR-Tools とは 仕様 解きたい最適化問題の定義 ソルバーの決定 最適化問題を解く API 版との比較 検証 検証内容 データ処理フロー データ準備 OR-Tools のインストール Python コード 検証結果 出力結果 処理プロセス OR-Tools とは OR-Tools は、Google が開発した、 最適化問題 を解くためのオープンソースライブラリです。 最適化問題とは、「限られたルールの中で、最適な答えを見つける問題」のことで、複数の配達先を回る最も効率的なルートの決定や、工場での無駄のない生産計画の作成などがあります。OR-Tools を使うことで、本来であれば組み合わせが多すぎて手作業では見つけられない最適解を、数学的に導き出すことができます。 参考 : OR-Tools について OR-Tools には、以下のような利用事例が考えられます。 物流・配送ルート最適化 車両の積載容量、配送指定時間などの制約を考慮し、最短ルートや最少車両数での配送計画を作成できます。 シフトスケジューリング 必要な人員数、従業員の希望休、スキル要件などの制約を考慮し、公平で効率的な勤務表を作成できます。 タスク割り当て 各人の処理能力、タスクの優先度などの制約を考慮し、作業時間の最小化やリソースの最適配置を実現できます。 生産計画 機械の稼働能力、製品ごとの所要時間、需要などの制約を考慮した最適な生産計画を作成できます。 仕様 解きたい最適化問題の定義 OR-Tools で最適化問題を解く際は、まず、どのような問題を解くのかを定義します。最適化問題は以下の3つの要素から定義されます。 要素 説明 例 決定変数 求めたい答え 生産数 制約 答えを求める上で考慮すべきルール 機械の処理能力 目的関数 何を最大/最小化したいかという目標 利益の最大化 上記の表の例は、「限られた機械の処理能力（制約）の中で、利益を最大化（目的関数）するための最適な生産数（決定変数）を求める」という問題を定義しています。 ソルバーの決定 OR-Tools は、 ソルバー と呼ばれる最適化アルゴリズムを用いて、定義した最適化問題を解きます。ソルバーは、「制約」を全て満たしながら「目的関数」が最適となる「決定変数」の値を探索します。ソルバーにはいくつかの種類があり、それぞれ得意・不得意が存在します。OR-Tools に標準搭載されているソルバーは以下の3種類です。 ソルバー 得意な問題 主な用途例 CP-SAT 「何個・どれを・いつ」という整数で答えが出る問題 生産計画、シフトスケジューリング、配送ルート最適化 GLOP 比率や量の配分など、小数で答えが出る問題 リソース配分、原材料の配合比率最適化 PDLP GLOP と同じく小数で答えが出るが、変数・制約が膨大な問題 数百万規模の変数・制約を持つ大規模な最適化 参考 : OR-Tools とその解法の引用方法 最適化問題を解く 以下の最適化問題を例に、ソルバーがどのように問題を解くのかを説明します。 要素 内容 決定変数 製品A、Bの生産数 制約 稼働可能時間は 100 分、製品Aの加工時間は 20 分/個、製品Bの加工時間は 30 分/個 目的関数 利益（製品Aは 1 個 500 円、製品Bは 1 個 800 円）の最大化 以下の3ステップで、ソルバーは最適化問題を解きます。 ① 制約による探索範囲の絞り込み 稼働時間上限（100 分）の制約から、各決定変数の上限を算出する 製品A（加工時間 20 分/個）：100 分 ÷ 20 分 = 最大 5 個 製品B（加工時間 30 分/個）：100 分 ÷ 30 分 = 最大 3 個 ② 決定変数への値の割り当て 絞り込んだ範囲の中で決定変数に値を1つずつ割り当て、目的関数の値を記録する 「製品Aを 0 個」に設定すると、稼働可能時間は残り 100 分であるため「製品Bは最大 3 個まで加工可能」となり、「製品A=0 / 製品B=3 / 利益 2,400 円」が記録される ③ 目的関数の値をもとにした最適解の更新 ②の処理を反復し、記録された目的関数の値を上回る結果が算出されるたびに、最適解を更新する 製品A 製品B 総利益 結果 0 個 3 個 2,400 円 最適解を更新 1 個 2 個 2,100 円 更新なし（2,400 円を下回るため） 2 個 2 個 2,600 円 最適解を更新 製品A = 3〜5 個のパターンにおいて、残りの稼働可能時間を最大限使っても暫定の最適解を上回らないため、 製品A=2・製品B=2（総利益 2,600 円） が最適解として確定する API 版との比較 Google が提供する最適化問題を解くためのツールとしては、OR-Tools の他に Operations Research API があります。これらのツールの主な違いは以下の通りです。 観点 OR-Tools Operations Research API 実行方法 ライブラリを用いた関数呼び出し（Python、C++、Java、C#） REST API、gRPC 計算処理担当 実行サーバー上 Google のクラウドインフラ上 カスタマイズ性 高い（制約・目的関数を自由設計） 低い（用意された設定の範囲内） 使用制限 なし リクエスト数・データサイズ・実行時間に上限あり 成熟度 安定版 アルファ / ベータ段階（2026年7月現在） 費用 無料 無料（安定版リリースで変更の可能性あり） 参考 : Operations Research API 検証 検証内容 Python の OR-Tools ライブラリを使い、製品の生産計画を作成します。今回は、CP-SAT ソルバーで以下の最適化問題を解きます。 要素 内容 決定変数 生産数 制約 各製品の生産数が需要数を超えないこと、各機械の稼働可能時間上限を超えないこと 目的関数 総利益の最大化（利益単価 × 生産数の合計） 製品マスタなどの源泉データの格納、および処理結果の出力先には BigQuery を採用します。 データ処理フロー 本検証では、BigQuery からデータを読み込み、OR-Tools で最適化計算を行った後、計算結果を BigQuery に書き込みます。 データ準備 BigQuery に以下のレコード内容でテーブルを作成します。 products（製品マスタ） 製品ID 利益単価（円/個） P001 1,000 P002 4,000 process_times（サイクルタイムマスタ） 製品ID 機械ID 加工時間（分/個） P001 M01 30 P002 M01 60 demand_forecast（需要予測） 製品ID 需要日 需要数 P001 2025-07-01 8 P002 2025-07-01 5 machine_calendar（設備稼働カレンダー） 機械ID 日付 稼働可能時間（分） M01 2025-07-01 360 OR-Tools のインストール 以下のコマンドでライブラリをインストールすることで、OR-Tools が使用できます。 pip install ortools Python コード OR-Tools を実行するためのコードを作成します。 from ortools.sat.python import cp_model import pandas as pd from google.cloud import bigquery # --- 設定 --- PROJECT_ID = "your-project-id" DATASET_ID = "your_dataset" plan_date = "2025-07-01" # --- BigQuery からデータ読み込み --- client = bigquery.Client(project=PROJECT_ID) products_df = client.query(f "SELECT * FROM `{PROJECT_ID}.{DATASET_ID}.products`" ).to_dataframe() process_df = client.query(f "SELECT * FROM `{PROJECT_ID}.{DATASET_ID}.process_times`" ).to_dataframe() demand_df = client.query(f "SELECT * FROM `{PROJECT_ID}.{DATASET_ID}.demand_forecast` WHERE `需要日` = '{plan_date}'" ).to_dataframe() demand_df[ "需要日" ] = pd.to_datetime(demand_df[ "需要日" ]) calendar_df = client.query(f "SELECT * FROM `{PROJECT_ID}.{DATASET_ID}.machine_calendar` WHERE `日付` = '{plan_date}'" ).to_dataframe() calendar_df[ "日付" ] = pd.to_datetime(calendar_df[ "日付" ]) # --- 辞書化 --- products = products_df[ "製品ID" ].tolist() machines = calendar_df[ "機械ID" ].unique().tolist() dates = sorted (demand_df[ "需要日" ].unique()) demand = {(r.製品ID, r.需要日): r.需要数 for r in demand_df.itertuples()} capacity = {(r.機械ID, r.日付): r.稼働可能時間_分 for r in calendar_df.itertuples()} process_time = {(r.製品ID, r.機械ID): r.加工時間_分 for r in process_df.itertuples()} profit = dict ( zip (products_df[ "製品ID" ], products_df[ "利益単価" ])) # --- ソルバーの決定 --- # CP-SAT を利用 model = cp_model.CpModel() # --- 最適化問題の定義 --- # 決定変数: 製品p・機械m・日付d の生産数 assign = { (p, m, d): model.NewIntVar( 0 , demand[(p, d)], f "x_{p}_{m}_{d}" ) for d in dates for p in products for m in machines if (p, m) in process_time and (p, d) in demand } # 制約1: 各製品の合計生産量 <= 需要数 for d in dates: for p in products: vars_ = [assign[(p, m, d)] for m in machines if (p, m, d) in assign] if vars_: model.Add( sum (vars_) <= demand.get((p, d), 0 )) # 制約2: 各機械の稼働可能時間を超えない for d in dates: for m in machines: vars_ = [assign[(p, m, d)] * int (process_time[(p, m)] * 10 ) for p in products if (p, m, d) in assign] if vars_: model.Add( sum (vars_) <= capacity.get((m, d), 0 ) * 10 ) # 目的関数: 利益単価 × 生産数 の合計を最大化 model.Maximize( sum (var * profit.get(p, 0 ) for (p, m, d), var in assign.items())) # --- 実行 --- solver = cp_model.CpSolver() status = solver.Solve(model) # --- BigQuery へ出力 --- if status in (cp_model.OPTIMAL, cp_model.FEASIBLE): results = [ { "生産日" : str (d.date()), "製品ID" : p, "機械ID" : m, "計画生産数" : solver.Value(var), "利益合計" : solver.Value(var) * profit.get(p, 0 ), } for (p, m, d), var in assign.items() if solver.Value(var) > 0 ] result_df = pd.DataFrame(results) job = client.load_table_from_dataframe( result_df, f "{PROJECT_ID}.{DATASET_ID}.production_plan" , job_config=bigquery.LoadJobConfig(write_disposition= "WRITE_TRUNCATE" ), ) job.result() 検証結果 出力結果 作成したコードを実行すると、BigQuery で以下の結果が出力されます。 production_plan（生産計画） 生産日 製品ID 機械ID 計画生産数 利益合計 2025-07-01 P001 M01 2 2,000 円 2025-07-01 P002 M01 5 20,000 円 処理プロセス solver.Solve(model) を呼び出すと、CP-SAT ソルバーが以下の流れで最適解を導き出します。 ① 制約による探索範囲の絞り込み 稼働時間上限（360 分）と需要上限の2つの制約から、各決定変数の上限を算出する P001（加工時間 30 分/個）：360 分 ÷ 30 分 = 最大 12 個 → 需要上限 8 個と比較して 0〜8 個 に確定 P002（加工時間 60 分/個）：360 分 ÷ 60 分 = 最大 6 個 → 需要上限 5 個と比較して 0〜5 個 に確定 ② 決定変数への値の割り当て 絞り込んだ範囲の中で決定変数に値を1つずつ割り当て、目的関数の値を記録する 「P001 の生産数を 0 個」に設定すると、「P002 は最大 5 個（需要上限）まで加工可能」となり、「P001=0 / P002=5 / 利益 20,000 円」が記録される ③ 目的関数の値をもとにした最適解の更新 ②の処理を反復し、最適解の探索を行う P001 P002 総利益 結果 0 個 5 個 20,000 円 最適解を更新 1 個 5 個 21,000 円 最適解を更新 2 個 5 個 22,000 円 最適解を更新 P001 = 3〜8 個のパターンにおいて、残りの稼働可能時間を最大限使っても暫定の最適解を上回らないため、 P001 = 2・P002 = 5（総利益 22,000 円） が最適解として確定する 河野 利紀 (記事一覧) クラウドソリューション部 デジタルワークプレイス課 2025年10月にG-genに入社。 神奈川在住で、Google Cloud をマスターするため日々エンジニアとして修行中。