アマゾン ウェブ サービス ジャパン (有志)
🐱

Agent Registry で動的に Harness (MCP/Skill) や A2A Agent を検索・実行する Agent の実装

Renya Kujirada
に公開
AWS
Model Context Protocol
Bedrock
Agent
A2A
tech

はじめに

AWS Japan AI/ML Specialist Solutions Architect の Kujirada です。

PJ 内で開発した Agent (sub agent), MCP Server, Skills などの Harness をどのように組織の資産として横断利用するかは、Agent の開発・運用における重要な課題です。開発した Harness などのリソースを一元的に管理・検索する仕組みが無ければ、リソースを再利用できず、類似する機能を重複開発してしまう状況になります。特に、Skills は、その性質上個人の資産になる傾向が高く、組織全体での再利用は困難です。

AWS Agent Registry は、この課題に対処するサービスであり、組織内で開発した Agent, MCP Server, Skills を一元的に管理・検索/再利用するための中央集約型カタログです。Agent Registry は、検索用の MCP Server をマネージドに提供しており、既存の MCP Client (Kiro / Claude Code 等) から容易にカタログを検索できます。

Agent Registry の検索用の MCP Server を活用することで、Anthropic の Tool Search Tool のように、必要なリソースだけを動的に読み込み、実行する汎用 Agent を実装することも可能です。これにより、トークン消費を削減しつつ、Registry に登録された多種多様なツール・Skills・Agent を利用することができ、汎用性の高い Agent を構築することができます。

本稿では、Strands Agent と Agent Registry の MCP Server を組み合わせ、MCP / Skills / A2A Agent を動的に検索・読み込みする Agent (Dynamic Load Agent) を実装し、その仕組みを解説します。実装は以下のリポジトリで公開しています。

https://github.com/ren8k/aws-registry-search-agent

検証内容

様々な問い合わせに対応可能な汎用 Agent を題材とします。Agent 起動時には Registry MCP の検索ツール 1 つしか持たず、ユーザー問い合わせに応じて適切なリソースを Registry から動的に検索・読み込み、実行できる構成を目指します。

この構成の利点は、Registry に MCP / Skills / A2A Agent を新規登録するだけで、Agent 側のコードや設定を変更せず即座に使えるようになる点です。Agent が事前知識として持つのは Registry を検索するためのツールのみで、利用可能なリソースは Agent Registry で管理します。

以降、Agent Registry の概要と Agent Registry でのリソースの登録方法を整理し、その後 Agent の実装方法について解説します。

Agent Registry とは

本節では、Agent Registry の概要、本稿で扱う 3 種類のリソース (A2A / MCP / Skills)、そしてリソースの登録方法を順に解説します。

概要

Agent Registry は、組織内で開発した Agent, MCP Server, Skills を一元的に管理・検索/再利用するための中央集約型カタログです。Agent Registry では、登録した全てのリソースに対して、スケール可能な形で、承認フローを含めた登録・発見のライフサイクルを実現することが可能になります。

Agent Registry は Registry と Registry Record という 2 つのリソースで構成されます。Registry Record に個々のリソースのメタデータを保存し、Registry で Registry Record を管理します。

リソース 説明
Registry Registry Record を格納するカタログ。承認方法 (approvalConfiguration) や認可方式 (IAM / JWT) を Registry 単位で設定する。
Registry Record カタログに登録される 1 つのリソース。MCP Server / A2A Agent / Skills / カスタムリソースを descriptorType で区別し、検索対象となる。

Registry Record に登録可能なリソースの種類 (サポートする descriptor type) は A2A, MCP, Skills, Custom Resource の 4 種です。

descriptorType 内容 スキーマ
A2A A2A Agent (AgentCard) A2A schemaVersion 0.3
MCP MCP Server (server + tools) Server Schema (registry.modelcontextprotocol.io 準拠) + Tools Schema
AGENT_SKILLS SKILL.md + 定義 (repository + package) agentskills.io 仕様 + Amazon 独自スキーマ
CUSTOM 任意の JSON スキーマ検証なし

各 Record は DRAFT → PENDING_APPROVAL → APPROVED のライフサイクルを経て検索可能になります。APPROVED の Record のみが SearchRegistryRecords API および MCP endpoint からの検索に露出する仕様です。

以降、A2A / MCP / Skills の 3 種について、実装を含めてそれぞれ解説します。

A2A

A2A (Agent2Agent) は、Agent が自身の能力をネットワーク越しに公開し、他の Agent から呼び出されるための標準的な方法を提供するプロトコルです。A2A の主要なアクターとして、A2A Server と A2A Client があります。A2A Server（Remote Agent）は、自身の能力やサービスエンドポイント URL を Agent Card (/.well-known/agent-card.json エンドポイント) として公開し、宣言したサービスエンドポイントでリクエストを受け付けます。A2A Client (呼び出し元の Agent) は、Agent Card を取得して A2A Server の能力を把握した上で、JSON-RPC 2.0 ベースの message/send メソッドで Remote Agent に対して指示を送信します。

A2A Server は AgentCore Runtime にデプロイすることができ、AgentCore SDK の bedrock_agentcore.runtime.a2a.serve_a2a (serve_a2a) を利用することで容易に A2A Server の実装とデプロイが可能です。

以下に、A2A Server の実装例 (都市ごとの天気を返す Agent) を示します。実装は、Strands Agent を StrandsA2AExecutor でラップし、AgentCard と一緒に serve_a2a に渡すだけです。StrandsA2AExecutor は、a2a-sdk の AgentExecutor を Strands Agent 用に実装されたクラスです。

assets/agents/weather/weather_a2a_agent.py
from bedrock_agentcore.runtime.a2a import serve_a2a
from strands import Agent, tool
from strands.models import BedrockModel
from strands.multiagent.a2a.executor import StrandsA2AExecutor


