A2A サンプル: Travel Planner OpenRouter

これは A2A（Agent2Agent）プロトコルに準拠した Python 実装です。 Google の公式 a2a-python SDK と OpenAI Python SDK に基づいて実装された旅行アシスタントのデモです。 OpenAI モデルの仕様に準拠した旅行アシスタントで、旅行計画サービスを提供できます。

ソースコード

A2A Travel Planner OpenRouter

プロジェクトアーキテクチャ

このプロジェクトは、A2A プロトコルを使用して相互運用可能な旅行計画エージェントを構築する方法を示しており、以下の主要コンポーネントが含まれています：

Travel Planner Agent: OpenAI 互換インターフェースに基づくコア旅行アシスタントロジック
Agent Executor: エージェントロジックを A2A サーバーに橋渡しする A2A プロトコルアダプター
A2A Server: 標準化されたエージェント間通信インターフェースを提供する A2A プロトコル準拠サーバー
Loop Client: A2A サーバーとやり取りするためのテストクライアント

ワークフローシーケンス図

sequenceDiagram
    participant Client
    participant A2AServer
    participant RequestHandler
    participant Executor as TravelPlannerAgentExecutor
    participant Agent as TravelPlannerAgent
    participant LLM as OpenAI-Compatible LLM

    Client->>A2AServer: エージェントカードをリクエスト
    A2AServer->>Client: エージェントカードを返す（スキル、能力）

    Note over Client,A2AServer: ユーザーが旅行計画をクエリ

    Client->>A2AServer: message/sendStream（ストリーミングリクエスト）
    A2AServer->>RequestHandler: ストリーミングリクエストをルーティング
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop ストリーミングレスポンス処理
        LLM-->>Agent: ストリーミングコンテンツチャンクを返す
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: SSE イベントをプッシュ
        A2AServer-->>Client: ストリームコンテンツ更新
    end
    
    LLM-->>Agent: 最終レスポンス完了
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: 最終 TaskArtifactUpdateEvent
    RequestHandler-->>A2AServer: 最終 SSE イベント
    A2AServer-->>Client: ストリーミングレスポンス完了

メインワークフロー

エージェントカード取得: クライアントは最初に A2A サーバーからエージェントカードを取得し、エージェントの能力とスキルを理解します
ストリーミングリクエスト処理: クライアントはユーザークエリを含むストリーミングメッセージリクエストを送信します
エージェント実行: エージェントエグゼキューターがリクエストを処理し、旅行プランナーエージェントのコアロジックを呼び出します
LLM インタラクション: エージェントは OpenAI 互換 LLM とストリーミング会話を行います
リアルタイムレスポンス: Server-Sent Events（SSE）を介してクライアントにリアルタイムでレスポンスをストリーミングします

はじめに

環境変数を設定します：

サンプルファイルをコピーして、API 認証情報を設定してください。

cp env.example .env

実際の値で .env ファイルを編集します：

# 必須：AI モデルサービスの API キー
API_KEY=your_actual_api_key_here

# オプション：モデル名（デフォルト：google/gemini-2.0-flash-001）
MODEL_NAME=google/gemini-2.0-flash-001

# オプション：API サービスのベース URL
BASE_URL=https://openrouter.ai/api/v1

依存関係をインストールしてサーバーを起動します：
```
uv venv
source .venv/bin/activate

uv sync
uv run .
```
新しいターミナルでループクライアントを実行します：
```
source .venv/bin/activate
uv run loop_client.py
```

設定

アプリケーションは設定に環境変数を使用します：

API_KEY（必須）：AI モデルサービスの API キー
MODEL_NAME（オプション）：使用するモデル名（デフォルト："google/gemini-2.0-flash-001"）
BASE_URL（オプション）：API サービスのベース URL（デフォルト："https://openrouter.ai/api/v1"）

技術的特徴

現在の実装

✅ A2A プロトコル準拠: Agent2Agent プロトコル仕様に完全準拠
✅ ストリーミングレスポンス: リアルタイムストリーミングコンテンツ生成をサポート
✅ OpenAI 互換: 任意の OpenAI 互換 API インターフェースをサポート
✅ モジュラー設計: エージェントロジックとプロトコル適応の明確な分離
✅ 環境設定: 柔軟な環境変数設定

将来の拡張計画

タスク状態管理の拡張

Google A2A LangGraph サンプルに基づいて、以下の機能を追加する予定です：

🔄 タスクライフサイクル管理: 完全なタスク状態追跡の実装（送信済み → 作業中 → 完了/失敗）
🔄 マルチターン会話サポート: ユーザーの明確化が必要な複雑な旅行計画シナリオをサポートする input_required 状態の追加
🔄 タスク永続化: 長時間実行される計画タスクのタスク状態永続化の実装
🔄 拡張エラーハンドリング: より詳細なエラー状態と回復メカニズム
🔄 タスクキャンセル: 進行中のタスクのキャンセルをサポート

状態管理の例

# 将来の状態管理実装例
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # より多くの情報が必要かどうかを検出
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="より多くの情報が必要です：具体的な目的地、日付、予算範囲を提供してください"
            )
        
        # 複雑なマルチステップ計画を実行
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="詳細な旅行計画を作成中..."
        )