10
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@har1101

Amazon S3 FilesをBedrock AgentCore Runtimeからマウントしてファイル操作してみた

10
Last updated at Posted at 2026-04-12

2026/05/05 追記：AgentCore Runtime×S3 Filesをネイティブマウントできるようになりました

詳細は以下記事をご覧ください！じんのさんありがとうございます^^

tl;dr

AgentCore RuntimeからS3 Filesへの接続は可能です。
ファイルの読み書き、ディレクトリ作成、削除、S3バケットへの自動同期まですべて動作しました。

確認項目 結果
AgentCoreコンテナ → Mount TargetへのTCP接続 OK
mount -t nfs4（標準NFS）でのマウント NG（IAM認証必須のため拒否される）
mount -t s3files（専用ヘルパー）でのマウント OK
ファイル読み書き・ディレクトリ作成・削除 OK
S3バケットへの自動同期 OK（約1分）
非rootユーザーでの読み書き OK（ワークスペース方式）
AIエージェントへの自然言語指示でファイル作成 OK

ただし、EC2でsudo mount -t s3filesするだけの手順とは異なり、AgentCore固有の工夫がいくつか必要でした。その差分を中心に解説します。

はじめに

こんにちは、ふくちと申します。

2026年4月7日にAmazon S3 FilesがGAとなりました。これはファイルおよびストレージ操作を使用してデータの読み取り・書き込み・整理などを行うことができるS3というサービス(機能?)のようです。

また、S3 Files登場のニュースブログにはこう書かれています。

S3は組織のあらゆるデータの中心的なハブとなり、本番アプリケーションの実行、機械学習モデルのトレーニング、エージェント型AIシステムの構築など、あらゆるAWSコンピューティングインスタンス、コンテナ、関数から直接アクセスできます。

そこで公式ドキュメントの手順を確認しに行ってみると…
image.png

なんでAgentCore Runtimeがないんだよ！教えはどうなってんだ教えは
ということで、自分で良い感じにマウントする手法を試してみたので、その方法を共有できればと思います！

上記の通りEC2・Lambda・ECS・EKSからマウントする手順はAWS公式ドキュメントで共有されているのと、クラメソさんの記事ではかなり詳細に解説があるので、S3 Filesについてもっと深堀りしたい方はこちらをご参照ください。

では、Bedrock AgentCore Runtimeからはどうでしょうか。AgentCoreのコンテナ環境にはEC2ほどの自由度がなく、公式にも手順は用意されていません。

本当にマウントできるのか、できるとしたら何が違うのか、気になったので実際に検証してみました。

この記事で扱う内容

  • S3 Filesの標準セットアップ手順（EC2向け）との対比
  • AgentCoreから接続する場合に必要な追加の工夫
  • rootマウント→非rootアプリ実行の構成
  • 自然言語でAIエージェントにファイル操作させるところまで

アーキテクチャ

今回構築したコア部分の構成は以下です。
image.png

VPC内のプライベートサブネットにAgentCore RuntimeのENIとS3 Filesのマウントターゲット(実体はENI)の2つが配置されます。
AgentCore Runtimeのコンテナ内ではefs-proxyがIAM認証とTLS暗号化を処理し、マウントターゲット経由でS3 Filesに接続します。

また、全体アーキテクチャは以下です。
image.png

以下ドキュメントの記載に則って、最小限のNAT Gateway+S3ゲートウェイVPCエンドポイントだけ追加して設定しています。

コンテナエージェントの場合、AgentCoreは定期的にECRからコンテナイメージを更新します。ECRはイメージレイヤーをAmazon S3に保存しています。S3ゲートウェイVPCエンドポイントがない場合、このトラフィックはNATゲートウェイを経由してルーティングされ、データ処理料金が発生します。S3ゲートウェイVPCエンドポイントは無料で、これらの料金をなくします。VPCがNAT経由でインターネットにアクセスできる場合でも、すべてのVPCモードコンテナエージェントにこのエンドポイントを追加することを強くお勧めします。

VPC上のAgentCore Runtimeに関して、もっとセキュアな環境にしたい場合は以下が参考になります。

S3 Filesの標準手順とAgentCoreでの差分

まず、EC2でS3 Filesをマウントする標準的な手順と、AgentCoreでやる場合に何が変わるかを整理します。

公式ドキュメント（Mounting S3 file systems on EC2）の手順をベースにしています。

また、再現性確保やコンソール作成における一部制約の都合上、AWS SDK(boto3) v1.42.85を用いて作成しています。
執筆当時、S3 FilesはまだCDK対応してないので、全部SDKで統一したという感じです。

# 標準手順（EC2） AgentCoreでの違い
1 VPC・セキュリティグループを作成 同じ
2 S3 Files/コンピュートのサービスロール作成 同じ
3 S3バケット作成 + バージョニング有効化 同じ
4 S3 Filesとマウントターゲット作成 同じ
5 amazon-efs-utils v3.0.0をインストール EC2ならdnf installでOKだが、AgentCoreはDockerマルチステージビルド
6 sudo mount -t s3files :/ /mnt/s3files でコンピュートからマウント EC2ならコマンド実行でOKだが、AgentCoreは一工夫必要

標準手順

上記手順で進めていきます。

1. VPC・セキュリティグループを作成しておく

以下の条件でVPCを作成しておきます。

リソース 条件
パブリックサブネット 1つ以上。ルートテーブルで0.0.0.0/0をインターネットゲートウェイに向ける
プライベートサブネット 1つ以上。ルートテーブルで0.0.0.0/0をNATゲートウェイに向け、かつS3ゲートウェイVPCエンドポイントのプレフィックスリストルートも設定する
インターネットゲートウェイ あり
NATゲートウェイ あり（パブリックサブネットに配置）
S3ゲートウェイVPCエンドポイント あり（プライベートサブネットのルートテーブルに紐づけ）

また、セキュリティグループも2つ作成しておきます。
以下はAgentCore Runtime用です。

ルール タイプ プロトコル ポート範囲 送信先/ソース
アウトバウンド NFS TCP 2049 S3 Filesのセキュリティグループ
アウトバウンド すべてのトラフィック すべて すべて 0.0.0.0/0

以下はS3 Files用です。

ルール タイプ プロトコル ポート範囲 送信先/ソース
インバウンド NFS TCP 2049 AgentCore Runtimeのセキュリティグループ
サンプルスクリプト
setup_vpc.py
#!/usr/bin/env python3
"""
Create a test VPC with public/private subnets, IGW, NAT GW, and security groups
for S3 Files x AgentCore connectivity testing.
"""

import sys
import time
import boto3
from config import (
    REGION, PROJECT_NAME, VPC_CIDR, SUBNETS,
    save_state, load_state, tag_spec,
)

ec2 = boto3.client("ec2", region_name=REGION)


def wait_for(description, check_fn, max_wait=300, interval=10):
    """Poll until check_fn returns True."""
    print(f"  Waiting for {description}...", end="", flush=True)
    elapsed = 0
    while elapsed < max_wait:
        if check_fn():
            print(" done")
            return
        print(".", end="", flush=True)
        time.sleep(interval)
        elapsed += interval
    raise TimeoutError(f"Timed out waiting for {description}")


def create_vpc(state):
    if state.get("vpc_id"):
        print(f"VPC already exists: {state['vpc_id']}")
        return

    print("Creating VPC...")
    resp = ec2.create_vpc(
        CidrBlock=VPC_CIDR,
        TagSpecifications=tag_spec("vpc", "vpc"),
    )
    vpc_id = resp["Vpc"]["VpcId"]
    ec2.modify_vpc_attribute(VpcId=vpc_id, EnableDnsSupport={"Value": True})
    ec2.modify_vpc_attribute(VpcId=vpc_id, EnableDnsHostnames={"Value": True})
    state["vpc_id"] = vpc_id
    save_state(state)
    print(f"  VPC created: {vpc_id}")


