Amazon Bedrock AgentCore に OpenAI GPT-4.1 mini を使ったエージェントをデプロイしてみた
こんにちは、森田です。
AWS Summit New York City で Amazon Bedrock AgentCore が発表されましたね。
Bedrock AgentCore とは
エージェントをセキュアかつスケーラブルに利用するために必要な機能・インフラを提供するサービスです。
特に興味深い点として、モデルは Amazon Bedrock に限らず、あらゆるモデルを利用できるようになっています。
本記事では、OpenAI gpt-4.1-mini を Bedrock AgentCore で動かしてみたいと思います。
やってみた
上記のような構成を作成していきます。
参考情報
上記のリポジトリを参考に進めていきます。
環境構築
uvを使って仮想環境を構築します。
また、リポジトリ内にあるrequirements.txtのライブラリをインストールします。
エージェントをローカルで実行
まずは、以下コードをローカルで実行し、正常に動作するかを確認します、
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 用に変更します。
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 のような他の主要なフレームワークにも対応しており、同様の方法で統合できます。
より詳細な情報については、公式ドキュメントをご参照ください。
BedrockAgentCoreAppでは、次の処理が自動的に行われます。
- ポート8080をリッスンするHTTPサーバーを作成
/invocations
リクエストを処理するために必要なエンドポイントを実装/ping
ヘルスチェックのエンドポイントを実装- AWS 標準に従ったエラー処理の管理
そのため、BedrockAgentCoreAppを使えば、AgentCore ランタイムの動作要件を満たしたAIエージェントサーバを起動することができるわけです。
また、今回のサンプルでは、BedrockAgentCoreApp を利用しましたが、動作要件を満たすことができれば、FastAPIなども利用可能なようです。
Bedrock AgentCore のデプロイ
まずは、必要となるロールの作成を行います。
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
を利用してデプロイを行います。
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が生成されます。
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リポジトリが作成されるようになります。
最後に以下スクリプトを実行してデプロイを完了されます。
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)
モデル呼び出し
それでは、デプロイしたモデルを呼び出してみます。
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は、現在プレビューでの利用となるため、デプロイ方法など今後変更となる可能性もあるため、くれぐれもご注意ください。