@tool
def lookup_city_weather(city: str) -> str:
    """Return a short, plausible Japanese weather line for a Japanese city.

    Args:
        city: Japanese city name (例: ``"東京"``, ``"大阪"``).
    """
    samples = {
        "東京": "晴れ時々曇り / 最高 18℃ 最低 9℃",
        "大阪": "曇り / 最高 17℃ 最低 10℃",
    }
    return samples.get(city, f"{city} の予報データは登録されていません")


agent = Agent(
    model=BedrockModel(model_id="global.anthropic.claude-sonnet-4-6"),
    name="Weather Forecast Agent",
    description="Provides brief Japanese-language weather forecasts ...",
    tools=[lookup_city_weather],
)


if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))

serve_a2a を利用することで、AgentCore Runtime や A2A で要求される以下の仕様を 1 行で満たすことが可能です。

serve_a2a の機能 内容
port 9000 / 0.0.0.0 で listen A2A の規定 port 9000 を既定値として持ち、Docker 環境を検知すると bind 先を 0.0.0.0 (ローカルでは 127.0.0.1) に自動で切替
/ping の自動配信 AgentCore Runtime が要求するヘルスチェック用エンドポイントを Starlette ルートとして自動で公開す
/.well-known/agent-card.json の自動配信 AgentCard discovery 用エンドポイントを a2a-sdk 経由で公開し、AgentCard の url を AGENTCORE_RUNTIME_URL 環境変数の値で上書き。
Runtime ヘッダの自動伝搬 session-id / request-id / workload access token などの Bedrock Runtime 固有ヘッダを BedrockAgentCoreContext 経由で tool 側から参照可能
Agent フレームワーク非依存 a2a-sdk の AgentExecutor を実装したクラスを渡すだけで、Strands / LangGraph / Google ADK などのいずれにも対応可能

MCP

MCP (Model Context Protocol) は、外部システムがツールを公開し、AI アプリケーションがそれを標準的な方法で利用するためのプロトコルです。MCP アーキテクチャは、Server / Host / Client のロールで構成されます。MCP Server は自身が提供する Tool を tools/list メソッドで公開し、JSON-RPC 2.0 ベースで Tool 実行 (tools/call) のリクエストを受け付けます。MCP Host (Claude Code 等の AI アプリケーション) は、接続する MCP Server ごとに MCP Client を生成し、MCP Client を介して Server から取得した Tool を LLM のコンテキストに供給します。

AgentCore で MCP Server をデプロイする方法はいくつかありますが、FastMCP を利用した AgentCore Runtime へのデプロイが容易です。AgentCore における MCP Server のデプロイ方法の詳細は以下のブログをご参照下さい。

https://zenn.dev/aws_japan/articles/001-bedrock-agentcore-remote-mcp

以下に、MCP Server の実装例 (商品の注文状況を確認するツール) を示します。実装は、Tool として公開したい関数を @mcp.tool() デコレータで登録するだけです。AgentCore Runtime は MCP Server コンテナが 0.0.0.0:8000/mcp パスで利用可能であることを前提としており、FastMCP のデフォルト port (8000) と path (/mcp) がそのまま AgentCore Runtime の要件に合致します。また、AgentCore Runtime の仕様に合わせて、mcp.run() には transport="streamable-http", host="0.0.0.0", stateless_http=True (基本的には stateless mode で問題ない) を明示する必要があります。

assets/mcp-servers/order-management/src/mcp_server.py
from fastmcp import FastMCP

mcp = FastMCP("order-management")

ORDERS = {"123": {"status": "shipped", "tracking": "TRK-789456"}}


@mcp.tool()
def get_order_status(orderId: str) -> dict:
    """Get the status of an order by order ID."""
    return ORDERS.get(orderId, {"error": "Order not found"})


@mcp.tool()
def update_order(orderId: str, action: str) -> dict:
    """Update an order - cancel it or change its shipping address."""
    ...  # 省略


if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=8000,
        stateless_http=True,
    )

Agent Skills

Agent Skills は、Agent にドメイン知識やワークフローを追加するための再利用可能な指示書です。各 Skill は SKILL.md (frontmatter 必須の Markdown) と任意のリソースファイル (scripts/, references/, assets/) で構成されます。

my-skill/
├── SKILL.md          # 必須: 指示文 + フロントマター
├── scripts/          # 任意: 実行可能スクリプト
├── references/       # 任意: 参考資料
└── assets/           # 任意: テンプレート、設定、データ

SKILL.md は frontmatter (name / description の YAML) と本文の Markdown で構成されます。Anthropic はこの構造を Progressive Disclosure (段階的開示) という設計原則で整理しており、Skill 内の情報を以下の 3 段階に分けて Agent のコンテキストへ読み込ませます。この仕組みにより、Agent は全ての Skill の手順をコンテキストに常駐させずに済み、必要な時に必要な深さの情報だけをロードできるため、Skills の総量が増えてもトークン消費を抑えられます。

段階 読み込むタイミング 読み込む内容
第 1 段階 Agent 起動時にメタデータのみシステムプロンプトに常駐 frontmatter の name と description のみ
第 2 段階 Agent が当該 Skill を必要と判断した時点で読み込む SKILL.md 本文
第 3 段階 本文中で参照される付随ファイルが必要になった時に読み込む references/, scripts/, assets/ 配下のリソース

以下に、Skills の記述例 (天気予報回答用テンプレート) を示します。name と description で「いつこの Skill を使うべきか」を Agent に伝え、本文に具体的な Response Template を記述する構成です。

assets/skills/weather-skill.md
---
name: weather-response-template
description: Use this skill whenever the user asks about weather, forecast, or 天気 / 予報. Provides a consistent Japanese-language response template for weather inquiries.
---

# Weather Response Template (Japanese)

## Purpose

