Strands Agents + Bedrockで、アラート対応サポートSlack Bot作ってみた

こんにちは！
株式会社カンリーでSREをやっている本間です。ほんまですよ！

Strands Agents と Bedrock を組み合わせて、以下のような機能を持つ Slack Bot を作成しました。本記事では、その実装方法や構成について紹介します。

• SlackのメンションでAIエージェントを起動
• メンションのあったスレッド（アラートを想定）のテキストを読み取る
• AIが内容を要約し、対応方針を提案、スレッドに返信する

本記事はやっていませんが、MCPを利用することでより広い範囲のサポートができそうな可能性を感じました…！

成果物

「@incident-sherpa このアラートの要約と対応方針をまとめてください」のようにメンションすると、アラート（今回はAWS Health Event）の要約と対応を提案してくれます。

（ちなみに、"sherpa（シェルパ）"とはヒマラヤ登山をガイドしてくれる案内人のことらしく、インシデント対応の“道案内役”として相性が良いと思い採用しました。ありがとう、ChatGPT）

これにより、AWS に詳しくないメンバーでもアラート内容をすぐ理解でき、属人化の抑制や SRE への問い合わせ削減につながります。今回はアラート対応を目的に実装しましたが、他の用途にも応用できそうです！

シーケンス図 / 構成のポイント

ざっくり構成ポイントは以下のとおりです

• Strands AgentsをECS Fargate上で動かす
  • Lambda も検討しましたが、レスポンス速度と、将来的に「会話しながら継続サポートする」用途を想定しているため、常時稼働可能な構成として Fargate を選択しました
  • 今後はBedrock AgentCoreに載せ替えも検討です
• Slack のメンションをトリガーに処理開始
  • Slack の Event Subscriptions を使用し、メンション時に指定エンドポイントへイベントが飛ぶ仕組みです
• 返信はSlack SDKを利用
  • 後述しますが、Strands Agents Tools は使用していません

Strands Agents / Strands Agents Tools

「Strands Agents」は、AWS が中心となって開発している AI エージェント向けのオープンソース Python SDK です。
ざっくり言うと、「モデル（LLM）」「ツール（API や関数）」「プロンプト」をセットで定義すると、その組み合わせでタスクを自動で進めてくれるエージェントを作れるライブラリです。

最小構成でもサクッと動かせるので、まずエージェントを触ってみたい人にちょうどいい感じです。

もっと詳しく知りたい人は、AWS が出しているこちらの記事を読んでみてください👇
https://aws.amazon.com/jp/blogs/news/introducing-strands-agents-an-open-source-ai-agents-sdk/

Strands Agents を さらに便利にしてくれるのが「Strands Agents Tools」です。
Tools はエージェントの“手足”みたいなもので、Slack に投稿したり、ファイル操作をしたり…といった処理を、エージェント自身が状況を見て選んで実行できるようになる“機能のまとまり（Python関数）”です。

リポジトリはこちら👇
https://github.com/strands-agents/tools

細かく指示しなくても、エージェントが必要な Tool を勝手に使ってくれるようです。

本当は Strands Agents Tools を使って、「推論した結果をエージェントがそのまま Slack に返信してくれる」という仕組みを作りたかったのですが、いろいろ試した結果うまくいかず……。

結局、前述のとおり素直に Slack SDK を使って返信する方式 に落ち着きました。

Slack API

Slack Bot を動かすために設定した内容を、ざっくりまとめます。

OAuth & Permissions

まずは Bot が必要とするスコープ（権限）を付与します。

■必須スコープ

スコープ	説明
app_mentions:read	メンション（app_mention）イベントを受け取るために必須
channels:history	スレッドの履歴を読むのに必要
channels:read	チャンネルの基本情報の取得
chat:write	ボットが返信するために必須
groups:read	プライベートチャンネルの基本情報取得

■任意スコープ（必要に応じて）

スコープ	説明
groups:history	プライベートチャンネルのスレッド履歴も読みたい場合
im:history	Bot との DM でも使えるようにしたい場合
mpim:history	グループ DM に対応したい場合
files:read	スレッドに添付されたファイル（ログなど）を扱いたい場合

Event Subscriptions

メンションイベントが飛んでくるエンドポイントをここで設定します。

ここで注意点がひとつ。
Slack は、イベント送信後 3 秒以内に 200 OK を返さないとエラー扱いになります。

We wait longer than 3 seconds to receive a valid response from your server.

参考：https://docs.slack.dev/apis/events-api/#failure

つまり、受け取ったら即レス（200 OK）→ バックグラウンドでエージェント処理
という非同期構成にしておく必要があります。

その他

一通り設定を終えたら、アプリをワークスペースにインストールします。
また、以下のシークレットも使用します。

Bot User OAuth Token（xoxb-〜）
→ OAuth & Permissions ページにあります

Signing Secret
→ Basic Information ページにあります

Signing Secret は、リクエストが本当に Slack から来たものか検証するために使う署名チェック用です。リクエストヘッダーX-Slack-Signatureを、この値を使って検証します。
セキュリティのため設定することを推奨します。

