AWS Agent Registry に MCP と エージェント を載せて、オーケストレーターから動的呼び出しするまで

"""関西弁変換エージェント（Strands Agents + Bedrock Claude Sonnet 4.5）"""

from strands import Agent
from strands.models import BedrockModel

# 関西弁変換のシステムプロンプト
KANSAI_SYSTEM_PROMPT = """\
あなたはコテコテの大阪のおっちゃんです。
入力された文章を、技術的な正確性を保ちながら、めっちゃ濃い関西弁に翻訳してください。

ルール:
- AWSサービス名や技術用語（CloudFront、ALB、ECS Fargate 等）はそのまま残す
- 元の意味を変えない
- 翻訳結果のみを返す（前置きや説明は不要）

関西弁のトーン:
- とにかくテンポ良く、親しみやすく、ちょっとうるさいくらいで
- 語尾は「やで」「やんけ」「やろ」「やなぁ」「やがな」「ちゃう？」を多用
- 文頭に「ほな」「せやから」「ええか」「あんな」を入れる
- 「めっちゃ」「ごっつ」「めちゃくちゃ」を強調語として使う
- 提案・推奨は「〜しとき」「〜したらええ」「〜にしとこ」
- 断定は「〜やで」「〜やがな」「〜に決まっとるやろ」
- 必要なら「で？」「知らんけど」を文末に添える

例:
- 「CloudFrontを使うのがおすすめです」
→ 「ほな CloudFront 使うんがええで、めっちゃ速なるからな」

- 「ALBの後ろにECSを並べましょう」
→ 「せやから ALB の後ろに ECS 並べたるんやで、これ鉄板やろ」

- 「Multi-AZにしましょう」
→ 「Multi-AZ にしとき、なんかあったらシャレならんで」
"""

def create_agent() -> Agent:
    """関西弁変換エージェントを作成する"""
    model = BedrockModel(
        model_id="jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name="ap-northeast-1",
    )

    return Agent(
        model=model,
        system_prompt=KANSAI_SYSTEM_PROMPT,
        callback_handler=None,
    )

def translate_to_kansai(text: str) -> str:
    """入力テキストを関西弁に翻訳する"""
    agent = create_agent()
    result = agent(text)
    return result.message["content"][0]["text"]

def main():
    """ローカル動作確認用"""
    user_input = (
        "CloudFront で画像配信して、ALB の後ろに ECS Fargate を"
        "並べる構成がおすすめです。データベースは Aurora MySQL の"
        "Multi-AZ にしましょう。"
    )

    print("=" * 60)
    print("入力:")
    print(user_input)
    print("=" * 60)

    output_text = translate_to_kansai(user_input)

    print("出力:")
    print(output_text)
    print("=" * 60)

if __name__ == "__main__":
    main()

"""関西弁変換エージェント A2A サーバー（AgentCore Runtime 向け）"""

from strands.multiagent.a2a import A2AServer
from agent import create_agent

def main():
    agent = create_agent()

    # A2A サーバーとしてラップ
    # AgentCore Runtime の A2A プロトコルでは port=9000 が必須
    a2a_server = A2AServer(
        agent=agent,
        host="0.0.0.0",
        port=9000,
    )

    # サーバー起動
    a2a_server.serve()

if __name__ == "__main__":
    main()

FROM public.ecr.aws/docker/library/python:3.12-slim
WORKDIR /app
# uv をインストール
RUN pip install --no-cache-dir uv
# 依存関係ファイルをコピーして先にインストール（キャッシュ効かせるため）
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
# アプリケーションコードをコピー
COPY agent.py a2a_server.py ./
# A2A サーバーがリッスンするポート（AgentCore Runtime A2A は 9000 必須）
EXPOSE 9000
# サーバー起動
CMD ["uv", "run", "python", "a2a_server.py"]

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "bedrock-agentcore.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "BedrockInvokeModel",
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream"
      ],
      "Resource": "*"
    },
    {
      "Sid": "ECRImageAccess",
      "Effect": "Allow",
      "Action": [
        "ecr:BatchGetImage",
        "ecr:GetDownloadUrlForLayer",
        "ecr:GetAuthorizationToken"
      ],
      "Resource": "*"
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents",
        "logs:DescribeLogStreams"
      ],
      "Resource": "*"
    }
  ]
}

