どうもこんにちは。
今回は、AWS CDK初心者がAWS CDKを使ってLambda関数をデプロイしてみたので、その手順を記事にしました。
はじめに
AWS CDKを使いたいと思ったきっかけは、大量のLambda関数とAmazon Bedrock AgentCore Gatewayを使用してMCPサーバ的なものを作りたいと思ったからです。
これを作るには、AWSマネジメントコンソール上でLambda関数を作成して、Bedrock AgentCore Gatewayを作成する必要がありました。
まぁ〜めんどくさい...!
なので、AWS CDKを使ってまとめてデプロイ&管理してしまおう!というふうに考えたので、AWS CDKを勉強し始めた所存です。
技術レベル
私の技術レベルは以下です。(今回の記事で必要な技術のみ)
- AWSマネジメントコンソール上でのLambda関数(Python)の作成 → できる
- AWSマネジメントコンソール上でのBedrock AgentCore Gatewayの作成 → できる
AWSマネジメントコンソール上では、Lambda関数の作成もAgentCore Gatewayの作成も習得済なのです。しかし、習得済であるからこそ、大量のLambda関数を作って各Lambda関数をAgentCore Gatewayのターゲットとして設定していくのがめんどくさかった...!
学習ステップを考えた!
もう ClaudeCodeに任せるか! とも考えましたが、ClaudeCodeに作らせたAWS CDKのコードの品質の良し悪しを判断できないのは如何なものなのかと思い、自力でAWS CDKを作成するところから始めようと思っているのでございます。
というわけで、以下のステップを踏んで、学習していきたいと思っています。
1つのLambda関数のデプロイを通して、CDKを学ぶ
- 1つのLambda関数をデプロイする
- 1つのLambda関数に 環境変数,タイムアウト値 を指定してデプロイする
- 1つのLambda関数に 既存のVPC,プライベートサブネット,セキュリティグループを指定してデプロイする
- 1つのLambda関数に登録済のLambdaレイヤーを紐づけてデプロイする
- Bedrock AgentCore Gatewayの1つのターゲットに1つのLamnda関数を紐づけてデプロイする
複数のLambda関数のデプロイを通して、CDKを学ぶ
上ができればこれもできるでしょうという精神!
- 複数のLambda関数に 既存のVPC,プライベートサブネット,セキュリティグループを指定してデプロイする
- Bedrock AgentCore Gatewayの複数のターゲットに複数のLambda関数を紐づけてデプロイする
今回のゴールはMCPサーバーを作成することなので、そのためのステップになっています。
ステップ1. 1つのLambda関数をデプロイする
1. 作業ディレクトリを用意して、CDKプロジェクトを初期化する
# 作業ディレクトリの用意
$ mkdir playground && cd $_
$ mkdir sample_aws_cdk
# CDKプロジェクトの初期化
$ cdk init --language python
$ source .venv/bin/activate
$ python -m pip install -r requirements.txt
2. app.pyを編集する
app.pyを開き、以下をコピペします。
#!/usr/bin/env python3
import os
import aws_cdk as cdk
from aws_cdk_sample.aws_cdk_sample_stack import AwsCdkSampleStack
app = cdk.App()
AwsCdkSampleStack(
app, "AwsCdkSampleStack",
env=cdk.Environment(
account=os.getenv('CDK_DEFAULT_ACCOUNT'),
region='ap-northeast-1'
),
description="サンプルのCDKスタック"
)
app.synth()
3. スタックを作成する
aws_cdk_sample/aws_cdk_sample_stack.pyにスタックを定義していきます。
まずは、シンプルなLambda関数です。
from aws_cdk import (
Stack,
aws_lambda as _lambda
)
from constructs import Construct
import os
class AwsCdkSampleStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
fn = _lambda.Function(
self,
"HelloWorldLambda",
runtime=_lambda.Runtime.PYTHON_3_13,
handler="lambda_function.lambda_handler",
code=_lambda.Code.from_asset(os.path.join(os.path.dirname(__file__), "lib/hello_world"))
)
4. Lambda関数を作成する
まず、以下を実行し、Lambda関数を作成します。
$ mkdir lib && mkdirr lib/hello_world && touch lib/hello_world/lambda_function.py
次に、作成したlib/hello_world/lambda_function.pyに以下をペーストします。
def lambda_handler(event, context):
return {
'statusCode': 200,
'body': 'Hello, World!'
}
5. スタックからAWS CloudFormationテンプレートを合成
記述したスタックから、AWS CloudFormationテンプレートを合成するコマンドを実行します。
$ cdk synth
6. デプロイ
以下のコマンドを実行します。
$ cdk deploy
これでデプロイが実行されます。
AWSマネジメントコンソールでLambda関数一覧を確認すると、Lambda関数が作成されているはずです。
ステップ2. 1つのLambda関数に 環境変数,タイムアウト値 を指定してデプロイする
1. スタックを編集する
aws_cdk_sample/aws_cdk_sample_stack.pyのスタックを編集していきます。以下をペーストします。
from aws_cdk import (
Stack,
Duration,
aws_lambda as _lambda
)
from constructs import Construct
import os
class AwsCdkSampleStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 共通の環境変数を定義
common_environment = {
"AWS_REGION_NAME": "ap-northeast-1",
"RDS_SECRET": "sample-rds-sm-01"
}
fn = _lambda.Function(
self,
"MyFunction",
runtime=_lambda.Runtime.PYTHON_3_13,
handler="lambda_function.lambda_handler",
code=_lambda.Code.from_asset(os.path.join(os.path.dirname(__file__), "lib/hello_world")),
timeout=Duration.minutes(1), # タイムアウト値を1分
memory_size=256, # メモリサイズを256MB
environment=common_environment # 環境変数を指定
)
2. AWS CloudFormationテンプレートを合成してデプロイ
記述したスタックから、AWS CloudFormation テンプレートを合成するコマンドを実行してデプロイします。
$ cdk synth
$ cdk deploy
これでLambda関数に環境変数やタイムアウト値が指定されているはずです。
ステップ3. 1つのLambda関数に 既存のVPC,プライベートサブネット,セキュリティグループを指定してデプロイする
ここが最初の関門かなぁ...
0. Lambda関数に指定したいVPC・PrivateSubnet・セキュリティグループのIDを取得する
AWSマネジメントコンソールからVPC・PrivateSubnet・セキュリティグループのIDを取得してください。
- VPC:
vpc-から始まる文字列 - PrivateSubnet:
subnet-から始まる文字列(アベイラビリティゾーンの確認しておく) - セキュリティグループ:
sg-から始まる文字列
1. スタックを編集する
aws_cdk_sample/aws_cdk_sample_stack.pyのスタックを編集していきます。以下をペーストします。
from aws_cdk import (
Stack,
Duration,
aws_lambda as _lambda,
aws_ec2 as ec2
)
from constructs import Construct
import os
class AwsCdkSampleStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 既存のVPCを参照
vpc = ec2.Vpc.from_lookup(
self,
"SampleVpc",
vpc_id="vpc-xxxxxxxxxxxxxxxxx"
)
# サブネットの選択(private-A, private-C)
subnet_private_a = ec2.Subnet.from_subnet_attributes(
self,
"SubnetPrivateA",
subnet_id="subnet-xxxxxxxxxxxxxxx",
availability_zone="ap-northeast-1a"
)
subnet_private_c = ec2.Subnet.from_subnet_attributes(
self,
"SubnetPrivateC",
subnet_id="subnet-xxxxxxxxxxxxxxx",
availability_zone="ap-northeast-1c"
)
subnet_selection = ec2.SubnetSelection(
subnets=[subnet_private_a, subnet_private_c]
)
# 既存のセキュリティグループを参照
security_group = ec2.SecurityGroup.from_security_group_id(
self,
"SampleSecurityGroup",
security_group_id="sg-xxxxxxxxxxxxx"
)
# 共通の環境変数
common_environment = {
"AWS_REGION_NAME": "ap-northeast-1",
"RDS_SECRET": "sample-rds-sm-01"
}
fn = _lambda.Function(
self,
"MyFunction",
runtime=_lambda.Runtime.PYTHON_3_13,
handler="lambda_function.lambda_handler",
code=_lambda.Code.from_asset(os.path.join(os.path.dirname(__file__), "lib/hello_world")),
timeout=Duration.minutes(1),
memory_size=256,
environment=common_environment,
vpc=vpc,
vpc_subnets=subnet_selection,
security_groups=[security_group],
allow_public_subnet=True # `allow_public_subnet`をTrueにしておく必要がある
)
2. AWS CloudFormationテンプレートを合成してデプロイ
記述したスタックから、AWS CloudFormation テンプレートを合成するコマンドを実行してデプロイします。
$ cdk synth
$ cdk deploy
これで、Lambda関数にVPC・プライベートサブネット・セキュリティグループが指定されているはずです。
ステップ4. Lamndaレイヤーをデプロイして、登録済のLamndaレイヤーを紐づけてデプロイする
AWS上にLamndaレイヤーを登録してARNを控えておいてください。
今回はSQLAlchemyというPythonライクにSQLを実行するライブラリをLambdaレイヤーに指定します。
1. スタックを編集
以下をペーストします。
from aws_cdk import (
Stack,
Duration,
aws_lambda as _lambda,
aws_ec2 as ec2
)
from constructs import Construct
import os
class AwsCdkSampleStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 既存のVPCを参照
vpc = ec2.Vpc.from_lookup(
self,
"SampleVpc",
vpc_id="vpc-xxxxxxxxxxxxxxxxx"
)
# サブネットの選択(private-A, private-C)
subnet_private_a = ec2.Subnet.from_subnet_attributes(
self,
"SubnetPrivateA",
subnet_id="subnet-xxxxxxxxxxxxxxx",
availability_zone="ap-northeast-1a"
)
subnet_private_c = ec2.Subnet.from_subnet_attributes(
self,
"SubnetPrivateC",
subnet_id="subnet-xxxxxxxxxxxxxxx",
availability_zone="ap-northeast-1c"
)
subnet_selection = ec2.SubnetSelection(
subnets=[subnet_private_a, subnet_private_c]
)
# 既存のセキュリティグループを参照
security_group = ec2.SecurityGroup.from_security_group_id(
self,
"SampleSecurityGroup",
security_group_id="sg-xxxxxxxxxxxxx"
)
# 共通の環境変数
common_environment = {
"AWS_REGION_NAME": "ap-northeast-1",
"RDS_SECRET": "sample-rds-sm-01"
}
# Lambdaレイヤーの作成
sqlalchemy_layer_arn = f"arn:aws:lambda:ap-northeast-1:{self.account}:layer:sqlalchemy:1"
# レイヤーの参照
sqlalchemy_layer = _lambda.LayerVersion.from_layer_version_arn(
self,
"SqlAlchemyLayer",
layer_version_arn=sqlalchemy_layer_arn
)
fn = _lambda.Function(
self,
"MyFunction",
runtime=_lambda.Runtime.PYTHON_3_13,
handler="lambda_function.lambda_handler",
code=_lambda.Code.from_asset(os.path.join(os.path.dirname(__file__), "lib/hello_world")),
timeout=Duration.minutes(1),
memory_size=256,
environment=common_environment,
layers=[sqlalchemy_layer], # Lambdaレイヤーを配列で指定
vpc=vpc,
vpc_subnets=subnet_selection,
security_groups=[security_group],
allow_public_subnet=True
)
2. AWS CloudFormationテンプレートを合成してデプロイ
記述したスタックから、AWS CloudFormation テンプレートを合成するコマンドを実行してデプロイします。
$ cdk synth
$ cdk deploy
ステップ5. Bedrock AgentCore Gatewayの1つのターゲットに1つのLamnda関数を紐づけてデプロイする
問題はここですよ...
CDKでLambda関数をデプロイする時みたいに記事が多くないので、参考資料が少なかった...
以下の記事とAPIリファレンスを参考にしてみました。
1. スタックを編集
fn = _lambda.Function(の記述の下から以下のように記述します。
# Gateway用のロールを作成
gateway_role = iam.Role(
self,
"GatewayRole",
assumed_by=iam.ServicePrincipal("bedrock-agentcore.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("BedrockAgentCoreFullAccess")
]
)
# Lambda関数にGateway用のロールを当てる
fn.grant_invoke(gateway_role)
# Gatewayを作成
gateway = bedrockagentcore.CfnGateway(
self,
"SampleGateway",
name="sample-gateway",
authorizer_type="AWS_IAM",
protocol_type="MCP",
description="Sample Gateway",
role_arn=gateway_role.role_arn
)
# Gatewayのターゲットを作成
gateway_target = bedrockagentcore.CfnGatewayTarget(
self,
"MyGatewayTarget",
name="lambda-target",
gateway_identifier=gateway.attr_gateway_identifier,
target_configuration=bedrockagentcore.CfnGatewayTarget.TargetConfigurationProperty(
mcp=bedrockagentcore.CfnGatewayTarget.McpTargetConfigurationProperty(
lambda_=bedrockagentcore.CfnGatewayTarget.McpLambdaTargetConfigurationProperty(
lambda_arn=fn.function_arn,
tool_schema=bedrockagentcore.CfnGatewayTarget.ToolSchemaProperty(
inline_payload=[
bedrockagentcore.CfnGatewayTarget.ToolDefinitionProperty(
name="my_tool",
description="My Lambda tool",
input_schema=bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
type="object",
properties={
"param": bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
type="string",
description="Input parameter"
)
}
)
)
]
)
)
)
),
credential_provider_configurations=[bedrockagentcore.CfnGatewayTarget.CredentialProviderConfigurationProperty(
credential_provider_type="GATEWAY_IAM_ROLE"
)],
description="Lambda function target"
)
2. AWS CloudFormationテンプレートを合成してデプロイ
記述したスタックから、AWS CloudFormation テンプレートを合成するコマンドを実行してデプロイします。
$ cdk synth
$ cdk deploy
ステップ6. 複数のLambda関数に 既存のVPC,プライベートサブネット,セキュリティグループを指定してデプロイする
ステップ5まで完了すればあとはLambda関数とGatewayのターゲットを増やしていくだけなので簡単です。変数名が競合しないように注意しておけば基本的には大丈夫なはずです。
StrandsAgentsからLambda×AgentCoreGateway製のMCPサーバを呼んでみる
作成されたBedrock AgentCore Gatewayの詳細を開くと、呼び出しコードが記載されていますので、それを参考にStrandsAgents側で呼び出し処理を記述します。
しかし、詳細に記載されているサンプルコードはJWTを使用する場合なのかな?
前述のステップで、IAM認証でAgentCore Gatewayを使用するように設定をしたので、aws_iam_streamablehttp_clientというライブラリを使用してMCPクライアントの初期化を行います。
環境変数でSAMPLE_MCP_GATEWAY_URLにデプロイしたAgentCore GatewayのURLを指定しておいてください。
from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
# Sample MCPクライアントの初期化
sample_mcp_client_factory = lambda: aws_iam_streamablehttp_client(
endpoint=SAMPLE_MCP_GATEWAY_URL,
aws_region=BEDROCK_AGENT_AWS_REGION,
aws_service="bedrock-agentcore",
terminate_on_close=False
)
sample_mcp_client = MCPClient(sample_mcp_client_factory)
# AgentにMCPを接続
sample_agent = Agent(
tools=[sample_mcp_client],
system_prompt="""
あなたはほげほげほげほげです。ツールを使用して、ほげほげしてください。
"""
)
これで、sample_agent.stream_async('こんにちは')とかを実行したりすると、レスポンスが返ってきます。
aws_iam_streamablehttp_clientを使用した具体的なMCPサーバーの呼び出しコードは以下の記事を参照してください!AgentCore RuntimeでデプロイしたMCPサーバとAgentCore GatewayでデプロイしたMCPサーバの接続方法は一緒です!
今後やること
AWS CDKでLambda関数とBedrock AgentCore Gatewayをデプロイできることが確認できたので、CDK上でLambda関数を量産してツールを作っていけば理想のMCPサーバーが完成することでしょう!
まとめ
AWS CDKを使ってLambda関数とBedrock AgentCore GatewayをデプロイしてMCPサーバーを構築しました。
cdk deployコマンドでデプロイに失敗した時のロールバックに時間かかるのがなぁ...
(Lambda関数の削除に時間がかかっているっぽいです。PrivateSubnetに繋いでいるから?)