AWS LambdaでMCPサーバーを動的インストール：サーバーレスAIエージェントの実装

はじめに
検証の背景と目的
アーキテクチャ概要
1. システム構成
2. 主要コンポーネント
実装の詳細
DynamoDB設定
1. テーブル構造
2. データ例
検証結果
1. 1. uvxコマンドによる動的インストール
2. 2. Zapier（StreamableHTTP）へのアクセス
技術的なポイントと工夫
苦労したポイント
パフォーマンス考察
まとめ
1. 本番環境への適用に向けて

はじめに

前回の記事では、Amazon Bedrock AgentsとMCPサーバーの統合について検証しました。本記事では、その技術をAWS Lambda上で実装し、MCPサーバーの動的インストールとサーバーレス実行を実現する方法について解説します。

検証の背景と目的

AIエージェントシステムを本番環境で運用する際、以下の課題があるのではないでしょうか。

スケーラビリティ: ユーザー数の増加に応じて自動的にスケールする必要がある
コスト効率: 使用した分だけ課金されるサーバーレスアーキテクチャが望ましい
柔軟性: ユーザーごとに異なるMCPサーバーセットを動的に利用できる必要がある

AWS Lambdaを活用することで、これらの課題を解決できます。

アーキテクチャ概要

システム構成

[ユーザーリクエスト]
    ↓
[Lambda関数]
    ├─ DynamoDBからMCPサーバー設定を取得
    ├─ MCPサーバーを動的インストール (uvx/npx)
    ├─ MCPクライアント生成
    └─ Bedrock Agentを実行
    ↓
[レスポンス]

主要コンポーネント

DynamoDB: ユーザーごとのMCPサーバー設定を保存
Lambda関数: MCPサーバーのインストールとエージェント実行
Bedrock Agent: LLMとMCPツールの統合実行

実装の詳細

1. Lambda関数のエントリーポイント

import os, json, asyncio, logging
from pathlib import Path
import boto3
import mcp_manager as mm
from mcp_manager import create_mcp_client, get_user_mcp_config, install_server
from bedrock_agent import agent_invoke

def lambda_handler(event, context):
    # Lambda実行環境の初期化
    for d in ("/tmp/.local/share", "/tmp/uv-cache", "/tmp/uv-tools", 
              "/tmp/.local/share/uv", "/tmp/.local/share/uv/tools"):
        Path(d).mkdir(parents=True, exist_ok=True)
    
    os.environ["HOME"] = "/tmp"
    os.environ["XDG_DATA_HOME"] = "/tmp/.local/share"
    os.environ["UV_CACHE_DIR"] = "/tmp/uv-cache"
    os.environ["UV_TOOLS_DIR"] = "/tmp/uv-tools"
    os.environ["PATH"] = "/var/task/bin:" + os.environ.get("PATH", "")
    
    mm.DATA_DIR = Path("/tmp/data")
    mm.DATA_DIR.mkdir(parents=True, exist_ok=True)
    
    os.environ.setdefault("BEDROCK_MODEL", 
                          "anthropic.claude-3-5-sonnet-20241022-v2:0")
    
    # 入力パラメータの取得
    user = event.get("user", "default")
    query = event.get("query", "Hello from Lambda.")
    
    # DynamoDBからMCPサーバー設定を取得
    ddb_servers = load_servers_from_ddb(user)
    ev_servers = event.get("servers")
    servers = ev_servers if isinstance(ev_servers, list) else ddb_servers
    
    # MCPサーバーのインストール
    for s in servers:
        try:
            install_server(user, s)
        except Exception:
            logging.exception(f"install_server で例外が発生: {s.get('name')}")
    
    # MCPクライアント生成とエージェント実行
    cfg = get_user_mcp_config(user)
    mcp_client = create_mcp_client(cfg)
    resp = asyncio.run(agent_invoke(mcp_client, query))
    
    return {
        "statusCode": 200,
        "body": json.dumps({"response": resp}, ensure_ascii=False)
    }

2. DynamoDBからの設定取得

def load_servers_from_ddb(user: str):
    """DynamoDBからユーザーのMCPサーバー設定を取得"""
    table_name = os.environ.get("SERVERS_TABLE")
    if not table_name:
        logging.warning("SERVERS_TABLE が未設定")
        return []
    
    ddb = boto3.resource("dynamodb")
    table = ddb.Table(table_name)
    
    try:
        resp = table.get_item(Key={"user": user})
        item = resp.get("Item")
        if not item:
            logging.info(f"DynamoDBに user={user} のレコードが見つかりません")
            return []
        
        servers = item.get("servers", [])
        
        # JSON文字列の場合はパース
        if isinstance(servers, str):
            servers = json.loads(servers)
        
        if not isinstance(servers, list):
            logging.warning(f"servers が配列ではありません: {type(servers)}")
            return []
        
        return servers
    except Exception:
        logging.exception("DynamoDB GetItem に失敗")
        return []