このスキルは、天気や気象予報に関する質問に答えるときの **日本語テンプレート** を提供する。
回答は簡潔で、必ず以下のセクションを含むこと。

## Response Template

​```
【現在の天気】
- 場所: {場所}
- 天候: {晴れ / 曇り / 雨 / 雪}
- 気温: {X}°C
- 湿度: {Y}%
...
​```

A2A / MCP / Skills の登録・検索方法

Registry 自体と各 Record を作成する手順を、boto3 ベースで解説します。具体的には、以下の図のように、OAuth 認証 の Agent Registry を作成し、IAM 認証の A2A Server, MCP Server と local 上の Skills (Markdown) を Record として登録する手順を示します。なお、Github で公開している実装では CDK で構築していますが、ここでは API レベルの動きを理解しやすいよう Python SDK のサンプルを使います。

Step 1: Registry の作成

bedrock-agentcore-control クライアントの create_registry を呼びます。

import boto3

region = "us-east-1"

# Control plane クライアント (Registry / Record の Create / Delete などに使う)
cp_client = boto3.client("bedrock-agentcore-control", region_name=region)
# Data plane クライアント (SearchRegistryRecords など)
dp_client = boto3.client("bedrock-agentcore", region_name=region)

resp = cp_client.create_registry(
    name="aws-registry-for-search-agent",
    description="Registry for aws-registry-search-agent demo",
    approvalConfiguration={"autoApproval": True},
    authorizerType="CUSTOM_JWT",
    authorizerConfiguration={
        "customJWTAuthorizer": {
            "discoveryUrl": discovery_url, # OIDC discovery URL
            "allowedClients": [cognito_client_id], # App Client ID
        }
    },
)
registry_id = resp["registryArn"].split("/")[-1]

authorizerType は Registry の検索 API (Search API) に対する Inbound Authorization 方式を指定します。AWS_IAM (デフォルト) と CUSTOM_JWT の 2 種類から選択可能で、本実装では Cognito を使った JWT 認可を行うため CUSTOM_JWT を指定しています。autoApproval=True を指定すると承認フローを省略でき、SubmitRegistryRecordForApproval API の実行だけで APPROVED に到達します。

Step 2-a: MCP / A2A record の登録 (sync 方式)

MCP / A2A record は、Registry の sync 機能 (synchronizationType=URL) を使うと容易に登録できます。具体的には、Registry 自身が指定 URL に SigV4 / OAuth で接続し、エンドポイントから tools list や AgentCard を自動取得して descriptor を populate する仕組みです。

MCP record の登録
gateway_url = "<Gateway MCP endpoint URL>"  # 例: https://<gateway-id>.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp
sync_role_arn = "<Registry sync IAM Role ARN>"

# MCP record (Gateway URL から server / tools metadata を sync)
cp_client.create_registry_record(
    registryId=registry_id,
    name="order_management_mcp",
    description="Order management MCP server - look up / update orders",
    descriptorType="MCP",
    synchronizationType="URL",
    synchronizationConfiguration={
        "fromUrl": {
            "url": gateway_url,
            "credentialProviderConfigurations": [
                {
                    "credentialProviderType": "IAM",
                    "credentialProvider": {
                        "iamCredentialProvider": {
                            "roleArn": sync_role_arn,
                            "service": "bedrock-agentcore",
                        },
                    },
                },
            ],
        },
    },
    recordVersion="1.0",
)
A2A record の登録
from urllib.parse import quote

runtime_arn = "<A2A Runtime ARN>"  # 例: arn:aws:bedrock-agentcore:us-east-1:<account-id>:runtime/<runtime-id>

# A2A record (Runtime の /.well-known/agent-card.json から AgentCard を sync)
agent_card_url = (
    f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/"
    f"{quote(runtime_arn, safe='')}/invocations/.well-known/agent-card.json"
)
cp_client.create_registry_record(
    registryId=registry_id,
    name="weather_forecast_a2a",
    description="A2A weather-forecast agent for Japanese cities.",
    descriptorType="A2A",
    synchronizationType="URL",
    synchronizationConfiguration={
        "fromUrl": {
            "url": agent_card_url,
            "credentialProviderConfigurations": [
                {
                    "credentialProviderType": "IAM",
                    "credentialProvider": {
                        "iamCredentialProvider": {
                            "roleArn": sync_role_arn,
                            "service": "bedrock-agentcore",
                        },
                    },
                },
            ],
        },
    },
    recordVersion="1.0",
)

この sync 方式の利点は、Record の内容をデプロイ済みの MCP Server / A2A Agent の出力に追従させられる点にあります。具体的には、MCP record では MCP Server から取得した server 定義と tool 一覧が、A2A record では /.well-known/agent-card.json が返す AgentCard が、それぞれ record の内容として書き込まれるため、Agent / MCP Server 実装と Registry 登録内容の乖離を構造的に防げます。

