Bedrock AgentCore Runtime に求められる要件を確認してみた

#Amazon Bedrock

#Amazon Bedrock Agents

森田力

2025.07.19

こんにちは、森田です。

先日以下の記事を執筆しましたが、AgentCore Runtime の構築に starter toolkitを利用していました。

この starter toolkit は AgentCore Runtime に求められる要件を気にせず、簡単に構築することができます。

一方で、要件を満たすことができれば、一般的なWebフレームワークでも AgentCore Runtime を構築することができます。

本記事では、AgentCore Runtime に求められる要件を確認し、FastAPIを使って AgentCore Runtime を構築していきます。

AgentCore Runtime に求められる要件

コンテナイメージ

コンテナイメージについては、AgentCore Runtimeは、コンテナとして動作させるため、イメージが必要となります。

また、そのイメージは、ECR上にデプロイする必要があります。

プラットフォーム

ARM64アーキテクチャ向けにエージェントをコンテナ化が必要がです。

そのため、イメージビルドする場合は、--platform=linux/arm64を指定しましょう。

ポート

アプリケーションはポート 8080 で実行させる必要があります。

アプリケーション

以下のエンドポイントを作成する必要があります。

invocationsエンドポイント(POST)
- エージェント呼び出し時に実行されます。
pingエンドポイント(GET)
- ヘルスチェック時に実行されます。

それでは、これらの要件を満たすように FastAPI を使って AgentCore Runtime を構築していきます。

環境構築

uvを使って環境構築を行います。

必要となるライブラリもインストールします。

setup

uv init --python 3.11
uv add "uvicorn[standard]" pydantic httpx strands-agents fastapi

エージェントの構築には、Strands Agents を使います。

https://strandsagents.com/latest/

エージェントサーバの構築

エージェントサーバでは、以下のエンドポイントを作成します。

invocationsエンドポイント
pingエンドポイント

invocationsエンドポイント

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(request: InvocationRequest):
    try:
        user_message = request.input.get("prompt", "")
        if not user_message:
            raise HTTPException(
                status_code=400, 
                detail="No prompt found in input. Please provide a 'prompt' key in the input."
            )

        result = strands_agent(user_message)
        response = {
            "message": result.message,
            "timestamp": datetime.utcnow().isoformat(),
            "model": "strands-agent",
        }

        return InvocationResponse(output=response)

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent processing failed: {str(e)}")

pingエンドポイント

@app.get("/ping")
async def ping():
    return {"status": "healthy"}

Dockerfileの作成

コンテナとして動作させるためDockerfileの作成を行います。

Dockerfile

# Use uv's ARM64 Python base image
> FROM --platform=linux/arm64 ghcr.io/astral-sh/uv:python3.11-bookworm-slim

WORKDIR /app

# Copy uv files
COPY pyproject.toml uv.lock ./

# Install dependencies (including strands-agents)
RUN uv sync --frozen --no-cache

# Copy agent file
COPY agent.py ./

# Expose port
EXPOSE 8080

# Run application
CMD ["uv", "run", "uvicorn", "agent:app", "--host", "0.0.0.0", "--port", "8080"]

デプロイ環境のプラットフォームについては、linux/arm64となるため、上記のように--platform=linux/arm64を指定します。

ローカルでの動作確認

ローカル環境でコンテナ上で動作確認を行います。

まずは、ビルドを行います。

ビルド

podman build --platform linux/arm64 -t my-agent:arm64 .

続いてビルドしたイメージを使ってコンテナを起動します。

なお、エージェントでは、Bedrockを実行しているため認証情報を渡すようにします。

コンテナ起動

podman run --platform linux/arm64 -p 8080:8080 \
  -e AWS_ACCESS_KEY_ID \
  -e AWS_SECRET_ACCESS_KEY \
  -e AWS_SESSION_TOKEN \
  my-agent:arm64

curlコマンドを実行して動作確認をしてみます。

ping

% curl http://localhost:8080/ping                 

{"status":"healthy"}%

invoke

% curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "input": {"prompt": "こんにちは"}
  }'
{"output":{"message":{"role":"assistant","content":[{"text":"こんにちは！お元気ですか？何かお手伝いできることがあれば、お気軽にお声かけください。"}]},"timestamp":"2025-07-19T14:53:47.476164","model":"strands-agent"}}%

ECRリポジトリへのpush

まずは、イメージをpushするためのECRリポジトリの作成を行います。

aws ecr create-repository --repository-name bedrock-agentcore/fastapi \
--region us-east-1

作成後、以下コマンドを実行してpushまで完了させます。

push.sh

# アカウントIDを取得
account_id=$(aws sts get-caller-identity --query Account --output text)

# ECRにログイン
aws ecr get-login-password --region us-east-1 | \
  podman login --username AWS --password-stdin "${account_id}.dkr.ecr.us-east-1.amazonaws.com"

# イメージをビルド
podman build --platform linux/arm64 \
  -t "${account_id}.dkr.ecr.us-east-1.amazonaws.com/bedrock-agentcore/fastapi:latest" \
  .