def create_subnets(state):
    if state.get("subnets"):
        print(f"Subnets already exist: {list(state['subnets'].keys())}")
        return

    print("Creating subnets...")
    subnets = {}
    for name, cfg in SUBNETS.items():
        resp = ec2.create_subnet(
            VpcId=state["vpc_id"],
            CidrBlock=cfg["cidr"],
            AvailabilityZone=cfg["az"],
            TagSpecifications=tag_spec("subnet", name),
        )
        subnet_id = resp["Subnet"]["SubnetId"]
        subnets[name] = subnet_id
        print(f"  {name}: {subnet_id} ({cfg['az']})")

    state["subnets"] = subnets
    save_state(state)


def create_internet_gateway(state):
    if state.get("igw_id"):
        print(f"IGW already exists: {state['igw_id']}")
        return

    print("Creating Internet Gateway...")
    resp = ec2.create_internet_gateway(
        TagSpecifications=tag_spec("internet-gateway", "igw"),
    )
    igw_id = resp["InternetGateway"]["InternetGatewayId"]
    ec2.attach_internet_gateway(InternetGatewayId=igw_id, VpcId=state["vpc_id"])
    state["igw_id"] = igw_id
    save_state(state)
    print(f"  IGW created and attached: {igw_id}")


def create_nat_gateway(state):
    if state.get("nat_gw_id"):
        print(f"NAT Gateway already exists: {state['nat_gw_id']}")
        return

    print("Creating NAT Gateway (this takes ~2 min)...")
    # Allocate EIP
    eip_resp = ec2.allocate_address(
        Domain="vpc",
        TagSpecifications=tag_spec("elastic-ip", "nat-eip"),
    )
    eip_id = eip_resp["AllocationId"]
    state["nat_eip_id"] = eip_id

    # Create NAT GW in public subnet
    nat_resp = ec2.create_nat_gateway(
        SubnetId=state["subnets"]["public-1a"],
        AllocationId=eip_id,
        TagSpecifications=tag_spec("natgateway", "nat"),
    )
    nat_gw_id = nat_resp["NatGateway"]["NatGatewayId"]
    state["nat_gw_id"] = nat_gw_id
    save_state(state)

    wait_for(
        "NAT Gateway to become available",
        lambda: ec2.describe_nat_gateways(NatGatewayIds=[nat_gw_id])
        ["NatGateways"][0]["State"] == "available",
        max_wait=300,
        interval=15,
    )
    print(f"  NAT Gateway: {nat_gw_id}")


def create_route_tables(state):
    if state.get("rt_private_id"):
        print(f"Route tables already configured")
        return

    vpc_id = state["vpc_id"]

    # Public route table: use the main route table
    print("Configuring route tables...")
    rt_resp = ec2.describe_route_tables(
        Filters=[{"Name": "vpc-id", "Values": [vpc_id]}, {"Name": "association.main", "Values": ["true"]}]
    )
    rt_main_id = rt_resp["RouteTables"][0]["RouteTableId"]

    # Add internet route to main RT (for public subnet)
    ec2.create_route(
        RouteTableId=rt_main_id,
        DestinationCidrBlock="0.0.0.0/0",
        GatewayId=state["igw_id"],
    )
    # Associate public subnet with main RT
    ec2.associate_route_table(
        RouteTableId=rt_main_id,
        SubnetId=state["subnets"]["public-1a"],
    )
    state["rt_public_id"] = rt_main_id

    # Private route table with NAT GW
    rt_priv = ec2.create_route_table(
        VpcId=vpc_id,
        TagSpecifications=tag_spec("route-table", "rt-private"),
    )
    rt_private_id = rt_priv["RouteTable"]["RouteTableId"]
    ec2.create_route(
        RouteTableId=rt_private_id,
        DestinationCidrBlock="0.0.0.0/0",
        NatGatewayId=state["nat_gw_id"],
    )

    # Associate private subnets
    for name, subnet_id in state["subnets"].items():
        if name.startswith("private"):
            ec2.associate_route_table(
                RouteTableId=rt_private_id,
                SubnetId=subnet_id,
            )

    state["rt_private_id"] = rt_private_id
    save_state(state)
    print(f"  Public RT: {rt_main_id}, Private RT: {rt_private_id}")


def create_security_groups(state):
    if state.get("sg_agentcore_id"):
        print(f"Security groups already exist")
        return

    vpc_id = state["vpc_id"]
    print("Creating security groups...")

    # AgentCore SG
    sg_ac = ec2.create_security_group(
        GroupName=f"{PROJECT_NAME}-agentcore",
        Description="AgentCore Runtime - allows outbound NFS and internet",
        VpcId=vpc_id,
        TagSpecifications=tag_spec("security-group", "sg-agentcore"),
    )
    sg_agentcore_id = sg_ac["GroupId"]

    # S3 Files Mount Target SG
    sg_mt = ec2.create_security_group(
        GroupName=f"{PROJECT_NAME}-s3files-mt",
        Description="S3 Files Mount Target - allows NFS from AgentCore",
        VpcId=vpc_id,
        TagSpecifications=tag_spec("security-group", "sg-s3files-mt"),
    )
    sg_s3files_mt_id = sg_mt["GroupId"]

    # AgentCore → S3 Files (NFS outbound)
    ec2.authorize_security_group_egress(
        GroupId=sg_agentcore_id,
        IpPermissions=[{
            "IpProtocol": "tcp",
            "FromPort": 2049,
            "ToPort": 2049,
            "UserIdGroupPairs": [{"GroupId": sg_s3files_mt_id}],
        }],
    )

    # S3 Files ← AgentCore (NFS inbound)
    ec2.authorize_security_group_ingress(
        GroupId=sg_s3files_mt_id,
        IpPermissions=[{
            "IpProtocol": "tcp",
            "FromPort": 2049,
            "ToPort": 2049,
            "UserIdGroupPairs": [{"GroupId": sg_agentcore_id}],
        }],
    )

    state["sg_agentcore_id"] = sg_agentcore_id
    state["sg_s3files_mt_id"] = sg_s3files_mt_id
    save_state(state)
    print(f"  AgentCore SG: {sg_agentcore_id}")
    print(f"  S3 Files MT SG: {sg_s3files_mt_id}")


def create_vpc_endpoints(state):
    """Create VPC endpoints needed for AgentCore in private subnets."""
    if state.get("vpc_endpoints"):
        print(f"VPC endpoints already exist")
        return

    vpc_id = state["vpc_id"]
    private_subnets = [
        sid for name, sid in state["subnets"].items() if name.startswith("private")
    ]

    print("Creating VPC endpoints...")
    endpoints = {}

    # S3 Gateway endpoint
    try:
        resp = ec2.create_vpc_endpoint(
            VpcId=vpc_id,
            ServiceName=f"com.amazonaws.{REGION}.s3",
            RouteTableIds=[state["rt_private_id"]],
            VpcEndpointType="Gateway",
            TagSpecifications=tag_spec("vpc-endpoint", "vpce-s3"),
        )
        endpoints["s3"] = resp["VpcEndpoint"]["VpcEndpointId"]
        print(f"  S3 Gateway: {endpoints['s3']}")
    except ec2.exceptions.ClientError as e:
        if "RouteAlreadyExists" in str(e):
            print("  S3 Gateway endpoint route already exists, skipping")
        else:
            raise

    state["vpc_endpoints"] = endpoints
    save_state(state)


