【AWS】AWS Agent Registryを使ってみた！

はじめに

Agent Registryが発表されましたね。早速試してみました。

この記事では、公式ドキュメントをもとにAgentCore Registryの概要を整理しつつ、実際にPythonからAPIを叩いて動作確認した内容をまとめます。

Agent Registryとは

Agent Registry は、MCPサーバー・AIエージェント・スキルを組織全体で一元管理・検索できるフルマネージドカタログサービスのようなものです。

人間からでも、AIエージェントからでも、「どんなツールが使えるか」を検索・発見できる仕組みを提供します。

また、AWSだけでなく他のクラウド、オンプレミスで構築されたものであっても登録可能であるとのことです。

また、私は勘違いしていたのですが、登録されたものを好き勝手に実行できる環境ではない点に注意です。
あくまで登録して管理するものだと捉えると無難です。
Registry経由で使うこともできますがそれはまた今度。

Agent Registryで何を管理するのか

Registryには以下の4種類のリソースを登録できます。

タイプ	説明
MCP Server	AIエージェントが実行できるツール群。MCP（Model Context Protocol）プロトコルに準拠
Agent	自律的に推論・計画・実行するエージェント。A2A（Agent-to-Agent）プロトコルに対応
Skill	エージェント間で共有・再利用可能な機能単位
Custom	上記以外のカスタムリソース。JSONスキーマで自由に定義

コンソール画面では以下のようになっています。

Registryの各機能について

公式ドキュメント（Registry Concepts）をもとに各機能を整理します。

Registry（レジストリ）の構成

Registryは「レコードの入れ物」です。AWSアカウント内に複数作れるので、たとえば次のような使い分けが可能です。

• 本番環境 / QA / 開発環境ごとに分ける
• チームや事業部ごとに分ける
• MCPサーバー専用・エージェント専用と種類で分ける

Agent Registry の主要ペルソナ

Agent Registryは、複数の役割（ペルソナ）によって運用される仕組みです。
ここでは代表的な4つのペルソナを整理します。

Administrator（管理者）

Registry全体を管理する責任者です。

• Registryの作成・構成（チーム単位 / 環境単位など）
• 認証方式の決定（IAM / JWT）
• 承認フローの設計（手動 or 自動）
• IAMによるアクセス制御
• EventBridge連携による通知・審査フロー構築

→ Registryの「ルール」と「ガバナンス」を定義する役割

Publisher（パブリッシャー）

エージェントやツールを作成し、Registryに登録する人です。

• Agent / MCP Server / Skill の作成
• Registry Recordの作成
• メタデータ・Capability・バージョン情報の定義
• Draftで編集し、承認申請
• フィードバックを受けて修正・再提出

→エージェントを「他の人が使える状態」にする役割

Curator / Approver（キュレーター）

Registryの品質を担保する役割です。

• 登録されたRecordの審査（セキュリティ・品質・メタデータ）
• 承認 / 却下の判断
• フィードバックの提供
• 不要・古いリソースのDeprecated化

→ Registryを「信頼できるカタログ」に保つ役割

Consumer（消費者）

エージェントやツールを利用する人、または他のエージェントです。

• Registryを検索（自然言語 / キーワード）
• 承認済みリソースの発見（Discovery）
• Agent / MCP Server の利用
• IAM / JWTで認証してアクセス

→ 必要な機能を持つエージェントを見つけて利用する役割

---

これらのペルソナによって、

• 作成（Publisher）
• 管理（Administrator）
• 審査（Curator）
• 利用（Consumer）

というライフサイクルで運用されます。

これにより、エージェントは単なるコードではなく
「組織で管理・共有されるリソース」として扱われるようになります。

動作イメージ

ここまででまとめたAgent Registryの動作イメージはこんな感じです。
かなりわかりやすく抽象化しています()
ペルソナによる責務の分離や、エージェントをはじめとしたリソースを組織単位で管理することで再利用可能にしていることがわかるかと思います。

なぜAgent Registryが必要なのか

では、なぜAgent Registryが重要になってくるのでしょうか。
そこもみていきます。

従来の課題