3. MCPサーバーの動的インストール

def install_server(user: str, server_info: Dict[str, Any]):
    # Lambda環境用のディレクトリ作成
    for p in ['/tmp/.local/share', '/tmp/uv-tools', '/tmp/uv-cache',
              '/tmp/.local/share/uv', '/tmp/.local/share/uv/tools']:
        Path(p).mkdir(parents=True, exist_ok=True)
    
    cfg = get_user_mcp_config(user)
    
    if "mcpServers" not in cfg:
        cfg["mcpServers"] = {}
    
    server_name = server_info.get("name")
    if not server_name:
        raise ValueError("server_info に 'name' キーがありません")
    
    if server_name in cfg["mcpServers"]:
        raise ValueError(f"MCPサーバー '{server_name}' は既に存在します")
    
    protocol = server_info.get("protocol")
    url = server_info.get("url", None)
    install = server_info.get("install", {})
    
    # 環境変数の補強（uvx/uv用）
    env_vars = install.get("env", {})
    if install.get("command") in ["uv", "uvx"]:
        env_vars["UV_CACHE_DIR"] = "/tmp/uv-cache"
        env_vars["UV_TOOLS_DIR"] = "/tmp/uv-tools"
        env_vars["HOME"] = "/tmp"
        env_vars["XDG_DATA_HOME"] = "/tmp/.local/share"
    
    # サーバー設定の保存
    if url is not None:
        server_config = {
            "protocol": protocol,
            "url": url,
            "headers": server_info.get("headers", {}),
            "auth": server_info.get("auth", {})
        }
    else:
        server_config = {
            "command": install.get("command"),
            "args": install.get("args", []),
            "env": env_vars
        }
    
    cfg["mcpServers"][server_name] = server_config
    save_user_mcp_config(user, cfg)

4. Bedrock Agentとの統合

async def agent_invoke(mcp_client, query):
    model = os.getenv('BEDROCK_MODEL', 
                      'anthropic.claude-3-5-sonnet-20241022-v2:0')
    agent = ConverseAgent(model)
    agent.tools = ConverseToolManager()
    
    agent.system_prompt = """You are a knowledgeable and reliable AI assistant.
    Please answer users' questions accurately and concisely.
    Tools should only be used when clearly necessary to address the user's question."""
    
    async with mcp_client:
        tools = await mcp_client.list_tools()
        for tool in tools:
            agent.tools.register_tool(
                name=tool.name,
                func=mcp_client.call_tool,
                description=tool.description,
                input_schema={'json': tool.inputSchema}
            )
        
        try:
            response = await agent.invoke_with_prompt(query)
            return response
        except Exception as e:
            print(f"Error occurred: {e}")
            raise

DynamoDB設定

テーブル構造

テーブル名: ServersTable
パーティションキー: user (String)
属性: servers (List)

データ例

{
  "user": "takayuki",
  "servers": [
    {
      "name": "fs",
      "install": {
        "command": "uvx",
        "args": ["mcp-server-filesystem", "--root", "/var/task"]
      }
    },
    {
      "name": "time",
      "install": {
        "command": "uvx",
        "args": ["mcp-server-time"]
      }
    },
    {
      "name": "gdrive",
      "install": {
        "command": "npx",
        "args": ["--yes", "@modelcontextprotocol/server-google-drive"]
      }
    }
  ]
}

検証結果

1. uvxコマンドによる動的インストール

テストイベント:

{
  "user": "test-user",
  "query": "あなたは何ができますか。",
  "servers": [
    {
      "name": "mcp_server",
      "install": {
        "command": "uv",
        "args": ["tool", "run", "mcp-server-time"],
        "env": {
          "HOME": "/tmp",
          "XDG_DATA_HOME": "/tmp/.local/share",
          "UV_CACHE_DIR": "/tmp/uv-cache",
          "UV_TOOLS_DIR": "/tmp/uv-tools"
        }
      }
    }
  ]
}

結果: 成功

{
  "statusCode": 200,
  "body": "{\"response\": \"私は時間に関する以下の2つの主要な機能を提供できます：\\n\\n1. 特定のタイムゾーンの現在時刻を取得\\n- 世界中の任意のタイムゾーンの現在時刻を確認できます\\n- 例：東京、ニューヨーク、ロンドンなどの現在時刻\\n\\n2. タイムゾーン間の時刻変換\\n- ある地域の時刻を別の地域の時刻に変換できます\\n- 例：日本時間の15:00をニューヨーク時間に変換する、など\"}"
}

