Content

こんにちは!みっちーです!



前回の記事「データサイエンスエージェントをAgentEngineにデプロイしてみた!」では、ADK(Agent Development Kit)で作ったAIエージェントを、Google CloudのAgent Engineにデプロイするところまでをご紹介しました。



クラウド上で自分だけのエージェントが動くようにはできましたが、せっかくなら、Webアプリケーションや社内の業務システムなど、色々なところから呼び出して使いたいですよね?



そこで今回は、前回応用編としてご紹介した、デプロイしたAgent Engineを外部のアプリケーションから簡単に呼び出せる「REST API」として公開する方法を、より詳細に解説します!具体的には、Cloud RunFastAPIを使って、APIを構築していきます。

前提条件

    • デプロイしたAgent EngineのリソースIDが控えてあること。

    • Agent Engineが使用するCloud Storageバケット名がわかっていること。

    • gcloud CLIがインストールされ、プロジェクトに対して認証済みであること。

    • Docker Desktopなどがインストールされ、Dockerコマンドが利用可能であること。

もし、まだの方は、前回の記事を参考にデプロイまでを完了させてください!

Agent EngineのAPIと外部連携の課題

「あれ?Agent Engineにデプロイしたら、もうAPIとして使えるんじゃないの?」と思った方もいるかもしれません。その通りなのですが、実は一つ、実用上で考慮すべき点があります。

Agent EngineのAPIは、一般的なWeb API(REST API)で使われる単純な「リクエストを送って、レスポンスを受け取る」という形式ではなく、gRPCという高度な技術に基づいた「ストリーミング形式」になっています。

これは、エージェントが思考している過程をリアルタイムで送り返したり、会話の履歴を効率的に管理したりするための仕組みで非常に高性能なのですが、多くのWebアプリケーションやツールが標準で対応しているREST APIとは少し毛色が異なります。そのため、外部システムから直接呼び出そうとすると、認証や接続の部分で少し複雑な実装が必要になってしまうのです。

ですが、この課題は、間に「通訳」を立てることでスマートに解決できます。その「通訳」の役割を担ってくれるのが、今回ご紹介するCloud Runです。

Cloud RunでAPIアダプターを構築する

ここからは、外部のアプリケーション(クライアント)とAgent Engineの間に入って通信を仲介してくれる「APIアダプター」を構築していきます。このアダプターが、gRPCの複雑な部分を隠蔽し、クライアントに対してはシンプルなREST APIを提供してくれます。

構成図

全体の構成は以下のようになります。

    1. 外部アプリケーションは、使い慣れたシンプルなHTTPSリクエスト(POSTメソッド)で、Cloud Run上のAPIアダプターに質問を送信します。

    1. Cloud Run APIアダプターは、リクエストを受け取ると、Vertex AI SDKを使ってAgent Engineとの認証やセッション管理、gRPC通信といった複雑な処理をすべて肩代わりします。

    1. Agent Engineは、質問を処理して、結果をアダプターに返します。

    1. Cloud Run APIアダプターは、Agent Engineからの応答を、きれいなJSON形式に整形して、外部アプリケーションに返します。

このようにCloud Runを間に挟むことで、Agent Engineの高度な機能を保ちつつ、様々なシステムから簡単に利用できるようになるのです。

準備するもの

早速、実装に進みましょう!今回は、adk-samples/python/agents/data-scienceディレクトリの中に、APIアダプター用のコードを追加していきます。まず、作業用のディレクトリを作成しましょう。

mkdir api-adapter
cd api-adapter

このディレクトリの中に、以下の3つのファイルを作成していきます。

    • main.py: APIのロジックを記述するFastAPIアプリケーション本体


    • Dockerfile: アプリケーションをコンテナ化するための設計図


    • requirements.txt: 必要なPythonライブラリの一覧

Dockerfileの作成

