🔥
Pydantic-AI+FastA2Aで楽々AIエージェントをホスト!(悲報有)
FastMCPの記事に引き続きFastMCPを検証しました。
はじめに
AIエージェント、かっこよく作って動かしたいけど、なんだか思ったより複雑なコードが必要そう...
そんな風に感じていませんか?ご安心ください!
FastA2Aを使えば、AIエージェントの構築とホスティングが驚くほどシンプルになります。
ひとまずA2A試してみて!みたいな指示があったときなどに役立つかも...?
実装方法
雲行き怪しそうなので手抜き気味に...
エージェントサーバー
OPENAI_API_KEYなどの環境変数を.envに設定してください。
from pydantic_ai import Agent
from dotenv import load_dotenv
load_dotenv(".env")
agent = Agent('openai:gpt-4.1-mini',
instructions='句点の代わりに♦,♠,♣,♥のどれかをつけてください')
app = agent.to_a2a()
クライアント
タスク送信
コード
import asyncio
import json
from pprint import pprint
from fasta2a.client import A2AClient, Message
from fasta2a.schema import TextPart
client = A2AClient(base_url="http://localhost:8010")
message = Message(
role="user",
parts=[TextPart(
type="text",text="簡潔に自己紹介してください。" # メッセージを入力してください。
)])
response_submitted = await client.send_task(message=message)
pprint(response_submitted)
返値
{'id': '3f7f467d-8fd4-4dbe-991f-f92700094782',
'jsonrpc': '2.0',
'result': {'history': [{'parts': [{'text': '簡潔に自己紹介してください。', 'type': 'text'}],
'role': 'user'}],
'id': '43f565a0-0dd2-4b02-af13-ec52993b7691',
'session_id': 'dc9cca98-ae06-4f7e-8220-f5c20ca78d52',
'status': {'state': 'submitted',
'timestamp': '***'}}}
タスク取得
未完了であればworkingstatusが返ってきます。
コード
response_completed = await client.get_task(response_submitted["result"]["id"])
pprint(response_completed)
返値
{'id': None,
'jsonrpc': '2.0',
'result': {'artifacts': [{'index': 0,
'name': 'result',
'parts': [{'text': 'こんにちは♦ 私はChatGPTです♠ '
'OpenAIによって開発されたAI言語モデルです♣ '
'さまざまな質問にお答えし、会話をサポートします♥',
'type': 'text'}]}],
'history': [{'parts': [{'text': '簡潔に自己紹介してください。', 'type': 'text'}],
'role': 'user'}],
'id': '43f565a0-0dd2-4b02-af13-ec52993b7691',
'session_id': 'dc9cca98-ae06-4f7e-8220-f5c20ca78d52',
'status': {'state': 'completed',
'timestamp': '***'}}}
まとめ
非常にFastなコードでA2Aが試せました!
A2A公式の読み解きは結構難しかったので、抽象化度が高いPydantic-AIとFastA2Aは良いライブラリだと思います。
AIエージェントハッカソン用プロジェクトでもマルチエージェント構築にA2Aを使おうかと思っていましたが、結構ハマりポイントがあったので手を出さないで良かったです笑
まだまだ発展途上といった感じですが、ここまでシンプルになるラッパーは他になさそうなので、メンテナンス続けてほしい...
Discussion