登録される A2A Record の例
A2A record
AgentCard (weather a2a agent)
{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text"
  ],
  "defaultOutputModes": [
    "text"
  ],
  "description": "Provides brief Japanese-language weather forecasts for Japanese cities. Returns condition (sunny/cloudy/rainy), temperature range, and short advice.",
  "name": "Weather Forecast Agent",
  "preferredTransport": "JSONRPC",
  "protocolVersion": "0.3.0",
  "skills": [
    {
      "description": "Return a short, plausible Japanese weather line for a Japanese city.\n\nThis tool does not call any real weather API. It produces a deterministic\ndummy forecast so the agent can demonstrate the A2A tool-calling path and\nto verify that the tool definition surfaces in the AgentCard ``skills[]``\nafter Registry synchronisation (ADR 0018).\n\nReturns:\n    A one-line Japanese forecast: ``condition / high-low temperature``.",
      "id": "lookup_city_weather",
      "name": "lookup_city_weather",
      "tags": []
    }
  ],
  "url": "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-east-1%3A869603330160%3Aruntime%2Faws_registry_search_agent_weather_a2a-mIwzN27IQp/invocations",
  "version": "0.1.0"
}
登録される MCP Record の例
MCP
MCP Server Schema (order management mcp)
{
  "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json",
  "name": "aws-registry-search-agent-nxch8wf7uq.gateway.bedrock-agentcore.us-east-1.amazonaws.com/aws-registry-search-agent",
  "description": "-",
  "version": "1.0.0",
  "remotes": [
    {
      "type": "streamable-http",
      "url": "https://aws-registry-search-agent-nxch8wf7uq.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp"
    }
  ]
}
MCP Tool Schema(order management mcp)
{
  "tools": [
    {
      "name": "order-management-target___get_order_status",
      "description": "Get the status and details of an order by order ID, including items, total, shipping address, and tracking info.",
      "inputSchema": {
        "type": "object",
        "properties": {
          "orderId": {
            "type": "string",
            "description": "Order ID to look up."
          }
        },
        "required": [
          "orderId"
        ],
        "additionalProperties": false
      },
      "outputSchema": {
        "additionalProperties": true,
        "type": "object"
      },
      "_meta": {
        "fastmcp": {
          "tags": []
        }
      }
    },
    {
      "name": "order-management-target___update_order",
      "description": "Update an order - cancel it, change its shipping address, or perform other modifications.",
      "inputSchema": {
        "type": "object",
        "properties": {
          "orderId": {
            "type": "string",
            "description": "Order ID to update."
          },
          "action": {
            "type": "string",
            "description": "Action: 'cancel' or 'change_address'."
          },
          "newAddress": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "description": "New shipping address (required if action=change_address)."
          }
        },
        "required": [
          "orderId",
          "action"
        ],
        "additionalProperties": false
      },
      "outputSchema": {
        "additionalProperties": true,
        "type": "object"
      },
      "_meta": {
        "fastmcp": {
          "tags": []
        }
      }
    }
  ]
}

Step 2-b: AGENT_SKILLS record の登録 (inline)

AGENT_SKILLS は sync 非対応のため、SKILL.md 本文を直接 inline で渡します。

with open("./skills/weather-skill.md") as f:
    skill_md = f.read()

cp_client.create_registry_record(
    registryId=registry_id,
    name="weather_response_skill",
    description="Use this skill whenever the user asks about weather ...",
    descriptorType="AGENT_SKILLS",
    descriptors={
        "agentSkills": {
            "skillMd": {"inlineContent": skill_md},
            "skillDefinition": {
                "schemaVersion": "0.1.0",
                "inlineContent": (
                    '{"repository": {"url": "https://github.com/example/my-skill",'
                    ' "source": "github"}}'
                ),
            },
        }
    },
    recordVersion="1.0",
)

Agent Registry の AGENT_SKILLS record は skillMd と skillDefinition という 2 つの descriptor を含みます。skillMd には SKILL.md 全文が、skillDefinition にはリポジトリの情報や PyPI / npm パッケージ宣言などの JSON メタデータが格納されます。skillDefinition の各フィールドの型は公式ドキュメントに記載されています。

両 descriptor とも optional であり、SKILL.md 本文だけを登録したい場合は skillDefinition を省略しても構いません。なお、skillMd.inlineContent は --- で囲まれた YAML frontmatter で始まる必要があり、frontmatter が無い SKILL.md を渡すと ValidationException で record 作成が失敗します。

登録される Skills Record の例
Skill Markdown (weather response skill)
---
name: weather-response-template
description: Use this skill whenever the user asks about weather, forecast, or 天気 / 予報. Provides a consistent Japanese-language response template for weather inquiries.
---

# Weather Response Template (Japanese)

## Purpose

このスキルは、天気や気象予報に関する質問に答えるときの **日本語テンプレート** を提供する。
回答は簡潔で、必ず以下のセクションを含むこと。

## Response Template

```
【現在の天気】
- 場所: {場所}
- 天候: {晴れ / 曇り / 雨 / 雪}
- 気温: {X}°C
- 湿度: {Y}%

【今日の予報】
- 最高気温: {X}°C / 最低気温: {Y}°C
- 降水確率: {Z}%

【備考】
{服装や外出時の注意点を 1 文で}
```

## Example

**Input**: 東京の天気を教えて

**Output**:
```
【現在の天気】
- 場所: 東京
- 天候: 晴れ
- 気温: 18°C
- 湿度: 45%

【今日の予報】
- 最高気温: 22°C / 最低気温: 14°C
- 降水確率: 10%

【備考】
日中は過ごしやすいですが、朝晩は冷え込むので薄手の上着をおすすめします。
```

## Notes

- 外部の天気 API を呼び出す手段が Agent に無い場合は、"最新の情報は気象庁 Web サイトで確認してください" と付記してテンプレートを埋める
- 場所が曖昧な場合は、ユーザーに 1 度だけ確認する
skill definition (weather response skill)
{
  "websiteUrl": "https://github.com/ren8k/aws-registry-search-agent",
  "repository": {
    "url": "https://github.com/ren8k/aws-registry-search-agent",
    "source": "github"
  }
}

Notes

  • 外部の天気 API を呼び出す手段が Agent に無い場合は、"最新の情報は気象庁 Web サイトで確認してください" と付記してテンプレートを埋める
  • 場所が曖昧な場合は、ユーザーに 1 度だけ確認する

Step 3: レコードの承認

Record 作成後は、autoApproval=True の Registry でも明示的に承認の Submit が必要です。レコードのステータスが 承認済み になると、search_registry_records API で検索することが可能になります。

for record_id in record_ids:
    cp_client.submit_registry_record_for_approval(
        registryId=registry_id,
        recordId=record_id,
    )

