どうもこんにちは。
今回は、AWSのLambda関数をMCPとして動かすことに成功したのでその方法を共有します。
この記事の前提条件
- PythonのStrandsAgentsを使用してAIエージェントを作成していること
Lambda関数とは
Lambda関数は、自分たちでサーバーの管理を行うことなくコードを実行できるコンピューティングサービスです。
EC2との違いとしては、常に動作し続けているのがEC2、使用する時だけ実行されるのがLambdaです。
上記の違いにより、料金の計算方法が異なります。
| サービス | 計算方法 |
|---|---|
| EC2 | サーバが起動している時間 |
| Lambda | コードが実行された時間（従量課金） |
Lambda関数でどんなプログラムを書いていた？
今回、MCPとして動かす対象のLambda関数は、RailsアプリケーションのDBの情報を取得する というAPIです。
RailsアプリケーションのDBは、RDSで起動されています。また、DBへの接続情報はSecretsManagerに格納されており、Lambda関数からboto3でシークレットの値を取得してDBに接続するようにしています。
Lambda関数では、Python3.13を使用し、DBへの接続はsqlalchemyを使用していました。
Lambda関数の情報はこの辺にしておきましょう。
Bedrock AgentCore Gatewayを使用
Amazon Bedrock AgentCore Gatewayでは、Lambda関数だけではなく、API GatewayやREST APIなどの外部のAPIなどもMCPとしてセットアップすることができます。
今回は、AWSマネジメントコンソール上でセットアップを行ったので、その手順を記載していきます。
1. Bedrock AgentCore Gatewayのトップページへ遷移
Gatewayのページに来たら、「ゲートウェイを作成」ボタンをクリックします。
2. 作成画面で必要情報を入力
2-1. 基本情報を入力
ゲートウェイ名は任意の値を入力し、「ターゲット」の項目の手前までは、デフォルトで大丈夫です。
2-2. ターゲットを設定
ターゲットでは、「ゲートウェイをどこに接続するか」を決定するかを設定します。
ターゲット名は任意の値を入力し、ターゲットタイプで「Lambda ARN」を選択します。
その後、Lambda関数のARNを入力し、ターゲットスキーマで「インラインスキーマを定義」を選択します。
2-3. インラインスキーマを設定
インラインスキーマでは、Lambda関数を使用すると何ができるのか、入力パラメータでは何が必要なのかを定義しておく必要があります。
今回は、以下のように定義をしています。（Lambda関数内で、WHERE句を使用して絞り込みができるようにコーディングをしているので、以下のようにinputSchemaを記述をしています。）
[
{
"name": "sapmle-app-incidents"
"description": "サンプルアプリケーションに保存されている障害情報一覧を取得する",
"inputSchema": {
"properties": {
"customer_ids": {
"type": "string"
},
"statuses": {
"type": "string"
},
"created_user_ids": {
"type": "string"
},
"updated_user_ids": {
"type": "string"
}
},
"type": "object"
}
}
]
3. 登録実行
「ゲートウェイを作成」ボタンをクリックして、ゲートウェイをセットアップ完了です。
4. サンプルコードを参考にAIエージェントのコードに記述していく
私はAIエージェントをStrandsAgentsを使用して書いているので、それ前提で記述します。
4-1. 必要なライブラリ
from strands import Agent
import logging
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
import os
import requests
import json
4-2. エージェントからMCP（ゲートウェイ）へ接続するための認証情報をセット
サンプルコードが以下のように定義されているので、まずはコピペします。TOKEN_URLはそのままでOKです。
CLIENT_IDとCLIENT_SECRETは、Cognitoのコンソール画面へ遷移して、作成されているユーザープールをクリックし、左メニューのアプリケーションクライアントをクリックして 確認します。
CLIENT_ID = "<YOUR_CLIENT_ID>"
CLIENT_SECRET = "<YOUR_CLIENT_SECRET>"
TOKEN_URL = "https://sample-agentcore-gateway-cognito.auth.ap-northeast-1.amazoncognito.com/oauth2/token"
4-3. 接続するためにコーディング
以下のようにStrandsAgentのコードが記述されているとします。
base_prompt = """
あなたはユーザからの質問に正確に回答するAIアシスタントです。必要に応じて、以下のツールを使用して調査し、正確に回答してください。
ツールを使用するにあたって、必要な情報が不足していればユーザに要求をしてください。
"""
super_strands_agent = Agent(
tools=all_tools, # ツールのリスト
system_prompt=base_prompt # プロンプト
)
1行上に、「MCPが接続できた場合に、ツールリストとプロンプトを拡張する」コードを追加します。
base_prompt = """
あなたはユーザからの質問に正確に回答するAIアシスタントです。必要に応じて、以下のツールを使用して調査し、正確に回答してください。
ツールを使用するにあたって、必要な情報が不足していればユーザに要求をしてください。
"""
##### ここ追加 #####
all_tools = []
sample_mcp_tools = set_sample_mcp()
if sample_mcp_tools:
all_tools.extend(sample_mcp_tools)
base_prompt += sample_mcp_prompt
print(f"✅ Total tools after SAMPLE MCP integration: {len(all_tools)}", flush=True)
###################
super_strands_agent = Agent(
tools=all_tools, # ツールのリスト
system_prompt=base_prompt # プロンプト
)
次に、set_sample_mcp関数を用意します。
def set_piman_mcp():
## ここは作成したゲートウェイのURLに書き換える
gateway_url = "https://saple-gateway.gateway.bedrock-agentcore.ap-northeast-1.amazonaws.com/mcp"
try:
## アクセストークンを取得
response = requests.post(
token_url,
data="grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}".format(client_id=client_id, client_secret=client_secret),
headers={'Content-Type': 'application/x-www-form-urlencoded'}
)
access_token = response.json()['access_token']
except Exception as e:
print(f"❌ Error fetching access token for SAMPLE MCP: {str(e)}", flush=True)
return []
## ヘッダーにアクセストークンを設定
headers = { "Authorization": f"Bearer {access_token}" }
try:
## MCPクライアントを定義
sample_mcp_client = MCPClient(
lambda: streamablehttp_client(
gateway_url,
headers=headers,
timeout=300,
terminate_on_close=False
)
)
## MCPクライアントと接続（サンプルコードには、このコードがなかった）
sample_mcp_client.__enter__()
except Exception as e:
print(f"❌ Error creating SAMPLE MCP client: {str(e)}", flush=True)
return []
try:
more_tools = True
tools = []
pagination_token = None
while more_tools:
## MCPサーバからツールリストを取得
tmp_tools = sample_mcp_client.list_tools_sync(pagination_token=pagination_token)
tools.extend(tmp_tools)
if tmp_tools.pagination_token is None:
more_tools = False
else:
more_tools = True
pagination_token = tmp_tools.pagination_token
return tools
except Exception as e:
print(f"❌ Error listing SAMPLE MCP tools: {str(e)}", flush=True)
return []
これで、MCPサーバ（ゲートウェイ）との接続は完了です。
5. ECR・Bedrock AgentCore Runtimeへデプロイ
エージェントのデプロイ方法については、以下の記事を参照いただければと思います。
まとめ
上記の手順を踏むことで、ゲートウェイ経由でLambda関数をツールとして接続することができます。
また、今回は認証をCognitoでのJWTを使用しましたが、「ツールを使用できる範囲をユーザによって制限したい場合」には、AgentCore Policyなどを使用してみるのが良いと思います。
以上