AIエージェント開発が組織全体に広がると、こんな問題が出てきます。

1. ツールがどこにあるか分からない

チームAが作ったMCPサーバー、チームBが作ったエージェント、それぞれがSlackやWikiに散らばっていて「どこに何があるか」を把握するのが大変です。

2. 重複開発

既存のツールを知らずに「同じようなもの」を別のチームが再開発。リソースの無駄と、メンテナンスコストの増加につながります。

3. ガバナンスの欠如

誰でも自由にツールを使えてしまうと、セキュリティレビューが通っていないツールが本番のエージェントに組み込まれるリスクがあります。

Registryによる解決

課題	Registryの対応
ツールが散在している	一元化されたカタログで全リソースを検索可能に
重複開発	既存ツールを検索・発見してから開発開始できる
ガバナンス不足	承認ワークフローにより、審査済みリソースのみ公開

想定される使用例

公式ドキュメントを参考に、実際のユースケースを考えてみます。

例1：エンタープライズAIプラットフォームの整備

大企業でAI活用が進んでいると、各部門がそれぞれMCPサーバーやエージェントを開発します。

データ分析チーム → BigQuery MCPサーバーを開発
マーケティングチーム → Salesforce連携エージェントを開発
セキュリティチーム → ログ分析エージェントを開発

これらをすべてRegistryに登録しておけば、新しいエージェントを作る際に「使えるツールはないか？」をRegistry検索一発で確認できます。AI Agentが自律的にRegistryを検索して、必要なツールを動的に組み合わせることも可能です。

例2：Agentの動的ツール発見

# Agentが起動時にRegistryを参照して使えるツールを自律的に選択
agent = Agent(model=MODEL_ID)
available_tools = search_registry("データ分析")
# → Registry から「data-analysis-mcp」「bigquery-mcp」が見つかる
agent.use_tools(available_tools)

Registryがあることでハードコードされたツール一覧ではなく、その時点でRegistryに承認されているツールを動的に組み込めます。

例3：開発環境の分離

production-registry  ← 本番用（厳格な審査）
staging-registry     ← ステージング用
dev-registry         ← 開発者が自由に実験

環境ごとにRegistryを分けることで、開発中のツールが本番エージェントに誤って使われることを防げます。

やってみた

以下、公式ドキュメントを参考に試してみました。

検証環境

項目	内容
OS	macOS
Python	3.12
パッケージ管理	uv
AWSリージョン	us-east-1
boto3	1.42.86
strands-agents	1.35.0
bedrock-agentcore	1.6.0

# 環境構築
uv init agentcore-registry-test
cd agentcore-registry-test
uv add boto3 "botocore[crt]" strands-agents bedrock-agentcore requests aws-requests-auth

実施手順

⚠️ 注意：現時点でboto3 SDKは未対応

まず最初のハマりポイントとして、2026年4月時点ではboto3（1.42.86）にAgentCore Registry APIが含まれていません。

import boto3
client = boto3.client('bedrock-agentcore-control', region_name='your-region')
print([m for m in dir(client) if 'registry' in m.lower()])
# → [] ← 空！Registry関連メソッドが存在しない

そのため、直接HTTP + SigV4署名でAPIを呼び出す必要があります。

import json
import boto3
import requests
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

REGION = "your-region"
BASE_URL = f"https://bedrock-agentcore-control.{REGION}.amazonaws.com"

session = boto3.session.Session(region_name=REGION)
credentials = session.get_credentials().get_frozen_credentials()

def signed_request(method, path, body=None):
    url = BASE_URL + path
    data = json.dumps(body) if body else ""
    request = AWSRequest(
        method=method, url=url, data=data,
        headers={"Content-Type": "application/json"},
    )
    SigV4Auth(credentials, "bedrock-agentcore", REGION).add_auth(request)
    prep = request.prepare()
    return requests.request(method=method, url=url, headers=dict(prep.headers), data=data)

STEP 1：Registry を作成する