# 検索可能化までは結果整合 (数秒〜数分)
resp = dp_client.search_registry_records(
    registryIds=[registry_arn],
    searchQuery="order management",
    maxResults=5,
)

ここまでで、Registry に MCP / A2A / Skills の record が登録され、SearchRegistryRecords API および専用 MCP endpoint から検索可能な状態になります。

本検証 (Dynamic Load Agent) における AWS 構成図

本実装の全体像を以下に示します。Agent は Strands Agent で実装しており、Agent 初期化時には、Agent Registry の MCP Server (検索ツール) のみを登録しています。Agent Registry には、2種類の A2A Agent (天気予報Agent, ニュース配信Agent), MCP Server (注文管理ツール), Skills (天気予報回答テンプレート) を登録しています。Agent Registry の Inbound 認証は OAuth、A2A / MCP Server の Inbound 認証は IAM (SigV4) としています。

Agent は、Harness の検索を行う際は Agent Registry に対してリクエストを行い、検索結果の A2A Agent や MCP Server を実行する際は AgentCore Runtime や AgentCore Gateway に対して直接リクエストを行います。

Dynamic Load Agent の仕組み

本稿の中心となる Agent 側の実装を解説します。Agent (src/aws_registry_search_agent/) は 2 ファイル構成で、agent.py が CLI エントリポイント、registry_tools.py が Registry MCP の検索結果を処理する Hook 実装です。

設計思想

設計の核は、ローカル @tool を 1 つも実装せず、Registry MCP が提供するツール search_registry_records 1 つだけを Agent に持たせ、Hook で検索結果 (Harness) を動的に読み込む点にあります。具体的には、ユーザーの問い合わせに応じて Agent は Registry を semantic search し、その検索結果を Strands の AfterToolCallEvent Hook で受け取り、検索ヒットした A2A Agent / MCP Server / Skills を Agent に動的注入します。また、注入と同時に、Hook は tool 結果を 1 行サマリに書き換え、巨大な registryRecords JSON (検索結果) が会話履歴に入らないように制御します。

このアプローチにより、必要なタイミングで必要なリソースを読み込む Progressive Disclosure を、A2A Agent / MCP Server / Skills などの Harness に対して実現できます。これにより、トークン消費を削減しつつ Registry に登録された多種多様なツール・Skills・Agent を効率良く利用できます。また、Registry に Harness を登録するだけで Agent 側のコードを修正せずに機能追加できる、汎用性の高い Agent を構築できます。

実装上の工夫点

設計を実装に落とし込む過程で、Strands SDK と Bedrock Converse API の仕様を踏まえた以下 4 点を工夫しています。各項目の詳細は以降の節で順に解説します。

工夫点 内容
ローカル @tool を持たない Agent 起動時の構成 検索を Registry MCP に委譲し、Agent は Registry MCP / AgentSkills Plugin / Hook の 3 点だけで起動する。
Hook で「検索 → 自動ロード」を構造的に強制 AfterToolCallEvent で検索結果をパースし、descriptorType ごとに MCP / A2A / Skills を動的注入する。
tool result 書き換えで registryRecords JSON を排除 Hook 内で event.result["content"] を 1 行サマリに上書きし、巨大な JSON を会話履歴に残さない。
A2A は A2AAgent + @tool 閉包で個別 tool 化 検索された Agent 毎に動的にツールを生成する。その際、エンドポイント URL を Python 閉包に隠蔽し、AgentCard の name/description を基に ツールとして定義する。

次節以降、各工夫点における実装を順に解説します。

Agent 起動時の構成

Agent 起動時、tools に渡すのは Agent Registry が提供する検索用 MCP Server に対する MCPClient 1 つだけです。Agent の引数 plugins=[AgentSkills(skills=[])] は Strands 公式の Skills 管理 Plugin を空で起動する構成で、内部で skills という 1 つのツールを Agent に登録し、<available_skills> XML を system prompt に動的注入します。引数 hooks=[loader] にはカスタム Hook を指定しており、動的に Agent, MCP Server, Skills を読み込みます。

src/aws_registry_search_agent/agent.py
import os
from typing import Any

from mcp.client.streamable_http import streamablehttp_client
from strands import Agent, AgentSkills
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient

from aws_registry_search_agent.registry_tools import (
    RegistryLazyLoader,
    fetch_registry_access_token,
)


def build_registry_mcp_client(
    registry_id: str, access_token: str, region: str = "us-east-1"
) -> MCPClient:
    url = f"https://bedrock-agentcore.{region}.amazonaws.com/registry/{registry_id}/mcp"
    headers = {"Authorization": f"Bearer {access_token}"}

    def transport() -> Any:
        return streamablehttp_client(url, headers=headers)

    return MCPClient(transport)


def build_agent() -> Agent:
    region = os.environ.get("AWS_REGION", "us-east-1")
    registry_id = os.environ["REGISTRY_ID"]
    oauth_config = {
        "cognito_domain": os.environ["COGNITO_DOMAIN"],
        "client_id": os.environ["COGNITO_CLIENT_ID"],
        "client_secret": os.environ["COGNITO_CLIENT_SECRET"],
        "scopes": os.environ["COGNITO_SCOPES"],
    }

    # Registry MCP は OAuth Bearer (Cognito Client Credentials) で接続
    token = fetch_registry_access_token(oauth_config)
    registry_mcp = build_registry_mcp_client(registry_id, token, region=region)

    # Gateway / A2A Runtime は SigV4 (IAM)。Loader が内部で auth を保持する。
    skills_plugin = AgentSkills(skills=[])
    loader = RegistryLazyLoader(region=region, skills_plugin=skills_plugin)

    agent = Agent(
        model=BedrockModel(model_id="global.anthropic.claude-sonnet-4-6"),
        system_prompt=SYSTEM_PROMPT,
        tools=[registry_mcp],          # Registry 自体が提供する MCP Client
        plugins=[skills_plugin],       # AgentSkills(空)
        hooks=[loader],                # 動的注入の中核
    )
    return agent

