Amazon Bedrock MCP Gateway, MCP Server(Lambda)の作成例

概要

本文書は、Amazon Bedrock AgentCore Gateway 経由で実行するMCPツール（lambda）を対象に、以下の3つの作業を実施した記録です。

1. MCP ツールの呼び出し実行（Lambda → Bedrock）
2. AWS Agent Registry への MCP サーバー登録
3. Lambda 関数に対するユニットテストの作成と実行

検証環境

● Claude Sonnet 4.6（モデルID: us.anthropic.claude-sonnet-4-6）
● Claude Code のバージョンは 2.1.197
● Claude CodeのMCPにはAWS MCP Serverをセット

以下の作業はすべて自然言語でclaudeに指示し、cliを実行してもらったもの。

利用した Gateway とツール名

項目	値
Gateway URL	https://fastapi-bedrock-gateway-jp-aaaaaaaaa.gateway.bedrock-agentcore.ap-northeast-1.amazonaws.com/mcp
ツール名	benchmark-jp-jp-target___benchmark_jp_jp

[AWS 解説]
Bedrock AgentCore Gateway は MCP プロトコル（streamable-http transport）を
ネイティブにサポートするマネージドエンドポイントです。
ツール名の ___ 区切りは「ターゲット名 + ツール関数名」を結合した
AgentCore の命名規則によるものです。

---

実行された Lambda ソースコード

ファイル: lambda_function_jp_jp.py

import json
import os

import boto3

MODEL_ID = os.getenv("ANTHROPIC_MODEL", "jp.anthropic.claude-sonnet-4-6")
BEDROCK_REGION = os.getenv("BEDROCK_REGION", "ap-northeast-1")


def lambda_handler(event, context):
    bytes_len = int(event.get("bytes", 100))
    client = boto3.client("bedrock-runtime", region_name=BEDROCK_REGION)
    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 4096,
        "messages": [{"role": "user", "content": f"文字列長{bytes_len}byteとなるような文字列を返して。文字列は0123456789012345のようにしてください。"}],
    }
    response = client.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps(body),
        contentType="application/json",
        accept="application/json",
    )
    result = json.loads(response["body"].read())
    return {
        "statusCode": 200,
        "body": json.dumps({"message": result["content"][0]["text"], "model": MODEL_ID}, ensure_ascii=False)
    }

[AWS 解説]

• MODEL_ID には 東京リージョン向けクロスリージョン推論プロファイル
  jp.anthropic.claude-sonnet-4-6 を使用しています。
  jp. プレフィックスが付いている場合、Bedrock は日本リージョン内で
  トラフィックを最適ルーティングします。
• boto3.client("bedrock-runtime", region_name=BEDROCK_REGION) は
  呼び出しごとにクライアントを生成していますが、Lambda のコンテナ再利用を
  活かすには関数スコープ外（グローバル）でクライアントを初期化すると
  ウォームスタート時のレイテンシをさらに削減できます。
• ensure_ascii=False を指定することで、日本語テキストが Unicode エスケープ
  されず可読性の高いレスポンスが返ります。

---

AWS Agent Registry への登録

Gateway 経由で呼び出せる MCP サーバーを AWS Agent Registry に登録し、
他のエージェントやシステムからディスカバリー可能な状態にします。

[AWS 解説]
AWS Agent Registry（bedrock-agentcore-control）は MCP サーバー、A2A エージェント、
カスタムリソースなどを一元管理するカタログサービスです。
Registry に登録されたレコードは tools/list を自動同期し、
APPROVED ステータスになると他サービスから検索・呼び出しが可能になります。

Registry（カタログ本体）

項目	値
Registry 名	fastapi-bedrock-jp-registry
Registry ID	hhhh
Registry ARN	arn:aws:bedrock-agentcore:ap-northeast-1:xxxxx:registry/hhhh
認証タイプ	AWS_IAM
自動承認	有効
リージョン	ap-northeast-1（東京）

[AWS 解説]
authorizer-type: AWS_IAM を指定することで、Registry の Search/Invoke API は
SigV4 署名による IAM 認証が必須となります。
autoApproval: true を設定しているため、レコード作成後に
submit-registry-record-for-approval を呼び出すと即座に APPROVED へ遷移します。

Registry Record（MCP サーバー登録）

項目	値
Record 名	fastapi-bedrock-gateway-jp
Record ID	aaaaa
Record ARN	arn:aws:bedrock-agentcore:ap-northeast-1:XXXXX:registry/YYYY/record/zzzzz
Descriptor Type	MCP
バージョン	1.0.0
同期タイプ	URL（Gateway URL から自動同期）
ステータス	APPROVED

