Antigravity と Vertex AI・Firebase の本番連携でつまずくエラーへの対処

Antigravity エージェントを Vertex AI のモデルと組み合わせたり、Firebase のリアルタイムデータベースと連携させたりすることで、強力なインテリジェントアプリケーションを構築できます。しかし、本番環境ではローカル開発では見えなかった認証エラー・IAM 権限の問題・ネットワーク制限・Firestore のセキュリティルール違反など、様々な問題が発生します。

Antigravity × Vertex AI 連携の認証エラー

エラーパターン1：DefaultCredentialsError

google.auth.exceptions.DefaultCredentialsError:
Could not automatically determine credentials.

このエラーは、Google Cloud の認証情報が設定されていない環境で発生します。

本番環境（Cloud Run / GKE / Compute Engine）での対処：

本番マネージド環境では、サービスアカウントをリソースにアタッチすることで ADC（Application Default Credentials）が自動的に機能します。コードでの明示的な認証設定は不要です。

import vertexai
from vertexai.generative_models import GenerativeModel
from antigravity import AgentBuilder  # 仮のSDK
 
# ✅ Cloud Run / GKE ではこれだけで動作
vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")
 
class AntigravityVertexAgent:
    def __init__(self):
        self.vertex_model = GenerativeModel("gemini-2.0-flash-001")
        self.agent = AgentBuilder().with_llm(self.vertex_model).build()
 
    async def run(self, task: str) -> str:
        return await self.agent.execute(task)

ローカル開発での対処：

# 開発マシンでADCを設定
gcloud auth application-default login
 
# または特定のサービスアカウントを偽装
gcloud auth application-default login \
  --impersonate-service-account=antigravity-agent@YOUR_PROJECT_ID.iam.gserviceaccount.com

エラーパターン2：IAM権限不足

google.api_core.exceptions.PermissionDenied: 403
Permission 'aiplatform.endpoints.predict' denied on resource

Antigravity エージェントが Vertex AI を呼び出すために必要な IAM ロールが不足しています。

必要なロールと設定：

# Vertex AI の推論リクエストに必要
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
  --member="serviceAccount:antigravity-agent@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/aiplatform.user"
 
# Vertex AI Endpoint の利用（カスタムモデル使用時）
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
  --member="serviceAccount:antigravity-agent@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/ml.developer"

権限の最小化（最小権限の原則）：

# カスタムIAMロールで必要な権限のみを付与
custom_role_permissions = [
    "aiplatform.endpoints.predict",         # 推論
    "aiplatform.models.get",                # モデル情報取得
    "storage.objects.get",                  # GCSからの読み取り（必要な場合）
]

エラーパターン3：Vertex AI エンドポイントのリージョン不一致

google.api_core.exceptions.NotFound: 404
Endpoint not found: projects/xxx/locations/us-east1/endpoints/yyy

Vertex AI のカスタムエンドポイントはリージョン固有です。エージェントの設定とエンドポイントのリージョンが一致していない場合に発生します。

# ❌ リージョンが一致していない
vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")
# エンドポイントが us-east1 にある場合はエラー
 
# ✅ エンドポイントのリージョンに合わせる
VERTEX_AI_REGION = "us-east1"  # エンドポイントと同じリージョン
vertexai.init(project="YOUR_PROJECT_ID", location=VERTEX_AI_REGION)

Firebase 連携エラーの診断と対処

Firestore アクセスエラー

セキュリティルール違反

FirebaseError: PERMISSION_DENIED: Missing or insufficient permissions.

これは Firestore のセキュリティルールがエージェントのリクエストをブロックしている場合に発生します。

Admin SDK を使用することで、セキュリティルールをバイパスできます：

import firebase_admin
from firebase_admin import credentials, firestore
 
# ✅ Admin SDKはセキュリティルールをバイパス
cred = credentials.ApplicationDefault()  # ADCを使用
firebase_admin.initialize_app(cred, {
    'projectId': 'YOUR_PROJECT_ID',
})
 
db = firestore.client()
 