# Registry 作成
response = signed_request("POST", "/registries", {
    "name": "my-first-registry",
    "description": "テスト用Registry",
    "authorizationConfig": {"authorizationType": "IAM"},
    "approvalConfiguration": {"autoApproval": False}  # 手動承認モード
})
# → 202 Accepted
# {
#   "registryArn": "arn:aws:bedrock-agentcore:us-east-1:xxxx:registry/hB2G7OqPFX075ATT"
# }

作成後すぐには使えず、CREATING → READY と状態遷移します（約30秒）。

# READY になるまで待機
import time
registry_id = "hB2G7OqPFX075ATT"
while True:
    resp = signed_request("GET", f"/registries/{registry_id}")
    status = resp.json().get("status")
    print(f"Status: {status}")
    if status == "READY":
        break
    time.sleep(5)

STEP 2：MCPサーバーをRegistryに登録する

天気情報を提供するMCPサーバーのメタデータを登録してみます。

import json

# MCPサーバーの定義
server_content = json.dumps({
    "name": "io.example/weather-server",
    "description": "天気情報を提供するMCPサーバー",
    "version": "1.0.0"
})

# ツール定義
tools_content = json.dumps({
    "tools": [
        {
            "name": "get_current_weather",
            "description": "指定した都市の現在の天気を取得します",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "都市名（例: Tokyo）"}
                },
                "required": ["city"]
            }
        },
        {
            "name": "get_forecast",
            "description": "指定した都市の5日間天気予報を取得します",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "都市名"},
                    "days": {"type": "integer", "description": "予報日数（1-5）"}
                },
                "required": ["city"]
            }
        }
    ]
})

# Registry に登録
response = signed_request("POST", f"/registries/{registry_id}/records", {
    "name": "weather-mcp-server",
    "description": "天気情報MCPサーバー",
    "recordVersion": "1.0",
    "descriptorType": "MCP",
    "descriptors": {
        "mcp": {
            "server": {"schemaVersion": "2025-12-11", "inlineContent": server_content},
            "tools": {"protocolVersion": "2024-11-05", "inlineContent": tools_content}
        }
    }
})
# → 202 Accepted
# {
#   "recordArn": "arn:aws:bedrock-agentcore:us-east-1:xxxx:registry/.../record/qrDz9L6MBSVo",
#   "status": "CREATING"
# }

作成直後のレコードは CREATING → DRAFT と遷移します。

STEP 3：承認ワークフロー

承認申請

DRAFT 状態のレコードを承認申請します。

record_id = "qrDz9L6MBSVo"

# 承認申請（APIパス注意: submit-for-approval）
response = signed_request(
    "POST",
    f"/registries/{registry_id}/records/{record_id}/submit-for-approval",
    {}
)
# → 202 Accepted
# {
#   "status": "PENDING_APPROVAL"
# }

承認（管理者として）

# 承認（statusReason が必須！）
response = signed_request(
    "PATCH",
    f"/registries/{registry_id}/records/{record_id}/status",
    {
        "status": "APPROVED",
        "statusReason": "セキュリティレビュー済み・本番利用可"
    }
)
# → 202 Accepted
# {
#   "status": "APPROVED"
# }

ハマりポイント：statusReason が必須です。なしだと 400 Bad Request になります。

承認後のレコード詳細を確認すると、ステータスが APPROVED になっています。

{
  "name": "weather-mcp-server",
  "status": "APPROVED",
  "statusReason": "セキュリティレビュー済み・本番利用可",
  "descriptorType": "MCP",
  "descriptors": {
    "mcp": {
      "server": { "inlineContent": "...", "schemaVersion": "2025-12-11" },
      "tools": { "inlineContent": "...", "protocolVersion": "2024-11-05" }
    }
  }
}

STEP 4：Strands Agentから Registry を利用する

AgentCore Registryと Strands Agents を組み合わせると、エージェントが動的にRegistryを検索してツール情報を取得できます。

from strands import Agent, tool
import json

MODEL_ID = "us.anthropic.claude-sonnet-4-5-20250929-v1:0"

