Amazon Bedrock AgentCore に OpenAI GPT-4.1 mini を使ったエージェントをデプロイしてみた

Amazon Bedrock AgentCore に OpenAI GPT-4.1 mini を使ったエージェントをデプロイしてみた

Clock Icon2025.07.17

こんにちは、森田です。

AWS Summit New York City で Amazon Bedrock AgentCore が発表されましたね。

https://dev.classmethod.jp/articles/bedrock-agent-core/

Bedrock AgentCore とは

エージェントをセキュアかつスケーラブルに利用するために必要な機能・インフラを提供するサービスです。

特に興味深い点として、モデルは Amazon Bedrock に限らず、あらゆるモデルを利用できるようになっています。

本記事では、OpenAI gpt-4.1-mini を Bedrock AgentCore で動かしてみたいと思います。

やってみた

Untitled (3).png

上記のような構成を作成していきます。

参考情報

https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/01-hosting-agent

上記のリポジトリを参考に進めていきます。

環境構築

uvを使って仮想環境を構築します。

また、リポジトリ内にあるrequirements.txtのライブラリをインストールします。

https://github.com/awslabs/amazon-bedrock-agentcore-samples/blob/main/01-tutorials/01-AgentCore-runtime/01-hosting-agent/03-strands-with-openai-model/requirements.txt

エージェントをローカルで実行

まずは、以下コードをローカルで実行し、正常に動作するかを確認します、

strands_agents_openai.py
from strands import Agent, tool
from strands_tools import calculator # Import the calculator tool
import argparse
import json
from strands.models.litellm import LiteLLMModel
import os


os.environ["OPENAI_API_KEY"] = "your-api-key"


# Create a custom tool 
@tool
def weather():
    """ Get weather """ # Dummy implementation
    return "sunny"

model = "gpt-4.1-mini"
litellm_model = LiteLLMModel(
    model_id=model, params={"max_tokens": 32000, "temperature": 0.7}
)


agent = Agent(
    model=litellm_model,
    tools=[calculator, weather],
    system_prompt="You're a helpful assistant. You can do simple math calculation, and tell the weather."
)

def strands_agent_open_ai(payload):
    """
    Invoke the agent with a payload
    """
    user_input = payload.get("prompt")
    response = agent(user_input)
    return response.message['content'][0]['text']

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("payload", type=str)
    args = parser.parse_args()
    response = strands_agent_open_ai(json.loads(args.payload))
    print(response)

コード作成後、実行すると以下のように weather tool を使っていることが確認できます。

実行結果
$ python strands_agents_openai.py '{"prompt": "What is the weather now?"}'
Tool #1: weather
The weather now is sunny. Is there anything else you would like to know or do?The weather now is sunny. Is there anything else you would like to know or do?

エージェントを Bedrock AgentCore 用に書き換える

先ほど作成した処理を Bedrock AgentCore 用に変更します。

strands_agents_openai.py
from strands import Agent, tool
from strands_tools import calculator # Import the calculator tool
import argparse
import json
from strands.models.litellm import LiteLLMModel
import os
> from bedrock_agentcore.runtime import BedrockAgentCoreApp

>app = BedrockAgentCoreApp()

os.environ["OPENAI_API_KEY"] = "your-api-key"

# Create a custom tool 
@tool
def weather():
    """ Get weather """ # Dummy implementation
    return "sunny"

model = "gpt-4.1-mini"
litellm_model = LiteLLMModel(
    model_id=model, params={"max_tokens": 32000, "temperature": 0.7}
)


agent = Agent(
    model=litellm_model,
    tools=[calculator, weather],
    system_prompt="You're a helpful assistant. You can do simple math calculation, and tell the weather."
)


> @app.entrypoint
> def strands_agent_open_ai(payload):
>    """
>    Invoke the agent with a payload
>    """
>    user_input = payload.get("prompt")
>    response = agent(user_input)
>    return response.message['content'][0]['text']

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

特筆すべき点としては、BedrockAgentCoreApp のインスタンスを作成し、@app.entrypoint デコレータを使ってエージェントを呼び出す関数を登録している点です。これにより、Bedrock AgentCore ランタイムでエージェントの処理を呼び出せるようになります。

このサンプルでは Strands Agents を利用していますが、Bedrock AgentCore は LangGraph のような他の主要なフレームワークにも対応しており、同様の方法で統合できます。

より詳細な情報については、公式ドキュメントをご参照ください。

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-how-it-works.html

BedrockAgentCoreAppでは、次の処理が自動的に行われます。

  • ポート8080をリッスンするHTTPサーバーを作成
  • /invocationsリクエストを処理するために必要なエンドポイントを実装
  • /pingヘルスチェックのエンドポイントを実装
  • AWS 標準に従ったエラー処理の管理

そのため、BedrockAgentCoreAppを使えば、AgentCore ランタイムの動作要件を満たしたAIエージェントサーバを起動することができるわけです。

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-service-contract.html

また、今回のサンプルでは、BedrockAgentCoreApp を利用しましたが、動作要件を満たすことができれば、FastAPIなども利用可能なようです。

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/getting-started-custom.html

Bedrock AgentCore のデプロイ

まずは、必要となるロールの作成を行います。

create_role.py
import sys
import os
import boto3
import json
import time
from boto3.session import Session

# Get the current notebook's directory
current_dir = os.path.dirname(os.path.abspath('__file__' if '__file__' in globals() else '.'))

# Navigate up to the utils.py location
utils_dir = os.path.join(current_dir, '..')
utils_dir = os.path.abspath(utils_dir)