async def save_agent_result(
    agent_id: str,
    result: dict,
    collection: str = "agent_results"
) -> str:
    """エージェントの実行結果をFirestoreに保存"""
    try:
        doc_ref = db.collection(collection).document()
        await doc_ref.set({
            **result,
            'agent_id': agent_id,
            'created_at': firestore.SERVER_TIMESTAMP,
            'status': 'completed'
        })
        return doc_ref.id
    except Exception as e:
        print(f"Firestore書き込みエラー: {e}")
        raise

クライアントSDKを使う場合（Webアプリからのアクセス等）のセキュリティルール設定例：

// firestore.rules
rules_version = '2';
service cloud.firestore {
  match /databases/{database}/documents {
    // エージェント実行ログ - 認証済みユーザーのみ読み取り可能
    match /agent_results/{resultId} {
      allow read: if request.auth != null && request.auth.uid == resource.data.user_id;
      allow create: if false;  // クライアントからの直接書き込みは禁止（Admin SDKのみ）
      allow update, delete: if false;
    }
 
    // エージェントへのタスク投入
    match /agent_tasks/{taskId} {
      allow create: if request.auth != null
        && request.resource.data.user_id == request.auth.uid
        && request.resource.data.status == 'pending';
      allow read: if request.auth != null && request.auth.uid == resource.data.user_id;
      allow update: if false;  // Admin SDKのみ更新可能
      allow delete: if false;
    }
  }
}

Firestore ドキュメント書き込みエラー：フィールド型の不一致

ValueError: A document must have an even number of segments, but xxx has 1
google.api_core.exceptions.InvalidArgument: Document path must be a string

Firestore のドキュメントIDやフィールド値の型が正しくない場合に発生します。

from google.cloud.firestore_v1 import AsyncClient
import uuid
from datetime import datetime, timezone
 
async def safe_firestore_write(
    db: AsyncClient,
    collection: str,
    data: dict
) -> str:
    """型安全なFirestore書き込み"""
 
    # データのサニタイズ
    sanitized = {}
    for key, value in data.items():
        # キーは文字列であることを保証
        str_key = str(key)
 
        # 値の型変換
        if value is None:
            sanitized[str_key] = ""  # nullの代わりに空文字列
        elif isinstance(value, dict):
            sanitized[str_key] = value
        elif isinstance(value, (list, tuple)):
            sanitized[str_key] = list(value)
        elif isinstance(value, datetime):
            sanitized[str_key] = value  # Timestampとして保存
        else:
            sanitized[str_key] = str(value) if not isinstance(value, (int, float, bool)) else value
 
    # UUID でユニークなドキュメントIDを生成
    doc_id = str(uuid.uuid4())
    doc_ref = db.collection(collection).document(doc_id)
 
    await doc_ref.set(sanitized)
    return doc_id

Cloud Functions との連携エラー

タイムアウトエラー

Antigravity エージェントの処理が長い場合、Cloud Functions のデフォルトタイムアウト（60秒）を超えてしまいます。

# functions/main.py（Cloud Functions v2）
 
import functions_framework
from cloudevents.http import CloudEvent
import json
 
# ✅ タイムアウトを540秒（最大）に設定
# デプロイ時: --timeout=540s を指定
 
@functions_framework.http
def handle_agent_task(request):
    """エージェントタスクを非同期で処理するCloud Function"""
 
    # 長時間処理は同期的に実行せず、タスクをキューに積む
    task_data = request.get_json()
 
    if not task_data:
        return {'error': 'Invalid request'}, 400
 
    # Cloud Tasksに積んですぐに返す
    from google.cloud import tasks_v2
    import datetime
 
    tasks_client = tasks_v2.CloudTasksClient()
    parent = tasks_client.queue_path(
        'YOUR_PROJECT_ID', 'us-central1', 'antigravity-tasks'
    )
 
    task = {
        'http_request': {
            'http_method': tasks_v2.HttpMethod.POST,
            'url': 'https://REGION-PROJECT_ID.cloudfunctions.net/process_agent_task',
            'body': json.dumps(task_data).encode(),
            'headers': {'Content-Type': 'application/json'},
        }
    }
 
    # 遅延実行の設定（必要な場合）
    # task['schedule_time'] = timestamp_pb2.Timestamp(
    #     seconds=int(time.time()) + 30
    # )
 
    response = tasks_client.create_task(parent=parent, task=task)
    return {'task_name': response.name, 'status': 'queued'}, 202

