【アップデート】Gemini Enterprise Agent RuntimeのAgent IdentityがGAとなりました #GoogleCloudNext

# SPIFFE URI（エージェント間認証・mTLS 通信に使用）
spiffe://TRUST_DOMAIN/resources/SERVICE/RESOURCE_PATH

# IAM プリンシパル識別子（IAM ポリシーへの記述に使用）
principal://TRUST_DOMAIN/NAMESPACE/AGENT_NAME

# 例（IAM プリンシパル識別子）
principal://agents.global.org-123456789012.system.id.goog/resources/aiplatform/projects/9876543210/locations/us-central1/reasoningEngines/my-test-agent

export PROJECT_ID="your-project-id"
export LOCATION="us-central1"
export STAGING_BUCKET="your-staging-bucket"      # デプロイ用バケット
export DATA_BUCKET="your-data-bucket"            # データ用バケット（別バケット推奨）

gcloud config set project $PROJECT_ID

gcloud services enable \
    aiplatform.googleapis.com \
    storage.googleapis.com

# デプロイ用バケットの作成
gcloud storage buckets create gs://$STAGING_BUCKET \
    --location=$LOCATION \
    --uniform-bucket-level-access

# データ用バケットの作成
gcloud storage buckets create gs://$DATA_BUCKET \
    --location=$LOCATION \
    --uniform-bucket-level-access

# サンプルデータファイルの作成とアップロード
cat > product_info.txt << 'EOF'
【商品名】Cloud Storage ストレージプラン
【価格】標準ストレージ: $0.020/GB/月
【概要】
Google Cloud Storage はグローバルに利用できるオブジェクトストレージです。
耐久性 99.999999999%（イレブンナイン）を誇り、
あらゆる規模のデータを安全に保管できます。
EOF

gcloud storage cp product_info.txt gs://$DATA_BUCKET/product_info.txt

mkdir agent-identity-demo && cd agent-identity-demo
uv init --bare --python 3.12
uv add "google-cloud-aiplatform[adk,agent_engines]" google-cloud-storage
touch agent.py deploy.py

$ tree agent-identity-demo
agent-identity-demo
├── agent.py
├── deploy.py
├── pyproject.toml
└── uv.lock

from google.adk.agents import Agent
from google.cloud import storage

def read_file_from_gcs(bucket_name: str, blob_name: str) -> str:
    """Cloud Storage からファイルの内容を読み込む

    Args:
        bucket_name: 読み込むバケット名
        blob_name: 読み込むオブジェクト名

    Returns:
        ファイルの内容（テキスト）
    """
    # Gemini Enterprise Agent Runtime 上では ADC が自動的に Agent Identity を使用する
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    return blob.download_as_text()

agent = Agent(
    model="gemini-2.5-flash",
    name="gcs_reader_agent",
    instruction=(
        "あなたは Cloud Storage に保存されたデータを読み込んで回答するアシスタントです。"
        "ユーザーから質問を受けたら、適切なファイルを read_file_from_gcs ツールで取得し、"
        "その内容に基づいて回答してください。"
    ),
    tools=[read_file_from_gcs],
)

import vertexai
from vertexai import types
from vertexai.agent_engines import AdkApp

from agent import agent

PROJECT_ID = "your-project-id"
LOCATION = "us-central1"
STAGING_BUCKET = "your-staging-bucket"

client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=dict(api_version="v1beta1"),  # Agent Identity には v1beta1 が必要
)

app = AdkApp(agent=agent)

remote_app = client.agent_engines.create(
    agent=app,
    config={
        "display_name": "gcs-reader-agent",
        "identity_type": types.IdentityType.AGENT_IDENTITY,
        "requirements": [
            "google-cloud-aiplatform[adk,agent_engines]",
            "google-cloud-storage",
            "cloudpickle",
            "pydantic",
        ],
        "extra_packages": ["agent.py"],
        "staging_bucket": f"gs://{STAGING_BUCKET}",
    },
)

print(f"Agent Engine ID: {remote_app.api_resource.name}")
print(f"Effective Identity: {remote_app.api_resource.spec.effective_identity}")

uv run deploy.py

Agent Engine ID: projects/PROJECT_NUMBER/locations/us-central1/reasoningEngines/AGENT_ENGINE_ID
Effective Identity: agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/us-central1/reasoningEngines/AGENT_ENGINE_ID

# Effective Identity の値を変数に設定
export AGENT_IDENTITY="agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/us-central1/reasoningEngines/AGENT_ENGINE_ID"

# データバケットへの objectViewer 権限を付与
gcloud storage buckets add-iam-policy-binding gs://$DATA_BUCKET \
    --member="principal://$AGENT_IDENTITY" \
    --role="roles/storage.objectViewer"

import vertexai

PROJECT_ID = "your-project-id"
LOCATION = "us-central1"
AGENT_ENGINE_ID = "your-agent-engine-id"  # deploy.py の出力から取得
DATA_BUCKET = "your-data-bucket"

client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=dict(api_version="v1beta1"),
)

remote_app = client.agent_engines.get(
    name=f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{AGENT_ENGINE_ID}"
)

session = remote_app.create_session(user_id="test-user")

response = remote_app.stream_query(
    user_id="test-user",
    session_id=session["id"],
    message=f"バケット {DATA_BUCKET} の product_info.txt の内容を読んで、価格を教えてください。",
)

for chunk in response:
    for part in chunk.get("content", {}).get("parts", []):
        if "text" in part:
            print(part["text"], end="", flush=True)

この商品の価格は、標準ストレージで月額 $0.020/GB です。

google.resumable_media.common.InvalidResponse: ('Request failed with status code', 403,
'Expected one of', <HTTPStatus.OK: 200>, <HTTPStatus.PARTIAL_CONTENT: 206>)