はじめに
前回、Strands AgentsとStreamlitを使って、複数のAWSアカウントを横断調査できるチャットボットをローカル環境で構築しました。
本記事では、このチャットボットをAmazon Bedrock AgentCore Runtime上で動かしてみます。
これからAgentCoreに入門してみようという方の参考になれば幸いです。
Bedrock AgentCore Runtime とは
Amazon Bedrock AgentCore Runtime は、AI エージェントまたはツールをデプロイして実行するための、安全でサーバーレスな専用のホスティング環境を提供します。
AgentCore Runtimeは、リクエストが来たときだけコンテナを起動して、処理が終わったら止まります。課金も処理中のみです。
また、AgentCoreの嬉しいポイントの1つが、セッション管理だと思います。
AgentCore Runtimeでは、セッションIDごとに専用のmicroVMが割り当てられ、同じセッションIDのリクエストは、実行環境が存続している間は同じmicroVMにルーティングされます。そのため、プロセス内の状態などをリクエスト間で引き継ぐことができ、マルチターンのエージェントを実装しやすくなります。（長期的に会話履歴やユーザー情報を保持したい場合は、AgentCore Memoryなどを組み合わせて利用します）
実装
動作環境
Windows 11 の Git Bash (MINGW64) で実行しています。利用しているAWSリージョンはap-northeast-1です。
| 項目 | バージョン |
|---|---|
| Node.js | 22.14.0 |
| Python | 3.12 |
| aws-cli | 2.33.13 |
| AWS CDK | 2.1101.0 |
| @aws/agentcore | 0.28.1 |
| bedrock-agentcore | 1.22.0 |
| strands-agents | 1.53.0 |
AgentCore CLI のインストール
AgentCore には CLI が2つあります。bedrock-agentcore-starter-toolkit は現在はサポート対象外です。
https://github.com/aws/bedrock-agentcore-starter-toolkit
npm install -g @aws/agentcore
プロジェクトの作成
次のコマンドを実行して、プロジェクトを作成します。
agentcore create \
--name awsChatbot \
--framework Strands \
--model-provider Bedrock \
--memory none \
--build CodeZip
| オプション | 意味 |
|---|---|
--name |
エージェント名 |
--framework |
エージェントのフレームワーク |
--model-provider |
モデルの提供元 |
--memory none |
AgentCore Memory を使わない |
--build CodeZip |
コードをzipにしてS3経由でデプロイ |
上記のコマンドを実行すると、次のようなファイル構成が作成されます。
awsChatbot/
├── agentcore/ # 設定とCDKプロジェクト
│ ├── agentcore.json
│ ├── aws-targets.json
│ └── cdk/
└── app/awsChatbot/
├── main.py # エージェントの入口
└── pyproject.toml # Pythonの依存
前回ローカルで動かす際に作成したコードを、中身は変えずに次のように配置します。
app/awsChatbot/
├── main.py # 後ほど編集
├── agent.py # 変更なし
├── tools/aws_read.py # 変更なし
├── config/accounts.yaml # 変更なし
└── pyproject.toml
pyproject.toml に、前編のツールが使うライブラリを足します。
dependencies = [
"aws-opentelemetry-distro",
"bedrock-agentcore >= 1.9.1",
"botocore[crt] >= 1.35.0",
"strands-agents >= 1.15.0",
"boto3 >= 1.43.78",
"jmespath >= 1.1.0",
"pyyaml >= 6.0.3",
]
AgentCore用の入口をつくる（main.py）
AgentCore Runtimeは、コンテナに対してHTTPリクエストを投げます。
エージェントは、以下の仕様を満たすコンテナ化されたアプリケーションとしてデプロイする必要があります。
ホスト: 0.0.0.0
ポート : 8080 - HTTP ベースのエージェント通信用の標準ポート
agent.py はエージェントを組み立てるだけなので、このHTTPリクエストを受け取る入口として main.py が必要になります。
雛形のmain.pyを、前編のagent.pyにあるcreate_agent()を呼ぶように書き換えます。
"""AgentCore Runtime のエントリポイント。"""
from collections import OrderedDict
from typing import Any
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from agent import create_agent
app = BedrockAgentCoreApp()
log = app.logger
MAX_CACHED_SESSIONS = 128
_agents: "OrderedDict[str, Any]" = OrderedDict()
def get_or_create_agent(session_id: str) -> Any:
"""セッションIDごとにAgentを使い回して、会話履歴を引き継ぐ。"""
if session_id in _agents:
_agents.move_to_end(session_id)
return _agents[session_id]
if len(_agents) >= MAX_CACHED_SESSIONS:
_agents.popitem(last=False)
_agents[session_id] = create_agent()
return _agents[session_id]
@app.entrypoint
async def invoke(payload: dict[str, Any], context: Any):
prompt = payload.get("prompt")
if not isinstance(prompt, str) or not prompt.strip():
raise ValueError("prompt は文字列で指定してください。")
session_id = getattr(context, "session_id", None) or "default-session"
agent = get_or_create_agent(session_id)
async for event in agent.stream_async(prompt.strip()):
if not isinstance(event, dict) or "event" not in event:
continue
cbs = event["event"].get("contentBlockStart")
if cbs is not None and not cbs.get("start"):
continue
yield event
if __name__ == "__main__":
app.run()
やっていることは主に3つです。
-
BedrockAgentCoreApp()を作って、入口にしたい関数に@app.entrypointを付ける - その中で
create_agent()を呼ぶ -
app.run()で待ち受ける
ローカルで動かしてみる
デプロイする前に、手元で動かして確認できます。
agentcore dev
ブラウザで agent inspector が開き、チャット画面で試せます。
デプロイ
agentcore deploy
CDK が CloudFormation スタックを作り、AgentCore Runtime とIAMロールを一式そろえてくれます。--dry-run を付けると、何も変更せずに確認だけできます。
実行後、無事Runtimeが作成されていることをコンソール上で確認できました。(なぜかawsChatbotが2つ連なっているのですが、原因が分からず...)
IAMの設定
デプロイすると実行ロールが作られます。このロールが各アカウントの調査用ロールをAssumeRoleできるようにしておく必要があります。
また、AgentCore の実行ロールは最小権限で作られます。Bedrock呼び出しやCloudWatchへの書き込みは入っていますが、sts:AssumeRole は入っていないので、自分で足す必要があります。
① 信頼ポリシーに実行ロールを足す
調査される側のアカウントに置いたロールの信頼ポリシーに、実行ロールを許可する条件を足します。
{
"Effect": "Allow",
"Principal": { "AWS": "arn:aws:iam::123456789012:root" },
"Action": "sts:AssumeRole",
"Condition": {
"ArnLike": {
"aws:PrincipalArn": "arn:aws:iam::123456789012:role/<実行ロール名>"
}
}
}
② 実行ロールに sts:AssumeRole を足す
実行ロール側には、どのロールを引き受けてよいかを書きます。
{
"Sid": "AssumeReadOnlyRolesInTargetAccounts",
"Effect": "Allow",
"Action": "sts:AssumeRole",
"Resource": [
"arn:aws:iam::111111111111:role/aws-chatbot-readonly",
"arn:aws:iam::222222222222:role/aws-chatbot-readonly"
]
}
動かしてみる
次のようなコマンドで動作確認できます。
無事動いてくれました。
agentcore invoke "kabu-appアカウントのEC2は何台?"
おわりに
前回ローカルで動いていたAIエージェントがAWS上で動くようになりました。
本当はWebアプリ化までしたかったですが、時間なかったのでまた今度。。。
RuntimeはAgentCoreの基本となるものですが、AgentCoreにはまだまだたくさんの機能があります。GatewayやMemory、Policyなども触ってみて記事化していきます。（今年のre:InventまでにAgentCoreマスターになる！！）