def create_private_route_table(state):
    """Create a route table for private subnets (no default route - no NAT/IGW needed)."""
    if state.get("rt_private_id"):
        print(f"Route table already configured")
        return

    vpc_id = state["vpc_id"]
    print("Creating private route table...")
    rt_priv = ec2.create_route_table(
        VpcId=vpc_id,
        TagSpecifications=tag_spec("route-table", "rt-private"),
    )
    rt_private_id = rt_priv["RouteTable"]["RouteTableId"]

    for name, subnet_id in state["subnets"].items():
        if name.startswith("private"):
            ec2.associate_route_table(
                RouteTableId=rt_private_id,
                SubnetId=subnet_id,
            )

    state["rt_private_id"] = rt_private_id
    save_state(state)
    print(f"  Private RT: {rt_private_id} (no default route)")


def main():
    state = load_state()
    print(f"=== Setting up test VPC for {PROJECT_NAME} (minimal, no NAT/VPCE) ===\n")

    create_vpc(state)
    create_subnets(state)
    # NAT Gateway, IGW, VPC Endpoint は不要（AgentCoreが内部経路を持つため）
    create_private_route_table(state)
    create_security_groups(state)

    print(f"\n=== VPC setup complete ===")
    print(f"VPC:              {state['vpc_id']}")
    print(f"Private subnets:  {state['subnets']['private-1a']}, {state['subnets']['private-1c']}")
    print(f"AgentCore SG:     {state['sg_agentcore_id']}")
    print(f"S3 Files MT SG:   {state['sg_s3files_mt_id']}")


if __name__ == "__main__":
    main()

2. S3 Files/コンピュートのサービスロール作成

それぞれ2つのサービルロールを作成します。
S3 Filesサービルロールにおいては、S3 FilesがS3バケットにアクセスするための権限を付与します。

見ていただくとわかると思いますが、S3バケットへのアクセス権限に加えて、EventBridgeの権限も必要になっています。
S3 FilesとS3バケットを同期するタイミングで、裏側でEventBridgeが使われているようですね。

S3Files用IAMポリシーのサンプル
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3BucketPermissions",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:ListBucketVersions"
            ],
            "Resource": "arn:aws:s3:::bucket",
            "Condition": {
                "StringEquals": {
                    "aws:ResourceAccount": "accountId"
                }
            }
        },
        {
            "Sid": "S3ObjectPermissions",
            "Effect": "Allow",
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:DeleteObject*",
                "s3:GetObject*",
                "s3:List*",
                "s3:PutObject*"
            ],
            "Resource": "arn:aws:s3:::bucket/*",
            "Condition": {
                "StringEquals": {
                    "aws:ResourceAccount": "accountId"
                }
            }
        },
        {
            "Sid": "UseKmsKeyWithS3Files",
            "Effect": "Allow",
            "Action": [
                "kms:GenerateDataKey",
                "kms:Encrypt",
                "kms:Decrypt",
                "kms:ReEncryptFrom",
                "kms:ReEncryptTo"
            ],
            "Condition": {
                "StringLike": {
                    "kms:ViaService": "s3.region.amazonaws.com",
                    "kms:EncryptionContext:aws:s3:arn": [
                        "arn:aws:s3:::bucket",
                        "arn:aws:s3:::bucket/*"
                    ]
                }
            },
            "Resource": "arn:aws:kms:region:accountId:*"
        },
        {
            "Sid": "EventBridgeManage",
            "Effect": "Allow",
            "Action": [
                "events:DeleteRule",
                "events:DisableRule",
                "events:EnableRule",
                "events:PutRule",
                "events:PutTargets",
                "events:RemoveTargets"
            ],
            "Condition": {
                "StringEquals": {
                    "events:ManagedBy": "elasticfilesystem.amazonaws.com"
                }
            },
            "Resource": [
                "arn:aws:events:*:*:rule/DO-NOT-DELETE-S3-Files*"
            ]
        },
        {
            "Sid": "EventBridgeRead",
            "Effect": "Allow",
            "Action": [
                "events:DescribeRule",
                "events:ListRuleNamesByTarget",
                "events:ListRules",
                "events:ListTargetsByRule"
            ],
            "Resource": [
                "arn:aws:events:*:*:rule/*"
            ]
        }
    ]
}

また、信頼関係においてPrincipalでelasticfilesystem.amazonaws.comを設定する必要があります。S3 Filesの裏側ではEFSが用いられているということですね。
(まぁ正直、VPCなしで動かせるようにしてほしいところですが…)

S3 Files用IAMポリシーの信頼関係
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowS3FilesAssumeRole",
            "Effect": "Allow",
            "Principal": {
                "Service": "elasticfilesystem.amazonaws.com"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringEquals": {
                    "aws:SourceAccount": "accountId"
                },
                "ArnLike": {
                    "aws:SourceArn": "arn:aws:s3files:region:accountId:file-system/*"
                }
            }
        }
    ]
}

続いて、AgentCore RuntimeサービルロールにはS3 Filesへのアクセス権限を付与します。
いくつかマネージドなポリシーも用意されているようなので、今回はAmazonS3FilesClientFullAccessを用います。

詳しい説明は以下をご参照ください。

サンプルスクリプト
setup_iam.py
#!/usr/bin/env python3
"""
Create IAM roles:
1. AgentCore Runtime role - for running the agent container
2. S3 Files service role - for S3 Files to access the S3 bucket
   (Principal: elasticfilesystem.amazonaws.com)
"""

import json
import sys
import boto3
from botocore.exceptions import ClientError
from config import REGION, PROJECT_NAME, S3_BUCKET_NAME, save_state, load_state

iam = boto3.client("iam", region_name=REGION)
sts = boto3.client("sts", region_name=REGION)

AGENTCORE_ROLE_NAME = f"{PROJECT_NAME}-agentcore-role"
S3FILES_ROLE_NAME = f"{PROJECT_NAME}-s3files-role"


def get_account_id():
    return sts.get_caller_identity()["Account"]


def _create_role_if_needed(role_name, trust_policy, description, state_key, state):
    if state.get(state_key):
        print(f"  Role already exists: {state[state_key]}")
        return state[state_key]

    try:
        resp = iam.create_role(
            RoleName=role_name,
            AssumeRolePolicyDocument=json.dumps(trust_policy),
            Description=description,
            Tags=[{"Key": "Project", "Value": PROJECT_NAME}],
        )
        role_arn = resp["Role"]["Arn"]
    except ClientError as e:
        if e.response["Error"]["Code"] == "EntityAlreadyExists":
            resp = iam.get_role(RoleName=role_name)
            role_arn = resp["Role"]["Arn"]
            print(f"  Role already exists: {role_arn}")
        else:
            raise

    state[state_key] = role_arn
    save_state(state)
    print(f"  Role ARN: {role_arn}")
    return role_arn