インフラ構成

ここでは、通常の Web アプリ構成に「今回の Bot を動かすために追加で必要だった部分」だけサクッと紹介します。

ドメイン

Slack の Event Subscriptions では、イベントを受け取るエンドポイントを登録する必要があります。
そのため、ここで設定したドメインを Slack の Event Subscriptions 側にも設定してあげる必要があります。

ECS Fargate

エージェントを動かすためのリクエスト受付には、FastAPI や Flask などの Web アプリケーションフレームワーク が必要です。

アプリで設定したポートに対して外部からアクセスできるよう、セキュリティグループでそのポートを解放しておきます。

IAM（ECS タスクロール）

エージェントが Bedrock を使えるように、Bedrock へのアクセス権限が必要です。

また、Bot User OAuth Token や Signing Secret を Secrets Manager に入れる場合は、
Secrets Manager の参照権限も付与しておきます。

iam.tf

resource "aws_iam_policy" "bedrock_access" {
  name        = "bedrock-access"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "bedrock:InvokeModel",
          "bedrock:InvokeModelWithResponseStream"
        ]
        Resource = [
          "*"
        ]
      }
    ]
  })
}

アプリケーションコード（FastAPI / Python）

ここでは、コードの中でもポイントになる部分だけを抜粋して紹介します。

Slackのメッセージ取得

AIにお願いしつつも、手こずった部分です。
どこが面倒だったかというと、メッセージの種類によってテキストの取得方法が変わってしまう部分です。

今回AWS Health Eventアラート（アプリ:Amazon Q）を対象にしましたが、それ以外にもDatadogのアラート、ファイル（チャンネルメールアドレスに送ったもの）、などなどケースによってオブジェクトの構成が変わってしまいます。

参考になるのかわからない＆網羅できていないかもですが、一応その部分のコードを置いておきます。
また、重複してメッセージが取れることがあり、その部分の対策も含まれています。

Slackメッセージ取得処理

(省略)
        # ボットのユーザーIDを取得（メンション除去用）
        bot_user_id = None
        try:
            auth_result = slack_client.auth_test()
            if auth_result["ok"]:
                bot_user_id = auth_result["user_id"]
        except Exception as e:
            print(f"Warning: Could not get bot user ID: {e}")

        # スレッドの会話履歴を取得
        conversation_history = []
        try:
            result = slack_client.conversations_replies(
                channel=channel,
                ts=thread_ts,
                limit=100,  # 最大100件のメッセージを取得（古い順）
                inclusive=True  # スレッドの最初のメッセージも含める
            )
            
            print(f"Retrieved {len(result.get('messages', []))} messages from Slack API")
            
            if result["ok"]:
                raw_messages = result.get("messages", [])
                                
                # メッセージは古い順で返ってくるので、そのまま追加
                seen_timestamps = set()  # 重複チェック用
                for message in raw_messages:
                    user_id = message.get("user", "")
                    bot_id = message.get("bot_id", "")
                    text = message.get("text", "")
                    timestamp = message.get("ts", "")
                    
                    # 重複チェック
                    if timestamp in seen_timestamps:
                        print(f"Duplicate message skipped: ts={timestamp}")
                        continue
                    seen_timestamps.add(timestamp)
                    
                    # textが空の場合、attachmentsやblocksから取得を試みる
                    if not text or text.strip() == "":
                        # attachmentsから取得
                        attachments = message.get("attachments", [])
                        if attachments:
                            attachment_texts = []
                            for att in attachments:
                                # attachmentの中のblocksを解析
                                if att.get("blocks"):
                                    blocks = att["blocks"]
                                    for block in blocks:
                                        block_type = block.get("type")
                                        
                                        if block_type == "section":
                                            # sectionブロックのtext
                                            if block.get("text"):
                                                text_obj = block["text"]
                                                if isinstance(text_obj, dict):
                                                    attachment_texts.append(text_obj.get("text", ""))
                                                else:
                                                    attachment_texts.append(str(text_obj))
                                            
                                            # sectionブロックのfields
                                            if block.get("fields"):
                                                for field in block["fields"]:
                                                    if isinstance(field, dict):
                                                        field_text = field.get("text", "")
                                                        if field_text:
                                                            attachment_texts.append(field_text)
                                        
                                        elif block_type == "header":
                                            # headerブロック
                                            if block.get("text"):
                                                text_obj = block["text"]
                                                if isinstance(text_obj, dict):
                                                    attachment_texts.append(f"# {text_obj.get('text', '')}")
                                                else:
                                                    attachment_texts.append(f"# {str(text_obj)}")
                                
                                # 古いスタイルのattachmentフィールド（blocksがない場合のフォールバック）
                                if not attachment_texts:
                                    # title（存在する場合）
                                    if att.get("title"):
                                        attachment_texts.append(f"## {att['title']}")
                                    
                                    # pretext（存在する場合）
                                    if att.get("pretext"):
                                        attachment_texts.append(att["pretext"])
                                    
                                    # text本文（存在する場合）
                                    if att.get("text"):
                                        attachment_texts.append(att["text"])
                                    
                                    # fields（重要な詳細情報）
                                    if att.get("fields"):
                                        for field in att["fields"]:
                                            title = field.get("title", "")
                                            value = field.get("value", "")
                                            if title and value:
                                                # Markdownのリストとして整形
                                                attachment_texts.append(f"- **{title}**: {value}")
                                    
                                    # fallbackがある場合（他の情報がない時のみ）
                                    if not attachment_texts and att.get("fallback"):
                                        attachment_texts.append(att["fallback"])
                            
                            text = "\n".join(attachment_texts)
                        
                        # blocksから取得（attachmentsがない場合）
                        if not text:
                            blocks = message.get("blocks", [])
                            if blocks:
                                block_texts = []
                                for block in blocks:
                                    block_type = block.get("type")
                                    
                                    if block_type == "section":
                                        # sectionブロックのtext
                                        if block.get("text"):
                                            text_obj = block["text"]
                                            if isinstance(text_obj, dict):
                                                block_texts.append(text_obj.get("text", ""))
                                            else:
                                                block_texts.append(str(text_obj))
                                        
                                        # sectionブロックのfields
                                        if block.get("fields"):
                                            for field in block["fields"]:
                                                if isinstance(field, dict):
                                                    field_text = field.get("text", "")
                                                    if field_text:
                                                        block_texts.append(field_text)
                                    
                                    elif block_type == "header":
                                        # headerブロック
                                        if block.get("text"):
                                            text_obj = block["text"]
                                            if isinstance(text_obj, dict):
                                                block_texts.append(f"# {text_obj.get('text', '')}")
                                            else:
                                                block_texts.append(f"# {str(text_obj)}")
                                    
                                    elif block_type == "rich_text":
                                        # rich_textブロック（より複雑な構造）
                                        if block.get("elements"):
                                            for element in block["elements"]:
                                                if element.get("type") == "rich_text_section":
                                                    section_texts = []
                                                    for el in element.get("elements", []):
                                                        if el.get("text"):
                                                            section_texts.append(el["text"])
                                                    if section_texts:
                                                        block_texts.append("".join(section_texts))
                                
                                text = "\n".join(block_texts)
                    
                    # ボットのメンション部分を除去
                    if bot_user_id and text:
                        text = text.replace(f"<@{bot_user_id}>", "").strip()
                    
                    # ユーザー識別（ボットかユーザーか）
                    if bot_id:
                        sender = f"bot:{bot_id}"
                    elif user_id:
                        sender = f"user:{user_id}"
                    else:
                        sender = "unknown"
                    
                    conversation_history.append({
                        "sender": sender,
                        "text": text,
                        "timestamp": timestamp
                    })
        except SlackApiError as e:
            print(f"Error fetching conversation history: {e.response['error']}")
            # エラーでも処理を続行（履歴なしで回答）