Hook 駆動の動的ロード

RegistryLazyLoader というカスタム Hook を定義しています。これは Strands の HookProvider プロトコルに準拠し、AfterToolCallEvent を起点に発火します。具体的には、search_registry_records の tool 実行が完了した直後に同期的に呼ばれ、検索結果から MCP / A2A / Skills を取り出して Agent に動的注入する役割を担います。

src/aws_registry_search_agent/registry_tools.py
class RegistryLazyLoader(HookProvider):
    def register_hooks(self, registry: HookRegistry, **_kwargs: Any) -> None:
        registry.add_callback(AfterToolCallEvent, self._on_after_tool_call)

    def _on_after_tool_call(self, event: AfterToolCallEvent) -> None:
        if event.selected_tool is None:
            return
        if event.selected_tool.tool_name != REGISTRY_SEARCH_TOOL_NAME:
            return  # search_registry_records 以外は無視

        records = self._parse_search_records(event.result)
        if not records:
            return

        mcp_tool_names, a2a_tool_names, skill_names = self._dispatch_records(
            event.agent, records
        )

        # 巨大な registryRecords JSON を 1 行サマリに差し替える
        event.result["content"] = [
            {
                "text": self._format_summary(
                    mcp_tool_names, a2a_tool_names, skill_names, len(records)
                )
            }
        ]

本実装の Hook は「descriptorType ごとの動的注入」と「tool result の 1 行サマリ書き換え」という 2 つの責務を 1 つのコールバックで同時にこなしています。1 つ目の責務は _dispatch_records が担い、descriptorType ごとに MCP / A2A / Skills を Agent に動的追加します。2 つ目の責務は event.result["content"] の上書きで、巨大な registryRecords JSON を 1 行サマリに置き換えて会話履歴から排除します。1 つ目の詳細は後続の MCP / A2A / AGENT_SKILLS の各節で、2 つ目の詳細は次節「tool result 書き換えで膨大な検索結果を会話履歴から排除」で解説します。

このように Hook の利用の利点は、LLM へのプロンプト設計が最小限で済む点にあります。「search_registry_records を呼ぶと自動的にロードされます」と system prompt に 1 行書くだけで、LLM は「検索したら即実行できる」という体験を得られます。「search したら必ず load する」という規律は Hook で強制されるため、LLM 側で注入手順を守る必要がありません。

tool result 書き換えで 膨大な検索結果を会話履歴から排除

Registry MCP の検索結果は record 1 件あたり数 KB の JSON で、複数 record がヒットすると数十 KB の冗長な情報が会話履歴に残り、LLM のコンテキストウィンドウが逼迫してしまいます。Anthropic の Blog Harnessing Claude's Intelligence でも、ツールの実行結果を全てコンテキストに戻すと Agent のレスポンスが遅く・高価になり得ると指摘されています。

本実装では Hook 内で、検索結果である event.result["content"] を 1 行サマリに上書きすることで、この課題を回避します。Strands の tool executor は、AfterToolCallEvent の Hook が完了した後の after_event.result を会話履歴に追加する仕様になっており、Hook 内で content を書き換えると、その書き換え後の値が会話履歴に追加されます。この仕様を利用し、event.result["content"] を "Found 3 record(s). Loaded MCP tools: get_order_status, update_order. Loaded Skills: weather-response-template." のような 1 行サマリに差し替えれば、LLM が見るのはサマリだけになり、巨大な registryRecords JSON はコンテキストに一切入りません。それでも次ターンでは新しい tool / Skill / A2A Agent は既に読み込まれているため、機能性とトークン効率を両立できます。

src/aws_registry_search_agent/registry_tools.py
def _on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    ...  # search_registry_records 以外は早期 return

    records = self._parse_search_records(event.result)
    if not records:
        return

    mcp_tool_names, a2a_tool_names, skill_names = self._dispatch_records(
        event.agent, records
    )
    # 巨大な registryRecords JSON を 1 行サマリに差し替える
    event.result["content"] = [
        {
            "text": self._format_summary(
                mcp_tool_names,
                a2a_tool_names,
                skill_names,
                len(records),
            )
        }
    ]


@staticmethod
def _format_summary(
    mcp_tool_names: list[str],
    a2a_tool_names: list[str],
    skill_names: list[str],
    record_count: int,
) -> str:
    parts = [f"Found {record_count} record(s)."]
    if mcp_tool_names:
        parts.append(f"Loaded MCP tools: {', '.join(mcp_tool_names)}.")
    if a2a_tool_names:
        parts.append(f"Loaded A2A agents: {', '.join(a2a_tool_names)}.")
    if skill_names:
        parts.append(f"Loaded Skills: {', '.join(skill_names)}.")
    if not mcp_tool_names and not a2a_tool_names and not skill_names:
        parts.append("Nothing new to load (already injected or unsupported).")
    return " ".join(parts)

MCP record の動的注入

MCP の動的注入は、検索結果の record の MCP Server endpoint url を基に MCPClient を生成し agent.tool_registry.process_tools に渡す構造です。

src/aws_registry_search_agent/registry_tools.py
def _create_mcp_client(self, url: str) -> MCPClient:
    region = self.region

    def transport() -> Any:
        return aws_iam_streamablehttp_client(
            endpoint=url,
            aws_service="bedrock-agentcore",
            aws_region=region,
        )

    return MCPClient(transport)