kansai-translator-agent/
├ agent.py             # 関西弁エージェント本体（既出）
├ a2a_server.py        # A2A サーバー（既出）
├ Dockerfile           # コンテナ定義（既出）
├ sigv4_auth.py        # httpx 用 SigV4 認証ヘルパー（新規）
├ registry_loader.py   # Registry から A2A エージェントを取得（新規）
└ orchestrator.py      # オーケストレーター本体（新規）

"""httpx 用の SigV4 認証ヘルパー"""

from __future__ import annotations

import boto3
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

class SigV4HttpxAuth(httpx.Auth):
    requires_request_body = True

    def __init__(self, region: str, service: str = "bedrock-agentcore") -> None:
        credentials = boto3.Session().get_credentials()
        if credentials is None:
            raise RuntimeError("No AWS credentials found.")
        self._signer = SigV4Auth(credentials, service, region)

    def auth_flow(self, request: httpx.Request):
        aws_request = AWSRequest(
            method=request.method,
            url=str(request.url),
            data=request.content,
            headers=dict(request.headers),
        )
        self._signer.add_auth(aws_request)

        for key, value in aws_request.headers.items():
            request.headers[key] = value

        yield request

"""AWS Agent Registry から MCP サーバーとエージェントを動的に取得する"""

from __future__ import annotations

import json
import logging
import os
from dataclasses import dataclass

import boto3
from strands_tools.a2a_client import A2AClientToolProvider

from sigv4_auth import SigV4HttpxAuth

logger = logging.getLogger(__name__)

@dataclass
class RegistryLoaderConfig:
    registry_id: str
    region_name: str = "ap-northeast-1"
    max_results: int = 10

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

    def load_a2a_agents(self, query: str) -> A2AClientToolProvider:
        """Registry から A2A エージェントを検索して A2AClientToolProvider を返す"""
        response = self._client.search_registry_records(
            registryIds=[self._config.registry_id],
            searchQuery=query,
            maxResults=self._config.max_results,
            filters={"descriptorType": {"$eq": "A2A"}},
        )

        # デバッグ: 生レスポンスをログ出力
        logger.info(
            "Raw search response: %s",
            json.dumps(response, default=str, ensure_ascii=False, indent=2),
        )

        records = response.get("registryRecords") or []
        logger.info("Fetched %d A2A record(s) from registry", len(records))

        urls: list[str] = []
        for record in records:
            url = self._extract_a2a_url(record)
            if url:
                urls.append(url)
                logger.info("Loaded A2A agent: %s -> %s", record.get("name"), url[:80])

        auth = SigV4HttpxAuth(region=self._config.region_name)
        return A2AClientToolProvider(
            known_agent_urls=urls,
            httpx_client_args={"auth": auth},
        )

    @staticmethod
    def _extract_a2a_url(record: dict) -> str | None:
        descriptors = record.get("descriptors") or {}
        a2a_desc = descriptors.get("a2a") or {}

        card_json = (a2a_desc.get("agentCard") or {}).get("inlineContent")
        if not card_json:
            return None

        card = json.loads(card_json)
        return card.get("url")

"""関西弁AWSアーキテクト オーケストレーター"""

from __future__ import annotations

import logging
import os

from strands import Agent
from strands.models import BedrockModel

from registry_loader import RegistryLoader, RegistryLoaderConfig

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

MODEL_ID = "jp.anthropic.claude-sonnet-4-5-20250929-v1:0"
REGION = "ap-northeast-1"
# Registry ID は環境変数から取得
REGISTRY_ID = os.environ["AGENT_REGISTRY_ID"]

ORCHESTRATOR_SYSTEM_PROMPT = """\
あなたはAWSアーキテクトのオーケストレーターです。
ユーザーからアプリ要件を受け取ったら、以下の手順で動作してください。

1. AWS構成提案を考える（必要に応じてAWS MCP Serverのツールを使う）
2. 提案内容を「関西弁変換エージェント」に渡して関西弁にしてもらう
3. 関西弁化された提案をユーザーに返す

注意:
- 標準語ではなく、必ず関西弁変換エージェントを経由した関西弁で回答する
- 関西弁変換エージェントは A2A 経由で呼び出す
"""