def create_agentcore_role(state):
    """Create IAM role for AgentCore Runtime."""
    account_id = state["account_id"]

    trust_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
                "Action": "sts:AssumeRole",
                "Condition": {
                    "StringEquals": {"aws:SourceAccount": account_id}
                },
            }
        ],
    }

    print(f"Creating AgentCore role: {AGENTCORE_ROLE_NAME}...")
    _create_role_if_needed(
        AGENTCORE_ROLE_NAME, trust_policy,
        "AgentCore Runtime role for S3 Files connectivity test",
        "iam_role_arn", state,
    )
    state["iam_role_name"] = AGENTCORE_ROLE_NAME

    # Attach inline policy
    account_id = state["account_id"]
    inline_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "S3FilesAccess",
                "Effect": "Allow",
                "Action": [
                    "s3:GetObject",
                    "s3:GetObjectVersion",
                    "s3:PutObject",
                    "s3:ListBucket",
                    "s3:DeleteObject",
                    "s3:GetBucketLocation",
                ],
                "Resource": [
                    f"arn:aws:s3:::{S3_BUCKET_NAME}",
                    f"arn:aws:s3:::{S3_BUCKET_NAME}/*",
                ],
            },
            {
                "Sid": "VPCNetworking",
                "Effect": "Allow",
                "Action": [
                    "ec2:CreateNetworkInterface",
                    "ec2:DescribeNetworkInterfaces",
                    "ec2:DeleteNetworkInterface",
                    "ec2:AssignPrivateIpAddresses",
                    "ec2:UnassignPrivateIpAddresses",
                    "ec2:DescribeSubnets",
                    "ec2:DescribeSecurityGroups",
                    "ec2:DescribeVpcs",
                ],
                "Resource": "*",
            },
            {
                "Sid": "CloudWatchLogs",
                "Effect": "Allow",
                "Action": [
                    "logs:CreateLogGroup",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents",
                ],
                "Resource": f"arn:aws:logs:{REGION}:{account_id}:*",
            },
            {
                "Sid": "BedrockAccess",
                "Effect": "Allow",
                "Action": [
                    "bedrock:InvokeModel",
                    "bedrock:InvokeModelWithResponseStream",
                ],
                "Resource": "*",
            },
        ],
    }

    print("  Attaching AgentCore inline policy...")
    iam.put_role_policy(
        RoleName=AGENTCORE_ROLE_NAME,
        PolicyName=f"{PROJECT_NAME}-agentcore-policy",
        PolicyDocument=json.dumps(inline_policy),
    )

    # Attach managed policy for S3 Files client access
    try:
        iam.attach_role_policy(
            RoleName=AGENTCORE_ROLE_NAME,
            PolicyArn="arn:aws:iam::aws:policy/AmazonS3FilesClientFullAccess",
        )
        print("  Attached AmazonS3FilesClientFullAccess managed policy")
    except ClientError as e:
        print(f"  Could not attach S3FilesClientFullAccess: {e}")
        print("  (This managed policy may not exist yet; inline policy covers S3 access)")

    save_state(state)


def create_s3files_service_role(state):
    """Create IAM role for S3 Files service (assumed by elasticfilesystem.amazonaws.com)."""
    account_id = state["account_id"]

    trust_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AllowS3FilesAssumeRole",
                "Effect": "Allow",
                "Principal": {"Service": "elasticfilesystem.amazonaws.com"},
                "Action": "sts:AssumeRole",
                "Condition": {
                    "StringEquals": {"aws:SourceAccount": account_id},
                    "ArnLike": {
                        "aws:SourceArn": f"arn:aws:s3files:{REGION}:{account_id}:file-system/*"
                    },
                },
            }
        ],
    }

    print(f"Creating S3 Files service role: {S3FILES_ROLE_NAME}...")
    _create_role_if_needed(
        S3FILES_ROLE_NAME, trust_policy,
        "S3 Files service role - allows EFS to access S3 bucket",
        "s3files_role_arn", state,
    )
    state["s3files_role_name"] = S3FILES_ROLE_NAME

    # Attach permissions policy per AWS docs
    permissions_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "S3BucketPermissions",
                "Effect": "Allow",
                "Action": ["s3:ListBucket", "s3:ListBucketVersions"],
                "Resource": f"arn:aws:s3:::{S3_BUCKET_NAME}",
                "Condition": {
                    "StringEquals": {"aws:ResourceAccount": account_id}
                },
            },
            {
                "Sid": "S3ObjectPermissions",
                "Effect": "Allow",
                "Action": [
                    "s3:AbortMultipartUpload",
                    "s3:DeleteObject*",
                    "s3:GetObject*",
                    "s3:List*",
                    "s3:PutObject*",
                ],
                "Resource": f"arn:aws:s3:::{S3_BUCKET_NAME}/*",
                "Condition": {
                    "StringEquals": {"aws:ResourceAccount": account_id}
                },
            },
            {
                "Sid": "EventBridgeManage",
                "Effect": "Allow",
                "Action": [
                    "events:DeleteRule",
                    "events:DisableRule",
                    "events:EnableRule",
                    "events:PutRule",
                    "events:PutTargets",
                    "events:RemoveTargets",
                ],
                "Condition": {
                    "StringEquals": {
                        "events:ManagedBy": "elasticfilesystem.amazonaws.com"
                    }
                },
                "Resource": ["arn:aws:events:*:*:rule/DO-NOT-DELETE-S3-Files*"],
            },
            {
                "Sid": "EventBridgeRead",
                "Effect": "Allow",
                "Action": [
                    "events:DescribeRule",
                    "events:ListRuleNamesByTarget",
                    "events:ListRules",
                    "events:ListTargetsByRule",
                ],
                "Resource": ["arn:aws:events:*:*:rule/*"],
            },
        ],
    }

    print("  Attaching S3 Files service permissions policy...")
    iam.put_role_policy(
        RoleName=S3FILES_ROLE_NAME,
        PolicyName=f"{PROJECT_NAME}-s3files-policy",
        PolicyDocument=json.dumps(permissions_policy),
    )
    save_state(state)


def main():
    state = load_state()
    account_id = get_account_id()
    state["account_id"] = account_id

    print(f"=== Setting up IAM roles (account: {account_id}) ===\n")

    create_agentcore_role(state)
    print()
    create_s3files_service_role(state)

    print(f"\n=== IAM setup complete ===")
    print(f"AgentCore role:    {state['iam_role_arn']}")
    print(f"S3 Files role:     {state['s3files_role_arn']}")


if __name__ == "__main__":
    main()

3. S3バケット作成 + バージョニング有効化

S3バケットを作成し、バージョニングを有効にします。バージョニングが無効だとS3 Files有効化時に失敗します。

s3 = boto3.client("s3", region_name="ap-northeast-1")

# バケット作成
s3.create_bucket(
    Bucket="my-s3files-bucket",
    CreateBucketConfiguration={"LocationConstraint": "ap-northeast-1"},
)

# バージョニング有効化（必須）
s3.put_bucket_versioning(
    Bucket="my-s3files-bucket",
    VersioningConfiguration={"Status": "Enabled"},
)

4. S3 Filesとマウントターゲット作成

S3 FilesのAPI操作には通常のs3クライアントではなく、専用のs3filesクライアントを使います。
ただし、boto3でs3filesクライアントを使うには、S3 FilesのGA（2026年4月7日）以降にリリースされたバージョンが必要です。v1.42.85以上であればおそらくOKです。

s3files = boto3.client("s3files", region_name="ap-northeast-1")

# ファイルシステム作成
resp = s3files.create_file_system(
    bucket="arn:aws:s3:::my-s3files-bucket",
    roleArn="arn:aws:iam::<account-id>:role/s3files-service-role",
)
fs_id = resp["fileSystemId"]

# マウントターゲット作成（サブネットごとに1つ）
s3files.create_mount_target(
    fileSystemId=fs_id,
    subnetId="subnet-xxxxxxxxx",
    securityGroups=["sg-xxxxxxxxx"],
)

コンソールからS3 Filesを作成すると、自動作成されたIAMロールが付与されます。
またこの時、自動的にマウントターゲットまで作成されるのですが、セキュリティグループにdefaultが付与されるので注意が必要です。

この点から、個人的には現在であればSDK/CLIで、将来的にはCDKなどでS3 Filesを作成することをおすすめします。

サンプルスクリプト
setup_s3files.py
#!/usr/bin/env python3
"""
Create S3 bucket, enable S3 Files filesystem, and create mount targets.
Uses the `s3files` service client (not `s3`).
"""

import sys
import time
import boto3
from botocore.exceptions import ClientError
from config import REGION, PROJECT_NAME, S3_BUCKET_NAME, save_state, load_state

s3 = boto3.client("s3", region_name=REGION)
s3files = boto3.client("s3files", region_name=REGION)


