再構築ではなく改修：レガシーエンタープライズサービスを変革するエージェント型オーバーレイ
本文の状態
日本語全文を表示中
詳細モードで約16分の本文を読めます。
同じ出来事の情報源
この情報源を基点に整理
AWS Machine Learning Blog
AWS は、安定した REST API やマイクロサービスを基盤としつつ、エージェント間通信（A2A）に対応させるための「エージェント型オーバーレイ」手法を提案している。これにより、既存システムを再構築せずに AI エージェント機能を追加できる。
Continue in AI NEW LAB
このニュースを、実務の判断につなげる
AI NEW LABで、試したことや先に確認したい条件を共有できます。まずはログインなしで読めます。
AI NEW LABで論点を見るSource Article
元記事を日本語で読む
本文に関係しない購読案内、埋め込み通知、サイト内プロモーションは除いています。
本投稿に示された意見は著者の見解であり、シスコの公式見解ではありません。
エンタープライズアーキテクチャは長年、REST APIs とマイクロサービスを中心に構築されてきました。これらのシステムは安定しており、十分にテストされ、生産環境に深く組み込まれています。これらは、構造化されたメッセージを通じて自律的に協力し、推論し、調整する新興の標準である Agent-to-Agent（A2A）通信のために設計されたものではありません。共通のエージェントプロトコルが存在しない状況ではこのアプローチは機能していましたが、その結果、多くの既存エージェントが新興の A2A フレームワークの外側に置かれたままとなっています。今日における課題は、従来のサービスに A2A を追加するだけではありません。REST ベースのエージェントを標準化されたエージェント間通信の世界へと取り込む必要があります。
AWS と著者とのこの技術的協力において、私たちは実用的な解決策である「エージェント型オーバーレイ」を提案します。エージェント型オーバーレイは、従来の REST ベースのサービスを A2A（Agent-to-Agent）相互作用に参加できるエージェントに変換する薄いラッパー層です。また、これらは Model Context Protocol (MCP) と互換性のあるツールとして REST API を公開します。これらを組み合わせることで、企業はビジネスロジックの書き換えやコードの重複、並行インフラの運用を行うことなく、既存の REST サービスに A2A 機能を追加できます。これにより、既存サービスをエージェントとして再利用することで、インフラにおけるエージェントのスプレッド（拡散）を削減します。私たちは、エージェント型オーバーレイの構築方法を示す参照アーキテクチャとサンプルコードを提供しています。
バックグラウンド：REST と A2A の比較
REST API は、決定論的なクライアント・サーバー統合のために設計されています。クライアントは定義されたエンドポイントを呼び出し、パラメータを渡して、通常は HTTP セマンティクスによって制御されるステートレスなリクエスト・レスポンスフローで予測可能な応答を受信します。これにより、明確な契約、強力な互換性、運用の簡素さを備えたビジネス機能（作成、読み取り、更新、削除など）を公開する際に REST が極めて優れていることがわかります。
A2A は自律型エージェント間の相互運用性を目的として設計されています。エージェントはメタデータ（エージェントカードなど）を通じて互いを見つけ、機能を交渉し、構造化されたメッセージ（通常は JSON-RPC を介して）を交換することで、多段階のタスクを調整します。REST が安定したサービスインターフェースと直接実行に最適化されているのに対し、A2A は推論駆動型の調整、タスク指向のメッセージング、およびエージェント間の協働に最適化されています。その結果、孤立したエンドポイントを呼び出すのではなく、複数のサービスにわたって計画し、委任し、アクションを組み合わせることのできるシステムが実現されます。
エージェントシステムへの移行における課題
REST API とエージェントシステムは直交するパラダイムに基づいているため、既存のサービスを標準化されたエージェント通信へ移行するのは困難です。しかし企業は、大規模な再構築なしに両方を利用する必要があります。A2A を通じた新しいエージェント通信が企業システム向けの調整モデルを導入しているものの、既存の企業システムと並行してエージェントインフラストラクチャを展開・運用する必要性により、普及は遅れています。この並列運用は運用の複雑さとコストを増大させ、AI の効果的な導入に対する障壁となっています。
A2A が標準化される以前、企業はエージェントを REST ベースまたは独自のプロトコルに基づくサービスとして一般的に展開していました。これらは、リクエスト・レスポンスエンドポイントにエージェント固有のロジックを組み込んだ従来の API として扱われていました。その結果、現在存在する多くのエージェントは A2A ネイティブではなく、これが新たな移行課題を生んでいます。すなわち、コアロジックを書き換えることなく、標準化された A2A プロトコルを使用してこれらのエージェントを相互運用可能にすることです。
上記の図は、REST ベースのアプリケーションを示しています。REST API スack はモノリシックな構造であるか、あるいは分散されたエンドポイントの集合である可能性があります。REST API コントローラーは、必ずしもアプリケーション外部へのリクエストを委譲する明示的なブローカーである必要はありません。これはフレームワーク自体の一部となり得ます。例えば、Flask アプリケーションでは、フレームワークがコントローラーの抽象化を標準で提供しており、@app.route() または Flask の RESTful 拡張機能 を使用した際に通常見られるような構造です。ここで重要なのは、一連のエンドポイントを用いて REST スack を捉えるという考え方です。
ソリューションのアプローチ
このセクションでは、レガシーなエンタープライズシステムにエージェント機能を付与するために採用できるさまざまなアプローチについて議論します。これらを、アジェンティック・オーバーレイ（注：エージェント機能のラッパー層）を使用するアプローチと比較します。
REST と A2A のスタックを別々に維持する: 一つの手法は、同じ機能を公開するための並行した二つの方法を開発・維持することです。これには以下のような意味が含まれます:
- 二組のエンドポイント: /api/v2/... および /a2a/....
- 認証（auth）、検証（validation）、エラーマッピングの二重実装（注意深く再利用しない限り）。
- 二つのデプロイパイプライン（ビルド、テスト、リリース、ロールバック）。
- 可観測性（observability）作業の倍増: 両方のパスに対するログ、メトリクス、トレーシング。
- 不整合のリスクが高まる。A2A は REST で実行される同じ操作に対して異なる出力を返す可能性がある。
- 時間とともにコストと運用複雑性が増大する。
分離されたスタックだが、ビジネスロジックは共有: 既存のエンドポイントをリファクタリングするとは、現在の REST API コード構造（および場合によっては動作）を変更し、それを新しいインターフェースである A2A で再利用可能にすることです。REST エンドポイントを現状のまま放置するのではなく、それらを再編成します。通常はビジネスロジックを共有サービスへ抽出し、コントローラーやハンドラーを更新してそのサービスを呼び出すようにします。外部の REST パスが同じまま残ったとしても、リファクタリングによって機能低下（regressions）、動作のドリフト、そして膨大なテスト負荷が生じる可能性があります。
コアとなる考え方：エージェント型オーバーレイ
*エージェント型オーバーレイ*とは、REST ベースのサービスが A2A（Agent-to-Agent）通信に参加できるようにする薄いラッパー層です。このオーバーレイは以下の機能を提供します。
- エージェントメッセージを REST ペイロードに変換し、その逆も行うこと。
- REST エンドポイントをエージェントタスクまたはツールとして公開すること。
最も重要なのは、A2A は新しい API ではないということです。既存の API に対する新しいインターフェースに過ぎません。基盤となる REST サービス自体は変更されません。
アプリケーション内へのエージェント型オーバーレイの追加
このアプローチでは、以下の図に示すように、/api/v2/... と /a2a/... の 2 つセットのエンドポイント（REST vs. A2A）が存在しますが、ビルド、テスト、リリース、ロールバックのためのデプロイパイプラインは単一です。このパターンにより、コアとなるビジネスロジックを書き換えることなく、従来の REST API エンドポイントをエージェント型エンドポイントに変換できます。サービスのデプロイプロセス自体も変更されません。同じホストとポートに対して新しいルートを追加するだけで済みますが、システムには増加したトラフィックを処理するためにスケールが必要になる場合があります。
ルーティングにはエージェントスキルそのものを適用することも可能です。MCP サーバーを使用して外部サービスを呼び出すこともできますが、エージェントスキルは、API を MCP サーバーのスキルとしてインポートすることなく、エージェントスコープ内でリクエストを直接ルーティングできます。必要な場合、個別の MCP サーバーを用意することなく、あらゆるエンドポイントをエージェントスキルとして公開できます。
このアプローチは、既存のサービスをエージェントとして再利用することで、インフラにおけるエージェントのスプレッドを削減します。このデザインパターンは、機能スコープが限定的で、REST ベースとエージェント機能の両方を必要とするスーパーバイザーエージェント（例：意図分類やルーティング）に対して特に効果的です。
エージェント型オーバーレイの実装例
概念実証として、このセクションでは、Flask を使用した従来の REST ベースの計算機サービスを、オーバーレイを使用してエージェントシステムへ移行する方法を示します。オーバーレイには、一般的なエージェントカード、エージェントメッセージエンドポイント、機能、スキル、ヘルスチェックなど、標準的な A2A コンポーネント（またはルート）を追加します。また、エージェント型メッセージを REST API メッセージに変換し、その後エージェントから REST 呼び出しを実行するメッセージ変換デザインパターンも導入します。A2A メッセージ翻訳ワークフローは以下の通りです：
- JSON-RPC 2.0 リクエストを受信する。
- A2A タスクを REST エンドポイントにマッピングする。
- 認証ヘッダーを転送する。
- 内部で REST エンドポイントを呼び出す。
- REST レスポンスを JSON-RPC フォーマットに変換する。
ステップ 0: リクエスト・レスポンス形式
このセクションでは、以下のセクションで示す計算機例を用いて、REST プロトコルと A2A プロトコルのリクエスト・レスポンス形式を比較します。
REST と A2A の入力リクエストの比較：
REST
A2A
翻訳全文
A2A リクエストトランスレーター - カルキュレータの例。
このモジュールは、既存のカルキュレータ REST API 上で A2A (JSON-RPC 2.0) の互換性を提供するために、リクエストトランスレーターパターンを実装しています。
A2A Spec 0.3 準拠:
- エージェントカード: GET /.well-known/agent-card.json
- JSON-RPC エンドポイント: POST /a2a
- メソッド: SendMessage, SendStreamingMessage
- メッセージ形式: { "message": { "parts": [{ "kind": "data", "data": {...} }] } }
デフォルトの A2A API URL (必要に応じて build_agent_card(url) で上書き)
A2A_API_URL = "http://localhost:5000/a2a"
EXECUTE_TIMEOUT_SECONDS = 30
JSON ファイルからスキルを読み込む
_SKILLS_FILE = Path(__file__).parent / "skills.json"
_SKILLS_CACHE: Optional[List[Dict[str, Any]]] = None
def _load_skills() -> List[Dict[str, Any]]:
"""
skills.json ファイルからスキルを読み込みます。
スキルは初回読み込み後にキャッシュされ、繰り返しファイルを読み込むのを防ぎます。
"""
global _SKILLS_CACHE
if _SKILLS_CACHE is not None:
return _SKILLS_CACHE
try:
with open(_SKILLS_FILE) as f:
_SKILLS_CACHE = json.load(f)
return _SKILLS_CACHE
except FileNotFoundError:
logger.error(f"スキルファイルが見つかりません: {_SKILLS_FILE}")
return []
except json.JSONDecodeError as e:
logger.error(f"スキルファイル内の JSON が無効です: {e}")
return []
def build_agent_card(api_url: Optional[str] = None) -> Dict[str, Any]:
"""設定可能な API URL を用いて、A2A エージェントカード (v0.3.0 フォーマット) を構築する。"""
if api_url is None:
api_url = A2A_API_URL
return {
"name": "Calculator Agent",
"description": "基本的な四則演算をサポートするシンプルな電卓エージェント",
"supportedInterfaces": [
{"url": api_url, "protocolBinding": "JSONRPC", "protocolVersion": "0.3"},
],
"provider": {
"organization": "Example Organization",
"url": "",
},
"version": "1.0.0",
"capabilities": {
"streaming": False,
"pushNotifications": False,
"extendedAgentCard": False,
},
"defaultInputModes": ["text/plain", "application/json"],
"defaultOutputModes": ["text/plain", "application/json"],
"skills": _load_skills(),
}
エージェントカードは動的に構築される
AGENT_CARD = build_agent_card()
ステップ 2: 内部 REST コールの実装
def invoke_rest_endpoint(
endpoint: str,
json_data: Optional[Dict] = None,
http_method: str = "POST"
) -> Tuple[Optional[Dict], int]:
"""
実際の HTTP リクエストを介して、内部 REST エンドポイントを呼び出す。
実行中のサーバーを呼び出すために requests.post/get を使用する。これにより、ミドルウェア、デコレータ、およびヘッダーが適切に処理されることを保証する。
引数:
endpoint: REST エンドポイントパス（例："/api/v1/calculate"）
json_data: POST/PUT リクエストのボディ
http_method: HTTP メソッド（GET、POST など）
戻り値:
(レスポンスデータ，ステータスコード) のタプル
"""
try:
base_url = request.host_url.rstrip("/")
url = f"{base_url}{endpoint}"
headers = {"Content-Type": "application/json"}
auth_header = request.headers.get("Authorization")
if auth_header:
headers["Authorization"] = auth_header
logger.info(f"Adapter: REST への委譲を実行中 {http_method} {url}")
if http_method.upper() == "POST":
response = http_requests.post(
url, json=json_data, headers=headers,
timeout=EXECUTE_TIMEOUT_SECONDS
)
elif http_method.upper() == "GET":
response = http_requests.get(
url, headers=headers,
timeout=EXECUTE_TIMEOUT_SECONDS
)
else:
response = http_requests.request(
http_method, url, json=json_data, headers=headers,
timeout=EXECUTE_TIMEOUT_SECONDS
)
logger.info(f"Adapter: REST から {response.status_code} を返却")
return response.json(), response.status_code
except http_requests.RequestException as e:
logger.error(f"Adapter: REST エンドポイント呼び出しエラー：{e}", exc_info=True)
return {"error": "Internal server error"}, 500
ステップ 3: メッセージペイロードの抽出実装 (A2A Spec 0.3)
def extract_message_payload(message: Dict) -> Optional[Dict]:
"""
A2A メッセージパーツからペイロードを抽出します（Spec 0.3 フォーマット）。
期待されるフォーマット:
{
"message": {
"parts": [{"kind": "data", "data": {"operation": "add", "operands": [5, 3]}}]
}
}
戻り値:
抽出されたデータペイロードを辞書型で返す。見つからない場合は None を返す。
"""
try:
parts = message.get("parts", [])
for part in parts:
if isinstance(part, dict) and part.get("kind") == "data":
return part.get("data")
return None
except Exception as e:
logger.error(f"メッセージペイロードの抽出エラー: {e}")
return None
def build_a2a_message(message_id: str, context_id: str, content: Any) -> Dict:
"""
A2A 準拠のメッセージオブジェクトを構築します（A2A Spec 0.3）。
レスポンスフォーマット:
{
"messageId": "uuid",
"contextId": "uuid",
"role": "agent",
"parts": [{"kind": "data", "data": {...}}],
"kind": "message",
"metadata": {}
}
"""
if isinstance(content, dict):
parts = [{"kind": "data", "data": content}]
else:
parts = [{"kind": "text", "text": str(content)}]
return {
"messageId": message_id,
"contextId": context_id,
"role": "agent",
"parts": parts,
"kind": "message",
"metadata": {}
}
ストリーミングのためのサーバー送信イベント (SSE) に関する注記:
直前の extract_message_payload() 関数は、SendMessage と SendStreamingMessage の両方で同じように動作します。
計算機のような即座に完了する操作では、どちらのメソッドも単一の結果を返します。一方、レポート生成やデータ分析などの長時間実行される操作においては、SSE ストリーミングによりサーバーが逐次的な更新を送信することが可能になります。
ステップ 4: JSON-RPC レスポンスビルダーの実装
JSON-RPC 2.0 仕様に準拠して定義されたエラーコード
class JsonRpcError:
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
def jsonrpc_error(code: int, message: str, data: Any = None, request_id: Any = None) -> Dict:
"""JSON-RPC 2.0 エラーレスポンスを構築する。"""
response = {
"jsonrpc": "2.0",
"error": {"code": code, "message": message},
"id": request_id
}
if data is not None:
response["error"]["data"] = data
return response
def jsonrpc_success(result: Any, request_id: Any = None) -> Dict:
"""JSON-RPC 2.0 成功レスポンスを構築する。"""
return {
"jsonrpc": "2.0",
"result": result,
"id": request_id
}
翻訳全文
ステップ 5: SendMessage — A2A から REST への委譲
def handle_send_message(data: Dict) -> Tuple[Any, int]:
"""
SendMessage の処理を行う。単純なパススルーとして /api/v1/calculate へ転送する。
このアダプタは、ペイロードの内容に基づいて検査やルーティングを行わない。
"""
request_id = data.get("id")
params = data.get("params", {})
message = params.get("message", {})
context_id = message.get("contextId") or generate_id()
message_id = generate_id()
payload = extract_message_payload(message)
if not payload:
return jsonify(jsonrpc_error(
JsonRpcError.INVALID_PARAMS,
"無効なパラメータ：message.parts にデータが見つかりません。",
request_id=r
原文を表示
The opinions expressed in this post are the authors’ views and not those of Cisco.
Enterprise architectures have long been centered on REST APIs and microservices. These systems are stable, well-tested, and deeply embedded in production environments. They weren’t designed for Agent-to-Agent (A2A) communication, the emerging standard for autonomous agents that collaborate, reason, and coordinate through structured messaging. That worked in the absence of a common agent protocol, but it means many existing agents now sit outside the emerging A2A framework. The challenge today is no longer only adding A2A to traditional services. You also need to bring these REST-based agents into a standardized agent-to-agent world.
In this technical collaboration between AWS and the authors, we present a pragmatic solution: *agentic overlays*. Agentic overlays are thin wrapper layers that transform traditional REST-based services into agents capable of participating in A2A interactions. They also expose REST APIs as tools compatible with the Model Context Protocol (MCP). Together, they let enterprises add A2A capabilities to existing REST services without rewriting business logic, without duplicating code, and without running parallel infrastructures. This reduces agent sprawl in the infrastructure by reusing existing services as agents. We provide reference architectures and sample code that show how to build agentic overlays.
Background: REST vs. A2A
REST APIs are designed for deterministic, client-server integration. A client calls a well-defined endpoint, passes parameters, and receives a predictable response, typically in a stateless request-response flow governed by HTTP semantics. This makes REST excellent for exposing business capabilities (such as create, read, update, and delete) with clear contracts, strong compatibility, and operational simplicity.
A2A is designed for interoperability between autonomous agents. Agents discover one another through metadata (such as an agent card), negotiate capabilities, and exchange structured messages (often through JSON-RPC) to coordinate multi-step tasks. Where REST optimizes for stable service interfaces and direct execution, A2A optimizes for reasoning-driven coordination, task-oriented messaging, and agent collaboration. The result is systems that can plan, delegate, and compose actions across multiple services rather than invoking isolated endpoints.
Challenges with moving towards agentic systems
REST APIs and agentic systems are based on orthogonal paradigms, which makes it hard for enterprises to move existing services into standardized agentic communication. Yet enterprises need to use both without a major overhaul. Although newer agent communication through A2A introduces coordination models for enterprise systems, adoption has been slowed by the need to deploy and operate agentic infrastructures alongside existing enterprise systems. This parallel operation increases operational complexity and cost, creating barriers to adopting AI effectively.
Before A2A was standardized, enterprises commonly deployed agents as REST-based or proprietary services. They treated them as conventional APIs with agent-specific logic embedded in request-response endpoints. As a result, many existing agents today aren’t A2A-native, which creates a new migration challenge: making these agents interoperate using standardized A2A protocols without rewriting their core logic.
The preceding figure shows a REST-based application. The REST API stack might be a monolith or a set of distributed endpoints. The REST API controller doesn’t need to be an explicit broker that delegates requests outside the application. It can be part of the framework itself. For example, in a Flask application, the framework provides the controller abstraction out of the box, as you typically see when using @app.route() or Flask’s RESTful extensions. The idea here is to capture the REST stack with a set of endpoints.
Solution approaches
In this section, we discuss the different approaches you could take to add an agentic capability to a legacy enterprise system. We compare them to the approach of using agentic overlays.
Maintain separate REST and A2A stacks: One approach is to develop and maintain two parallel ways to expose the same capabilities. This could mean:
- Two sets of endpoints: /api/v2/... and /a2a/....
- Two implementations of auth, validation, and error mapping (unless carefully reused).
- Two deployment pipelines (build, test, release, rollback).
- Double observability work: logs, metrics, and tracing for both paths.
- Higher risk of inconsistency. A2A may return a different output for the same operation carried out by REST.
- Higher cost and operational complexity over time.
Separated stacks, but shared business logic: Refactoring existing endpoints means you change your current REST API code structure (and sometimes behavior) so it can be reused by a new interface such as A2A. Instead of leaving REST endpoints as-is, you reorganize them, usually by extracting business logic into shared services and updating controllers and handlers to call those services. Even if the external REST paths remain the same, refactoring can introduce regressions, behavior drift, and a large test burden.
The core idea: Agentic overlays
An *agentic overlay* is a thin wrapper layer that lets REST-based services participate in A2A communication. The overlay:
- Transforms an agentic message into a REST payload, and the reverse.
- Exposes REST endpoints as agent tasks or tools.
Most importantly, A2A isn’t a new API. It’s a new interface to an existing API. The underlying REST service remains unchanged.
Adding an agentic overlay within the application
In this approach, you have two sets of endpoints, /api/v2/... and /a2a/... (REST vs. A2A), as shown in the following diagram, but a single deployment pipeline for build, test, release, and rollback. With this pattern, traditional REST API endpoints can be transformed into agentic endpoints without rewriting the core business logic. The deployment process doesn’t change for the service. For the same host and same port, you add new routes, although the systems might need to be scaled to handle increased traffic.
You can apply the agent skills themselves for routing. An MCP server can be used to invoke external services, but agent skills can route requests within the agent scope directly without importing APIs into an MCP server as skills. Whatever endpoints you have can be exposed as agent skills without needing a separate MCP server.
This approach reduces agent sprawl in the infrastructure by reusing existing services as agents. This design pattern works well for supervisor agents that need both REST-based and agentic capabilities with limited functional scope, such as intent classification and routing.
Agentic overlay example implementation
As a proof of concept, this section shows how to port an example legacy REST-based calculator service that uses Flask into an agentic system using an overlay. For the overlay, we add the standard A2A components (or routes), such as a well-known agent card, agent message endpoint, capabilities, skills, and health. We also introduce a message transformation design pattern that converts agentic messages to REST API messages, then issues REST invocation calls from the agent. The A2A message translation workflow is as follows:
- Receives JSON-RPC 2.0 requests.
- Maps A2A tasks to REST endpoints.
- Forwards authentication headers.
- Calls REST endpoints internally.
- Translates REST responses to JSON-RPC format.
Step 0: Request-response format
This section compares the request-response format for the REST and A2A protocols, using the calculator example demonstrated in the following sections.
REST vs. A2A input request:
| REST | A2A |
|---|---|
| { “operation”: “add”, “operands”: [5, 3] } | { “jsonrpc”: “2.0”, “method”: “SendMessage”, “params”: { “message”: { “role”: “user”, “parts”: [ { “kind”: “data”, “data”: { “operation”: “add”, “operands”: [5, 3] } } ] } }, “id”: 1 } |
REST vs. A2A output response:
| REST | A2A |
|---|---|
| {“result”: 8} | { “jsonrpc”: “2.0”, “result”: { “messageId”: “uuid”, “contextId”: “uuid”, “role”: “agent”, “parts”: [{“kind”: “data”, “data”: {“result”: 8}}], “kind”: “message”, “metadata”: {} }, “id”: 1 } |
Step 1: Set up the agent
In this step, you create the calculator agent example with a well-known agent card and agent skills loaded. The build_agent_card function builds the agent card dynamically.
"""
A2A Request Translator - Calculator Example.
This module implements the Request Translator Pattern to provide A2A
(JSON-RPC 2.0) compatibility over the existing Calculator REST API.
A2A Spec 0.3 Compliance:
- Agent Card: GET /.well-known/agent-card.json
- JSON-RPC endpoint: POST /a2a
- Methods: SendMessage, SendStreamingMessage
- Message format: { "message": { "parts": [{ "kind": "data", "data": {...} }] } }
"""
# Default A2A API URL (override via build_agent_card(url) if needed)
A2A_API_URL = "http://localhost:5000/a2a"
EXECUTE_TIMEOUT_SECONDS = 30
# Load skills from JSON file
_SKILLS_FILE = Path(__file__).parent / "skills.json"
_SKILLS_CACHE: Optional[List[Dict[str, Any]]] = None
def _load_skills() -> List[Dict[str, Any]]:
"""
Load skills from the skills.json file.
Skills are cached after first load to avoid repeated file reads.
"""
global _SKILLS_CACHE
if _SKILLS_CACHE is not None:
return _SKILLS_CACHE
try:
with open(_SKILLS_FILE) as f:
_SKILLS_CACHE = json.load(f)
return _SKILLS_CACHE
except FileNotFoundError:
logger.error(f"Skills file not found: {_SKILLS_FILE}")
return []
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON in skills file: {e}")
return []
def build_agent_card(api_url: Optional[str] = None) -> Dict[str, Any]:
"""Build the A2A Agent Card (v0.3.0 format) with configurable API URL."""
if api_url is None:
api_url = A2A_API_URL
return {
"name": "Calculator Agent",
"description": "Simple calculator supporting basic arithmetic operations",
"supportedInterfaces": [
{"url": api_url, "protocolBinding": "JSONRPC", "protocolVersion": "0.3"},
],
"provider": {
"organization": "Example Organization",
"url": "",
},
"version": "1.0.0",
"capabilities": {
"streaming": False,
"pushNotifications": False,
"extendedAgentCard": False,
},
"defaultInputModes": ["text/plain", "application/json"],
"defaultOutputModes": ["text/plain", "application/json"],
"skills": _load_skills(),
}
# Agent Card built dynamically
AGENT_CARD = build_agent_card()Step 2: Implement the internal REST caller
def invoke_rest_endpoint(
endpoint: str,
json_data: Optional[Dict] = None,
http_method: str = "POST"
) -> Tuple[Optional[Dict], int]:
"""
Call internal REST endpoint via real HTTP request.
Uses requests.post/get to call the running server. This ensures
any middleware, decorators, and headers are properly exercised.
Args:
endpoint: REST endpoint path (e.g. "/api/v1/calculate")
json_data: Request body for POST/PUT
http_method: HTTP method (GET, POST, etc.)
Returns:
Tuple of (response_data, status_code)
"""
try:
base_url = request.host_url.rstrip("/")
url = f"{base_url}{endpoint}"
headers = {"Content-Type": "application/json"}
auth_header = request.headers.get("Authorization")
if auth_header:
headers["Authorization"] = auth_header
logger.info(f"Adapter: Delegating to REST {http_method} {url}")
if http_method.upper() == "POST":
response = http_requests.post(
url, json=json_data, headers=headers,
timeout=EXECUTE_TIMEOUT_SECONDS
)
elif http_method.upper() == "GET":
response = http_requests.get(
url, headers=headers,
timeout=EXECUTE_TIMEOUT_SECONDS
)
else:
response = http_requests.request(
http_method, url, json=json_data, headers=headers,
timeout=EXECUTE_TIMEOUT_SECONDS
)
logger.info(f"Adapter: REST returned {response.status_code}")
return response.json(), response.status_code
except http_requests.RequestException as e:
logger.error(f"Adapter: Error calling REST endpoint: {e}", exc_info=True)
return {"error": "Internal server error"}, 500Step 3: Implement message payload extraction (A2A Spec 0.3)
def extract_message_payload(message: Dict) -> Optional[Dict]:
"""
Extract payload from A2A message parts (Spec 0.3 format).
Expected format:
{
"message": {
"parts": [{"kind": "data", "data": {"operation": "add", "operands": [5, 3]}}]
}
}
Returns:
Extracted data payload as dict, or None if not found
"""
try:
parts = message.get("parts", [])
for part in parts:
if isinstance(part, dict) and part.get("kind") == "data":
return part.get("data")
return None
except Exception as e:
logger.error(f"Error extracting message payload: {e}")
return None
def build_a2a_message(message_id: str, context_id: str, content: Any) -> Dict:
"""
Build an A2A-compliant message object (A2A Spec 0.3).
Response format:
{
"messageId": "uuid",
"contextId": "uuid",
"role": "agent",
"parts": [{"kind": "data", "data": {...}}],
"kind": "message",
"metadata": {}
}
"""
if isinstance(content, dict):
parts = [{"kind": "data", "data": content}]
else:
parts = [{"kind": "text", "text": str(content)}]
return {
"messageId": message_id,
"contextId": context_id,
"role": "agent",
"parts": parts,
"kind": "message",
"metadata": {}
}Note on Server-Sent Events (SSE) for streaming:
The preceding extract_message_payload() function works the same way for both SendMessage and SendStreamingMessage.
For operations that are instant (like our calculator), both methods return a single result. For long-running operations (for example, report generation or data analysis), SSE streaming allows the server to push incremental updates.
Step 4: Implement JSON-RPC response builders
# The error codes defined as per JSON-RPC 2.0 specification
class JsonRpcError:
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
def jsonrpc_error(code: int, message: str, data: Any = None, request_id: Any = None) -> Dict:
"""Build a JSON-RPC 2.0 error response."""
response = {
"jsonrpc": "2.0",
"error": {"code": code, "message": message},
"id": request_id
}
if data is not None:
response["error"]["data"] = data
return response
def jsonrpc_success(result: Any, request_id: Any = None) -> Dict:
"""Build a JSON-RPC 2.0 success response."""
return {
"jsonrpc": "2.0",
"result": result,
"id": request_id
}Step 5: SendMessage — A2A-to-REST delegation
def handle_send_message(data: Dict) -> Tuple[Any, int]:
"""
Handle SendMessage -- dumb pass-through to /api/v1/calculate.
The adapter does NOT inspect or route based on payload contents.
"""
request_id = data.get("id")
params = data.get("params", {})
message = params.get("message", {})
context_id = message.get("contextId") or generate_id()
message_id = generate_id()
payload = extract_message_payload(message)
if not payload:
return jsonify(jsonrpc_error(
JsonRpcError.INVALID_PARAMS,
"Invalid params: No data found in message.parts.",
request_id=r
関連記事
今日のまとめ
AIデイリーブリーフで今日の重要ニュースをまとめ読み