Amazon Bedrock AgentsとMCPサーバーの統合検証：権限制御とプロトコル対応

AIエージェントシステムの実用化において、適切な権限制御と外部ツールとの柔軟な連携は重要な課題です。本記事では、Amazon Bedrock Agentsを用いたマルチエージェント構成において、ユーザーごとのエージェント利用制限と、Model Context Protocol（MCP）サーバーの複数プロトコル対応について検証してみました。

検証の背景と目的
検証1: Bedrock Agentsにおける権限制御の実装
検証2: MCPサーバーの複数プロトコル対応
苦労ポイント
まとめ

検証の背景と目的

企業でAIエージェント導入を検討する際、以下の要件が求められることが多いのではないでしょうか。

権限制御: ユーザーの役割や権限に応じて、利用可能なエージェント機能を制限する
柔軟な外部連携: 様々な通信プロトコルに対応し、多様な外部サービスと統合する

私が開発に携わるSCSKのAIサービスInfoWeaveにおいてもMCPサーバー対応を検討しており、これらの要件を満たす実装方式を実際に検証しました。

検証1: Bedrock Agentsにおける権限制御の実装

実装方式

Supervisorエージェントに対するリクエスト時に、promptSessionAttributesを活用して利用不可能なCollaborator（サブエージェント）を指定し、システムプロンプトでその制約を強制する方式を採用しました。

システムプロンプトの設計

以下はシステムプロンプトの例です。promptSessionAttributesにリストされているサブエージェントを利用しないよう強く指示しています。

# Supervisor Agent Instruction
あなたは「Supervisor Agent」です。
目的は **ユーザーの要求を正確に理解し、最短経路で解決するために
最適なサブエージェント／ツールをリストアップする** ことです。

## 基本行動
1. promptSessionAttributesにリストされている Collaborator は利用不可の為、
   **絶対に** 利用しない。
2. 利用不可の Collaborator を確認する。
3. 各タスクを完遂するために、以下の Collaborator を状況に応じて組み合わせる。
   - task-creation
   - task-validation
   - function1
   - function2
   - function3
   - function4

実装コード

以下は実装コードの例です。

import boto3
import json
import uuid

def lambda_handler(event, context):
    input_text = "今日の東京の天気を教えてください。"

    agent_id = 'xxxxxxxx'
    agent_alias_id = 'yyyyyyyy'
    session_id = str(uuid.uuid1())

    # promptSessionAttributesで利用不可のCollaboratorを指定
    session_state = {
        "promptSessionAttributes": {
            "non-use-agent1": "function4"
        }
    }

    client = boto3.client("bedrock-agent-runtime")

    response = client.invoke_agent(
        inputText=input_text,
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        enableTrace=False,
        sessionState=session_state
    )

    event_stream = response['completion']
    for event in event_stream:
        if 'chunk' in event:
            data = event['chunk']['bytes'].decode("utf-8")
            print(data)

出力結果

Start RequestId: xxxxxxxxxx-xxxxxxxx-xxxx-xxxx Version: $LATEST
1. ユーザーの要求を分析：
 - 目的: 東京の今日の天気情報を取得
 - 必要な情報: 場所（東京）、日時（今日）は明確
2. タスク分解と必要な Collaborator の選出:
 - タスク状況の確認と分解
 - 天気情報の取得（Web APIまたはブラウザ利用）
 - 結果の検証
3. 利用不可の Collaborator:
 - function4 は利用不可
4. 実施計画:
天気情報取得には function1 が最適
(1) task-creation
(2) function1
(3) task-validation
利用不可な Collaborator は「function4」です
END RequestId: xxxxxxxxxx-xxxxxxxx-xxxx-xxxx

※実際にはfunction1～function4には、例えば外部API呼び出し機能やWeb検索機能等の機能が備わっています。ユーザー要望の「東京の天気を調べる」為に、使用可能な機能の中からどの機能を呼び出せばいいのかをSupervisorが判断して組み合わせています。
例えば今回の例だとfunction4に天気情報検索API、function1にWeb検索機能が割当てられている場合、直感的にfunction4を使いたくなりますが使用不可リストに入っているため、function1のWeb検索を利用してユーザー要望を達成する、という流れになります。

検証結果

結論: promptSessionAttributesに利用不可のCollaboratorを指定することで、指定されたCollaboratorは使用しないようにすることができました。Supervisorは、promptSessionAttributesでリストされているCollaboratorを除外した上で、最適なCollaboratorを選択し実行計画を立てることを確認しました。もちろん利用可能なCollaboratorを指定することで、リストされたCollaboratorのみを使用することも可能です。

技術的なポイントと工夫

