AWS Agent RegistryからSkillsを動的に読み込んでStrands Agentに注入してみた

uv init skills-dynamic-import-strands -p 3.13
cd skills-dynamic-import-strands
uv add "boto3>=1.42.87" "strands-agents>=1.36.0" "strands-agents-tools>=0.5.0" "pyyaml>=6.0"

.
├── pyproject.toml
├── register_skill.py          # SKILL.mdをRegistryへ登録する
├── registry_skill_loader.py   # Registryから検索してSkillに変換する
├── agent.py                   # スキル有/無でAgentを組み立てて比較する
└── skills/
    └── doc-reviewer/
        └── SKILL.md

uv run python3 -c "
import boto3, json
c = boto3.client('bedrock-agentcore-control', region_name='us-east-1')
resp = c.create_registry(
    name='skills-auto-approve',
    description='Agent Skills registry with auto-approval enabled',
    approvalConfiguration={'autoApproval': True},
)
print(json.dumps(resp, indent=2, default=str))
"

---
name: doc-reviewer
description: 仕様書やドキュメントを日本語で構造化レビューし、要点・リスク・改善提案を優先度付きで返す
---

あなたはドキュメントレビュー担当の日本語アシスタントです。

入力として渡された文書を、必ず以下のMarkdown見出し構成で出力してください。
順序・見出し名・絵文字は変えないこと。

## 🎯 要点 (TL;DR)
- 箇条書きで最大3項目、1項目あたり40文字以内で端的にまとめる

## ⚠️ リスク
- 箇条書きで最大5項目。各項目は「リスク内容 — 想定される影響」の形式で書く

## 🛠 改善提案
- 箇条書きで最大5項目。必ず冒頭に `[P1]`/`[P2]`/`[P3]` の優先度タグを付ける
- `[P1]` は即時対応が必要なもの、`[P2]` は次スプリント相当、`[P3]` はいつかやれば良いもの

上記以外のセクションは追加しないこと。冗長な前置きや結論段落は書かないこと。

"""Register a local SKILL.md to AWS Agent Registry and approve it immediately."""
from __future__ import annotations

import argparse
import logging
import os
import re
import time
from pathlib import Path

import boto3
import yaml

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n(.*)$", re.DOTALL)

def _wait_until_not(control, registry_id: str, record_id: str, transient: str, *, timeout: float = 30.0) -> str:
    deadline = time.monotonic() + timeout
    while True:
        status = control.get_registry_record(
            registryId=registry_id,
            recordId=record_id,
        )["status"]
        if status != transient:
            return status
        if time.monotonic() > deadline:
            raise TimeoutError(f"record {record_id} stuck in {transient}")
        time.sleep(1.0)

def parse_skill_md(path: Path) -> tuple[str, str, str]:
    text = path.read_text(encoding="utf-8")
    match = FRONTMATTER_RE.match(text)
    if not match:
        raise ValueError(f"{path} must start with YAML frontmatter")
    meta = yaml.safe_load(match.group(1)) or {}
    name = meta.get("name")
    if not name:
        raise ValueError(f"{path} frontmatter must define 'name'")
    return name, meta.get("description", ""), text

def register_skill(registry_id: str, skill_md_path: Path, record_version: str = "1.0", region_name: str = "us-east-1") -> str:
    name, description, inline_content = parse_skill_md(skill_md_path)
    control = boto3.client("bedrock-agentcore-control", region_name=region_name)

    created = control.create_registry_record(
        registryId=registry_id,
        name=name,
        description=description,
        recordVersion=record_version,
        descriptorType="AGENT_SKILLS",
        descriptors={"agentSkills": {"skillMd": {"inlineContent": inline_content}}},
    )
    record_id = created["recordArn"].rsplit("/", 1)[-1]
    logger.info("Created record %s (status=%s)", record_id, created["status"])

    _wait_until_not(control, registry_id, record_id, "CREATING")
    control.submit_registry_record_for_approval(registryId=registry_id, recordId=record_id)
    status = _wait_until_not(control, registry_id, record_id, "SUBMITTING")
    logger.info("Record %s reached status %s", record_id, status)

    return record_id

export AGENT_REGISTRY_ID="xxx"
export AWS_REGION="us-east-1"
uv run python register_skill.py skills/doc-reviewer/SKILL.md

