【調査レポ9】Google Antigravity SDKでバックグラウンド定期実行！Periodic Triggerによる自動タスク化と通知システム

import os
import sys
import asyncio
import datetime

# 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
except ImportError:
    print("エラー: google.antigravity パッケージが見つかりません。pip install google-antigravity で導入してください。", file=sys.stderr)
    sys.exit(1)

# 3. バックグラウンドで定期実行するAIタスクの関数
async def run_periodic_monitoring(agent: Agent, task_id: int):
    """一定間隔ごとに呼び出され、監視レポートを自動生成するタスク関数。"""
    now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"\n▶ 定期タスク #{task_id} を開始します...")
    print(f"==========================================")
    print(f" [定期タスク発火ログ] 実行時刻: {now_str}")
    print(f"==========================================")

    prompt = (
        f"現在の時刻は {now_str} です。"
        "システムの定期健康診断を行い、稼働状態が正常である旨の簡潔な監視レポートを作成してください。"
    )

    try:
        response = await agent.chat(prompt)
        result_text = await response.text()
        print(f"{result_text}\n")
    except Exception as e:
        print(f"エラー: タスク #{task_id} の実行中に例外が発生しました: {e}", file=sys.stderr)

# 4. メイン処理：定期ループの制御
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("--- 定期自動実行（Periodic Task Automation）デモを開始します ---")
    print("※ 5秒間隔で2回自動タスクを実行します。本番運用では cron やタイマーを設定します。\n")

    config = LocalAgentConfig(
        system_instructions=(
            "あなたはシステム運用を自動化する監視AIアシスタントです。"
            "指示された日時におけるシステムの監視レポートを簡潔にまとめて出力してください。"
        ),
        model="gemini-3.1-flash-lite"
    )

    # Agent の初期化と定期ループのシミュレーション
    async with Agent(config) as agent:
        # デモとして 5 秒間隔で 2 回自動発火させるループ
        for i in range(1, 3):
            await run_periodic_monitoring(agent, task_id=i)
            if i < 2:
                print("⏳ 次の定期チェックまで 5 秒待機します...")
                await asyncio.sleep(5)

    print("--- 定期自動実行デモが正常に終了しました ---")

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 periodic_agent_demo.py

--- 定期自動実行（Periodic Task Automation）デモを開始します ---
※ 5秒間隔で2回自動タスクを実行します。本番運用では cron やタイマーを設定します。

▶ 定期タスク #1 を開始します...

==========================================
 [定期タスク発火ログ] 実行時刻: 2026-08-25 06:51:54
==========================================
定期システム監視レポートを作成しました。現在のシステム稼働状態は以下の通りです。

システム稼働状態レポート
- 実行日時: 2026-08-25 06:51:54
- ステータス: 正常 (ALL SYSTEMS OPERATIONAL)
- CPU/メモリ使用率: 正常範囲内
- ネットワーク応答: 正常

システムは安定して稼働しています。

⏳ 次の定期チェックまで 5 秒待機します...

▶ 定期タスク #2 を開始します...

==========================================
 [定期タスク発火ログ] 実行時刻: 2026-08-25 06:51:59
==========================================
定期システム監視レポートを作成しました。現在のシステム稼働状態は以下の通りです。

システム稼働状態レポート
- 実行日時: 2026-08-25 06:51:59
- ステータス: 正常 (ALL SYSTEMS OPERATIONAL)
- データベース接続: 正常
- レスポンスタイム: 良好

引き続き安定した稼働を確認しました。

--- 定期自動実行デモが正常に終了しました ---