# イメージをECRにプッシュ
podman push "${account_id}.dkr.ecr.us-east-1.amazonaws.com/bedrock-agentcore/fastapi:latest"

AgentCore Runtimeへのデプロイ

以下スクリプトを実行して、デプロイを行います。

deploy.py

import boto3
import json
import time

def create_agentcore_role(agent_name, region='us-east-1'):
    iam_client = boto3.client('iam', region_name=region)
    sts_client = boto3.client('sts', region_name=region)
    account_id = sts_client.get_caller_identity()["Account"]
    role_name = f'agentcore-{agent_name}-role'

    # IAMポリシー
    role_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "BedrockPermissions",
                "Effect": "Allow",
                "Action": [
                    "bedrock:InvokeModel",
                    "bedrock:InvokeModelWithResponseStream"
                ],
                "Resource": ["*"]
            },
            {
                "Sid": "ECRImageAccess",
                "Effect": "Allow",
                "Action": [
                    "ecr:BatchGetImage",
                    "ecr:GetDownloadUrlForLayer"
                ],
                "Resource": [
                    f"arn:aws:ecr:{region}:{account_id}:repository/*"
                ]
            },
            {
                "Sid": "LogsCreateAndDescribe",
                "Effect": "Allow",
                "Action": [
                    "logs:DescribeLogStreams",
                    "logs:CreateLogGroup"
                ],
                "Resource": [
                    f"arn:aws:logs:{region}:{account_id}:log-group:/aws/bedrock-agentcore/runtimes/*"
                ]
            },
            {
                "Sid": "LogsDescribeGroups",
                "Effect": "Allow",
                "Action": [
                    "logs:DescribeLogGroups"
                ],
                "Resource": [
                    f"arn:aws:logs:{region}:{account_id}:log-group:*"
                ]
            },
            {
                "Sid": "LogsPutEvents",
                "Effect": "Allow",
                "Action": [
                    "logs:CreateLogStream",
                    "logs:PutLogEvents"
                ],
                "Resource": [
                    f"arn:aws:logs:{region}:{account_id}:log-group:/aws/bedrock-agentcore/runtimes/*:log-stream:*"
                ]
            },
            {
                "Sid": "ECRTokenAccess",
                "Effect": "Allow",
                "Action": [
                    "ecr:GetAuthorizationToken"
                ],
                "Resource": ["*"]
            },
            {
                "Sid": "XRayAccess",
                "Effect": "Allow",
                "Action": [
                    "xray:PutTraceSegments",
                    "xray:PutTelemetryRecords",
                    "xray:GetSamplingRules",
                    "xray:GetSamplingTargets"
                ],
                "Resource": ["*"]
            },
            {
                "Sid": "CloudWatchMetrics",
                "Effect": "Allow",
                "Action": ["cloudwatch:PutMetricData"],
                "Resource": ["*"]
            },
            {
                "Sid": "GetAgentAccessToken",
                "Effect": "Allow",
                "Action": [
                    "bedrock-agentcore:GetWorkloadAccessToken",
                    "bedrock-agentcore:GetWorkloadAccessTokenForJWT",
                    "bedrock-agentcore:GetWorkloadAccessTokenForUserId"
                ],
                "Resource": [
                    f"arn:aws:bedrock-agentcore:{region}:{account_id}:workload-identity-directory/default",
                    f"arn:aws:bedrock-agentcore:{region}:{account_id}:workload-identity-directory/default/workload-identity/{agent_name}-*"
                ]
            }
        ]
    }

    # AssumeRoleポリシー
    assume_role_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AssumeRolePolicy",
                "Effect": "Allow",
                "Principal": {
                    "Service": "bedrock-agentcore.amazonaws.com"
                },
                "Action": "sts:AssumeRole"
            }
        ]
    }

    # ロール作成
    try:
        role = iam_client.create_role(
            RoleName=role_name,
            AssumeRolePolicyDocument=json.dumps(assume_role_policy)
        )
        time.sleep(10)
    except iam_client.exceptions.EntityAlreadyExistsException:
        print(f"Role '{role_name}' already exists. Recreating...")
        # 既存ロール削除
        policies = iam_client.list_role_policies(RoleName=role_name)
        for policy_name in policies['PolicyNames']:
            iam_client.delete_role_policy(RoleName=role_name, PolicyName=policy_name)
        iam_client.delete_role(RoleName=role_name)
        time.sleep(10)
        role = iam_client.create_role(
            RoleName=role_name,
            AssumeRolePolicyDocument=json.dumps(assume_role_policy)
        )
        time.sleep(10)
    except Exception as e:
        print(f"Error creating role: {e}")
        return None

    # インラインポリシー付与
    try:
        iam_client.put_role_policy(
            RoleName=role_name,
            PolicyName="AgentCorePolicy",
            PolicyDocument=json.dumps(role_policy)
        )
        time.sleep(10)
    except Exception as e:
        print(f"Error attaching policy: {e}")
        return None

    print(f"Role '{role_name}' created and policy attached.")
    return role['Role']['Arn']