[AWS 解説]
synchronizationType: URL を指定することで、Bedrock AgentCore が
登録先の MCP エンドポイントに initialize / tools/list を定期的に呼び出し、
ツール定義を自動同期します。これにより Lambda 側でツールを追加・変更した場合も
Registry のメタデータが自動更新されます。
同期には fastapi-bedrock-gateway-role の IAM ロールを SigV4 署名に使用しています。

---

MCP サーバー定義（Registry 自動取得）

Registry が Gateway から取得・保存した MCP サーバースキーマとツール定義です。

サーバースキーマ

{
  "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json",
  "name": "aaaaa.gateway.bedrock-agentcore.ap-northeast-1.amazonaws.com/fastapi-bedrock-gateway-jp",
  "description": "-",
  "version": "1.0.0",
  "remotes": [
    {
      "type": "streamable-http",
      "url": "https://fastapi-bedrock-gateway-jp-aaaaa.gateway.bedrock-agentcore.ap-northeast-1.amazonaws.com/mcp"
    }
  ]
}

[AWS 解説]
スキーマバージョン 2025-12-11 は MCP プロトコルの最新安定版です。
remotes[].type: streamable-http は HTTP ベースの双方向ストリーミングトランスポートを
示し、Server-Sent Events（SSE）による非同期レスポンスにも対応しています。

ツール定義（tools/list 取得結果）

{
  "tools": [
    {
      "name": "benchmark-jp-jp-target___benchmark_jp_jp",
      "description": "東京リージョンLambdaから東京リージョンBedrockを呼び出すベンチマーク実行(hello x5)",
      "inputSchema": {
        "type": "object",
        "properties": {
          "bytes": { "description": "レスポンス文字列の長さ(byte)", "type": "integer" }
        },
        "required": ["bytes"]
      }
    },
  ]
}

---

ユニットテスト

テスト設計方針

Lambda 関数が本番 AWS に接続しない状態で完全に検証できるよう、
以下のライブラリを組み合わせています。

ライブラリ	バージョン	用途
pytest	9.1.1	テストフレームワーク（収集・実行・レポート）
pytest-cov	7.1.0	カバレッジ計測
moto	5.2.2	@mock_aws で boto3 が本物の AWS に接続しないことを保証
unittest.mock	標準ライブラリ	patch で boto3.client を差し替え、Bedrock レスポンスを完全制御

[AWS 解説]
moto の @mock_aws は AWS SDK の HTTP レイヤーをインターセプトし、
テスト中に誤って本番 AWS リソースへアクセスすることを防ぎます。
ただし moto 5.2.2 時点では Bedrock Runtime の invoke_model レスポンスボディが
{} のみ返るため、Anthropic Messages 形式の完全なレスポンスは
unittest.mock.patch で差し替える設計としています。

テストコード

ファイル: test_lambda_function_jp_jp.py

"""
ユニットテスト: lambda_function_jp_jp.py

ライブラリ:
  - pytest        : テストフレームワーク
  - unittest.mock : AWS Bedrock クライアントのモック
  - moto          : AWS リソース全体の分離（boto3.client 呼び出しが本物の AWS に届かないことを保証）
"""

import importlib
import io
import json
import os
import sys
from unittest.mock import MagicMock, patch

import pytest
from moto import mock_aws

# テスト対象モジュールのパスを解決
MODULE_DIR = os.path.dirname(__file__)
if MODULE_DIR not in sys.path:
    sys.path.insert(0, MODULE_DIR)


# --------------------------------------------------------------------------- #
# ヘルパー
# --------------------------------------------------------------------------- #

def _make_bedrock_response(text: str) -> dict:
    """invoke_model が返す boto3 レスポンスオブジェクトを模倣する。"""
    payload = json.dumps({"content": [{"type": "text", "text": text}]}).encode()
    mock_body = MagicMock()
    mock_body.read.return_value = payload
    return {"body": mock_body}


def _make_lambda_event(bytes_value=None) -> dict:
    """Lambda イベントを生成する。bytes_value が None の場合はキーを含まない。"""
    return {"bytes": bytes_value} if bytes_value is not None else {}


def _load_handler(model_id: str = "jp.anthropic.claude-sonnet-4-6",
                  region: str = "ap-northeast-1"):
    """環境変数を設定してモジュールを再ロードし、handler 関数を返す。"""
    os.environ["ANTHROPIC_MODEL"] = model_id
    os.environ["BEDROCK_REGION"] = region
    # 環境変数を反映させるためにモジュールを再インポート
    import lambda_function_jp_jp as mod
    importlib.reload(mod)
    return mod.lambda_handler