(省略)

Bedrock に処理を投げる部分

model_id や prompt はまだ調整段階のものなので、実際の環境に合わせて適宜変更してください！

エージェントの役割（ロール）を定義して、プロンプトを渡すだけでちゃんと結果が返ってくるの便利ですね👏

from strands import Agent
from strands.models import BedrockModel
(省略)
				
        # BedrockModelを作成
        bedrock_model = BedrockModel(
            model_id="<model_id>",
            region_name="<region_name>",
            temperature=0.3,
        )
        
        # エージェントを作成
        agent = Agent(
            model=bedrock_model,
            system_prompt="""あなたはインシデント対応を支援するAIアシスタントです。
以下の点を理解して回答してください：
1. Slackスレッドの会話履歴全体が提供されます
2. スレッドの最初にはシステムアラートやエラー通知が含まれている可能性があります
3. ユーザーの質問に対して、会話履歴とアラート内容を踏まえて回答してください
4. 具体的な対応方法や調査手順を提案してください""",
        )
        
        # 会話履歴を含めてエージェントを実行
        prompt = f"""以下はSlackスレッドの会話履歴です：
{formatted_history}
---
最新のメッセージで、あなたにメンションがありました。
上記の会話履歴全体を理解し、特にスレッドの最初にあるアラートやエラー情報を参照して、適切に回答してください。"""
        
        response = agent(prompt)
(省略)

今後の拡張

今後は、こんな機能も追加していきたいと思っています。

• GitHub のコード参照（MCP GitHub）を使って、コードベースで影響範囲を推論
• Backlog に自動チケット作成
• AWS リソースや Datadog の情報を参照しながら、Slack 上である程度調査を進められるようにする
• 初動対応の自動化（ChatOps 的な方向）

まだまだやれることは多そうです！

おわりに

Strands Agents を使ってみて、AIエージェントが手軽に実装できることを実感しました。

SRE チームでは、今後も運用負荷を下げてくれる AI エージェントをどんどん作っていく予定です。
最後まで読んでいただき、ありがとうございました！