A2AのTaskライフサイクル入門｜進捗・ストリーミング・キャンセル設計

小型サーバーと無字の計時器、保護された停止スイッチを置いたタスク管理机 A2A
カテゴリー
A2A
公開日
2026.09.09

はじめに

調査エージェントやコード生成は数分かかることがあります。A2Aでは処理をTaskとして表し、状態更新や成果物をイベントとして扱えます。ストリーミングを有効にすれば、クライアントはSSE経由で途中経過を受信できます。

この記事の目的はAPIを暗記することではなく、再接続やキャンセルでも壊れない状態機械を設計することです。

この記事は、記事14の最小A2Aサーバーを理解した人、またはPythonの非同期処理を少し触った人を対象にしています。読み終えると、長時間処理の状態、途中経過、切断後の復元、協調的キャンセルを別々に設計できるようになります。

具体例で考える

10個の文書を調査するエージェントを想像してください。依頼を受けた直後はsubmitted、調査中はworking、結果を保存できたらcompletedです。途中で質問が必要ならinput-required、回復できないエラーならfailed、利用者が停止を求めた場合はcanceledへ進みます。

  • Task：開始から完了まで追跡する1件の仕事
  • SSE：サーバーからブラウザへイベントを順次届けるHTTP方式
  • terminal state：完了、失敗、キャンセルなど、それ以上処理を進めない最終状態
  • 状態機械：どの状態からどの状態へ移れるかをルール化したもの

扱う内容と必要環境

この記事では、キャンセル信号を保持するWorkRegistryと、処理の区切りで停止を確認するanalyze()を作ります。その後、A2A Python SDKのTaskUpdaterへ状態更新を渡す考え方と、SSEが切れてもTask Storeから結果を復元する設計を確認します。

  • Python 3.12
  • a2a-sdk[http-server]==1.1.2
  • asyncioのasync／awaitを読める程度の基礎知識
  • SSE検証を行う場合はStarlette、Uvicorn、HTTPクライアント

この記事の掲載コード自体は標準ライブラリだけで動く設計部品です。A2Aの通信形式を含む完全なサーバーではないため、プロトコル接続の最小構成は前の記事を先に確認してください。

状態遷移を先に決める

代表的な流れは次の通りです。

submitted -> working -> completed
                    -> failed
                    -> input-required -> working
                    -> canceled

terminal stateへ到達したTaskを再びworkingへ戻さない、という不変条件をアプリ側でも守ります。状態更新と成果物保存を別々に行う場合、片方だけ成功しないようトランザクションまたは再実行可能な処理にします。

Executorで進捗を通知する考え方

Python SDKではExecutorがEventQueueへ状態イベントやartifactを追加します。実際の処理関数にはasyncio.Eventなどのキャンセル信号を渡します。

import asyncio
from collections.abc import Awaitable, Callable

class WorkRegistry:
    def __init__(self):
        self._cancel: dict[str, asyncio.Event] = {}

    def start(self, task_id: str) -> asyncio.Event:
        signal = asyncio.Event()
        self._cancel[task_id] = signal
        return signal

    def cancel(self, task_id: str) -> bool:
        signal = self._cancel.get(task_id)
        if signal is None:
            return False
        signal.set()
        return True

async def analyze(
    items: list[str],
    signal: asyncio.Event,
    analyze_one: Callable[[str], Awaitable[str]],
) -> list[str]:
    results = []
    for item in items:
        if signal.is_set():
            raise asyncio.CancelledError()
        results.append(await analyze_one(item))
    return results

analyze_oneは文書1件を処理する非同期関数です。外から渡す形にすると、テストでは待ち時間の短い模擬関数へ交換できます。

SDK 1.1.2ではTaskUpdaterをEventQueueへ接続し、submit()、start_work()、complete()、cancel()などで状態イベントを送れます。TaskUpdaterはterminal stateへ到達した後の状態更新をRuntimeErrorで拒否します。ビジネス処理をSDK型から分離しておくと、状態遷移と実処理を別々にテストできます。

Streamingは履歴保存の代わりではない

SSE接続は切れるものとして扱います。ネットワーク断の間に完了しても、Task Storeから現在状態と成果物を取得できるようにします。イベントを再購読できる場合も、イベントIDと保持期限を決めます。

進捗率は正確に計算できる処理だけに出します。不明なのに90%で長時間止まる表示より、「12件中4件完了」のような単位が明確な情報が有用です。

input-required

処理に追加情報が必要ならTaskをinput-requiredへ移します。質問文だけでなく、何を回答すべきか、入力形式、機密情報を送ってよい経路かを示します。

回答受領時には以下を再検査します。

  • Taskの所有者と回答者が一致する
  • Taskが現在input-requiredである
  • 回答がスキーマに一致する
  • 有効期限を超えていない
  • 同じ回答を二重適用しない

キャンセル

キャンセル要求は「停止を試みる」操作です。外部API呼び出しやモデル推論が即時停止できるとは限りません。キャンセルを受けたら新しい副作用を開始せず、安全な区切りで停止します。

completedとcancel要求が競合する場合の優先順位を決め、compare-and-setなどで一度だけterminal stateを確定します。

テスト項目

  • 正常完了まで状態が順方向に進む
  • 接続を途中で切ってもTaskを再取得できる
  • input-required後の回答で処理を再開できる
  • working中のキャンセルで新しい副作用が止まる
  • completed直前のキャンセル競合が一貫した結果になる
  • 同じmessage/task要求の再送で成果物が重複しない
  • 別ユーザーのTaskを取得・キャンセルできない

最初の検証では、各項目の処理前に短い待ち時間を入れ、「10件中3件完了」のような進捗を送る模擬処理を使います。SSE接続を途中で閉じ、再接続後にTask Storeから同じ進捗または最終結果を取得できれば、ストリームだけに状態を依存していないと確認できます。

この設計を確認するため、StarletteとUvicornで補助HTTPサーバーを起動し、実際のSSE接続を2イベント受信後に切断しました。切断後もバックグラウンド処理は継続し、通常のGETでcompletedと成果物を取得できました。再接続したSSEでも完了状態を受信できています。別タスクでは処理中のキャンセルがcanceledとなり、完了済みタスクへのキャンセルはHTTP 409で拒否されました。

この補助サーバーは、切断・復元・協調的キャンセルという設計を検証するためのものです。A2AのJSON-RPCメソッドやイベント形式を再現した実装ではありません。A2A SDK側については、前節のTaskUpdaterによる状態イベントとterminal state後の更新拒否を別に検証しています。

よくあるトラブル

  • 切断すると進捗が分からなくなる：現在状態をSSE接続内だけに保持している
  • キャンセルしたのに次の処理が始まる：各作業の境界でキャンセル信号を確認していない
  • completedからworkingへ戻る：最終状態を更新できない条件を保存層にも設ける
  • 同じ成果物が複数できる：要求IDを使った重複防止がない
  • 進捗率が長時間止まる：割合ではなく「全10件中3件」のような実数表示を検討する

数秒で終わる単純処理にTaskやSSEを導入すると複雑さが増えます。長時間処理、再接続、追加質問のいずれも必要ない場合は通常の応答が適しています。

関連記事

まとめ

A2Aのストリーミングは利用者体験を改善しますが、信頼性の中心はTask Storeと状態遷移です。terminal stateの不変条件、冪等性、再接続、認可、協調的キャンセルを先に設計すると、長時間エージェントを安定して運用できます。

参考リンク

コメント

タイトルとURLをコピーしました