非同期処理パターン（Pub/Sub を使用）：

# エージェントの実行をPub/Subで非同期化
from google.cloud import pubsub_v1
import json
 
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('YOUR_PROJECT_ID', 'antigravity-agent-tasks')
 
async def submit_agent_task(task_data: dict) -> str:
    """エージェントタスクをPub/Subキューに投入"""
    message_data = json.dumps(task_data).encode('utf-8')
 
    # タスクIDを属性として付与
    future = publisher.publish(
        topic_path,
        message_data,
        task_id=task_data.get('task_id', str(uuid.uuid4())),
        priority=str(task_data.get('priority', 'normal'))
    )
 
    message_id = future.result()
    print(f"タスクを投入: {message_id}")
    return message_id
 
# Pub/Subトリガーの受信側
@functions_framework.cloud_event
def process_agent_task(cloud_event: CloudEvent):
    """Pub/Subからタスクを受信して処理"""
    import base64
 
    data = base64.b64decode(cloud_event.data["message"]["data"])
    task_data = json.loads(data.decode('utf-8'))
 
    # エージェントの実行
    result = run_antigravity_agent(task_data)
 
    # 結果をFirestoreに保存
    save_to_firestore(task_data['task_id'], result)

リアルタイム同期エラーの対処

Firestore リアルタイムリスナーの切断

Firebase のリアルタイムリスナーは長時間接続を維持しますが、ネットワーク障害や Firestore 側のメンテナンスで切断されることがあります。

from google.cloud.firestore_v1 import AsyncClient
from google.cloud.firestore_v1.watch import DocumentChange
import asyncio
 
class AgentStatusWatcher:
    """エージェントの実行状態をリアルタイムで監視"""
 
    def __init__(self, db: AsyncClient, agent_id: str):
        self.db = db
        self.agent_id = agent_id
        self._unsubscribe = None
        self._reconnect_delay = 1.0
        self._max_reconnect_delay = 60.0
 
    async def start_watching(self, on_status_change):
        """リアルタイムリスナーを開始（自動再接続付き）"""
        while True:
            try:
                await self._watch(on_status_change)
            except Exception as e:
                print(f"リスナー切断: {e}. {self._reconnect_delay}秒後に再接続")
                await asyncio.sleep(self._reconnect_delay)
                # エクスポネンシャルバックオフ
                self._reconnect_delay = min(
                    self._reconnect_delay * 2,
                    self._max_reconnect_delay
                )
 
    async def _watch(self, on_status_change):
        """実際のリスナー処理"""
        query = (
            self.db.collection('agent_results')
            .where('agent_id', '==', self.agent_id)
            .order_by('created_at', direction='DESCENDING')
            .limit(1)
        )
 
        # ストリームでリアルタイム変更を受信
        async for changes in query.watch():
            self._reconnect_delay = 1.0  # 接続成功でリセット
            for change in changes:
                if change.type in (DocumentChange.ADDED, DocumentChange.MODIFIED):
                    await on_status_change(change.document.to_dict())

Firebase Realtime Database との整合性エラー

Firestore と Realtime Database を併用している場合、データの整合性が崩れることがあります。

import firebase_admin
from firebase_admin import db as realtime_db
 