# --------------------------------------------------------------------------- #
# フィクスチャ
# --------------------------------------------------------------------------- #

@pytest.fixture(autouse=True)
def aws_env(monkeypatch):
    """テストごとにダミーの AWS 認証情報を設定し、本物の AWS に接続しないようにする。"""
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
    monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "ap-northeast-1")


# --------------------------------------------------------------------------- #
# 正常系テスト
# --------------------------------------------------------------------------- #

class TestNormalCases:
    """正常系: Bedrock が正常にレスポンスを返す場合。"""

    @mock_aws
    def test_returns_200_status_code(self):
        """statusCode が 200 であることを確認する。"""
        handler = _load_handler()
        with patch("boto3.client") as mock_client_constructor:
            mock_client_constructor.return_value.invoke_model.return_value = (
                _make_bedrock_response("0")
            )
            result = handler(_make_lambda_event(bytes_value=1), {})

        assert result["statusCode"] == 200

    @mock_aws
    def test_response_body_contains_message_and_model(self):
        """レスポンス body に message と model キーが含まれることを確認する。"""
        handler = _load_handler()
        with patch("boto3.client") as mock_client_constructor:
            mock_client_constructor.return_value.invoke_model.return_value = (
                _make_bedrock_response("0")
            )
            result = handler(_make_lambda_event(bytes_value=1), {})

        body = json.loads(result["body"])
        assert "message" in body
        assert "model" in body

    @mock_aws
    def test_response_message_matches_bedrock_output(self):
        """Bedrock が返したテキストが message に格納されることを確認する。"""
        expected_text = "01234567890123456789"
        handler = _load_handler()
        with patch("boto3.client") as mock_client_constructor:
            mock_client_constructor.return_value.invoke_model.return_value = (
                _make_bedrock_response(expected_text)
            )
            result = handler(_make_lambda_event(bytes_value=20), {})

        body = json.loads(result["body"])
        assert body["message"] == expected_text

    @mock_aws
    def test_bytes_parameter_is_passed_to_bedrock(self):
        """bytes 引数の値が Bedrock へのプロンプトに含まれることを確認する。"""
        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("x" * 50)

        with patch("boto3.client", return_value=mock_bedrock):
            handler(_make_lambda_event(bytes_value=50), {})

        call_kwargs = mock_bedrock.invoke_model.call_args
        body_sent = json.loads(call_kwargs.kwargs.get("body") or call_kwargs.args[0] if call_kwargs.args else call_kwargs.kwargs["body"])
        prompt = body_sent["messages"][0]["content"]
        assert "50" in prompt

    @mock_aws
    def test_default_bytes_when_event_has_no_bytes_key(self):
        """event に bytes キーがない場合はデフォルト値 100 が使われることを確認する。"""
        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("0" * 100)

        with patch("boto3.client", return_value=mock_bedrock):
            result = handler(_make_lambda_event(), {})

        # デフォルト 100 が使われてもエラーにならず 200 を返す
        assert result["statusCode"] == 200
        call_kwargs = mock_bedrock.invoke_model.call_args
        body_sent = json.loads(call_kwargs.kwargs["body"])
        assert "100" in body_sent["messages"][0]["content"]

    @mock_aws
    def test_large_bytes_value(self):
        """bytes=1000 の大きな値でも正常に動作することを確認する。"""
        handler = _load_handler()
        with patch("boto3.client") as mock_client_constructor:
            mock_client_constructor.return_value.invoke_model.return_value = (
                _make_bedrock_response("0" * 1000)
            )
            result = handler(_make_lambda_event(bytes_value=1000), {})

        assert result["statusCode"] == 200
        body = json.loads(result["body"])
        assert len(body["message"]) == 1000


# --------------------------------------------------------------------------- #
# 環境変数テスト
# --------------------------------------------------------------------------- #

