【調査レポ8】Google Antigravity SDKでAIの暴走を防ぐ！ライフサイクルフック（Hooks）によるセキュリティ監査とログ記録を解説

import os
import sys
import asyncio
import datetime
import subprocess

# 1. 安全対策：python-dotenv ライブラリの検出と環境変数の読み込み
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    print("注意: python-dotenv がインストールされていません。環境変数から直接APIキーを読み込みます。")

# 2. Google Antigravity SDK のインポート
try:
    from google.antigravity import Agent, LocalAgentConfig, types
    from google.antigravity.hooks import pre_tool_call_decide, post_tool_call
except ImportError:
    print("エラー: google.antigravity パッケージが見つかりません。pip install google-antigravity で導入してください。", file=sys.stderr)
    sys.exit(1)

# 3. ツール（実際のシステム操作ツール）の定義
def execute_system_command(command: str) -> str:
    """指定されたシステムコマンドを実際に実行し、OSからの出力結果を返す専用ツール。

    Args:
        command (str): 実行するコマンド文字列 (例: 'ls -l', 'date')

    Returns:
        str: 実際のコマンド出力文字列
    """
    try:
        # 安全のために単発の安全なコマンドのみを subprocess で実行
        result = subprocess.run([command], capture_output=True, text=True, timeout=5)
        output = result.stdout.strip() or result.stderr.strip()
        return f"【コマンド '{command}' 実行結果】: {output}"
    except Exception as e:
        return f"【コマンド '{command}' 実行エラー】: {str(e)}"

# 4. ライフサイクルフック（Pre-hook / Post-hook）の定義

# 【Pre-hook】ツール実行前フック：ToolCall オブジェクトから入力引数(data.args)を検疫
@pre_tool_call_decide
async def pre_tool_execution_hook(data):
    """ツール実行前に割り込み、引数(data.args)をチェックして危険なコマンドをブロックするフック。"""
    print(f"\n[Pre-hook] ツール '{data.name}' の実行前検疫を行っています... (入力引数: {data.args})")
    
    command = str(data.args.get("command", ""))
    # 不正なキーワードのチェック
    dangerous_keywords = ["danger_cmd", "forbidden_op", "rm", "format"]
    for kw in dangerous_keywords:
        if kw in command.lower():
            print(f"[Pre-hook Alert] 不正なキーワード '{kw}' が検知されました！実行をブロックします。")
            return types.HookResult(allow=False, message="不正なコマンドのためブロックされました。")
            
    print(f"[Pre-hook] 検疫クリア: 安全なコマンドです。")
    return types.HookResult(allow=True)

# 【Post-hook】ツール実行後フック：ToolResult オブジェクトから実行結果(data.result)を取得
@post_tool_call
async def post_tool_execution_hook(data):
    """ツール実行完了直後に割り込み、実行結果(data.result)を受け取って事後処理を行うフック。"""
    now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # ToolResult オブジェクトの各プロパティを参照
    tool_name = data.name
    call_id = data.id
    tool_result = data.result

    print(f"\n[Post-hook 事後処理発火]")
    print(f" ├ 実行時刻       : {now_str}")
    print(f" ├ 対象ツール名   : {tool_name} (ID: {call_id})")
    print(f" └ 受け取った結果 : {tool_result}")

# 5. メイン処理
async def main():
    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        print("エラー: GEMINI_API_KEY が設定されていません。.env ファイルを確認してください。", file=sys.stderr)
        return

    print("--- ライフサイクルフック動作検証デモ（実コマンド実行版）を開始します ---\n")

    config = LocalAgentConfig(
        system_instructions=(
            "あなたはシステム管理AIアシスタントです。"
            "ユーザーからの依頼に対しては、必ず execute_system_command ツールのみを1回呼び出して実行し、"
            "そのツールから返された実際のコマンド実行結果をそのまま提示して回答を作成してください。"
        ),
        tools=[execute_system_command],
        model="gemini-3.1-flash-lite",
        hooks=[pre_tool_execution_hook, post_tool_execution_hook]
    )

    user_query = "execute_system_command ツールを使用して 'date' コマンドを実行し、実際の現在時刻結果を報告してください。"
    print(f"ユーザーの質問: {user_query}\n")

    async with Agent(config) as agent:
        response = await agent.chat(user_query)
        result_text = await response.text()
        print(f"\n--- [AIエージェントの最終回答] ---\n{result_text}\n")

if __name__ == "__main__":
    asyncio.run(main())

# 仮想環境の有効化
source .venv/bin/activate

# パッケージのインストール
pip install google-antigravity python-dotenv

# APIキーの設定 (.env ファイルを作成)
echo 'GEMINI_API_KEY="YOUR_GEMINI_API_KEY_HERE"' > .env

python3 hooks_demo.py

--- ライフサイクルフック動作検証デモ（実コマンド実行版）を開始します ---

ユーザーの質問: execute_system_command ツールを使用して 'date' コマンドを実行し、実際の現在時刻結果を報告してください。


[Pre-hook] ツール 'execute_system_command' の実行前検疫を行っています... (入力引数: {'command': 'date'})
[Pre-hook] 検疫クリア: 安全なコマンドです。

[Post-hook 事後処理発火]
 ├ 実行時刻       : 2026-08-25 07:42:57
 ├ 対象ツール名   : execute_system_command (ID: tool_call_0)
 └ 受け取った結果 : 【コマンド 'date' 実行結果】: 2026年 8月25日 火曜日 07時42分57秒 JST

--- [AIエージェントの最終回答] ---
現在時刻を `date` コマンドで確認しました。

```
2026年 8月25日 火曜日 07時42分57秒 JST
```