def create_bucket(state):
    if state.get("s3_bucket"):
        print(f"S3 bucket already exists: {state['s3_bucket']}")
        return

    print(f"Creating S3 bucket: {S3_BUCKET_NAME}...")
    try:
        s3.create_bucket(
            Bucket=S3_BUCKET_NAME,
            CreateBucketConfiguration={"LocationConstraint": REGION},
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "BucketAlreadyOwnedByYou":
            print("  Bucket already exists (owned by you)")
        else:
            raise

    # バージョニング有効化（S3 Files必須）
    s3.put_bucket_versioning(
        Bucket=S3_BUCKET_NAME,
        VersioningConfiguration={"Status": "Enabled"},
    )
    print(f"  Versioning enabled")

    state["s3_bucket"] = S3_BUCKET_NAME
    save_state(state)
    print(f"  Bucket created: {S3_BUCKET_NAME}")


def create_file_system(state):
    if state.get("s3files_fs_id"):
        print(f"S3 Files filesystem already exists: {state['s3files_fs_id']}")
        return

    account_id = state["account_id"]
    bucket_arn = f"arn:aws:s3:::{S3_BUCKET_NAME}"
    s3files_role_arn = state["s3files_role_arn"]

    print(f"Creating S3 Files filesystem on bucket {S3_BUCKET_NAME}...")
    print(f"  Bucket ARN:  {bucket_arn}")
    print(f"  Role ARN:    {s3files_role_arn}")

    try:
        resp = s3files.create_file_system(
            bucket=bucket_arn,
            roleArn=s3files_role_arn,
        )
        fs_id = resp["fileSystemId"]
        status = resp.get("status", "unknown")
        print(f"  Filesystem ID: {fs_id} (status: {status})")
    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        if "AlreadyExists" in error_code or "Conflict" in error_code:
            print("  Filesystem already exists, retrieving...")
            fs_list = s3files.list_file_systems()
            for fs in fs_list.get("fileSystems", []):
                if S3_BUCKET_NAME in fs.get("bucket", ""):
                    fs_id = fs["fileSystemId"]
                    break
            else:
                raise RuntimeError("Could not find existing filesystem")
        else:
            raise

    state["s3files_fs_id"] = fs_id
    save_state(state)


def wait_for_filesystem(state):
    """Wait for the S3 Files filesystem to become available."""
    fs_id = state["s3files_fs_id"]
    print(f"  Waiting for filesystem {fs_id} to become available...", end="", flush=True)

    for _ in range(60):
        try:
            resp = s3files.get_file_system(fileSystemId=fs_id)
            status = resp.get("status", "unknown")
            if status.upper() in ("AVAILABLE", "ACTIVE", "ENABLED"):
                print(f" done ({status})")
                return
            print(f".", end="", flush=True)
        except Exception:
            print(".", end="", flush=True)
        time.sleep(10)

    print(" (proceeding anyway - filesystem may still be creating)")


def create_mount_targets(state):
    if state.get("mount_targets"):
        print(f"Mount targets already exist: {list(state['mount_targets'].keys())}")
        return

    fs_id = state["s3files_fs_id"]
    sg_mt_id = state["sg_s3files_mt_id"]
    mount_targets = {}

    print("Creating mount targets (takes ~5 min each)...")
    for subnet_name in ["private-1a", "private-1c"]:
        subnet_id = state["subnets"][subnet_name]
        print(f"  Creating mount target in {subnet_name} ({subnet_id})...")
        try:
            resp = s3files.create_mount_target(
                fileSystemId=fs_id,
                subnetId=subnet_id,
                securityGroups=[sg_mt_id],
            )
            mt_id = resp.get("mountTargetId", "created")
            ip_addr = resp.get("ipv4Address", resp.get("ipAddress", "pending"))
            mount_targets[subnet_name] = {
                "mount_target_id": mt_id,
                "subnet_id": subnet_id,
                "ip_address": ip_addr,
            }
            print(f"    Mount target: {mt_id}, IP: {ip_addr}")
        except ClientError as e:
            error_msg = str(e)
            if "Conflict" in error_msg or "AlreadyExists" in error_msg:
                print(f"    Mount target already exists in {subnet_name}")
            else:
                raise

    state["mount_targets"] = mount_targets
    save_state(state)


def wait_for_mount_targets(state):
    """Wait for mount targets to become available and retrieve their IPs."""
    fs_id = state["s3files_fs_id"]
    print("  Waiting for mount targets to become available...", end="", flush=True)

    for _ in range(40):
        try:
            resp = s3files.list_mount_targets(fileSystemId=fs_id)
            targets = resp.get("mountTargets", [])
            all_available = all(
                t.get("status", "").upper() in ("AVAILABLE", "ACTIVE")
                for t in targets
            )
            if all_available and targets:
                print(" done")
                # Update state with IPs
                for mt in targets:
                    ip = mt.get("ipv4Address", mt.get("ipAddress"))
                    subnet_id = mt.get("subnetId")
                    if ip and subnet_id:
                        for name, info in state.get("mount_targets", {}).items():
                            if info.get("subnet_id") == subnet_id:
                                info["ip_address"] = ip
                                info["mount_target_id"] = mt.get("mountTargetId", info["mount_target_id"])
                save_state(state)

                print("  Mount target details:")
                for name, info in state["mount_targets"].items():
                    print(f"    {name}: IP={info['ip_address']}, ID={info['mount_target_id']}")
                return

            print(".", end="", flush=True)
        except Exception as e:
            print(f"x", end="", flush=True)
        time.sleep(15)

    print(" (timed out - check status manually)")
    print(f"  aws s3files list-mount-targets --file-system-id {fs_id} --region {REGION}")


def main():
    state = load_state()
    if not state.get("vpc_id"):
        print("ERROR: VPC not set up. Run setup_vpc.py first.")
        sys.exit(1)
    if not state.get("s3files_role_arn"):
        print("ERROR: S3 Files IAM role not set up. Run setup_iam.py first.")
        sys.exit(1)

    print(f"=== Setting up S3 Files for {PROJECT_NAME} ===\n")

    create_bucket(state)
    create_file_system(state)
    wait_for_filesystem(state)
    create_mount_targets(state)
    wait_for_mount_targets(state)

    print(f"\n=== S3 Files setup complete ===")
    print(f"Bucket:     {state['s3_bucket']}")
    print(f"FS ID:      {state['s3files_fs_id']}")
    for name, info in state.get("mount_targets", {}).items():
        print(f"Mount ({name}): IP={info.get('ip_address')}")
    print(f"\nMount command (inside container):")
    print(f"  sudo mount -t s3files {state['s3files_fs_id']}:/ /mnt/s3files")


if __name__ == "__main__":
    main()

5. amazon-efs-utils v3.0.0をインストール

ここからEC2と手順が少し変わってきます。また、Dockerfileの編集がメインになってきます。

EC2(Amazon Linux 2023)であれば、セッションマネージャーなどで接続して、クラスメソッドさんの記事を参考にamazon-efs-utils v3.0.0以上をインストールすればOKです。

# 上記記事よりコマンド引用
% sudo dnf install -y git rpm-build make rust cargo openssl-devel \
    cmake gcc gcc-c++ perl python3-devel golang
% cd /tmp && git clone --depth 1 --branch v3.0.0 https://github.com/aws/efs-utils.git
% cd efs-utils && sudo make rpm
% sudo dnf install -y build/amazon-efs-utils*rpm
% rpm -q amazon-efs-utils
amazon-efs-utils-3.0.0-1.amzn2023.x86_64

しかしAgentCore CLIが生成するDockerfileのベースイメージはDebian系（ghcr.io/astral-sh/uv:python3.12-bookworm-slim）です。
amazon-efs-utilsのaptパッケージは提供されていないため、apt-get installではインストールできません。

ダメ元でpublic.ecr.aws/amazonlinux/amazonlinux:2023を用いてみると、dnf install amazon-efs-utilsでv3.0.0がインストールできていることをCodeBuildのログで確認しました。

# CodeBuildのログより
amazon-efs-utils    aarch64  3.0.0-4.amzn2023    amazonlinux  6.5 M

ということで、Amazon Linux 2023をベースイメージにすれば、dnf install一発でamazon-efs-utilsが入ります。

Dockerfile
FROM public.ecr.aws/amazonlinux/amazonlinux:2023

# amazon-efs-utils (v3.0.0) + Python + 必要ツールを一括インストール
RUN dnf install -y \
        amazon-efs-utils \
        python3.12 python3.12-pip python3.12-devel \
        procps iproute nmap-ncat \
        tar gzip && \
    dnf clean all && \
    mkdir -p /mnt/s3files /var/log/amazon/efs

# uvのインストール
RUN curl -LsSf https://astral.sh/uv/install.sh | sh && \
    ln -s /root/.local/bin/uv /usr/local/bin/uv

ARG UV_DEFAULT_INDEX
ARG UV_INDEX

WORKDIR /app

ENV UV_SYSTEM_PYTHON=1 \
    UV_COMPILE_BYTECODE=1 \
    UV_NO_PROGRESS=1 \
    PYTHONUNBUFFERED=1 \
    DOCKER_CONTAINER=1 \
    UV_DEFAULT_INDEX=${UV_DEFAULT_INDEX} \
    UV_INDEX=${UV_INDEX} \
    PATH="/app/.venv/bin:$PATH"

RUN useradd -m -u 1000 bedrock_agentcore

COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project

COPY --chown=bedrock_agentcore:bedrock_agentcore . .
RUN uv sync --frozen --no-dev

# entrypoint.shでrootマウント→非rootアプリ起動（後述）
COPY entrypoint.sh /app/entrypoint.sh
RUN chmod +x /app/entrypoint.sh

# AgentCore Runtimeのサービスコントラクトで定義されているポート
EXPOSE 8080 8000 9000

ENTRYPOINT ["/app/entrypoint.sh"]
CMD ["opentelemetry-instrument", "python", "-m", "main"]

ただ、Amazon Linux 2023以外を使いたい場合もあるかと思います。
その時はマルチステージビルドでバイナリをコピーする方式を採用しました。

# Stage 1: Amazon Linux 2023からefs-utilsを取得
FROM public.ecr.aws/amazonlinux/amazonlinux:2023 AS efs-utils
RUN dnf install -y amazon-efs-utils && \
    mkdir -p /export/sbin /export/bin /export/etc && \
    cp -r /usr/sbin/mount.efs /usr/sbin/mount.s3files \
          /usr/sbin/efs-proxy /usr/sbin/efs_utils_common \
          /usr/sbin/mount_efs /usr/sbin/mount_s3files \
          /export/sbin/ && \
    cp /usr/bin/amazon-efs-mount-watchdog /export/bin/ && \
    cp -r /etc/amazon/efs /export/etc/

# Stage 2: アプリケーション
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        nfs-common stunnel4 python3 procps iproute2 netcat-openbsd \
        openssl ca-certificates && \
    rm -rf /var/lib/apt/lists/*

COPY --from=efs-utils /export/sbin/ /usr/sbin/
COPY --from=efs-utils /export/bin/ /usr/bin/
COPY --from=efs-utils /export/etc/ /etc/amazon/

ハマりポイント: efs_utils_commonの配置パス

amazon-efs-utilsのPythonモジュール(efs_utils_common等)は/usr/lib/python*/site-packages/ではなく/usr/sbin/に配置されます。
rpm -ql amazon-efs-utilsでインストール先を確認して判明しました。

6. sudo mount -t s3files :/ /mnt/s3files でコンピュートからマウント

最後に、コンピュートリソースからS3 Filesをマウントします。
EC2であればsudo mount -t s3filesで終わりですが、AgentCoreのコンテナ環境では少し事情が異なります。

AgentCore CLIが生成するデフォルトDockerfileにはUSER bedrock_agentcore（UID 1000）があり、非rootユーザーで実行されます。
しかしmountコマンドにはroot権限が必要です。

かといってコンテナ全体をrootで動かすのはセキュリティ上望ましくありません。そこで、entrypoint.shというスクリプトを作って3段階の処理を行う構成にしました。

  1. rootでS3 Filesをマウント
  2. 非rootユーザー用のワークスペースディレクトリを作成して所有権を移譲
  3. 非rootユーザーに切り替えてアプリを起動
entrypoint.sh
#!/bin/bash
set -e

MOUNT_POINT="${NFS_MOUNT_POINT:-/mnt/s3files}"
FS_ID="${NFS_FS_ID:-}"
APP_USER="bedrock_agentcore"
WORKSPACE="${MOUNT_POINT}/workspace"

mkdir -p "$MOUNT_POINT"

# Step 1: rootでマウント
if [ -n "$FS_ID" ] && [ -x /usr/sbin/mount.s3files ]; then
    mount -t s3files "$FS_ID":/ "$MOUNT_POINT" 2>&1 && \
        echo "S3 Files mount SUCCESS" || \
        echo "S3 Files mount FAILED"
fi

# Step 2: ワークスペース準備
if mountpoint -q "$MOUNT_POINT" 2>/dev/null; then
    mkdir -p "$WORKSPACE"
    chown "$APP_USER":"$APP_USER" "$WORKSPACE"
fi

# Step 3: 非rootでアプリ起動
exec su -s /bin/bash "$APP_USER" -c "exec $*"

この構成により、非rootユーザーは/mnt/s3files/workspace/に読み書きでき、それ以外のディレクトリは読み取りのみとなります。

パス 所有者 非rootユーザーの権限
/mnt/s3files/ root 読み取りのみ
/mnt/s3files/workspace/ bedrock_agentcore 読み書き可能

このスクリプトをDockerfile内で動かすことで、AgentCore Runtime起動の段階でS3 Filesとの接続や非ルートユーザー作成などが自動的に行われます。

参考：AgentCore CLIでのプロジェクト作成とデプロイ

あとはエージェントを作成して、AgentCore Runtimeへデプロイするだけです。
エージェントにはファイル操作用のツールを渡してあげましょう。今回はテスト用に色々確認用ツールも渡しています。

サンプルエージェントコード
main.py
import json
import os
import subprocess

os.environ["BYPASS_TOOL_CONSENT"] = "true"

from strands import Agent, tool
try:
    from strands_tools import file_read, file_write, shell, editor
    STRANDS_TOOLS_AVAILABLE = True
except ImportError:
    STRANDS_TOOLS_AVAILABLE = False
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model

app = BedrockAgentCoreApp()
log = app.logger

# NFS configuration from environment
NFS_FS_ID = os.environ.get("NFS_FS_ID", "")
NFS_MOUNT_IP = os.environ.get("NFS_MOUNT_IP", "")
MOUNT_POINT = os.environ.get("NFS_MOUNT_POINT", "/mnt/s3files")

tools = []


def _run(cmd: str, timeout: int = 30) -> dict:
    """Run shell command and return result."""
    try:
        r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
        return {"cmd": cmd, "exit": r.returncode, "out": r.stdout.strip(), "err": r.stderr.strip()}
    except subprocess.TimeoutExpired:
        return {"cmd": cmd, "exit": -1, "out": "", "err": "timeout"}


@tool
def check_environment() -> str:
    """Check the container environment: OS, user, capabilities, NFS tools, network."""
    results = {
        "uname": _run("uname -a"),
        "whoami": _run("whoami && id"),
        "capabilities": _run("cat /proc/1/status | grep -i cap"),
        "nfs_tools": _run("which mount.nfs4 || which mount.nfs || echo NO_NFS_TOOLS"),
        "network": _run("ip addr show | grep -E 'inet '"),
        "nfs_config": {"fs_id": NFS_FS_ID, "mount_ip": NFS_MOUNT_IP, "mount_point": MOUNT_POINT},
    }
    return json.dumps(results, indent=2)

tools.append(check_environment)


@tool
def test_nfs_connectivity() -> str:
    """Test TCP connectivity to NFS mount target on port 2049."""
    results = {}
    if NFS_MOUNT_IP:
        results["tcp_2049"] = _run(
            f"nc -zv -w5 {NFS_MOUNT_IP} 2049 2>&1", timeout=10
        )
    else:
        results["tcp_2049"] = {"error": "NFS_MOUNT_IP not set"}
    return json.dumps(results, indent=2)

tools.append(test_nfs_connectivity)


@tool
def mount_s3files() -> str:
    """Attempt to mount S3 Files filesystem via NFS."""
    results = {}
    _run(f"mkdir -p {MOUNT_POINT}")

    # Try mount -t nfs4 (using IP)
    if NFS_MOUNT_IP:
        results["nfs4_mount"] = _run(
            f"mount -t nfs4 -o nfsvers=4.1,rsize=1048576,wsize=1048576,hard,timeo=600,retrans=2 "
            f"{NFS_MOUNT_IP}:/ {MOUNT_POINT} 2>&1",
            timeout=30,
        )

    # Fallback: try mount -t s3files (using FS ID)
    if not results.get("nfs4_mount", {}).get("exit") == 0 and NFS_FS_ID:
        results["s3files_mount"] = _run(
            f"mount -t s3files {NFS_FS_ID}:/ {MOUNT_POINT} 2>&1",
            timeout=30,
        )

    results["verify"] = _run(f"mount | grep -E 'nfs|s3files' || echo 'NO MOUNT FOUND'")
    results["df"] = _run(f"df -h {MOUNT_POINT}")
    return json.dumps(results, indent=2)

tools.append(mount_s3files)


@tool
def test_file_io() -> str:
    """Test file read/write on mounted S3 Files."""
    results = {}
    results["write"] = _run(f"echo 'Hello from AgentCore' > {MOUNT_POINT}/test.txt 2>&1")
    results["read"] = _run(f"cat {MOUNT_POINT}/test.txt 2>&1")
    results["ls"] = _run(f"ls -la {MOUNT_POINT}/ 2>&1")
    results["cleanup"] = _run(f"rm -f {MOUNT_POINT}/test.txt 2>&1")
    return json.dumps(results, indent=2)

tools.append(test_file_io)


@tool
def run_full_s3files_test() -> str:
    """Run complete S3 Files connectivity test: env check, connectivity, mount, file I/O."""
    report = {}
    report["1_environment"] = json.loads(check_environment())
    report["2_connectivity"] = json.loads(test_nfs_connectivity())
    report["3_mount"] = json.loads(mount_s3files())

    mount_ok = (
        report["3_mount"].get("nfs4_mount", {}).get("exit") == 0
        or report["3_mount"].get("s3files_mount", {}).get("exit") == 0
    )
    if mount_ok:
        report["4_file_io"] = json.loads(test_file_io())
    else:
        report["4_file_io"] = {"skipped": "mount failed"}

    report["summary"] = {
        "nfs_tools": "NO_NFS_TOOLS" not in report["1_environment"].get("nfs_tools", {}).get("out", ""),
        "tcp_ok": report["2_connectivity"].get("tcp_2049", {}).get("exit") == 0,
        "mount_ok": mount_ok,
        "io_ok": report.get("4_file_io", {}).get("write", {}).get("exit") == 0,
    }
    return json.dumps(report, indent=2)

tools.append(run_full_s3files_test)

# Add strands community tools for general file operations
if STRANDS_TOOLS_AVAILABLE:
    tools.extend([file_read, file_write, shell, editor])


_agent = None

def get_or_create_agent():
    global _agent
    if _agent is None:
        _agent = Agent(
            model=load_model(),
            system_prompt=f"""You are an S3 Files connectivity test agent running inside AgentCore Runtime.
You can test S3 Files NFS connectivity AND perform general file operations.

Configuration:
- File System ID: {NFS_FS_ID}
- Mount Target IP: {NFS_MOUNT_IP}
- Mount Point: {MOUNT_POINT}
- Workspace: {MOUNT_POINT}/workspace (you have write permission here)
- S3 Files is mounted at {MOUNT_POINT} (read-only for non-root areas, writable under workspace)

IMPORTANT: You are running as non-root user (bedrock_agentcore).
- You can READ files anywhere under {MOUNT_POINT}
- You can WRITE files under {MOUNT_POINT}/workspace/ (owned by you)
- For other directories, use shell to create and chown if needed

Available capabilities:
1. S3 Files connectivity testing (check_environment, test_nfs_connectivity, mount_s3files, test_file_io, run_full_s3files_test)
2. General file operations (file_read, file_write, editor) - use these to create, read, edit files
3. Shell command execution (shell) - for any other operations

When asked to create files, prefer {MOUNT_POINT}/workspace/ as the default location.
When asked to test connectivity, use run_full_s3files_test.
Always respond in the same language the user used.""",
            tools=tools,
        )
    return _agent


@app.entrypoint
async def invoke(payload, context):
    log.info("Invoking S3 Files test agent...")

    agent = get_or_create_agent()
    stream = agent.stream_async(payload.get("prompt"))

    async for event in stream:
        if "data" in event and isinstance(event["data"], str):
            yield event["data"]


if __name__ == "__main__":
    app.run()

今回はAgentCore CLIを使ってプロジェクトを作成しました。S3 Filesを使用する都合上、VPCモードとContainerビルドを指定します。

npm install -g @aws/agentcore

agentcore create \
  --name S3FilesTestAgent \
  --defaults \
  --build Container \
  --network-mode VPC \
  --subnets subnet-xxxxx,subnet-yyyyy \
  --security-groups sg-zzzzz

agentcore.jsonにファイルシステムの情報を環境変数として追加します。

{
  "envVars": [
    {"name": "NFS_FS_ID", "value": "fs-xxxxxxxxx"},
    {"name": "NFS_MOUNT_IP", "value": "xx.xx.xx.xx"},
    {"name": "NFS_MOUNT_POINT", "value": "/mnt/s3files"}
  ]
}

agentcore deploy -y

デプロイが完了したら、エージェントにファイルを作ってと指示すれば、任意のファイルを作成してくれます。そしてそのファイルは約1分後にS3バケット内へ反映されます。

image.png

image.png

image.png

コマンド実行による動作確認とその結果

ひとまずこれでAgentCore Runtime上のエージェントがS3 Filesを操作できるようになりました。

ただ、このままだとどうしても内部がどうなっているのかわからなかったので、Claude Codeと色々調査してみました。興味がある方は見ていってください。

内部の調査ではagentcore invoke --execを使います。これでコンテナ内のコマンドを実行して動作確認できます。

このコマンドは内部的にAWS APIのInvokeAgentRuntimeCommandを呼び出しており、AgentCore CLIのソースコードで確認できます。

1. 環境確認

--execはコンテナのrootで実行されます。エージェントアプリ自体はbedrock_agentcore（UID 1000）で動作します。

$ agentcore invoke --exec 'bash -c "uname -a && whoami"'
Linux localhost 6.1.158-15.288.amzn2023.aarch64 ... aarch64 GNU/Linux
root

rpm -ql amazon-efs-utilsでパッケージに含まれるファイル一覧を確認できます。

$ agentcore invoke --exec 'bash -c "rpm -ql amazon-efs-utils"'
/etc/amazon/efs/efs-utils.conf        # EFS設定ファイル
/etc/amazon/efs/efs-utils.crt         # TLS証明書
/etc/amazon/efs/s3files-utils.conf    # S3 Files設定ファイル
/usr/bin/amazon-efs-mount-watchdog    # マウントの死活監視デーモン
/usr/sbin/efs-proxy                   # TLS+IAM認証を処理するローカルプロキシ
/usr/sbin/efs_utils_common/           # 共通Pythonライブラリ（14モジュール）
/usr/sbin/mount.efs                   # EFS用マウントヘルパー
/usr/sbin/mount.s3files               # S3 Files用マウントヘルパー
/usr/sbin/mount_efs/                  # EFS固有ロジック
/usr/sbin/mount_s3files/              # S3 Files固有ロジック
...（他にライセンスファイル、manページ等）

マルチステージビルドでコピーしたefs-utilsが動作しています。v3.0.0になっていることもここで確認できます。

$ agentcore invoke --exec 'bash -c "mount.s3files --version"'
/usr/sbin/mount.s3files Version: 3.0.0

2. TCP接続テスト

AgentCoreのENIからマウントターゲットのENIへTCPポート2049で接続できました。

$ agentcore invoke --exec 'bash -c "nc -zv -w5 <マウントターゲットのIP> 2049 2>&1"'
Connection to <マウントターゲットのIP> 2049 port [tcp/nfs] succeeded!

3. mount -t nfs4では接続できない

mount -t nfs4はLinux標準のNFS v4クライアントでマウントするコマンドです。EFSではこれで接続できますが、S3 Filesではどうでしょうか。

$ agentcore invoke --exec 'bash -c "mount -t nfs4 -o nfsvers=4.1 <マウントターゲットのIP>:/ /mnt/s3files 2>&1"'
mount.nfs4: access denied by server while mounting <マウントターゲットのIP>:/

これは公式ドキュメントに「S3 Files always mounts a file system using TLS encryption and IAM authentication and these cannot be disabled.」とあるとおり、S3 FilesではIAM認証とTLS暗号化が必須であり無効にできないためです。

標準のNFSクライアントは認証情報を送信する仕組みを持たないので拒否されます。
S3 Filesのマウントにはamazon-efs-utilsの提供するmount -t s3filesが必要です。このヘルパーがIAM認証とTLS暗号化を自動的に処理してくれます。

4. mount -t s3filesなら成功

続いて、現在マウントされているファイルシステムを確認します。

# mount: 現在マウントされているファイルシステムの一覧を表示するコマンド
# grep s3files: その中からs3filesを含む行だけ抽出

$ agentcore invoke --exec 'bash -c "mount | grep s3files"'
127.0.0.1:/ on /mnt/s3files type nfs4 (rw,relatime,vers=4.2,rsize=1048576,
wsize=1048576,namlen=255,hard,noresvport,proto=tcp,port=20557,timeo=600,
retrans=2,sec=sys,clientaddr=127.0.0.1,local_lock=none,addr=127.0.0.1)

entrypoint.shでの自動マウントが成功していました。
接続先が127.0.0.1（localhost）になっているのは、efs-proxyがコンテナ内でNFSプロキシとして動作し、アプリからのNFS通信を受けてTLS+IAM認証付きでマウントターゲットに転送しているためです。

括弧内のオプション（nfsvers=4.2, rsize/wsize=1048576, hard, timeo=600, retrans=2, noresvport）はmount.s3filesが自動的に設定する最適値で、以下ドキュメントの「How does the mount helper work?」セクションに各値の説明が記載されています。

また、ファイルシステムのディスク使用量も確認してみると、容量8.0EBとなっていました。S3の実質無制限ストレージが反映されているようです。

$ agentcore invoke --exec 'bash -c "df -h /mnt/s3files"'
Filesystem      Size  Used Avail Use% Mounted on
127.0.0.1:/     8.0E     0  8.0E   0% /mnt/s3files

5. 双方向の同期タイミング

S3 FilesとS3バケットの同期は双方向で行われます。それぞれの方向でどのくらいの時間がかかるか計測しました。

S3バケット → S3 Filesへの反映

S3バケットにファイルを直接アップロードし、S3 Files側に見えるまでの時間を計測します。

# S3バケットにファイルを直接アップロード
$ aws s3 cp - s3://s3files-agentcore-test-bucket/sync-test.txt <<< "S3バケット側から追加"
# アップロード完了: 01:43:14

# 9秒後にS3 Files側で確認 → 既に反映されている
$ agentcore invoke --exec 'bash -c "cat /mnt/s3files/sync-test.txt"'
S3バケット側から追加

アップロードから9秒後の初回チェック時点で既にNFS側から読み取れました。公式ドキュメントの「Synchronization typically occurs within seconds」と一致します。

S3 Filesでの削除 → S3バケットへの反映

S3 Files側でファイルを削除し、S3バケットから消えるまでの時間を計測します。

# S3 Files側でファイルを削除
$ agentcore invoke --exec 'bash -c "rm -f /mnt/s3files/sync-test.txt"'
# 削除実行: 01:46:30

# 10秒後〜50秒後: まだS3バケットに残っている
$ aws s3 ls s3://s3files-agentcore-test-bucket/sync-test.txt
2026-04-13 01:43:15         77 sync-test.txt

# 66秒後: S3バケットからも消えた
$ aws s3 ls s3://s3files-agentcore-test-bucket/sync-test.txt
（出力なし = 削除が反映された）

削除がS3バケット側に反映されるまで約60秒かかりました。これはアップロード時も大体同じようです。

同期タイミングまとめ

方向 反映時間（実測） 公式ドキュメントの記載
S3バケット → NFS 数秒（9秒以内） "Synchronization typically occurs within seconds"
NFS → S3バケット（書き込み） 約1分 "it may take ~1 minute to synchronize changes back to your S3 bucket"
NFS → S3バケット（削除） 約1分（66秒） 同上

S3→NFSは数秒と高速ですが、NFS→S3は書き込み・削除ともに約1分かかります。リアルタイム性が求められる場合はこの遅延を考慮する必要があります。

ファイルの読み書きが正常に動作し、約1分後にS3バケットにも同期されました。これは公式ドキュメントの「it may take ~1 minute to synchronize changes back to your S3 bucket」と一致します。削除操作もS3側に反映されます。

6. 非rootユーザーでのファイル操作

これは普通にエージェントにプロンプトを投げるだけです。先程確認済みです。

# エージェントプロンプトでユーザー確認
$ agentcore invoke "whoamiとidを実行して" --stream
→ bedrock_agentcore, uid=1000

# workspaceへの書き込み
$ agentcore invoke "/mnt/s3files/workspace/にgreeting.txtを作成して" --stream
→ ファイル作成成功

# S3にも同期される
$ aws s3 cp s3://s3files-agentcore-test-bucket/workspace/greeting.txt -
こんにちは、非rootユーザーから書き込みました

まとめ

AgentCore RuntimeからS3 Filesへの接続は可能でした！

S3 Filesの標準セットアップ手順(バケット作成、ファイルシステム作成、マウントターゲット作成、セキュリティグループ設定)はEC2と同じです。

AgentCore固有の追加対応として必要だったのは以下の3つです。

  1. amazon-efs-utils v3.0.0のDockerマルチステージビルドによるインストール（Debian系にaptパッケージがないため）
  2. rootマウント→非rootアプリ実行の3段階entrypoint（コンテナ内でのmount権限の確保とセキュリティの両立）
  3. NAT GatewayまたはVPC Endpointはコンテナ起動時に必要

S3 FilesとAgentCoreはどちらも新しいサービスですが、この記事の構成で問題なくファイル読み書き・S3同期・AIエージェントからの自然言語操作まで動作しています。
今後のネイティブ統合にも期待しつつ、現時点でも実用可能なことが確認できました。

検証に使ったコードは以下のリポジトリにまとめてあります。

10
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
10
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?