class TestEnvironmentVariables:
    """環境変数 ANTHROPIC_MODEL / BEDROCK_REGION が反映されることを確認する。"""

    @mock_aws
    def test_model_id_from_env_var(self):
        """ANTHROPIC_MODEL 環境変数のモデル ID が Bedrock 呼び出しに使われることを確認する。"""
        custom_model = "us.anthropic.claude-opus-4-8"
        handler = _load_handler(model_id=custom_model)
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hello")

        with patch("boto3.client", return_value=mock_bedrock):
            result = handler(_make_lambda_event(bytes_value=5), {})

        call_kwargs = mock_bedrock.invoke_model.call_args.kwargs
        assert call_kwargs["modelId"] == custom_model

        body = json.loads(result["body"])
        assert body["model"] == custom_model

    @mock_aws
    def test_bedrock_region_from_env_var(self):
        """BEDROCK_REGION 環境変数のリージョンが boto3.client 呼び出しに使われることを確認する。"""
        handler = _load_handler(region="us-east-1")
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hello")

        with patch("boto3.client") as mock_client_constructor:
            mock_client_constructor.return_value = mock_bedrock
            handler(_make_lambda_event(bytes_value=5), {})

        mock_client_constructor.assert_called_once_with(
            "bedrock-runtime", region_name="us-east-1"
        )

    @mock_aws
    def test_default_model_id_when_env_not_set(self, monkeypatch):
        """ANTHROPIC_MODEL 未設定時にデフォルトモデル ID が使われることを確認する。"""
        monkeypatch.delenv("ANTHROPIC_MODEL", raising=False)
        handler = _load_handler()  # reload でデフォルト値に戻す
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hello")

        with patch("boto3.client", return_value=mock_bedrock):
            result = handler(_make_lambda_event(bytes_value=1), {})

        body = json.loads(result["body"])
        # デフォルトは jp.anthropic.claude-sonnet-4-6
        assert "anthropic" in body["model"]


# --------------------------------------------------------------------------- #
# 異常系テスト
# --------------------------------------------------------------------------- #

class TestErrorCases:
    """異常系: Bedrock API がエラーを返す場合。"""

    @mock_aws
    def test_bedrock_client_error_propagates(self):
        """Bedrock が例外を発生させた場合に例外が伝播することを確認する。"""
        import botocore.exceptions

        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.side_effect = botocore.exceptions.ClientError(
            {"Error": {"Code": "ThrottlingException", "Message": "Rate exceeded"}},
            "InvokeModel",
        )

        with patch("boto3.client", return_value=mock_bedrock):
            with pytest.raises(botocore.exceptions.ClientError) as exc_info:
                handler(_make_lambda_event(bytes_value=1), {})

        assert exc_info.value.response["Error"]["Code"] == "ThrottlingException"

    @mock_aws
    def test_bedrock_returns_empty_content_raises_error(self):
        """Bedrock が content 配列を返さない場合に KeyError/IndexError が発生することを確認する。"""
        handler = _load_handler()
        broken_payload = json.dumps({"content": []}).encode()
        mock_body = MagicMock()
        mock_body.read.return_value = broken_payload
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = {"body": mock_body}

        with patch("boto3.client", return_value=mock_bedrock):
            with pytest.raises((KeyError, IndexError)):
                handler(_make_lambda_event(bytes_value=1), {})

    @mock_aws
    def test_bytes_parameter_must_be_castable_to_int(self):
        """bytes が int にキャストできない値の場合に ValueError が発生することを確認する。"""
        handler = _load_handler()
        with patch("boto3.client"):
            with pytest.raises((ValueError, TypeError)):
                handler({"bytes": "not_a_number"}, {})


# --------------------------------------------------------------------------- #
# Bedrock リクエスト形式テスト
# --------------------------------------------------------------------------- #

class TestBedrockRequestFormat:
    """Bedrock に送信するリクエストの形式が正しいことを確認する。"""

    @mock_aws
    def test_request_contains_anthropic_version(self):
        """リクエストに anthropic_version が含まれることを確認する。"""
        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hi")

        with patch("boto3.client", return_value=mock_bedrock):
            handler(_make_lambda_event(bytes_value=2), {})

        body_sent = json.loads(mock_bedrock.invoke_model.call_args.kwargs["body"])
        assert body_sent["anthropic_version"] == "bedrock-2023-05-31"

    @mock_aws
    def test_request_contains_max_tokens(self):
        """リクエストに max_tokens が含まれることを確認する。"""
        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hi")

        with patch("boto3.client", return_value=mock_bedrock):
            handler(_make_lambda_event(bytes_value=2), {})

        body_sent = json.loads(mock_bedrock.invoke_model.call_args.kwargs["body"])
        assert "max_tokens" in body_sent
        assert body_sent["max_tokens"] > 0

    @mock_aws
    def test_request_content_type_is_json(self):
        """invoke_model の contentType が application/json であることを確認する。"""
        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hi")

        with patch("boto3.client", return_value=mock_bedrock):
            handler(_make_lambda_event(bytes_value=2), {})

        call_kwargs = mock_bedrock.invoke_model.call_args.kwargs
        assert call_kwargs["contentType"] == "application/json"
        assert call_kwargs["accept"] == "application/json"

    @mock_aws
    def test_messages_role_is_user(self):
        """リクエストの messages[0].role が 'user' であることを確認する。"""
        handler = _load_handler()
        mock_bedrock = MagicMock()
        mock_bedrock.invoke_model.return_value = _make_bedrock_response("hi")

        with patch("boto3.client", return_value=mock_bedrock):
            handler(_make_lambda_event(bytes_value=2), {})

        body_sent = json.loads(mock_bedrock.invoke_model.call_args.kwargs["body"])
        assert body_sent["messages"][0]["role"] == "user"

