A2Aプロトコルでマルチエージェントアプリを作ってみた｜ライター×校閲者ワークフロー

# pyproject.toml
[project]
name = "a2a-demo"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "a2a-sdk[http-server]>=0.3.0",
    "openai>=1.0.0",
    "httpx>=0.28.0",
    "uvicorn>=0.34.0",
    "python-dotenv>=1.0.0",
]

AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_API_KEY=your-api-key
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4o
AZURE_OPENAI_API_VERSION=2024-02-15-preview

import os
from typing import Literal
from openai import AzureOpenAI
from pydantic import BaseModel


class WriterResponse(BaseModel):
    """ライターエージェントのレスポンス形式"""
    status: Literal["input_required", "completed", "error"] = "input_required"
    content: str


class WriterAgent:
    """ブログ記事の初稿を書くエージェント"""

    SYSTEM_PROMPT = """あなたは技術ブログのライターです。
指定されたテーマについて、読みやすく分かりやすいブログ記事の初稿を書いてください。

## 記事の構成
1. 導入（テーマの背景と記事の目的）
2. 本文（技術的な説明、具体例）
3. まとめ（要点の整理）

## 注意事項
- 技術的に正確な内容を心がける
- 初心者にも分かりやすい表現を使う
- 適切な見出しを付ける
- 500〜800文字程度で簡潔にまとめる"""

    def __init__(self):
        self.client = AzureOpenAI(
            azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
            api_key=os.environ["AZURE_OPENAI_API_KEY"],
            api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2024-02-15-preview"),
        )
        self.deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]

    async def invoke(self, theme: str) -> WriterResponse:
        """テーマに基づいてブログ記事の初稿を生成する"""
        if not theme or theme.strip() == "":
            return WriterResponse(
                status="input_required",
                content="記事のテーマを指定してください。",
            )

        response = self.client.chat.completions.create(
            model=self.deployment_name,
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": f"テーマ: {theme}"},
            ],
            temperature=0.7,
            max_tokens=2000,
        )

        return WriterResponse(
            status="completed",
            content=response.choices[0].message.content,
        )

"""Writer Agent Executor - A2Aプロトコル用のエグゼキューター"""

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import Part, TaskState, TextPart
from a2a.utils import new_agent_text_message, new_task

from writer_agent.agent import WriterAgent


class WriterAgentExecutor(AgentExecutor):
    """ライターエージェントのA2Aエグゼキューター"""

    def __init__(self):
        self.agent = WriterAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        query = context.get_user_input()
        task = context.current_task

        if not task:
            task = new_task(context.message)
            await event_queue.enqueue_event(task)

        updater = TaskUpdater(event_queue, task.id, task.context_id)

        # 処理中ステータスを送信
        await updater.update_status(
            TaskState.working,
            new_agent_text_message("記事を執筆中です...", task.context_id, task.id),
        )

        # ライターエージェントを実行
        result = await self.agent.invoke(query)

        if result.status == "completed":
            # 成果物として記事を追加
            await updater.add_artifact(
                [Part(root=TextPart(text=result.content))],
                name="draft_article",
            )
            await updater.complete()
        else:
            await updater.update_status(
                TaskState.input_required,
                new_agent_text_message(result.content, task.context_id, task.id),
                final=True,
            )

"""Writer Agent Server - ライターエージェントのA2Aサーバー"""

import uvicorn
from dotenv import load_dotenv
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentSkill

from writer_agent.agent import WriterAgent
from writer_agent.agent_executor import WriterAgentExecutor

load_dotenv()


def main():
    skill = AgentSkill(
        id="write_article",
        name="ブログ記事ライター",
        description="指定されたテーマでブログ記事の初稿を執筆します",
        tags=["blog", "writing", "article"],
        examples=[
            "Pythonの非同期処理について記事を書いて",
            "Dockerの基本概念を解説する記事を書いて",
        ],
    )

    agent_card = AgentCard(
        name="Writer Agent",
        description="技術ブログの初稿を執筆するエージェント",
        url="http://localhost:10001/",
        version="1.0.0",
        default_input_modes=WriterAgent.SUPPORTED_CONTENT_TYPES,
        default_output_modes=WriterAgent.SUPPORTED_CONTENT_TYPES,
        capabilities=AgentCapabilities(streaming=False),
        skills=[skill],
    )

    request_handler = DefaultRequestHandler(
        agent_executor=WriterAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    server = A2AStarletteApplication(
        agent_card=agent_card,
        http_handler=request_handler,
    )

    print("Starting Writer Agent on http://localhost:10001")
    uvicorn.run(server.build(), host="0.0.0.0", port=10001)


if __name__ == "__main__":
    main()

class ReviewerAgent:
    """ブログ記事を校閲するエージェント"""

    SYSTEM_PROMPT = """あなたは技術ブログの校閲者です。
与えられた記事を以下の観点でチェックし、修正案を提示してください。

## チェック項目
1. **誤字脱字**: タイポや文法ミスがないか
2. **技術的正確性**: 技術的な説明が正確か
3. **読みやすさ**: 文章が分かりやすいか
4. **トーン&マナー**: 技術ブログとして適切なトーンか

## 出力形式（JSON）
{
    "issues": ["問題点1", "問題点2", ...],
    "suggestions": ["改善提案1", "改善提案2", ...],
    "corrected_article": "修正後の記事全文"
}"""

"""Orchestrator - ライターと校閲者エージェントを連携させるクライアント"""

import asyncio
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import MessageSendParams, SendMessageRequest


async def get_agent_client(base_url: str) -> tuple[A2AClient, httpx.AsyncClient]:
    """エージェントのA2Aクライアントを取得する"""
    httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(120.0))
    resolver = A2ACardResolver(httpx_client=httpx_client, base_url=base_url)
    agent_card = await resolver.get_agent_card()
    client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
    return client, httpx_client


