はじめに
普段の業務では数十個のAWSアカウント環境の運用を担当しています。
日々の業務の中で、「複数のAWSアカウントを横断して調査し、質問に答えてくれるチャットボットがあったら便利そう」と思い、まずはプライベートのAWS環境で作ってみることにしました。（ついでにAIエージェント構築の入門も兼ねて）
本テーマは前後編に分けて投稿する予定で、前編である本記事では、Strands AgentsとStreamlitを使って、複数のAWSアカウントを横断して調査できるチャットボットをローカル環境で構築します。
後編では、このチャットボットをBedrock AgentCore Runtime上で動作させ、Webアプリとして利用できる形にしてみます。
それDevOps Agentでできるのでは？というツッコミが飛んできそうですが、たしかにそうだと思います。ただ、このチャットボットでやりたいことが増えたときにツールを追加して解決できるのが自作することの強みだと思っています。
作ったもの
画像のように、日本語で質問するだけでAWSアカウントの調査をしてくれるチャットボットです。
自分のAWS環境を読み取り専用で調査することが可能で、リソースの作成・更新・削除は一切できません。
構成
処理の流れはざっくり7ステップで、
① 質問を受け取る
Streamlit → エージェント
② エージェントがLLMに「この質問に答えるには何をすればいいか」を聞く
→ LLMが「aws_read をこの引数で呼べ」と指示を返す
{"service": "s3", "operation": "list_buckets",
"accounts": ["dev-app1"], "query": "Buckets[].{Name:Name}"}
③ 指示どおり aws_read を呼び出す
④ AssumeRoleして、対象アカウントに並列で問い合わせる
⑤ 結果をJSON文字列で受け取る
[{"account_id": "xxxxxxxxxxxx", "account_name": "dev-app1",
"region": "ap-northeast-1",
"result": [{"Name": "my-bucket-1"}, {"Name": "my-bucket-2"}]}]
⑥ その結果をLLMに渡して、日本語の回答を書いてもらう
⑦ 画面に表示する
という流れになっています。
ファイル構成は以下のようになっています。中身の具体的な話は次章で説明します。
.
├── app.py # Streamlitのチャット画面
├── agent.py # エージェントの組み立て
├── tools/
│ └── aws_read.py # AWSに問い合わせるツール（546行）
├── config/
│ └── accounts.yaml # 調査対象アカウントの一覧
└── infra/ # 調査用IAMロール（Terraform）
├── providers.tf # アカウントごとのprovider定義
├── main.tf # 各アカウントへのロール配布
└── modules/
└── readonly-role/ # ロール本体の定義
実装
実装にはClaudeCodeを利用しています。
動作環境
| 項目 | 内容 |
|---|---|
| Python | 3.12 |
| AWS CLI | v2 |
| AWSアカウント | Amazon BedrockでClaudeモデルが有効化されている |
実際に動かしたときの主なライブラリのバージョンは以下のとおりです。
| ライブラリ | バージョン | 用途 |
|---|---|---|
| strands-agents | 1.53.0 | エージェントのフレームワーク |
| streamlit | 1.62.0 | チャット画面 |
| boto3 | 1.43.78 | AWSのAPI呼び出し |
| jmespath | 1.1.0 | 取得結果の絞り込み |
コード全体はGitHubに置いています。
エージェント本体（agent.py）
エージェントを組み立てているファイルです。
agent.py の全文（クリックで展開）
from __future__ import annotations
import os
from typing import Any
from strands import Agent
from strands.models import BedrockModel
from tools.aws_read import ALLOWED_OPERATION_PREFIXES, aws_read, load_config
DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-6"
DEFAULT_REGION = "ap-northeast-1"
def _format_account_table(accounts: list[dict[str, Any]]) -> str:
"""アカウント一覧をMarkdownの表にする。"""
lines = [
"| アカウント名 | アカウントID | 説明 |",
"| --- | --- | --- |",
]
for account in accounts:
name = account.get("name", "")
description = account.get("description", "") or "（説明なし）"
lines.append(f"| {name} | {account['id']} | {description} |")
return "\n".join(lines)
def build_system_prompt(config: dict[str, Any] | None = None) -> str:
"""調査対象アカウントを埋め込んだシステムプロンプトを組み立てる。"""
config = config or load_config()
accounts = config["accounts"]
default_region = config["default_region"]
prefixes = " / ".join(ALLOWED_OPERATION_PREFIXES)
return f"""あなたはAWS環境の読み取り専用調査アシスタントです。回答は必ず日本語で行います。
# 調査できるAWSアカウント
{_format_account_table(accounts)}
デフォルトのリージョンは {default_region} です。
質問文に出てくる「本番の」「開発の」「学習用の」「ハンズオンの」といった表現は、
上の表の説明を手がかりに、どのアカウントを指しているか判断してください。
どのアカウントか判断がつかない場合は、全アカウントを対象に調べるか、ユーザーに聞き返してください。
# 使えるツール
`aws_read` の1つだけです。読み取り専用のAWS API（{prefixes} で始まる操作）を、
複数のアカウントとリージョンに対して並列に実行して結果をJSONで返します。
# 必ず守ること
1. **回答は必ず aws_read の実行結果だけを根拠にしてください。**
あなたの一般知識や推測でリソース名・台数・設定値を答えてはいけません。
データが取得できなかった場合は「取得できなかった」と正直に伝えてください。
ツールを一度も呼ばずに具体的な調査結果を答えるのは禁止です。
2. **アカウント横断の質問は、1回のツール呼び出しにまとめてください。**
アカウントごとに `aws_read` を呼び分けてはいけません。`accounts` に複数指定するか
`"all"` を渡せば、ツール側が並列に全アカウントを調べます。
3. **結果が大きくなりそうな操作では、必ず `query`（JMESPath）で必要なフィールドだけに絞ってください。**
例: EC2の台数を数えたいなら
`query="Reservations[].Instances[].{{id:InstanceId,state:State.Name}}"` のようにする。
絞らずに全項目を取ると結果が巨大になり、途中で切り詰められて正しく答えられなくなります。
4. **回答の末尾に、根拠として実行した操作を1〜2行で添えてください。**
例:
> 根拠: ec2.describe_instances を dev-app1 / ap-northeast-1 に対して実行
# エラーが返ってきたときの対応
- 一部のアカウントだけ失敗した場合は、成功したぶんの結果で回答しつつ、
どのアカウントがなぜ失敗したかを必ず明記してください。黙って省略してはいけません。
# 回答のしかた
- 台数や一覧は、表や箇条書きで読みやすくまとめてください。
- 「古いものを指摘して」のような評価を求められたら、まず取得した事実（バージョン等）を示し、
そのうえで評価やおすすめの対応を述べてください。
- 対象が0件だった場合は「0件でした」とはっきり伝えてください。何も言わずに済ませないこと。
"""
def create_agent(config: dict[str, Any] | None = None) -> Agent:
"""調査用エージェントを生成する。"""
config = config or load_config()
model = BedrockModel(
model_id=os.environ.get("BEDROCK_MODEL_ID", DEFAULT_MODEL_ID),
region_name=os.environ.get("AWS_REGION", DEFAULT_REGION),
streaming=True,
temperature=0.2,
max_tokens=8192,
)
return Agent(
model=model,
tools=[aws_read],
system_prompt=build_system_prompt(config),
callback_handler=None,
)
create_agent()の部分がメインで、ここでエージェントを作っています。
やっていることはシンプルで、Agent() にモデル・ツール・システムプロンプトを渡すだけです。
-
modelはLLMが利用するモデルを指定し、今回はClaude Sonnet 4.6を利用しています。 -
toolsはエージェントが必要に応じて呼び出せる機能を指定するものです。今回はaws_readの1つだけです（次節で説明） -
system_promptはbuild_system_prompt()で組み立てたシステムプロンプトを指定しています。
build_system_prompt()では、システムプロンプトとして、エージェントの役割や調査できるアカウント一覧・守ってほしいルールを、1つの文字列にまとめています。
工夫点としては、LLMが自分の知識で答えようとするのを防ぐために、必ずツールを呼ぶように書いているのと、調査結果の根拠も記載させることで、どのように結果を導いたのかわかりやすくしています。
AWSに問い合わせるツール（tools/aws_read.py）
エージェントに持たせている唯一のツールになります。
コードは関数の定義の部分など一部省略しているので、全文が気になる方はGithubをご確認ください
aws_read() の中身（クリックで展開）
@tool
def aws_read(
service: str,
operation: str,
params: dict | None = None,
accounts: list[str] | None = None,
regions: list[str] | None = None,
query: str | None = None,
) -> str:
"""複数のAWSアカウントに対して読み取り専用のAPIを実行し、結果をJSONで返す。
アカウントとリージョンへの展開はこのツールの中で並列に行われる。
複数アカウントを調べたいときは、アカウントごとに呼び分けるのではなく
accounts に複数指定して1回で呼ぶこと。
Args:
service: AWSのサービス名。boto3のクライアント名と同じ。例: ec2, rds, s3
operation: 実行する操作名をsnake_caseで。describe_ / list_ / get_ で始まるものだけ実行できる。例: describe_instances
params: 操作に渡すパラメータ。boto3と同じPascalCaseのキーで指定する。
accounts: 調査対象。アカウントIDかアカウント名のリスト。all で全アカウント。例: ["dev-app1"]
regions: 対象リージョンのリスト。省略時は設定ファイルのdefault_region。
query: 結果に適用するJMESPath式。必要なフィールドだけに絞るために積極的に使うこと。
Returns:
[{account_id, account_name, region, result}] 形式のJSON文字列。
失敗したアカウントは result の代わりに error が入る。
"""
# 1. 読み取り専用の強制
if not operation.startswith(ALLOWED_OPERATION_PREFIXES):
return _error_payload(
f"操作 '{operation}' は許可されていません。"
f"このツールは読み取り専用で、{' / '.join(ALLOWED_OPERATION_PREFIXES)} "
"で始まる操作しか実行できません。",
rejected_operation=operation,
)
params = params or {}
# 2. 設定と対象の解決
try:
config = load_config()
resolved_accounts = _resolve_accounts(accounts, config)
targets = _resolve_targets(resolved_accounts, regions, service, config)
except Exception as exc:
return _error_payload(str(exc))
# 3. アカウント×リージョンを並列に実行
entries: list[dict[str, Any]] = []
with concurrent.futures.ThreadPoolExecutor(
max_workers=min(MAX_WORKERS, len(targets))
) as executor:
futures = [
executor.submit(_fetch_one, account, region, service, operation, params)
for account, region in targets
]
for future in futures:
entries.append(future.result())
# 4. JMESPathでの絞り込み
if query:
compiled = jmespath.compile(query)
for entry in entries:
if "result" in entry:
entry["result"] = compiled.search(entry["result"])
# 5. サイズ上限の適用
return _serialize_with_limit(entries)
@tool を付けるだけでツールになる
関数の上に@toolを付けるだけで、ただのPython関数をLLMが呼べるようになります。
@tool
def aws_read(service: str, operation: str, ...) -> str:
処理の内容
5ステップに分けて処理が行われています。
1. 読み取り専用を強制
if not operation.startswith(ALLOWED_OPERATION_PREFIXES):
return _error_payload(...)
ALLOWED_OPERATION_PREFIXES は ("describe_", "list_", "get_") です。これ以外で始まる操作名は、ここで即座に弾かれます。
実際に利用するIAMロールは読み取り権限しか持たないため、書き込み処理はエラーになりますが、そもそも書き込みの場合はAWSへの接続すら発生しないようにしています。
2. どのアカウントが調査対象か決める
config = load_config()
resolved_accounts = _resolve_accounts(accounts, config)
targets = _resolve_targets(resolved_accounts, regions, service, config)
LLMがAWSアカウントIDでなく、「dev-app1」というアカウント名と返すことがあるので、accounts.yaml を引いて、実際のアカウントIDに変換しています。
3. 複数アカウントを並列で調査
with concurrent.futures.ThreadPoolExecutor(max_workers=...) as executor:
futures = [
executor.submit(_fetch_one, account, region, service, operation, params)
for account, region in targets
]
for future in futures:
entries.append(future.result())
アカウント×リージョンの組を作って、まとめて並列に投げています。
アカウントごとにツールを呼び分けると、そのたびにLLMとの往復が発生して時間がかかるため、時短のためです。
実際にAWSを叩いているのは _fetch_one() の中です。
credentials = _assume_role(account["id"], region)
client = boto3.client(service, region_name=region, **credentials)
method = getattr(client, operation)
response = method(**params)
4. JMESPathで結果を絞る
if query:
compiled = jmespath.compile(query)
for entry in entries:
entry["result"] = compiled.search(entry["result"])
JMESPathはJSONから必要な部分だけを取り出す問い合わせ言語で、AWS CLI の --query オプションと同じものです。
aws s3api list-buckets --query "Buckets[].Name"
そのまま渡すとコンテキストを食い潰してしまうところを、必要な部分に絞っています。例えば、S3バケット一覧で試したところ、CreationDateやBucketArnが捨てられ、2165文字が742文字まで減りました。
5. 大きすぎる結果は切り詰める
return _serialize_with_limit(entries)
def _serialize_with_limit(entries: list[dict[str, Any]]) -> str:
"""結果をJSON文字列にする。上限を超えるぶんは捨てて truncated を立てる。"""
payload = _dumps(entries)
if len(payload) <= MAX_RESULT_CHARS:
return payload
notice_size = len(_dumps(_TRUNCATION_NOTICE))
budget = MAX_RESULT_CHARS - notice_size - 4 # 括弧とカンマのぶん
kept: list[dict[str, Any]] = []
used = 2 # 角括弧のぶん
for entry in entries:
chunk_size = len(_dumps(entry)) + 1
if used + chunk_size <= budget:
kept.append(entry)
used += chunk_size
continue
# 1件目すら入らないときは、中身を削ってでも何か返す
if not kept:
kept.append(_shrink_entry(entry, budget - used))
break
kept.append(_TRUNCATION_NOTICE)
return _dumps(kept)
queryで絞ってもなお大きいことがあるので、最後に上限をかけます。アカウント単位で入るところまで詰めて、あとは捨てます。
MAX_RESULT_CHARS = 50_000
捨てたときは、配列の最後にこれを足します。
_TRUNCATION_NOTICE = {
"truncated": True,
"note": (
"結果が大きすぎるため途中で切り詰めました。"
"query パラメータ（JMESPath）で必要なフィールドだけに絞り込んで再実行してください。"
),
}
切り詰めたこととどうすればいいかを一緒に返すことで、LLMが自分でqueryを書き直して再実行してくれるようにしています。
画面（app.py）
Streamlitで作成しています。
app.py の全文（クリックで展開）
"""Streamlit製のチャットUI。`uv run streamlit run app.py` で起動する。"""
from __future__ import annotations
import asyncio
import os
from typing import Any, Iterator
import streamlit as st
from dotenv import load_dotenv
# Agentを組み立てる前に .env を読む必要がある
load_dotenv()
from agent import DEFAULT_MODEL_ID, create_agent # noqa: E402
from tools.aws_read import load_config # noqa: E402
st.set_page_config(page_title="AWS調査チャットボット", page_icon="☁️", layout="wide")
def iter_agent_events(agent: Any, prompt: str) -> Iterator[dict[str, Any]]:
"""非同期イテレータの stream_async を、同期のジェネレータとして扱えるようにする。"""
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
events = agent.stream_async(prompt)
while True:
try:
yield loop.run_until_complete(events.__anext__())
except StopAsyncIteration:
break
finally:
asyncio.set_event_loop(None)
loop.close()
def describe_tool_use(tool_use: dict[str, Any]) -> str:
"""ツール呼び出しの中身を、UIに出せる短い日本語にする。"""
tool_input = tool_use.get("input") or {}
if isinstance(tool_input, str):
# 引数はストリーミング中に少しずつ届くので、途中では壊れたJSON文字列になっている
return "AWSに問い合わせる準備をしています…"
service = tool_input.get("service", "?")
operation = tool_input.get("operation", "?")
accounts = tool_input.get("accounts")
target = ", ".join(accounts) if isinstance(accounts, list) else "全アカウント"
return f"{service}.{operation} を {target} に対して実行中…"
config = load_config()
with st.sidebar:
st.header("設定")
st.caption("モデル")
st.code(os.environ.get("BEDROCK_MODEL_ID", DEFAULT_MODEL_ID), language=None)
st.caption("Bedrockのリージョン")
st.code(os.environ.get("AWS_REGION", "ap-northeast-1"), language=None)
st.divider()
st.caption(f"調査対象アカウント（{len(config['accounts'])}件）")
for account in config["accounts"]:
st.markdown(f"**{account.get('name')}** `{account['id']}` \n{account.get('description', '')}")
st.divider()
if st.button("会話をリセット", use_container_width=True):
st.session_state.pop("agent", None)
st.session_state.pop("messages", None)
st.rerun()
st.title("☁️ AWS調査チャットボット")
st.caption("自然言語で複数のAWSアカウントを横断調査します（読み取り専用）")
if "messages" not in st.session_state:
st.session_state.messages = []
if "agent" not in st.session_state:
st.session_state.agent = create_agent(config)
# これまでのやりとりを描き直す
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
if prompt := st.chat_input("例: dev-app1アカウントのEC2インスタンスの台数を教えて"):
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
with st.chat_message("assistant"):
status = st.status("考えています…", expanded=False)
placeholder = st.empty()
answer = ""
try:
for event in iter_agent_events(st.session_state.agent, prompt):
if "current_tool_use" in event:
status.update(label=describe_tool_use(event["current_tool_use"]))
if "data" in event:
if not answer:
status.update(label="AWSからの応答をまとめています", state="complete")
answer += event["data"]
placeholder.markdown(answer + "▌")
placeholder.markdown(answer)
status.update(label="完了", state="complete")
except Exception as exc:
status.update(label="エラー", state="error")
answer = f"エラーが発生しました: `{exc}`"
placeholder.error(answer)
st.session_state.messages.append({"role": "assistant", "content": answer})
チャット画面は数行で作れる
Streamlitにはチャット用の部品が用意されています。
if prompt := st.chat_input("例: dev-app1アカウントのEC2インスタンスの台数を教えて"):
with st.chat_message("user"):
st.markdown(prompt)
st.chat_input() が下部の入力欄、st.chat_message() が吹き出しです。
待ち時間に何が起きているか見せる
待ち時間に何も表示されていないと、ちゃんと動いているのか不安になってしまうので、今何をしているのかを表示するようにしています。
for event in iter_agent_events(st.session_state.agent, prompt):
if "current_tool_use" in event:
status.update(label=describe_tool_use(event["current_tool_use"]))
if "data" in event:
answer += event["data"]
placeholder.markdown(answer + "▌")
current_tool_use はツールを呼ぼうとしているときに届くイベントです。中身を読んで s3.list_bucketsをdev-app1に対して実行中… のような文言に変えています。
調査したいアカウントを記載
config/accounts.yaml には調査したいアカウントを並べます。
accounts:
- id: "111111111111"
name: prod-app1
description: 本番のアプリ1系
- id: "222222222222"
name: dev-app1
description: 開発環境
default_region: ap-northeast-1
各AWSアカウントでIAMロールを作成
調査に利用するIAMロールを各アカウントで作成します。
権限はAWS管理ポリシーの ReadOnlyAccess だけです。全サービスのdescribe/list/get系がまとめて付きます。
信頼ポリシーには、誰がこのロールを借りられるかを書きます。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"AWS": "arn:aws:iam::123456789012:root"
},
"Action": "sts:AssumeRole",
"Condition": {
"ArnLike": {
"aws:PrincipalArn": "arn:aws:iam::123456789012:role/chatbot-runner"
}
}
}
]
}
123456789012 はチャットボットを動かしている側のアカウントIDです。chatbot-runner の部分には、チャットボットが実際に使っているロール名が入ります。ローカルで動かすならログインに使っているロール、EC2に載せるならインスタンスに付けたロールです。
動かしてみる
uv run streamlit run app.py
でローカルで起動できます。
何かチャットを入力し、応答が返れば成功です！
おわりに
まずはローカルでAIエージェントを動かすところまで実装することができました。
次回は、Bedrock AgentCore Runtime上で動かし、Webアプリにすることで誰でも利用できる状態にしてみようと思います。
最後までお読みいただきありがとうございました！