1. システムプロンプトでの強制力
単にpromptSessionAttributesに情報を渡すだけでなく、システムプロンプトで「絶対に利用しない」という強い表現を使用することで、LLMの判断を確実に制御しています。

2. 出力フォーマットの明示
利用不可のCollaboratorを明示的に出力させることで、権限制御が正しく機能していることを可視化し、デバッグやログ分析を容易にしています。

検証2: MCPサーバーの複数プロトコル対応

MCPプロトコルの種類

Model Context Protocol（MCP）は、AIエージェントと外部ツールを接続するための標準プロトコルです。主に以下の3つの通信方式があります。

STDIO: 標準入出力を使用した通信（ローカル実行向け）
SSE (Server-Sent Events): HTTPベースの一方向ストリーミング
StreamableHTTP: HTTPベースの双方向通信

実装アーキテクチャ

FastMCPライブラリを使用し、複数のプロトコルに対応したMCPクライアントを実装しました。

from typing import Dict, Any
from fastmcp import FastMCP, Client
from fastmcp.client.transports import (
    UvxStdioTransport,
    NpxStdioTransport,
    FastMCPTransport,
    StreamableHttpTransport,
    SSETransport
)

def create_mcp_client(config: Dict[str, Any]) -> Client:
    composite_server = FastMCP()

    for prefix, server_cfg in config.get('mcpServers', {}).items():
        # StreamableHTTP型
        if "url" in server_cfg and server_cfg.get("protocol") == "StreamableHttp":
            url = server_cfg["url"]
            headers = server_cfg.get("headers", {})
            transport = StreamableHttpTransport(url=url, headers=headers)
            composite_server.mount(prefix=prefix, server=FastMCP.as_proxy(transport))

        # SSE型
        elif "url" in server_cfg and server_cfg.get("protocol") == "sse":
            url = server_cfg["url"]
            headers = server_cfg.get("headers", {})
            transport = SSETransport(url=url, headers=headers)
            composite_server.mount(prefix=prefix, server=FastMCP.as_proxy(transport))

        # STDIO型
        elif "command" in server_cfg:
            tool_command = server_cfg.get('command')
            tool_args = server_cfg.get('args', [])
            env_vars = server_cfg.get('env', {})

            if tool_command == 'uvx':
                transport = UvxStdioTransport(
                    tool_name=tool_args[0],
                    tool_args=tool_args[1:],
                    env_vars=env_vars
                )
            elif tool_command == 'npx':
                transport = NpxStdioTransport(
                    package=tool_args[1],
                    args=tool_args[2:],
                    env_vars=env_vars
                )

            composite_server.mount(prefix=prefix, server=FastMCP.as_proxy(transport))

    transport = FastMCPTransport(mcp=composite_server)
    client = Client(transport)
    return client

検証したMCPサーバー

StreamableHTTP: Zapier MCP Server — 以下２機能を設定・有効化

Gmail検索機能
Googleカレンダー連携

SSE: 自作MCPサーバー — 以下２機能を実装

エコーツール
現在時刻取得ツール

SSE型MCPサーバーの実装例

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("minimal-mcp-server")

@mcp.tool()
def echo(message: str) -> str:
    """入力されたメッセージをそのまま返す簡単なツール"""
    return f"Echo: {message}"

@mcp.tool()
def get_current_time() -> str:
    """現在の日時を取得するツール"""
    from datetime import datetime
    now = datetime.now()
    return f"現在の日時: {now.strftime('%Y-%m-%d %H:%M:%S')}"

if __name__ == "__main__":
    mcp.run(transport="sse")

検証結果

結論: StreamableHTTPとSSEの両方のプロトコルで、MCPサーバーとの通信が正常に動作することを確認しました。

Zapier（StreamableHTTP）経由でのGmail検索が成功
自作SSEサーバーからの時刻取得が成功
複数のMCPサーバーを同時に利用可能

技術的なポイントと工夫

1. プロトコルの自動判別
設定ファイルの構造から適切なTransportクラスを自動的に選択できるため、ユーザーは意識せず複数プロトコルを利用できます。
2. 統一的なインターフェース
FastMCPのcomposite_serverパターンを使用することで、異なるプロトコルのMCPサーバーを単一のクライアントから透過的に利用できます。
3. 設定の柔軟性
JSON形式の設定ファイルで、プロトコルタイプ・認証情報・ヘッダー等を柔軟に指定できる設計です。

{
  "mcpServers": {
    "zapier-mcp-sh": {
      "protocol": "StreamableHttp",
      "url": "https://mcp.zapier.com/api/mcp/s/...",
      "headers": {},
      "auth": {}
    },
    "test-mcp-sse": {
      "protocol": "sse",
      "url": "http://127.0.0.1:8000/sse",
      "headers": {},
      "auth": {}
    }
  }
}