# Add to sys.path
sys.path.insert(0, utils_dir)


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

    # 修正された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}-*"
                ]
            }
        ]
    }

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

    assume_role_policy_document_json = json.dumps(
        assume_role_policy_document
    )
    role_policy_document = json.dumps(role_policy)

    try:
        agentcore_iam_role = iam_client.create_role(
            RoleName=agentcore_role_name,
            AssumeRolePolicyDocument=assume_role_policy_document_json
        )
        time.sleep(10)
    except iam_client.exceptions.EntityAlreadyExistsException:
        print(f"Role '{agentcore_role_name}' already exists. Deleting and recreating.")
        try:
            policies = iam_client.list_role_policies(RoleName=agentcore_role_name)
            for policy_name in policies['PolicyNames']:
                iam_client.delete_role_policy(RoleName=agentcore_role_name, PolicyName=policy_name)
            
            iam_client.delete_role(RoleName=agentcore_role_name)
            time.sleep(10) # 削除が反映されるのを待つ
            
            agentcore_iam_role = iam_client.create_role(
                RoleName=agentcore_role_name,
                AssumeRolePolicyDocument=assume_role_policy_document_json
            )
            time.sleep(10) # 作成が反映されるのを待つ
        except Exception as e:
            print(f"Error while recreating role: {e}")
            return None


    print(f"Attaching policy to role '{agentcore_role_name}'")
    try:
        iam_client.put_role_policy(
            PolicyDocument=role_policy_document,
            PolicyName="AgentCorePolicy",
            RoleName=agentcore_role_name
        )
        print("Policy attached successfully.")
    except Exception as e:
        print(f"Error attaching policy: {e}")
        return None

    return agentcore_iam_role


agent_name="strands_agents_openai"
agentcore_iam_role = create_agentcore_role(agent_name=agent_name)

このスクリプトを実行することでagentcore-strands_agents_openai-roleが作成されます。

続いて、AgentCore Runtime starter toolkitを利用してデプロイを行います。

create_config.py
from bedrock_agentcore_starter_toolkit import Runtime
from boto3.session import Session
import boto3
import json
boto_session = Session()
region = boto_session.region_name

agentcore_runtime = Runtime()

response = agentcore_runtime.configure(
    entrypoint="strands_agents_openai.py",
    execution_role="作成したroleのARN",
    auto_create_ecr=True,
    requirements_file="requirements.txt",
    region=region
)

execution_roleにはcreate_role.pyで作成したロールのARNを指定します。

このスクリプトを実行することでデプロイするAgentCoreが生成されます。

.bedrock_agentcore.yaml
default_agent: strands_agents_openai
agents:
  strands_agents_openai:
    name: strands_agents_openai
    entrypoint: /***/strands_agents_openai.py
    platform: linux/arm64
    container_runtime: podman
    aws:
      execution_role: arn:aws:iam::**:role/agentcore-strands_agents_openai-role
      account: '**'
>      region: us-east-1
      ecr_repository: **.dkr.ecr.us-east-1.amazonaws.com/bedrock_agentcore-strands_agents_openai
>      ecr_auto_create: true
      network_configuration:
        network_mode: PUBLIC
      protocol_configuration:
        server_protocol: HTTP
      observability:
        enabled: true
    bedrock_agentcore:
      agent_id: strands_agents_openai-zMZascABQD
      agent_arn: arn:aws:bedrock-agentcore:us-east-1:**:runtime/strands_agents_openai-zMZascABQD
      agent_session_id: 13296795-4ec5-415e-a7c9-92c06463ff2f
    authorizer_configuration: null
    oauth_configuration: null

特に、意図したリージョンになっているかは確認しましょう。
また、ecr_auto_createはtrueとしておくことで、自動でECRリポジトリが作成されるようになります。

最後に以下スクリプトを実行してデプロイを完了されます。

deploy.py
from bedrock_agentcore_starter_toolkit import Runtime
import time

agentcore_runtime = Runtime()

launch_result = agentcore_runtime.launch()
status_response = agentcore_runtime.status()
status = status_response.endpoint['status']
end_status = ['READY', 'CREATE_FAILED', 'DELETE_FAILED', 'UPDATE_FAILED']
while status not in end_status:
    time.sleep(10)
    status_response = agentcore_runtime.status()
    status = status_response.endpoint['status']
    print(status)

モデル呼び出し

それでは、デプロイしたモデルを呼び出してみます。

invoke.py
import os, json
import boto3
import logging
logger = logging.getLogger(__name__)

agent_arn = "作成したAgent RuntimeのARN"

agentcore_client = boto3.client(
    'bedrock-agentcore',
    region_name="us-east-1"  # Specify the region where your agent is hosted
)

boto3_response = agentcore_client.invoke_agent_runtime(
    agentRuntimeArn=agent_arn,
    qualifier="DEFAULT",
    payload=json.dumps({"prompt": "How much is 2X2?"})
)
if "text/event-stream" in boto3_response.get("contentType", ""):
    content = []
    for line in boto3_response["response"].iter_lines(chunk_size=1):
        if line:
            line = line.decode("utf-8")
            if line.startswith("data: "):
                line = line[6:]
                logger.info(line)
                content.append(line)
    print("\n".join(content))
else:
    try:
        events = []
        for event in boto3_response.get("response", []):
            events.append(event)
    except Exception as e:
        events = [f"Error reading EventStream: {e}"]
    print(json.loads(events[0].decode("utf-8")))

以下のように期待した結果が出力されました。

$ python invoke.py
2 times 2 is 4.

さいごに

触ってみた感想としては、お手軽とはならないものの、モデルやソースコードを柔軟に変更できる点は非常に魅力的だと感じました。

また、Bedrock AgentCoreには、Observability、Identityなどの機能もあるため、また別の機会に触ってみたいと思います。

なお、Bedrock AgentCoreは、現在プレビューでの利用となるため、デプロイ方法など今後変更となる可能性もあるため、くれぐれもご注意ください。

Share this article

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.