INFO:__main__:Created record xxx (status=CREATING)
INFO:__main__:Record xxx reached status APPROVED

"""Load Agent Skills dynamically from AWS Agent Registry."""
from __future__ import annotations

import logging
from dataclasses import dataclass

import boto3
from strands.vended_plugins.skills import Skill

logger = logging.getLogger(__name__)

@dataclass
class RegistrySkillLoaderConfig:
    registry_id: str
    region_name: str = "us-east-1"
    max_results: int = 10

class RegistrySkillLoader:
    def __init__(self, config: RegistrySkillLoaderConfig) -> None:
        self._config = config
        self._client = boto3.client("bedrock-agentcore", region_name=config.region_name)

    def search(self, query: str) -> list[Skill]:
        response = self._client.search_registry_records(
            registryIds=[self._config.registry_id],
            searchQuery=query,
            maxResults=self._config.max_results,
            filters={"descriptorType": {"$eq": "AGENT_SKILLS"}},
        )
        records = response.get("registryRecords") or []
        logger.info("Fetched %d skill record(s) from registry", len(records))
        return [skill for record in records if (skill := self._to_skill(record))]

    def _to_skill(self, record: dict) -> Skill | None:
        descriptors = record.get("descriptors") or {}
        markdown = ((descriptors.get("agentSkills") or {}).get("skillMd") or {}).get("inlineContent")
        if markdown:
            return Skill.from_content(markdown)
        name = record.get("name", "unknown")
        description = record.get("description") or ""
        logger.warning("Skill '%s' has no skillMd content; falling back", name)
        return Skill(name=name, description=description, instructions=description)

"""Compare Strands Agent output with and without Skills loaded from AWS Agent Registry."""
from __future__ import annotations

import argparse
import logging
import os

from strands import Agent
from strands.models import BedrockModel
from strands.vended_plugins.skills import AgentSkills

from registry_skill_loader import RegistrySkillLoader, RegistrySkillLoaderConfig

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)

HAIKU_MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"

def build_model() -> BedrockModel:
    return BedrockModel(
        model_id=HAIKU_MODEL_ID,
        region_name=os.environ.get("AWS_REGION", "us-east-1"),
        temperature=0.0,
    )

def build_plain_agent() -> Agent:
    return Agent(model=build_model())

def build_skill_agent(search_query: str) -> Agent:
    loader = RegistrySkillLoader(RegistrySkillLoaderConfig(
        registry_id=os.environ["AGENT_REGISTRY_ID"],
        region_name=os.environ.get("AWS_REGION", "us-east-1"),
    ))
    skills = loader.search(search_query)
    logger.info("Loaded skills: %s", [s.name for s in skills])
    return Agent(model=build_model(), plugins=[AgentSkills(skills=skills)])

SAMPLE_DOC = """\
# 新機能: 社内ナレッジ検索ボット v0.3 仕様書 (ドラフト)

- 利用者: 全社員 (約1200名)
- 想定QPS: 平均3, ピーク20
- LLM: Claude Haiku / Sonnet を動的切替
- データソース: Confluence, Google Drive, Slack 過去ログ
- 認証: Slack OAuth のみ
- データ取り込み: 夜間バッチで全件再同期
- ログ: CloudWatch に会話ログを平文保存 (検索用)
- リリース予定: 来週金曜日, 段階ロールアウトなし
- 監視: なし (まずは出してみる)
- 予算: LLMコストは月上限なし
"""

PROMPT = f"下記のドキュメントをレビューしてください。\n\n{SAMPLE_DOC}"

def run(mode: str) -> None:
    print(f"MODE: {mode}")
    agent = build_plain_agent() if mode == "plain" else build_skill_agent("仕様書をレビューして要点・リスク・改善提案を返すスキル")
    print(agent(PROMPT).message)

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("mode", choices=["plain", "skills"])
    run(parser.parse_args().mode)

if __name__ == "__main__":
    main()

AGENT_REGISTRY_ID="xxx" AWS_REGION="us-east-1" uv run python agent.py plain

AGENT_REGISTRY_ID="xxx" AWS_REGION="us-east-1" uv run python agent.py skills

registry_skill_loader INFO: Fetched 1 skill record(s) from registry
__main__ INFO: Loaded skills: ['doc-reviewer']
...
doc-reviewer スキルを使用して、このドキュメントを構造化レビューします。
Tool #1: skills