どんな環境でも同じように動くコンテナを作成するための設計図です。ベースイメージには、安定性の高いPython 3.12を指定します。

# ベースイメージとして公式のPythonイメージを使用
FROM python:3.12-slim

# 作業ディレクトリを設定
WORKDIR /app

# 必要なライブラリをインストールするためのファイルをコピー
COPY requirements.txt .

# pipでライブラリをインストール
RUN pip install --no-cache-dir -r requirements.txt

# アプリケーションのソースコードをコピー
COPY . .

# コンテナがリッスンするポートを指定
EXPOSE 8080

# コンテナ起動時に実行するコマンド
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

ライブラリの準備 (requirements.txt)

アプリケーションが必要とするPythonライブラリを記述します。

fastapi
uvicorn
pydantic
google-cloud-aiplatform

FastAPIアプリケーションの作成 (main.py)

APIの心臓部となるFastAPIのコードです。Cloud Run上で安定して動作させるため、Agent Engineの初期化を非同期で行い、会話の継続性を保つためのセッション管理機能を実装します。

import os
import logging
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import vertexai
from vertexai import agent_engines

# --- ロギング設定 ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# --- アプリケーションの状態管理 ---
# この辞書に、lifespanで初期化されたオブジェクトを格納する
app_state = {}

# --- FastAPIのライフサイクル管理 ---
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
アプリケーションの起動・終了時に実行される処理を定義します。
起動時にVertex AI Agent Engineを初期化し、app_stateに格納します。
初期化に失敗した場合、アプリケーションは起動しません。
"""
logger.info("アプリケーションの起動処理を開始します...")
try:
# 1. 環境変数の読み込みと検証
project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
location = os.getenv("GOOGLE_CLOUD_LOCATION")
bucket = os.getenv("GOOGLE_CLOUD_STORAGE_BUCKET")
agent_resource_id = os.getenv("AGENT_RESOURCE_ID")

if not all([project_id, location, bucket, agent_resource_id]):
raise ValueError(
"以下の環境変数が設定されていません: "
"GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, "
"GOOGLE_CLOUD_STORAGE_BUCKET, AGENT_RESOURCE_ID"
)

# 2. Vertex AI SDKの初期化
# これはブロッキング処理のため、非同期のイベントループをブロックしないように
# asyncio.to_threadを使って別スレッドで実行します。
logger.info(f"Vertex AI SDKを初期化中 (Project: {project_id}, Location: {location})...")
await asyncio.to_thread(
vertexai.init,
project=project_id,
location=location,
staging_bucket=f"gs://{bucket}"
)

# 3. Agent Engineクライアントの取得
logger.info(f"Agent Engineを取得中 (Resource ID: {agent_resource_id})...")
agent = await asyncio.to_thread(
agent_engines.get,
agent_resource_id
)

# 4. 取得したクライアントをアプリケーションの状態として保持
app_state["remote_agent"] = agent
logger.info("Agentの初期化が正常に完了しました。アプリケーションがリクエストを受け付けます。")

except Exception as e:
logger.critical(f"アプリケーションの初期化中に致命的なエラーが発生しました: {e}", exc_info=True)
# ここで発生した例外はFastAPIによって捕捉され、アプリケーションの起動が中止されます。
# Cloud Run環境では、コンテナが異常終了したとみなされ、設定に応じて再起動が試みられます。
raise

# `yield`でFastAPIアプリケーション本体の処理に制御を移す
yield

# --- アプリケーション終了時の処理 ---
# 今回は特にクリーンアップ処理は不要
logger.info("アプリケーションをシャットダウンします。")


# --- FastAPIアプリケーションのインスタンス化 ---
app = FastAPI(
title="Data Agent API",
description="Vertex AI Agent Engineと対話するためのAPI",
version="1.0.0",
lifespan=lifespan # 上で定義したlifespanハンドラを登録
)

# --- CORSミドルウェアの設定 ---
# すべてのオリジンからのリクエストを許可(開発用)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

# --- Pydanticモデル定義(APIのI/Fを定義) ---
class QueryRequest(BaseModel):
user_id: str
query: str
session_id: str | None = None

class QueryResponse(BaseModel):
response: str
session_id: str

class HealthResponse(BaseModel):
status: str

# --- APIエンドポイント定義 ---
@app.get("/health", response_model=HealthResponse, tags=["Management"])
def health_check():
"""
サービスが正常に起動しているかを確認するヘルスチェックエンドポイント。
このエンドポイントが200 OKを返す場合、Agentの初期化は成功しています。
"""
return {"status": "ok"}


@app.post("/query", response_model=QueryResponse, tags=["Agent"])
async def query_agent(request: QueryRequest):
"""
Agent Engineに問い合わせを行い、応答を取得します。
セッションIDが指定されていない場合は、新しいセッションを自動的に作成します。
"""
# lifespanで初期化が完了しているため、app_stateから常にagentインスタンスを取得できる
remote_agent = app_state.get("remote_agent")
if not remote_agent:
# 基本的にこのエラーは発生しないはずだが、念のためハンドリング
raise HTTPException(status_code=503, detail="サービスが利用できません。")

current_session_id = request.session_id

try:
# セッションIDがない場合は新しいセッションを作成
if not current_session_id:
logger.info(f"新規セッションを作成します (user_id: {request.user_id})")
# SDKのメソッドはブロッキングするため、to_threadで実行
new_session = await asyncio.to_thread(
remote_agent.create_session, user_id=request.user_id
)
current_session_id = new_session['id']
logger.info(f"新規セッション作成完了: {current_session_id}")

# Agentに問い合わせをストリーミング
logger.info(f"Agentに問い合わせ中 (session_id: {current_session_id})")
response_stream = await asyncio.to_thread(
remote_agent.stream_query,
user_id=request.user_id,
session_id=current_session_id,
message=request.query,
)

# ストリームから応答を連結して完全なレスポンスを作成
full_response = "".join(
_parse_agent_event(event) for event in response_stream
)

logger.info(f"Agentからの最終応答長: {len(full_response)}")
if not full_response:
logger.warning("Agentからの応答が空でした。")
full_response = "すみません、応答を生成できませんでした。"

return QueryResponse(response=full_response, session_id=current_session_id)

except Exception as e:
logger.error(f"Agentへの問い合わせ中にエラーが発生しました: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"内部サーバーエラー: {str(e)}")

def _parse_agent_event(event: dict | object) -> str:
"""Agentからのイベントをパースして、テキスト部分を抽出するヘルパー関数。"""
match event:
# モデルからのテキスト応答を想定したケース
# {"content": {"parts": [{"text": "..."}]}} という構造にマッチ
case {"content": {"parts": [{"text": str(text)}]}}:
return text

# ツールからの出力などを想定したケース
# {"output": "..."} という構造にマッチ
case {"output": str(output)}:
return str(output) # 念のため文字列に変換

# 古い形式や、予期せぬオブジェクト形式の応答を想定したケース
case _ if hasattr(event, "text"):
return str(event.text)

# いずれにもマッチしない、未対応の形式
case _:
logger.warning(f"未対応のイベント形式です: {event}")
return ""

【重要】Staging Bucketについて

ここで一つ、非常に重要なポイントがあります。コード内のvertexai.init()staging_bucketというパラメータを渡している点です。

data-scienceエージェントは、BigQueryへのクエリ実行やデータ分析のために、内部でPythonコードを動的に生成し、実行します。このとき、生成したコードなどの中間ファイルを一時的に保存する場所として、Cloud Storageバケット(Staging Bucket)が必須となります。

Cloud Runのような制約のある環境では、このバケットを明示的に指定しないと、SDKが初期化処理を完了できずにハングアップし、APIが無応答になる原因となります。単純な応答のみを行うエージェントでは不要な場合もありますが、data-scienceエージェントのような高機能なエージェントでは必ず指定するようにしましょう。

Cloud Runへのデプロイと動作確認

【最重要】IAM権限の設定

デプロイを成功させるための最も重要なステップです。Cloud RunサービスがAgent Engineや、エージェントが利用する他のGoogle Cloudサービス(今回はBigQuery)と正しく連携するためには、適切なIAM権限を事前に付与しておく必要があります。

Cloud Runは、デフォルトではプロジェクトの「Compute Engine default service account」として動作します。このサービスアカウント([PROJECT_NUMBER]-compute@developer.gserviceaccount.com)に対して、以下の2つのロールを付与します。

    • Vertex AI ユーザー (roles/aiplatform.user): Agent Engineを呼び出すために必須です。

    • BigQuery データ閲覧者 (roles/bigquery.dataViewer): data-scienceエージェントがBigQueryのテーブルを読み取るために必須です。

以下のコマンドを実行して、権限を付与してください。

# --- ご自身の環境に合わせて設定してください ---
export GCP_PROJECT_ID="YOUR_GCP_PROJECT_ID"
# --- 設定ここまで ---

# プロジェクト番号からサービスアカウントのメールアドレスを特定
export PROJECT_NUMBER=$(gcloud projects describe ${GCP_PROJECT_ID} --format="value(projectNumber)")
export SERVICE_ACCOUNT_EMAIL="${PROJECT_NUMBER}-compute@developer.gserviceaccount.com"

# Vertex AI ユーザーロールを付与
gcloud projects add-iam-policy-binding ${GCP_PROJECT_ID} \
--member="serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
--role="roles/aiplatform.user"

# BigQuery データ閲覧者ロールを付与
gcloud projects add-iam-policy-binding ${GCP_PROJECT_ID} \
--member="serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
--role="roles/bigquery.dataViewer"

これらの権限設定を先に行っておくことで、後のデプロイがスムーズに進みます。

Cloud Runへソースコードから直接デプロイ

いよいよデプロイです!作成したAPIアダプターのソースコードを、Cloud Runにデプロイします。gcloud CLIの–sourceフラグを使うと、コンテナのビルドからデプロイまでを全自動で行ってくれます。

まず、デプロイに必要な情報を環境変数に設定します。YOUR_GCP_PROJECT_IDYOUR_AGENT_ID の部分は、ご自身の環境に合わせて書き換えてください。エージェントIDは、前回の記事でデプロイした際に控えておいた、末尾のID(例: 123456789…のような数値)です。

api-adapter ディレクトリ内で以下のコマンドを実行してください)

# --- ご自身の環境に合わせて設定してください ---
export GCP_PROJECT_ID="YOUR_GCP_PROJECT_ID"
export AGENT_ID="YOUR_AGENT_ID"
export REGION="us-central1" # エージェントをデプロイしたリージョン
export GCS_BUCKET_NAME="YOUR_GCS_BUCKET_NAME" # 例: my-project-staging-bucket
# --- 設定ここまで ---

# デプロイに必要な変数を設定
export SERVICE_NAME="data-science-adapter"
export FULL_AGENT_RESOURCE_ID="projects/${GCP_PROJECT_ID}/locations/${REGION}/reasoningEngines/${AGENT_ID}"

# Cloud Runへソースコードから直接デプロイを実行
gcloud run deploy ${SERVICE_NAME} \
--source . \
--platform=managed \
--project=${GCP_PROJECT_ID} \
--region ${REGION} \
--allow-unauthenticated \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${GCP_PROJECT_ID},GOOGLE_CLOUD_LOCATION=${REGION},AGENT_RESOURCE_ID=${FULL_AGENT_RESOURCE_ID},GOOGLE_CLOUD_STORAGE_BUCKET=${GCS_BUCKET_NAME}"

このコマンドのポイントは以下の通りです。

    • –source .: このフラグによって、カレントディレクトリ(.)のソースコードを使い、ビルドからデプロイまでを自動で行うよう指示します。

    • –allow-unauthenticated: 今回はテストのため、誰でもアクセスできる公開APIとしてデプロイします。本番環境では、IAMなどを使って適切にアクセス制御を行ってください。

    • –set-env-vars: main.pyで必要とした環境変数を、Cloud Runサービスに設定しています。ここでは、組み立てた完全なリソースIDを渡しているのがポイントです。

デプロイが完了すると、サービスのURLが出力されます。これが、あなたのAIエージェントに繋がる専用のAPIエンドポイントです!

Service [data-science-adapter] deployed.
URL: https://data-science-adapter-xxxxxxxxxx-uc.a.run.app

APIの動作確認(セッション維持と日本語対応)

最後に、APIが正しく機能するかを確認します。ここでは、エージェントとの会話を継続させる「セッション維持」と、日本語のようなマルチバイト文字を正しく送信する方法も合わせて解説します。

ステップ1: リクエスト用のJSONファイルを作成する

curlコマンドの引数に直接JSONを書くと、特に日本語が含まれる場合に文字コードの問題が発生しやすくなります。これを避けるため、リクエストの内容を一度JSONファイルとして保存します。api-adapterディレクトリに、request.jsonという名前で以下のファイルを作成してください。

{
"user_id": "api-test-user",
"query": "こんにちは!どんなデータがありますか?"
}

ステップ2: 最初の質問を送信し、セッションIDを控える

作成したJSONファイルを使って、APIに最初の質問を投げます。成功すると、応答としてJSONが返ってきます。この中に含まれる session_id の値(“…”の部分)を、次のステップで使うのでコピーして控えておいてください。

# Cloud RunのURLを環境変数に設定
export API_URL="https://data-science-adapter-xxxxxxxxxx-uc.a.run.app" # ご自身のURLに書き換えてください

# APIにリクエストを送信
curl -X POST "${API_URL}/query" \
-H "Content-Type: application/json" \
-d @request.json

すると、以下のような応答が返ってきます。この中に含まれる session_id の値(“…”の部分)を、次のステップで使うのでコピーして控えておいてください。

{"response":"こんにちは!利用可能なデータセットは...","session_id":"..."}

ステップ3: 会話を続ける(セッションを維持する)

次に、先ほど控えたsession_idを使って、会話を続けてみましょう。エージェントが前の会話の内容を覚えているか確認できます。先ほど作成したrequest.jsonを以下のように編集します。“session_id”のキーを追加し、値には先ほどコピーしたセッションIDを貼り付けてください。また、次の質問も変更してみましょう。

{
"user_id": "api-test-user",
"query": "trainテーブルに対して分析を開始したい。",
"session_id": "ここに先ほどコピーしたセッションIDを貼り付け"
}

ファイルを保存したら、ステップ2と同じコマンドを実行して、再度APIにリクエストを送信します。

# APIにリクエストを送信
curl -X POST "${API_URL}/query" \
-H "Content-Type: application/json" \
-d @request.json

エージェントが正しく応答し、売上が最も高い商品についての分析結果を返してくれれば、セッションが正常に維持されている証拠です。

やりましたね!これで、あなたのデータサイエンスエージェントは、会話の文脈を理解できる、本格的なREST APIとして機能するようになりました。

トラブルシューティング

もしAPIから応答がない、またはエラーが返ってくる場合は、以下の点を確認してみてください。

1. ヘルスチェックとCloud Runの起動状態を確認する

まず、/healthエンドポイントにアクセスして、APIが正常に応答するか確認します。{“status”:”ok”} という応答があれば、アプリケーションのコンテナは起動しています。もし応答がない、またはエラーが返ってくる場合は、そもそもコンテナの起動に失敗している可能性があります。その場合は次のステップに進み、Cloud Runのログを確認してください。起動に失敗する最も一般的な原因は、IAM権限の不足です。

curl https://YOUR_SERVICE_URL/health

2. Cloud Runのログを確認する

次に、Cloud Runのログを確認して、具体的なエラーメッセージを探します。

    • Google Cloudコンソールにアクセスし、ナビゲーションメニューから「Cloud Run」を選択します。


    • デプロイしたサービス(例: data-science-adapter)をクリックします。


    • サービスの詳細ページで、「ログ」タブを選択します。


    • ここで、アプリケーションの起動ログやリクエスト処理中のエラーログをリアルタイムで確認できます。重大度(Severity)でフィルタリングすると、エラーを特定しやすくなります。

ログを確認することで、IAM権限エラーや、予期しない応答イベントの構造などを特定できます。

セキュリティに関する重要な注意点

本ブログは、テストを簡単にするために–allow-unauthenticatedフラグを付けて、誰でもアクセスできる公開APIとしてサービスをデプロイしました。しかし、この設定は本番環境では非推奨です。悪意のある第三者によるAPIの不正利用や、意図しない高額な請求に繋がる危険性があります。

本番環境でAPIを安全に運用するためには、以下のような認証・認可の仕組みを導入することを推奨します。

    • IAMによるアクセス制御: –allow-unauthenticatedフラグを削除してデプロイし、特定のサービスアカウントやユーザーにのみ「Cloud Run 起動元」(roles/run.invokerロールを付与します。これにより、許可されたIDを持つアプリケーションやユーザーだけがAPIを呼び出せるようになります。

    • API Gatewayの利用: Cloud Runの前にAPI Gatewayを配置し、APIキーの検証やJWT(JSON Web Token)による認証を必須とします。これにより、より高度なアクセス管理、レート制限、モニタリングが可能になります。

    • Identity-Aware Proxy (IAP) の利用: 特定のGoogleアカウントからのアクセスのみを許可したい場合に有効です。IAPを有効にすることで、認証されたユーザーのみがCloud Runサービスにアクセスできるようになります。

ユースケースに応じて適切なセキュリティ対策を講じることが、AIエージェントを安全に運用する上で不可欠です。

まとめ

今回は、Agent EngineにデプロイしたAIエージェントを、Cloud RunFastAPIを使って外部アプリケーションから簡単に利用できるREST APIとして公開する方法を解説しました。

特に、data-scienceのような高機能なエージェントをデプロイする際は、適切なIAM権限の設定と、エージェントからの多様な応答形式への対応が成功の鍵となります。

gRPCベースのAgent Engineの前にCloud Runで作成した「APIアダプター」を置くことで、認証や通信の複雑さを吸収し、クライアント側はシンプルなI/Fでエージェントの強力な機能を利用できるようになります。この構成は、AIエージェントを実際の業務システムやWebサービスに組み込む際の、非常に強力なデザインパターンです。

今回作成したAPIは、単純に質問を投げて応答を待つだけのものでしたが、今後の展望としては、以下のような応用が考えられます。

    • セッション管理の実装: 過去の会話履歴を覚えてもらうために、user_idごとに会話のセッションを管理する機能を追加する。

    • ストリーミング対応: Agent Engineの思考プロセスをリアルタイムでクライアントに送信するために、WebSocketなどを使ってAPIをストリーミング対応にする。

    • 認証・認可の強化: API GatewayやIdentity Platformと連携し、特定のユーザーやアプリケーションのみが利用できるようにセキュリティを強化する。

ぜひ皆さんもCloud RunとAgent Engineを組み合わせて、AIエージェントを様々な場面で活用できるようにしていきましょう!

2025年8月22日 Agent EngineをREST APIとして使えるようにしてみた!

Category Google Cloud

ご意見・ご相談・料金のお見積もりなど、
お気軽にお問い合わせください。

お問い合わせはこちら