async def send_message(client: A2AClient, message: str) -> str:
    """エージェントにメッセージを送信して結果を取得する"""
    payload = {
        "message": {
            "role": "user",
            "parts": [{"kind": "text", "text": message}],
            "messageId": uuid4().hex,
        },
    }
    request = SendMessageRequest(
        id=str(uuid4()),
        params=MessageSendParams(**payload)
    )
    response = await client.send_message(request)
    return extract_result_text(response)


async def run_editorial_workflow(theme: str):
    """編集ワークフローを実行する"""
    print(f"テーマ: {theme}")

    # Step 1: ライターエージェントに接続
    writer_client, writer_http = await get_agent_client("http://localhost:10001")

    # Step 2: ライターに記事を依頼
    draft_article = await send_message(writer_client, theme)
    print("--- 初稿 ---")
    print(draft_article)

    # Step 3: 校閲者エージェントに接続
    reviewer_client, reviewer_http = await get_agent_client("http://localhost:10002")

    # Step 4: 校閲者に記事をレビュー依頼
    review_result = await send_message(reviewer_client, draft_article)
    print("--- 校閲結果 ---")
    print(review_result)

    # クリーンアップ
    await writer_http.aclose()
    await reviewer_http.aclose()

# ターミナル1: ライターエージェント
cd demo
PYTHONPATH=. uv run python -m writer_agent

# ターミナル2: 校閲者エージェント
cd demo
PYTHONPATH=. uv run python -m reviewer_agent

uv run python orchestrator/main.py "grpcとはなにか"

============================================================
A2A Protocol Demo: ライター × 校閲者 ワークフロー
============================================================

テーマ: grpcとはなにか

[Step 1] ライターエージェントに接続中...
✓ ライターエージェントに接続しました

[Step 2] 記事の初稿を依頼中...
✓ 初稿を受け取りました

--- 初稿（抜粋） ---
# gRPCとはなにか？初心者でもわかる解説

## 導入：gRPCの背景とこの記事の目的

近年、マイクロサービスやクラウドアプリケーションの普及に伴い、
サービス間の通信技術がますます重要になっています。
そんな中で注目されているのが「gRPC」という通信プロトコルです。
gRPCはGoogleが開発した高速で効率的なRPCフレームワークで、
多くの企業や開発者に支持されています。

### 双方向ストリーミング

gRPCは一方向または双方向のストリーミング通信もサポート。
これにより、リアルタイムのデータ送受信や大容量データの分割送信が可能です。
--- 初稿ここまで ---

[Step 3] 校閲者エージェントに接続中...
✓ 校閲者エージェントに接続しました

[Step 4] 記事の校閲を依頼中...
✓ 校閲結果を受け取りました

--- 校閲結果 ---
【発見した問題点】
  1. gRPCはGoogleが開発したオープンソースであることを明記したほうが良い
  2. 「gRPCは一方向または双方向のストリーミング通信もサポート」の文が
     やや簡潔すぎて意味が取りづらい
  3. 対応言語をもう少し具体的に挙げると親切
  4. 全体的に文章は読みやすいが、技術用語の説明がもう少し丁寧だと初心者に親切

【改善提案】
  1. gRPCがオープンソースとして広く利用されていることを明示する
  2. ストリーミング通信の種類（クライアント/サーバー/双方向）を
     具体的に説明し、用途例を簡単に示す
  3. C#, Node.js, PHPなどの対応言語も補足する
  4. 専門用語には簡単な注釈を加え、初心者が理解しやすいようにする

【修正後の記事（抜粋）】
### ストリーミング通信

gRPCは以下の3種類のストリーミング通信をサポートしています。

- クライアントストリーミング：クライアントからサーバーへ連続したデータを送信
- サーバーストリーミング：サーバーからクライアントへ連続したデータを送信
- 双方向ストリーミング：クライアントとサーバーが同時にデータを送受信

これにより、リアルタイムのデータ送受信や大容量データの分割送信が可能になります。
--- 校閲結果ここまで ---

============================================================
ワークフロー完了！
============================================================

# ライターエージェント
curl http://localhost:10001/.well-known/agent.json | jq

# 校閲者エージェント
curl http://localhost:10002/.well-known/agent.json | jq

{
  "name": "Writer Agent",
  "description": "技術ブログの初稿を執筆するエージェント",
  "url": "http://localhost:10001/",
  "version": "1.0.0",
  "skills": [
    {
      "id": "write_article",
      "name": "ブログ記事ライター",
      "description": "指定されたテーマでブログ記事の初稿を執筆します"
    }
  ],
  "capabilities": {
    "streaming": false
  }
}