@tool
def search_registry(query: str) -> str:
    """AgentCore Registry からキーワードでツールを検索します。

    Args:
        query: 検索キーワード（例: 天気, database）

    Returns:
        検索結果
    """
    resp = signed_request("GET", f"/registries/{registry_id}/records")
    records = resp.json().get("registryRecords", [])
    query_lower = query.lower()

    matched = [r for r in records
               if query_lower in r.get("name", "").lower()
               or query_lower in r.get("description", "").lower()]

    if not matched:
        return f"'{query}' に一致するツールが見つかりませんでした"

    results = []
    for r in matched:
        results.append(f"✅ {r['name']} - {r.get('description')}")
    return f"検索結果 ({len(matched)}件):\n" + "\n".join(results)


@tool
def get_tool_detail(tool_name: str) -> str:
    """Registryに登録されたツールの詳細を取得します。

    Args:
        tool_name: ツール名

    Returns:
        ツールの詳細情報（パラメータ等）
    """
    resp = signed_request("GET", f"/registries/{registry_id}/records")
    records = resp.json().get("registryRecords", [])
    target = next((r for r in records if r.get("name") == tool_name), None)
    if not target:
        return f"'{tool_name}' が見つかりません"

    record_id_local = target.get("recordId")
    detail = signed_request(
        "GET", f"/registries/{registry_id}/records/{record_id_local}"
    ).json()

    mcp = detail.get("descriptors", {}).get("mcp", {})
    tools_data = json.loads(mcp.get("tools", {}).get("inlineContent", "{}"))

    lines = [f"📦 {detail['name']} v{detail.get('recordVersion')}"]
    for t in tools_data.get("tools", []):
        props = t.get("inputSchema", {}).get("properties", {})
        lines.append(f"\n  • {t['name']}: {t.get('description')}")
        for param, info in props.items():
            lines.append(f"    - {param}: {info.get('description')}")
    return "\n".join(lines)


# Agent にRegistryツールを渡す
agent = Agent(
    model=MODEL_ID,
    system_prompt="AgentCore Registryのアシスタントです。ツールを使って質問に答えてください。",
    tools=[search_registry, get_tool_detail],
)

# 実際に質問してみる
result = agent("天気関連のツールはありますか？使い方も教えてください。")
print(result)

動かしてみる

登録されているものを確認する

まずRgistryに登録されているものを確認します。
先ほど作成、登録したものが確認できますね。

新しくRegistryに登録してみる

次に新しく登録してみます。
ここで新しく作成されたメール送信ツールは承認待ち状態となります。

承認待ち状態のツールはコンソール上から確認できます。

こいつを「承認者へ送信」その後「承認」とすることで利用可能となります。

さっき責務の分離ガ〜とか言ってたのにこの流れを私一人でやっては意味ないのですが、
本来は別々の人間がやるべきです笑

Strands+Registryで作成したツールを利用してみる

せっかくなので作成したメース送信ツールを使ってみたいと思います。
以下のコードで実装しました。Strandsを作成し、Registry経由でツールを利用します。
実際にメールを送るため、SESでのアドレスの検証が必要となっています。

email_agent.py

"""
Registry から email-sender を発見して使う Strands Agent
SESを利用しているため、事前に宛先アドレスの検証が必要。
実行: uv run python email_agent.py
"""
import json
import boto3
import requests
import smtplib
from email.mime.text import MIMEText
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from strands import Agent, tool
from strands.models.bedrock import BedrockModel

# --- 設定 ---
REGION = "your-regin"
BASE_URL = f"https://bedrock-agentcore-control.{REGION}.amazonaws.com"
REGISTRY_ID = "hogehoge"
MODEL_ID = "us.anthropic.claude-sonnet-4-5-20250929-v1:0"

session = boto3.session.Session(region_name=REGION)
credentials = session.get_credentials().get_frozen_credentials()

# Strands に明示的にリージョンを指定
bedrock_model = BedrockModel(model_id=MODEL_ID, region_name=REGION)


def signed_request(method, path, body=None):
    url = BASE_URL + path
    data = json.dumps(body) if body else ""
    req = AWSRequest(method=method, url=url, data=data,
                     headers={"Content-Type": "application/json"})
    SigV4Auth(credentials, "bedrock-agentcore", REGION).add_auth(req)
    prep = req.prepare()
    return requests.request(method=method, url=url, headers=dict(prep.headers), data=data)