---

テスト実行結果

実行コマンド

python3 -m pytest test_lambda_function_jp_jp.py -v \
  --cov=lambda_function_jp_jp --cov-report=term-missing

[AWS 解説]
--cov=lambda_function_jp_jp により Lambda 関数ファイル単体の
カバレッジを計測します。--cov-report=term-missing でカバーされていない
行番号も出力されるため、テスト漏れを素早く特定できます。

結果

============================= test session starts ==============================
platform linux -- Python 3.12.12, pytest-9.1.1, pluggy-1.6.0 -- /usr/bin/python3
cachedir: .pytest_cache
rootdir: /root/dynatrace3/fastapi-bedrock
plugins: anyio-4.13.0, cov-7.1.0
collecting ... collected 16 items

test_lambda_function_jp_jp.py::TestNormalCases::test_returns_200_status_code                     PASSED [  6%]
test_lambda_function_jp_jp.py::TestNormalCases::test_response_body_contains_message_and_model    PASSED [ 12%]
test_lambda_function_jp_jp.py::TestNormalCases::test_response_message_matches_bedrock_output     PASSED [ 18%]
test_lambda_function_jp_jp.py::TestNormalCases::test_bytes_parameter_is_passed_to_bedrock        PASSED [ 25%]
test_lambda_function_jp_jp.py::TestNormalCases::test_default_bytes_when_event_has_no_bytes_key   PASSED [ 31%]
test_lambda_function_jp_jp.py::TestNormalCases::test_large_bytes_value                           PASSED [ 37%]
test_lambda_function_jp_jp.py::TestEnvironmentVariables::test_model_id_from_env_var              PASSED [ 43%]
test_lambda_function_jp_jp.py::TestEnvironmentVariables::test_bedrock_region_from_env_var        PASSED [ 50%]
test_lambda_function_jp_jp.py::TestEnvironmentVariables::test_default_model_id_when_env_not_set  PASSED [ 56%]
test_lambda_function_jp_jp.py::TestErrorCases::test_bedrock_client_error_propagates              PASSED [ 62%]
test_lambda_function_jp_jp.py::TestErrorCases::test_bedrock_returns_empty_content_raises_error   PASSED [ 68%]
test_lambda_function_jp_jp.py::TestErrorCases::test_bytes_parameter_must_be_castable_to_int      PASSED [ 75%]
test_lambda_function_jp_jp.py::TestBedrockRequestFormat::test_request_contains_anthropic_version PASSED [ 81%]
test_lambda_function_jp_jp.py::TestBedrockRequestFormat::test_request_contains_max_tokens        PASSED [ 87%]
test_lambda_function_jp_jp.py::TestBedrockRequestFormat::test_request_content_type_is_json       PASSED [ 93%]
test_lambda_function_jp_jp.py::TestBedrockRequestFormat::test_messages_role_is_user              PASSED [100%]

================================ tests coverage ================================
Name                       Stmts   Miss  Cover   Missing
--------------------------------------------------------
lambda_function_jp_jp.py      12      0   100%
--------------------------------------------------------
TOTAL                         12      0   100%
============================== 16 passed in 2.01s ==============================

テストケースサマリー

クラス	件数	内容
TestNormalCases	6	正常系（200 応答・body 構造・デフォルト値・大きな入力値）
TestEnvironmentVariables	3	環境変数 ANTHROPIC_MODEL / BEDROCK_REGION の反映
TestErrorCases	3	異常系（Throttling・空レスポンス・型エラー）
TestBedrockRequestFormat	4	Bedrock API リクエストの形式検証
合計	16	全 PASSED、カバレッジ 100%

[AWS 解説]
ThrottlingException のテストは本番運用で特に重要です。
Bedrock の invoke_model は同時実行数やトークン毎分（TPM）の
クォータ超過時にこの例外を返すため、Lambda 側に指数バックオフによる
リトライロジックを実装する際の基礎テストとなります。
カバレッジ 100% は、Lambda の 12 ステートメント全行が
いずれかのテストケースで実行されたことを示します。