def main() -> None:
    # Registry から A2A エージェント（kansai-translator）を取得
    config = RegistryLoaderConfig(
        registry_id=REGISTRY_ID,
        region_name=REGION,
    )
    loader = RegistryLoader(config)
    a2a_provider = loader.load_a2a_agents("kansai translator")

    # Strands Agent を作成（A2Aツールを統合）
    model = BedrockModel(model_id=MODEL_ID, region_name=REGION)
    agent = Agent(
        model=model,
        tools=a2a_provider.tools,
        system_prompt=ORCHESTRATOR_SYSTEM_PROMPT,
        callback_handler=None,
    )

    # ユーザー入力
    user_input = (
        "ECサイトを作りたい。月間100万PV、画像が多い、"
        "予算は月50万円以内。AWSでどう構成すればいい？"
    )

    print("=" * 60)
    print("ユーザー入力:")
    print(user_input)
    print("=" * 60)

    result = agent(user_input)

    print("\n回答:")
    print(result.message["content"][0]["text"])
    print("=" * 60)

if __name__ == "__main__":
    main()

export AGENT_REGISTRY_ID="<your-registry-id>"
uv run python orchestrator.py

============================================================
ユーザー入力:
ECサイトを作りたい。月間100万PV、画像が多い、予算は月50万円以内。AWSでどう構成すればいい？
============================================================

回答:
完璧や！関西弁変換エージェントから返事もろたで。これをそのままお伝えするわ：

---
ほな ECサイトのAWS構成、ガツンと提案したるで！

## 要件まとめ
- 月間100万PV（約33,000PV/日、1日あたり約2,300リクエスト/時）やな
- 画像がめっちゃ多いんやろ
- 予算：月50万円以内や

## 推奨構成

### 1. フロントエンド・静的コンテンツ配信
**Amazon CloudFront + S3**
- CloudFront で画像やCSS/JSなんかの静的コンテンツをキャッシュ配信しとき、これ鉄板やで
- オリジンにS3バケット置いて画像ぶち込んどくんや
- 画像最適化は CloudFront Functions でWebPに自動変換したらええがな、めっちゃ軽なるで
- 推定コスト：月5-8万円（データ転送量によるけどな）

### 2. Webアプリケーション層
**ALB + ECS Fargate**
- Application Load Balancer で負荷分散や、これ当たり前やろ
- ECS Fargate でコンテナ実行しとき（2-4タスク、Auto Scaling設定しとくんやで）
- タスク定義：2vCPU、4GBメモリ × 2-4タスクでいこか
- 推定コスト：月8-12万円や

### 3. データベース
**Amazon RDS（MySQL/PostgreSQL）**
- db.t3.medium か db.t4g.medium にしとこ
- Multi-AZ構成で高可用性確保や、なんかあったらシャレならんで
- 読み取り負荷がごっつ高いんやったらリードレプリカ追加も検討したらええわ
- 推定コスト：月5-8万円やな

### 4. 画像処理・最適化
**Lambda + S3**
- 画像アップロード時に自動的にサムネイル生成・リサイズしたるんや
- CloudFront 経由で配信することでコスト削減できるで、賢いやろ
- 推定コスト：月1-2万円

### 5. セッション管理・キャッシュ
**Amazon ElastiCache（Redis）**
- cache.t3.micro か cache.t4g.small や
- セッション情報や商品情報のキャッシュに使うんやで
- 推定コスト：月1-2万円やな

### 6. セキュリティ
**AWS WAF + Certificate Manager**
- WAF でDDoS攻撃や不正アクセス防御しとき、これ必須やがな
- Certificate Manager でSSL/TLS証明書（無料やで、ラッキーやろ）
- 推定コスト：月2-3万円

### 7. 監視・ログ
**CloudWatch + CloudWatch Logs**
- アプリケーションとインフラの監視や
- アラート設定で異常検知しとかなあかんで
- 推定コスト：月1-2万円やな

## 月額概算コスト
- CloudFront + S3：5-8万円
- ECS Fargate：8-12万円
- RDS：5-8万円
- Lambda：1-2万円
- ElastiCache：1-2万円
- WAF：2-3万円
- CloudWatch：1-2万円

**合計：約23-37万円/月や**
予算50万円以内に余裕で収まるで、スケーラビリティもバッチリ確保できるがな！

## コスト最適化のポイント
1. CloudFront のキャッシュヒット率を90%以上に維持しとき、これ重要やで
2. S3 Intelligent-Tiering で古い画像のストレージコスト削減したらええがな
3. RDS のリザーブドインスタンスで最大40%割引や、使わな損やで
4. 不要な環境（開発・検証環境）は夜間・休日停止しとこ、これ基本やろ

## スケーリング余地
あんな、この構成やったら月間300-500万PVまで余裕で対応できるで、めっちゃ拡張性あるやろ！