def create_agent_runtime(agent_runtime_name, container_uri, role_arn, region='us-east-1'):
    client = boto3.client('bedrock-agentcore-control', region_name=region)
    try:
        response = client.create_agent_runtime(
            agentRuntimeName=agent_runtime_name,
            agentRuntimeArtifact={
                'containerConfiguration': {
                    'containerUri': container_uri
                }
            },
            networkConfiguration={"networkMode": "PUBLIC"},
            roleArn=role_arn
        )
        print("Agent Runtime created successfully!")
        print(f"Agent Runtime ARN: {response['agentRuntimeArn']}")
        print(f"Status: {response['status']}")
        return response
    except Exception as e:
        print(f"Error creating agent runtime: {e}")
        return None

if __name__ == "__main__":
    region = 'us-east-1'
    agent_name = "strands_agents_fastapi"
    agent_runtime_name = "strands_agents_fastapi"
    repo_name = 'bedrock-agentcore/fastapi'

    sts = boto3.client('sts', region_name=region)
    account_id = sts.get_caller_identity()['Account']
    container_uri = f'{account_id}.dkr.ecr.{region}.amazonaws.com/{repo_name}:latest'

    # 1. IAMロール作成
    role_arn = create_agentcore_role(agent_name, region=region)
    if not role_arn:
        print("Failed to create IAM role. Exiting.")
        exit(1)

    # 2. Agent Runtime作成
    create_agent_runtime(agent_runtime_name, container_uri, role_arn, region=region)

実行結果

% python deploy.py
Agent Runtime created successfully!
Agent Runtime ARN: arn:aws:bedrock-agentcore:us-east-1:アカウントID:runtime/strands_agents_fastapi-mYgIt42Juw
Status: CREATING

AgentCore Runtime の呼び出し

最後にデプロイした AgentCore Runtime を呼び出します。

invoke.py

import boto3
import json
import uuid

runtime_session_id = str(uuid.uuid4())

agentRuntimeArn = 'arn:aws:bedrock-agentcore:us-east-1:アカウントID:runtime/strands_agents_fastapi-xxx'

agent_core_client = boto3.client('bedrock-agentcore', region_name='us-east-1')
payload = json.dumps({
    "input": {"prompt": "こんにちは"}
})

response = agent_core_client.invoke_agent_runtime(
    agentRuntimeArn=agentRuntimeArn,
    runtimeSessionId=runtime_session_id,
    payload=payload,
    qualifier="DEFAULT"
)

response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)

実行結果

% python invoke.py
Agent Response: {'output': {'message': {'role': 'assistant', 'content': [{'text': 'こんにちは！お元気ですか？何かお手伝いできることがあれば、お気軽にお声かけください。'}]}, 'timestamp': '2025-07-19T15:22:52.718364', 'model': 'strands-agent'}}

要件を満たしていない場合の挙動

正常に動作することが確認できましたので、要件を満たしていない場合の挙動を確認してみます。

invocationsエンドポイント削除時

invocationsエンドポイントをコメントアウトして再度イメージをビルド、AgentCore Runtime の更新を行います。

実行結果

% python invoke.py
Agent Response: {'output': {'message': {'role': 'assistant', 'content': [{'text': 'こんにちは！お元気ですか？何かお手伝いできることがあれば、お気軽にお声かけください。'}]}, 'timestamp': '2025-07-19T15:22:52.718364', 'model': 'strands-agent'}}
botocore.errorfactory.RuntimeClientError: An error occurred (RuntimeClientError) when calling the InvokeAgentRuntime operation: Received error (404) from runtime. Please check your CloudWatch logs for more information.

404のエラーが返却されました。

CloudWatchを確認してみますが、エンドポイント自体が存在しないため、エラーも上がってきていません。

スクリーンショット 2025-07-20 1.07.36.png

invocationsエンドポイント実行エラー時

今度は、以下のように意図的にエラーを発生させるようにします。

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(request: InvocationRequest):
    raise ValueError("Error Test")

この場合は、500エラーが返却され、CloudWatch上にもエラーが記録されます。

botocore.errorfactory.RuntimeClientError: An error occurred (RuntimeClientError) when calling the InvokeAgentRuntime operation: Received error (500) from runtime. Please check your CloudWatch logs for more information.

スクリーンショット 2025-07-20 1.14.39.png

さいごに

既存のフレームワークに慣れている人の場合は、starter toolkitを使わない選択肢もありだと思います。

また、求められる要件としても、一般的なWebフレームワークであれば満たせると思うので、ぜひ慣れているフレームワークで試してみると良いでしょう。

Bedrock AgentCore Runtime に求められる要件を確認してみた

AgentCore Runtime に求められる要件

コンテナイメージ

プラットフォーム

ポート

アプリケーション

環境構築

エージェントサーバの構築

Dockerfileの作成

ローカルでの動作確認

ECRリポジトリへのpush

AgentCore Runtimeへのデプロイ

AgentCore Runtime の呼び出し

要件を満たしていない場合の挙動

invocationsエンドポイント削除時

invocationsエンドポイント実行エラー時

さいごに

関連記事

主なカテゴリ

AWSで探す

注目のテーマ

プロダクトやサービスで探す

特集やシリーズから探す

お問い合わせ

運営会社