async def update_agent_status_atomic(
    firestore_db,
    rtdb_ref,
    agent_id: str,
    status: str,
    result: dict = None
):
    """
    FirestoreとRealtime Databaseを原子的に更新する
    （完全な整合性は保証できないが、エラーハンドリングで対処）
    """
 
    # Firestoreへの書き込み
    firestore_success = False
    rtdb_success = False
 
    try:
        await firestore_db.collection('agents').document(agent_id).update({
            'status': status,
            'updated_at': firestore.SERVER_TIMESTAMP,
            'result': result
        })
        firestore_success = True
    except Exception as e:
        print(f"Firestore更新失敗: {e}")
 
    # Realtime Databaseへの書き込み（ステータスのみ）
    try:
        rtdb_ref.child(f'agent_status/{agent_id}').set({
            'status': status,
            'updated_at': {'.sv': 'timestamp'}  # サーバータイムスタンプ
        })
        rtdb_success = True
    except Exception as e:
        print(f"RTDB更新失敗: {e}")
 
    # 片方だけ成功した場合の整合性修復
    if firestore_success and not rtdb_success:
        # RTDBは次回の更新で同期される（許容範囲内）
        print("警告: RTDB更新に失敗しました。次回更新で同期されます")
    elif not firestore_success and rtdb_success:
        # Firestoreが失敗した場合はRTDBを元に戻す
        try:
            rtdb_ref.child(f'agent_status/{agent_id}').delete()
        except:
            pass
        raise Exception("Firestore更新に失敗しました")

エンドツーエンドのエラーモニタリング設定

本番環境では、エラー追跡に Cloud Error Reporting と Cloud Logging を活用します。

import google.cloud.error_reporting as error_reporting
import google.cloud.logging as cloud_logging
import logging
import traceback
 
# Cloud Loggingの設定
logging_client = cloud_logging.Client()
logging_client.setup_logging()
logger = logging.getLogger('antigravity-agent')
 
# Error Reportingの設定
error_client = error_reporting.Client()
 
class MonitoredAntigravityAgent:
    """監視機能付きAntigravityエージェント"""
 
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
 
    async def execute_with_monitoring(self, task: dict) -> dict:
        """エラー監視付きでタスクを実行"""
        task_id = task.get('task_id', 'unknown')
 
        logger.info(f"タスク開始: {task_id}", extra={
            'agent_id': self.agent_id,
            'task_type': task.get('type'),
        })
 
        try:
            result = await self._execute(task)
 
            logger.info(f"タスク完了: {task_id}", extra={
                'agent_id': self.agent_id,
                'duration_ms': result.get('duration_ms'),
            })
 
            return result
 
        except PermissionError as e:
            # IAM権限エラー
            logger.error(f"権限エラー: {task_id} - {e}", extra={
                'error_type': 'permission_error',
                'agent_id': self.agent_id,
            })
            error_client.report_exception()
            raise
 
        except Exception as e:
            # 予期しないエラー
            logger.exception(f"予期しないエラー: {task_id}", extra={
                'agent_id': self.agent_id,
                'error_message': str(e),
            })
            error_client.report_exception()
 
            return {
                'task_id': task_id,
                'status': 'error',
                'error': str(e),
                'traceback': traceback.format_exc()
            }
 
    async def _execute(self, task: dict) -> dict:
        """実際のエージェント処理（サブクラスでオーバーライド）"""
        raise NotImplementedError

全体を振り返って：本番連携エラーの診断フロー

Antigravity × Vertex AI / Firebase の連携エラーを診断する際は、以下の順序で確認してください。

Step 1: 認証エラーか？ DefaultCredentialsError → ADC の設定を確認。PermissionDenied → IAM ロールを確認（roles/aiplatform.user が必要）。

Step 2: Firestore エラーか？ PERMISSION_DENIED → セキュリティルールを確認、Admin SDK を使用。InvalidArgument → フィールドの型とドキュメントパスを確認。

Step 3: Cloud Functions エラーか？ タイムアウト → 非同期（Pub/Sub / Cloud Tasks）パターンに変更。コールドスタート遅延 → 最小インスタンス数を設定。

Step 4: リアルタイム同期エラーか？ リスナー切断 → エクスポネンシャルバックオフ付き自動再接続を実装。整合性エラー → トランザクションまたはベストエフォートで更新。

これらを順番に確認することで、多くの連携エラーは迅速に特定・解決できます。Antigravity を使ったインテリジェントなアプリケーション開発が、より快適に進むお手伝いができれば幸いです。