def _inject_mcp_records(
    self, agent: Agent, mcp_records: list[dict[str, Any]]
) -> list[str]:
    providers: list[MCPClient] = []
    for record in mcp_records:
        url = self._extract_mcp_url(record)
        if not url or url in self._injected_mcp_urls: # 二重ロード防止のためのガード
            continue
        providers.append(self._create_mcp_client(url))
        self._injected_mcp_urls.add(url)
    if not providers:
        return []
    added: list[str] = agent.tool_registry.process_tools(providers)
    return added

本実装で使っている aws_iam_streamablehttp_client は AWS 公式の mcp-proxy-for-aws パッケージが提供するヘルパーで、boto3 のデフォルトチェーンによる credential 解決と SigV4 署名を接続時に自動で行います。本実装側に credential を取得・凍結する補助コードは不要です。

agent.tool_registry.process_tools([provider]) は、ToolProvider 実装や @tool 関数などを Agent に動的に登録する Strands の標準 API です。具体的には、MCPClient のような ToolProvider を渡すと、内部で load_tools() を呼んで MCP server から tools 一覧を取得し、各 tool を MCPAgentTool でラップして Agent の ToolRegistry に登録します。登録された tool は次の invocation で Bedrock Converse API の toolConfig.tools[] に自動的に載ります。

_injected_mcp_urls: set[str] は二重ロード防止のためのガードです。MCPAgentTool.supports_hot_reload=False の仕様により、同名 tool を再登録すると register_tool が ValueError を投げるため、URL 単位でのスキップが必要になります。

A2A record の動的注入 (A2AAgent + @tool 閉包方式)

A2A record の動的注入では、検索結果の record の AgentCard の情報を基に作成した A2AAgent (Strands SDK のクライアントラッパ) を @tool 閉包でラップし、1 つの A2A Agent につき 1 つの個別 tool として注入します。注入の方法は、MCP record の動的注入と同様に、agent.tool_registry.process_tools に渡すだけです。

src/aws_registry_search_agent/registry_tools.py
def _build_a2a_tool(
    self, url: str, tool_name: str, description: str
) -> AgentTool:
    loader = self

    @tool(name=tool_name, description=description)
    async def _remote(input: str) -> str:
        auth = loader._build_a2a_sigv4_auth()
        async with httpx.AsyncClient(
            auth=auth, timeout=_A2A_INVOKE_TIMEOUT
        ) as http_client:
            remote = A2AAgent(
                endpoint=url,
                name=tool_name,
                description=description,
                client_config=ClientConfig(httpx_client=http_client),
            )
            result = await remote.invoke_async(input)
        message = result.message or {}
        content = message.get("content") if isinstance(message, dict) else None
        if isinstance(content, list) and content:
            first = content[0]
            if isinstance(first, dict) and isinstance(first.get("text"), str):
                return str(first["text"])
        return str(result)

    return _remote

このパターンは Strands 公式ドキュメントの As a Tool パターンと一致します。endpoint (URL) / tool_name / description を Python 閉包で捕獲することで、LLM からは weather_forecast_a2a(input="...") のような普通の個別 tool として見え、URL を LLM に生成させる必要がありません。加えて、Bedrock Converse API の toolConfig.tools[] に A2A Agent ごとに固有の name と description が正規に載るため、複数の A2A Agent がある場合でも description で識別できます。

AGENT_SKILLS record の動的注入

AGENT_SKILLS record の動的注入では、検索結果の record に格納された SKILL.md 本文 (descriptors.agentSkills.skillMd.inlineContent) を Skill.from_content(md) でパースして Skill オブジェクトに変換し、Strands SDK の AgentSkills Plugin を利用して set_available_skills(既存 + 新規) で読み込みます。MCP / A2A と異なり、tool_registry.process_tools ではなく Plugin の API を介して登録する点が特徴です。

AgentSkills は Agent に Skills を統合するための Strands 公式 Plugin で、Agent Skills 仕様 に準拠し、Skill の name と description だけを system prompt の <available_skills> XML ブロックに注入する Discovery、Agent が skills(skill_name=...) ツールを呼んだ時に SKILL.md 本文と関連リソースを返す Activation、Agent が指示に従って処理する Execution、の 3 段階で Progressive Disclosure を実現します。本実装では、Agent 起動時に空の AgentSkills Plugin を Agent(plugins=[AgentSkills(skills=[])]) で登録しておき、Hook 内でその中身を動的に詰め直しています。<available_skills> XML は invocation の前に毎回再構築されるため、Hook で set_available_skills を呼ぶだけで次のターンに新しい Skill の name と description が system prompt に反映されます。

src/aws_registry_search_agent/registry_tools.py
def _inject_skill_records(self, skill_records: list[dict[str, Any]]) -> list[str]:
    existing_names = {s.name for s in self.skills_plugin.get_available_skills()}
    combined: list[SkillSource] = list(self.skills_plugin.get_available_skills())
    added: list[str] = []
    for record in skill_records:
        md = self._extract_skill_md(record)
        if not md:
            continue
        try:
            skill = Skill.from_content(md)
        except ValueError:
            continue
        if skill.name in existing_names:
            continue
        combined.append(skill)
        existing_names.add(skill.name)
        added.append(skill.name)
    if added:
        self.skills_plugin.set_available_skills(combined)
    return added

全体フロー (時系列)

MCP Server の動的注入を例に、ユーザーが「注文 123 のステータスを教えて」と入力した場合の時系列を以下に示します。

ここで注目すべき点は、Hook が同期的に発火するため、ターン 1 終了時点で MCP tool が既に Agent に注入されていることです。具体的には、AfterToolCallEvent は tool 実行直後に同期コールバックとして呼ばれるため、process_tools(...) で登録した tool が次の Agent invocation で即座に LLM に見えます。会話履歴は維持されたまま、次ターンで新 tool / Skill / A2A Agent が利用可能になる、という挙動が成立します。A2A や Skills の場合も同じ流れで、ターン 1 で search_registry_records が呼ばれて Hook が動き、ターン 2 では注入された A2A tool (例: weather_forecast_a2a) や Skill (例: skills(skill_name="weather-response-template")) を直接呼び出せる状態になります。