2. Zapier（StreamableHTTP）へのアクセス

テストイベント:

{
  "user": "test-user",
  "query": "あなたは何ができますか。",
  "servers": [
    {
      "name": "mcp_server",
      "protocol": "StreamableHttp",
      "url": "https://mcp.zapier.com/api/mcp/s/.../mcp",
      "headers": {},
      "auth": {}
    }
  ]
}

結果: 成功

{
  "statusCode": 200,
  "body": "{\"response\": \"私は主に以下の機能を提供できます：\\n\\n1. Gmailのメール検索\\n- メールの送信者、受信者、件名、内容などで検索\\n- 添付ファイルの有無での検索\\n- 日付による検索\\n- ラベルによる検索など\\n\\n2. Googleカレンダーの予定検索\\n- 特定の期間の予定を検索\\n- イベント名や説明文での検索\\n- 定期的な予定の展開\\n- 複数のカレンダーからの検索\"}"
}

結論: Lambdaからの動的インストールとZapierへのアクセスの両方が正常に動作することを確認しました。

技術的なポイントと工夫

1. Lambda環境の制約への対応

Lambda関数は読み取り専用のファイルシステムを持ち、書き込み可能なのは /tmp ディレクトリのみです。この制約に対応するため、以下の工夫を行いました。

# 環境変数の設定
os.environ["HOME"] = "/tmp"
os.environ["XDG_DATA_HOME"] = "/tmp/.local/share"
os.environ["UV_CACHE_DIR"] = "/tmp/uv-cache"
os.environ["UV_TOOLS_DIR"] = "/tmp/uv-tools"

これにより、uvxやnpxが/tmp配下にパッケージをインストールできるようになります。

2. 依存ライブラリのパッケージング

Lambda Layerまたはデプロイパッケージに以下のライブラリを含める必要があります：

python3.12 -m pip install \
  "pydantic[email]==2.11.10" \
  fastmcp \
  mcp \
  pydantic-settings \
  uv \
  uvx \
  -t ~/lambda_build

重要: Lambdaのランタイムバージョンとビルド環境のPythonバージョンを一致させる必要があります（今回の検証ではPython 3.12）。

3. Node.jsツールのサポート

npxコマンドを使用するため、Lambda環境にNode.jsをバンドルする必要があります：

# Node.jsとnpmのインストール
curl -fsSL https://rpm.nodesource.com/setup_18.x | sudo bash -
sudo yum install -y nodejs

# パッケージのインストール
npm install @modelcontextprotocol/server-filesystem --prefix ~/lambda_build

Lambda関数のPATH環境変数に、Node.jsバイナリのパスを追加します：

os.environ["PATH"] = "/var/task/bin:" + os.environ.get("PATH", "")

4. IAMロールの設定

Lambda実行ロールに以下の権限が必要です。

Bedrock: bedrock:InvokeModel（InvokeModelだけで十分、FullAccessは過剰）
DynamoDB: dynamodb:GetItem（GetItemだけで十分、FullAccessは過剰）

本番環境では、最小権限の原則に従い、必要最小限の権限のみを付与することを推奨します。

5. コールドスタート対策

Lambda関数の初回実行時（コールドスタート）は、MCPサーバーのインストールに時間がかかります。以下の対策が考えられます。

Provisioned Concurrency: 事前にウォームアップされたインスタンスを確保
キャッシュ戦略: よく使われるMCPサーバーをLambda Layerに事前インストール
非同期処理: インストール処理を並列化

苦労したポイント

1. Lambda環境でのuvx実行

uvxコマンドは通常、ユーザーのホームディレクトリにツールをインストールしますが、Lambdaでは /tmp のみが書き込み可能です。環境変数 HOME、XDG_DATA_HOME、UV_CACHE_DIR、UV_TOOLS_DIR を適切に設定することで解決しましたが、この組み合わせを見つけるまでに試行錯誤が必要でした。

2. 依存ライブラリのバージョン互換性

FastMCPとMCPライブラリのバージョン互換性に注意が必要でした。特に、Pydanticのバージョン（2.11.10）を明示的に指定することで、依存関係の問題を回避しました。

3. 非同期処理の扱い

Bedrock AgentとMCPクライアントは非同期処理を使用しますが、Lambda関数のエントリーポイントは同期関数です。 asyncio.run()を使用して非同期処理を同期的に実行する必要がありました。

resp = asyncio.run(agent_invoke(mcp_client, query))

4. DynamoDBのデータ型変換

DynamoDBから取得したデータは、DynamoDB固有の型（{'S': 'value'}など）で返されることがあります。boto3のresourceインターフェースを使用することで、自動的にPythonネイティブ型に変換されますが、JSON文字列として保存されている場合の処理も考慮する必要がありました。