# ========================================
# STEP 1: Registry から email-sender を発見
# ========================================
def discover_email_tool_from_registry() -> dict | None:
    """Registryを検索してemail-senderのメタデータを取得する"""
    resp = signed_request("GET", f"/registries/{REGISTRY_ID}/records")
    if resp.status_code != 200:
        return None

    records = resp.json().get("registryRecords", [])
    target = next((r for r in records if r.get("name") == "email-sender"), None)
    if not target:
        return None

    # 詳細を取得
    record_id = target.get("recordId")
    detail = signed_request("GET", f"/registries/{REGISTRY_ID}/records/{record_id}").json()

    mcp = detail.get("descriptors", {}).get("mcp", {})
    tools_raw = mcp.get("tools", {}).get("inlineContent", "{}")
    tools_data = json.loads(tools_raw)

    return {
        "name": detail.get("name"),
        "status": detail.get("status"),
        "tools": tools_data.get("tools", []),
    }


print("🔍 Registry から email-sender を探しています...")
email_tool_info = discover_email_tool_from_registry()

if not email_tool_info:
    print("❌ email-sender が Registry に見つかりません")
    exit(1)

print(f"✅ 発見！ [{email_tool_info['status']}] {email_tool_info['name']}")
print(f"   利用可能なツール: {[t['name'] for t in email_tool_info['tools']]}")


# ========================================
# STEP 2: Strands Agent のツールを定義
# ========================================
@tool
def send_email(to: str, subject: str, body: str) -> str:
    """メールを送信します。

    Args:
        to: 送信先メールアドレス
        subject: メールの件名
        body: メールの本文

    Returns:
        送信結果メッセージ
    """
    # Registry から取得したメタデータを表示
    print(f"\n  📧 [email-sender] メール送信")
    print(f"     To: {to}")
    print(f"     Subject: {subject}")
    print(f"     Body: {body[:50]}...")

    ses = boto3.client("sesv2", region_name="your-region")
    ses.send_email(
        FromEmailAddress="ここにメアド",
        Destination={"ToAddresses": [to]},
        Content={
            "Simple": {
                "Subject": {"Data": subject},
                "Body": {"Text": {"Data": body}},
            }
        },
    )
    return f"✅ メール送信完了\n  宛先: {to}\n  件名: {subject}"


# ========================================
# STEP 3: Strands Agent を作成
# ========================================
agent = Agent(
    model=bedrock_model,
    system_prompt=(
        "あなたはメール送信アシスタントです。\n"
        "ユーザーの指示に従い send_email ツールを使ってメールを送ってください。\n"
        "必要な情報（宛先・件名・本文）が不足している場合は確認してください。"
    ),
    tools=[send_email],
)

# ========================================
# STEP 4: 実行
# ========================================
print("\n" + "="*50)
print("🤖 email-sender Agent 起動")
print("="*50)
print("'quit' で終了\n")

while True:
    try:
        user_input = input("あなた > ").strip()
    except (EOFError, KeyboardInterrupt):
        print("\n終了します")
        break

    if not user_input or user_input.lower() in ["quit", "exit", "q"]:
        print("終了します")
        break

    print("\nAgent > ", end="", flush=True)
    response = agent(user_input)
    print(f"{response}\n")

実際に届いたメールは以下になります。

ここまででRegistryの動作確認とStrandsからの利用を試してみました。

さいごに

Agent Registryを実際に試してみました。
これまた面白い機能が出てきたな〜という感じです。

AIエージェントが組織で本格的に活用されるようになると、ツールやエージェントを管理、整理する仕組みは必須になりますよね。
Agent Registryはそのための基盤として、シンプルで使いやすい設計になっていると感じました。
とはいえ、組織内でAIそのものがまだ全然浸透していない場合はどうするんだろうな〜と思ったり...

ちなみに、「AgentCore Registry」ではなく「Agent Registry」でいいんでしょうか？
ドキュメントでは「Agent Registry」となっているから、こっちが正なのかな...
とはいえマネコンがAgentCoreのページと一緒にあるから名前がややこしい〜〜〜

参考リンク