検証

2 つのシナリオで Agent の動作を検証しました。シナリオ 1 では MCP record の動的注入を、シナリオ 2 では A2A record と AGENT_SKILLS record の同時動的注入を扱い、検索 → 自動ロード → 即時実行のサイクルが各 descriptorType で成立することを確認します。

本検証は、本 README の手順で必要なリソースをCDKでデプロイ後に、run-agent.sh で実行可能です。

シナリオ 1: MCP record の動的ロード

「注文 123 のステータスを教えて」というユーザー発話に対して、Agent は Registry を "order management" で検索し、Order Management MCP Server を動的にロードした上で get_order_status を実行します。

> 注文 123 のステータスを教えて
注文ステータスを確認するためのツールを検索します！
Tool #1: search_registry_records # searchQuery="order management" で検索
注文管理ツールが見つかりました！注文番号 123 のステータスを取得します。
Tool #2: order-management-target___get_order_status
注文 123 の詳細はこちらです！

---

📦 **注文 123 のステータス**

- **ステータス**: 発送済み ✅
- **商品**: Widget Pro × 2（各 $49.99）
- **合計金額**: $99.98
- **配送先**: 123 Main St, New York, NY 10001
- **追跡番号**: TRK-789456

---

シナリオ 2: A2A record と AGENT_SKILLS record の同時動的ロード

「東京の天気を教えて」というユーザー発話に対して、Agent は "weather forecast" で検索し、Weather Forecast Agent (A2A on Runtime) と回答テンプレート用の Skill を 1 回の検索で同時にロードします。その後、次のターンで、その 2 つを並列実行しています。Strands では、LLM が assistant message に複数の toolUse ブロック (今回で言うと Skill 取得と A2A 呼び出し) を出力した場合、それらを並列実行します。

> 東京の天気を教えて
**「東京」**の天気をお調べしますね！天気情報のツールを検索します。
Tool #1: search_registry_records # searchQuery="weather forecast" で検索
天気予報エージェントが見つかりました！東京の天気を取得します。
Tool #2: skills

Tool #3: Weather_Forecast_Agent
東京の天気をお伝えします！

---

【現在の天気】
- 場所: 東京
- 天候: ☀️ 晴れ時々曇り

【今日の予報】
- 最高気温: **18°C** / 最低気温: **9°C**
- 降水確率: 低め

【備考】
朝晩は冷え込みますので、羽織れる上着を一枚持参されることをおすすめします。🧥

---

おわりに

本稿では、AWS Agent Registry の基礎や、A2A Agent / MCP Server / Skills の基礎と Agent Registry への登録方法を整理し、Agent Registry から動的にリソースを取得可能な Agent の実装パターンを解説しました。Agent の実装として、Strands Agent の Hook (AfterToolCallEvent) で 、MCP / Skills / A2A の 3 種を Agent に動的注入する仕組みと、コンテキストエンジニアリングのために tool 結果をサマリに書き換える設計を紹介しました。

この Agent の構成の利点は、Agent 側のコードを再デプロイせず、Registry に新しい MCP / Skills / A2A Agent を登録するだけで Agent から利用可能になる点にあります。組織横断で MCP / Skills / Agent を共通基盤として運用したい場合や、複数チームの成果物を 1 つの Agent から横断利用するハブを構築したい場合に、本稿の実装パターンが参考になれば幸いです。

補足: AgentCore SDK の serve_a2a で AgentCard の skills[] を反映する

AgentCore SDK には、Strands などのフレームワークで実装した Agent を A2A Server として AgentCore Runtime にデプロイするためのヘルパー関数 serve_a2a が用意されています。serve_a2a(executor) のように Agent を渡すだけで A2A Server を起動できますが、その際に自動生成される AgentCard の skills[] には、Agent の name と description しか含まれず、Strands Agent に @tool で登録したツールの詳細が反映されない仕様になっています。

そのため、@tool の内容を AgentCard の skills[] に反映したい場合は、AgentCard を明示的に組み立てて serve_a2a の第 2 引数に渡す必要があります。検証時の実装では、以下のように Strands Agent の agent.tool_registry.get_all_tools_config() から skills[] を組み立てて渡しています。

assets/agents/weather/weather_a2a_agent.py
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from bedrock_agentcore.runtime.a2a import serve_a2a
from strands.multiagent.a2a.executor import StrandsA2AExecutor

# agent = Agent(...)  # Strands Agent (@tool 登録済み) を構築

card = AgentCard(
    name=AGENT_NAME,
    description=AGENT_DESCRIPTION,
    url=os.environ.get("AGENTCORE_RUNTIME_URL", "http://127.0.0.1:9000/"),
    version="0.1.0",
    capabilities=AgentCapabilities(streaming=True),
    default_input_modes=["text"],
    default_output_modes=["text"],
    skills=[
        AgentSkill(
            id=cfg["name"],
            name=cfg["name"],
            description=cfg["description"],
            tags=[],
        )
        for cfg in agent.tool_registry.get_all_tools_config().values()
    ],
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent), card)

本トピックの詳細は以下の issue / 公式サンプルを参照ください。

参考

AWS / AgentCore

Strands Agents

Anthropic Engineering Blogs

プロトコル / 仕様

実装

GitHubで編集を提案
アマゾン ウェブ サービス ジャパン (有志)
アマゾン ウェブ サービス ジャパン (有志)

この Publication に投稿している記事は、アマゾン ウェブ サービス ジャパン合同会社または Amazon Web Services, Inc. 所属社員による個人の見解であり、所属する組織の公式見解ではありません。参加したい従業員の方は、Sugiyama Suguru までお知らせください。

Discussion