Claude Codeと征く、社内情報検索Slack botのStrands Agents対応への道（検証編）

# はじめに

こんにちは、**yu_Matsu**です！
先月は生成AI界隈で大きな発表が多く、頭がパンクしそうになる日々が続きましたが、皆様はいかがでしたでしょうか。

今回なその中でも **Strands Agents** と **Claude Code** のキャッチアップを兼ねて、社内情報検索Slack botのアップデートの検証を行ったため、その体験記を記事にしました。



# Strands Agentsとは

**Strands Agents** は、2025年5月16日に発表されたばかりの **AWS公式のAIエージェントのSDK** になります。エージェントのSDKといえば、LangChain/LangGraph や MastraAI などがよく利用されていますが、遂にAWSからも提供されることとなりました！

https://strandsagents.com/latest/

特徴として、「わずか数行のコードで AI エージェントを構築・実行するモデル駆動型アプローチを採用したオープンソース SDK」と謳われている通り、本当に簡単にAIエージェントを実装することが出来ます。

以下は公式ドキュメントのQuickStartから拝借した、最小単位のコード例になります。たったこれだけでAIエージェントが実現出来てしまうのは驚きです！

```Python
from strands import Agent

# Create an agent with default settings
agent = Agent()

# Ask the agent a question
agent("Tell me about agentic AI")
```

もちろん、ツールの利用やマルチエージェントも対応しており、従来と比べるとより直感的に実装できるようになっていますので、AIエージェントの敷居がまた一段下がったのではないでしょうか。

# Claude Codeとは

この記事のもう一つのトピックである **Claude Code** ですが、こちらは先月の Anthropic社の開発者向けイベントである **Claude with Code** で正式リリースが発表された**エージェント型コーディングツール**になります。

ベータ版ではCLIからの利用のみでしたが、正式リリースに伴い、VSCodeなどのIDE上での利用も可能になりました。

https://docs.anthropic.com/ja/docs/claude-code/overview

以下のコマンドでインストールすることができ、

```
npm install -g @anthropic-ai/claude-code
```

Claude Codeを起動したいプロジェクトのディレクトリに移動し、以下のコマンドを実行することで起動できます

```
claude
```

初回起動後は認証が必要になりますので、詳しくは下記のドキュメントをご参考ください。

https://docs.anthropic.com/ja/docs/claude-code/getting-started

セットアップが完了すると、以下のようなメッセージが表示されます。Claude Codeへの指示は下の入力欄に入力して実行する感じになります。

![スクリーンショット 2025-05-24 22.11.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/24a6580d-2ad2-4a37-a4d4-5fa0ff8c425f.png)

VSCodeの例になりますが、エディタ右上の３点リーダーを押下すると「**Run Claude Code**」が現れるようになっており、CLIからではなくここからでも起動できるようになっています。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/357fecc1-7e5b-4894-b9cc-0c1181853148.png" width=300>





  
# 既存のSlack botのコード
本題に入る前に、まずは既存のSlack botのコードのサンプルを以下に掲載します。

**Slack bolt**を利用して実装した簡単なSlack botであり、裏側では**Knowledge bases for Amazon Bedrock（以降「Knowledge bases」）** の Retriever を利用した RAG 搭載のLLMチェーンが動いています。

メッセージがストリーミングで表示されるようにしているため、少し途中の処理が複雑になっていますがご了承ください。

```Python:app.py
import os
import asyncio
import logging
from typing import Dict, List
from dotenv import load_dotenv

# slack bolt周り
from slack_bolt import App
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from slack_bolt.async_app import AsyncApp

# langchain周り
from langchain_aws import ChatBedrockConverse
from langchain_aws import AmazonKnowledgeBasesRetriever
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# Load environment variables from .env file
load_dotenv()

# 環境変数の設定
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_APP_TOKEN = os.environ["SLACK_APP_TOKEN"]
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
KNOWLEDGE_BASE_ID = os.environ["KNOWLEDGE_BASE_ID"]  # Amazon Knowledge Base ID

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 非同期Slackアプリの初期化
app = AsyncApp(token=SLACK_BOT_TOKEN)

# Amazon Knowledge Baseリトリーバーの設定
retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KNOWLEDGE_BASE_ID,
    retrieval_config={
        "vectorSearchConfiguration": {
            "numberOfResults": 4  # 検索結果の数
        }
    },
    region_name=AWS_REGION
)

# BedrockのAIモデルの初期化
bedrock_model = ChatBedrockConverse(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    region_name=AWS_REGION
)

# ドキュメントをフォーマットする関数
def format_docs(docs):
    """取得したドキュメントを文字列に整形"""
    return "\n\n".join(doc.page_content for doc in docs)

# プロンプトテンプレート
prompt = ChatPromptTemplate.from_template("""
あなたは専門的なアシスタントです。与えられたコンテキストに基づいて質問に答えてください。

コンテキスト:
{context}

質問: {question}

回答は日本語で行ってください。
""")

# RAGのLLMチェーン
chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | prompt
    | bedrock_model
    | StrOutputParser()
)

# メンションされたときの処理
@app.event("app_mention")
async def handle_mention(event, say):
    """Botがメンションされたときの処理"""
    user_id = event["user"]
    channel_id = event["channel"]
    text = event["text"]
    
    # <@BOT_ID>の部分を削除してメッセージのみを取得
    message_text = text.split(">", 1)[1].strip() if ">" in text else text
    
    try:
        # スレッドで初期メッセージ送信（タイピング中の表示代わり）
        response = await say(
            text="考え中...",
            thread_ts=event["ts"]
        )
        message_ts = response["ts"]
        
        # ストリーミング処理用の状態管理
        full_response = ""
        update_interval = 1.0  # 更新間隔（秒）
        last_update = 0
        
        # RAGチェーンを非同期ストリーミングで実行
        async for chunk in rag_chain.astream(message_text):
            # 新しいチャンクを追加（文字列として処理）
            if isinstance(chunk, str):
                full_response += chunk
            else:
                full_response += str(chunk)
            
            # 一定間隔でSlackメッセージを更新
            current_time = asyncio.get_event_loop().time()
            if current_time - last_update >= update_interval and full_response.strip():
                await app.client.chat_update(
                    channel=channel_id,
                    ts=message_ts,
                    text=full_response
                )
                last_update = current_time
        
        # 最終的な完全な応答を更新
        if full_response.strip():
            await app.client.chat_update(
                channel=channel_id,
                ts=message_ts,
                text=full_response
            )
        else:
            await app.client.chat_update(
                channel=channel_id,
                ts=message_ts,
                text="応答を生成できませんでした。"
            )
    except Exception as e:
        logger.error(f"エラーが発生しました: {e}")
        await say(f"エラーが発生しました: {str(e)}")

# メイン関数
async def main():
    # AsyncSocketModeHandlerの初期化
    handler = AsyncSocketModeHandler(app, SLACK_APP_TOKEN)
    # アプリの起動
    await handler.start_async()

# スクリプトとして実行された場合
if __name__ == "__main__":
    asyncio.run(main())
```

RAGのデータソースとなるドキュメントは、サンプルとして「**広報東京都2025年5月号**」のPDF版を利用しました。

https://www.koho.metro.tokyo.lg.jp/2025/05/index.html

動作イメージは以下のような感じです。（Slack botの設定等は本題から逸れるため省略します。）

Slack bot（今回は「テスト用ボット」という名前）にメンション付きでメッセージを送ると、スレッドが作成されて会話が開始します。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/3b0346d2-77b4-4da8-902c-8f7501cc98f7.png" width=500>

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/7dc33328-4f08-449f-b395-15ed283bbea0.png" width=350>

上記PDFの内容に関して答えてくれるかどうか試してみます。『「夢のみち」道路愛護ポスターコンクール』というコンクールの応募締め切りについて聞いてみます。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/22c30d7b-9e2d-4e80-a685-9a273149a58d.png" width=250>

正解は「6月27日」ですが、問題なく回答しています。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/f4474d19-4644-450b-9fbf-75991fa13a86.png" width=350>


以降で、このSlack botのコードを**Strands Agentsで書き換え**つつ、**MCPサーバーも利用できる**ようにしていきます！



# まずはStrands Agentsに慣れよう

早速 Claude Code と共に進めていきたいのですが、まずはお互いにStrands Agentsに慣れたいと思います。

Strands Agentsの説明でも触れた通り、非常に簡単なプログラムでAIエージェントを実装できます。以下のような test.py というベースラインのプログラム（ほぼQuickStartのもの）を作成し、機能を付け加えてみたいと思います。

```Python:test.py
import boto3
from strands import Agent
from strands.models import BedrockModel

# Create a BedrockModel
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    region_name='us-east-1',
    temperature=1.0,
)

agent = Agent(model=bedrock_model)

if __name__ == "__main__":
    result = agent("こんにちは！")
    print(result.message)
```

### Strands Agents で RAG を実装したい

今回の目的は RAG 搭載の Slack bot ですので、まずはRAGの利用方法に関して Claude Code に聞いてみます。

```:Claude Codeへの指示
Amazon BedrockのSDKである Strands Agents を利用して、簡単なAIエージェントをtest.pyというファイルに実装しました。
これに Knowledge bases for Amazon Bedrock の Retriever を利用したRAG機能を追加したいと思います。
Strands Agents は最新のSDKになりますので、必要に応じてWebで検索を行いながら test.py を改修して下さい
```

早速 Claude Code がタスクを実行し始めました。


![スクリーンショット 2025-05-28 23.18.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/6ff31280-4c48-4e7f-8bc7-b14ea9bc3188.png)

複雑なタスクであれば、Claude Code はまずブレークダウンを行い、 TODO リストを作成してからタスクに取り掛かっているように見えます。人間と同じような動きができていて、すごいなぁと感心していました。

また、Web検索を行う場合は、その検索キーワードとともに検索実行中であることも表示されています。

どうやら Strands Agents の公式ドキュメントにたどり着いたようです。
さらに検索してもいいかどうかを聞かれたので、「**1. Yes**」を選択しました。（ちなみに、これ以降Strands Agentsのドキュメントは承認なしで検索してもよければ「**2.**」を選択します。）

![スクリーンショット 2025-05-28 23.20.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/03639ece-5008-4b4c-9c0e-86b1af28e146.png)

見守っていると、Strands Agentsのドキュメントの調査が完了し、次のステップに移りました。どうやらboto3のRetrieverの調査をしているようです。

![スクリーンショット 2025-05-28 23.21.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/5c197e2a-aaec-476a-adbc-cc846ce3ecff.png)

一通り調査が完了して、調査内容に基づいてtest.pyを修正してくれたようです。
VSCode上で Claude Code を実行している場合は、２枚目の画像のように修正前と修正後の差分を表示してくれます。

![スクリーンショット 2025-05-28 23.23.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/ed423da9-fbb6-4085-884b-3e4bdd6d97e6.png)

![スクリーンショット 2025-06-01 11.57.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/c442ea9d-072f-4033-9d83-1600b60d390d.png)

結果を見てみると、**カスタムツール**でRAGを実現しているように見受けられます。確かに、Strands Agents では ```@tool``` デコレータを利用することでエージェントが利用するツールを自分で定義できます。

https://strandsagents.com/0.1.x/user-guide/concepts/tools/tools_overview/#building-loading-tools

しかし、少しドキュメントを漁ってみると、実は Strands Agents では「**retrieve**」という Retriever のツールが提供されていることがわかりました。**strands-agents-tools**というパッケージから利用可能です。

https://strandsagents.com/0.1.x/user-guide/concepts/tools/example-tools-package/

ですので、Claude Codeの変更を「**No**」で棄却し、追加で２枚目画像のような指示をしてみました。調査の手助けとなるように、retrieveツールのGitHubのURLを合わせて渡しています。

![スクリーンショット 2025-05-28 23.23.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/9747bf2b-2be7-4bd3-b636-cec1860c634a.png)

![スクリーンショット 2025-05-28 23.37.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/f6412a6d-c294-47ae-9d9a-a99946b0e87d.png)

調査の結果、retrieveツールを利用した実装にたどり着いてくれたようです。
先ほどと同様に変更差分が表示されるので、問題がなければ「**Yes**」を選択すると、変更が反映されます。

![スクリーンショット 2025-05-28 23.38.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/55a3e6f6-0676-4451-824c-a1267125a135.png)

修正された test.py が以下になります。

```Python:test.py
import os
import sys
import logging
from typing import Optional
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands_tools import retrieve
from dotenv import load_dotenv

load_dotenv()

# Enables Strands debug log level
logger = logging.getLogger("strands")


# Knowledge Base ID (環境変数から取得)
KNOWLEDGE_BASE_ID = os.getenv("KNOWLEDGE_BASE_ID", "YOUR_KNOWLEDGE_BASE_ID")

# Create a BedrockModel
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    region_name='us-east-1'
)

# Create agent with retrieve tool for RAG
agent = Agent(
    model=bedrock_model,
    tools=[retrieve]
)

if __name__ == "__main__":
    print("RAG対応Slackボットテスト開始")
    print(f"Knowledge Base ID: {KNOWLEDGE_BASE_ID}")
    print("-" * 50)
    
    # Test basic interaction
    result = agent("こんにちは！")
    print("基本応答:")
    print(result.message)
    print("-" * 50)
    
    # Test RAG functionality with retrieve tool
    rag_query = "｢夢のみち｣路愛護ポスターコンクール の応募締切日を教えてください。"
    rag_result = agent(rag_query)
    print("RAG応答:")
    print(rag_result.message)
```

実際に上記コードを実行してみた結果が以下になります。問題なくドキュメントの情報を回答できている事が分かるかと思います。

![スクリーンショット 2025-06-01 14.33.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/425ddef8-70a2-41a7-b52b-f855e555e378.png)


このように、Strands Agents で Knowledge bases での RAG を実現する場合、ナレッジベースIDを環境変数に設定さえしておけば、　エージェントの定義で ```tools=[retrieve]``` を追加するだけで良いので、かなり便利です。

:::note
**Tips:** retrieveツールに関して、上記のような利用する場合は、検索結果の採用数や、スコアの閾値といったパラメータを設定することが出来ないことに注意して下さい。
スコアの閾値に関しては環境変数に**MIN_SCORE**を設定すれば良いみたいですが、検索結果の採用数は環境変数で設定できないため、工夫が必要になりそうです。
:::

### MCPサーバーも利用できるようにしたい

すでに既存のSlack botをStrands Agentsに置き換えるには十分なのですが、せっかくなので**MCPサーバー**も利用できるようにしたいです。引き続きtest.pyを利用して、Claude Code と共に検証したいと思います。

Claude Codeに以下のような指示を投げてみました。

```:Claude Codeへの指示
test.pyに関して、Strands AgentsでRAGの実装まで完了しましたが、これに加えMCPサーバーも利用出来るようにしたいです。
利用するMCPサーバーはStandard I/Oタイプの想定で、コマンド（command）と引数（args）は環境変数等で指定できるようにして下さい。
```
同様にTODOリストを作成してタスクを実行し始めました。

![スクリーンショット 2025-05-29 11.34.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/8f8c6fbb-7b97-40df-a63f-0b3dc4826727.png)

Strands AgentsのMCPツール利用のドキュメントに自力でたどり着いたので、「Yes」を選択します。

![スクリーンショット 2025-05-29 11.35.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/f6c5b117-1162-461a-970c-e46176619e81.png)

https://strandsagents.com/0.1.x/user-guide/concepts/tools/mcp-tools/


過程は省略しますが、何度か Claude Code に追加の指示を出した後、test.pyの修正が完了しました。（Claude Codeだけではうまく対応できなかった箇所は人手で修正しています。）

![スクリーンショット 2025-05-29 16.42.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/6efeb1a0-2315-4b91-8302-69277f7c012d.png)

```Python:test.py（ところどころ人手で修正しています）
import os
import sys
import logging
from typing import Optional
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.tools.mcp import MCPClient
from strands_tools import retrieve
from mcp.client.stdio import stdio_client, StdioServerParameters
from dotenv import load_dotenv

load_dotenv()

# Enables Strands debug log level
logger = logging.getLogger("strands")

# Configuration
KNOWLEDGE_BASE_ID = os.getenv("KNOWLEDGE_BASE_ID", "YOUR_KNOWLEDGE_BASE_ID")
MCP_SERVER_COMMAND = os.getenv("MCP_SERVER_COMMAND", "uvx")
MCP_SERVER_ARGS = os.getenv("MCP_SERVER_ARGS", "awslabs.aws-documentation-mcp-server@latest")

# Knowledge Base ID (環境変数から取得)
logger.info(f"Knowledge Base ID: {KNOWLEDGE_BASE_ID}")
logger.info(f"MCP Server: {MCP_SERVER_COMMAND} {MCP_SERVER_ARGS}")


def create_mcp_client():
    # MCP client setup
    try:
        mcp_client = MCPClient(lambda: stdio_client(
            StdioServerParameters(
                command=MCP_SERVER_COMMAND, 
                args=MCP_SERVER_ARGS.split(",")
            )
        ))
        
        return mcp_client
    except Exception as e:
        logger.warning(f"Failed to connect to MCP server: {e}")
        logger.info("Continuing with RAG tools only")


def main():
    # Create a BedrockModel
    bedrock_model = BedrockModel(
        model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
        region_name='us-east-1'
    )


    with create_mcp_client() as mcp_client:
        # Base Tools
        tools = [retrieve]

        # Add MCP tools if available
        try:
            mcp_tools = mcp_client.list_tools_sync()
            tools.extend(mcp_tools)
            logger.info(f"Added {len(mcp_tools)} MCP tools")
        except Exception as e:
            logger.warning(f"Failed to list MCP tools: {e}")
            logger.info("Continuing with RAG tools only")

        # Create agent with RAG and MCP tools
        agent =  Agent(
            model=bedrock_model,
            tools=tools,
        )

        print("RAG対応Slackボットテスト開始")
        print(f"Knowledge Base ID: {KNOWLEDGE_BASE_ID}")
        print("-" * 50)
        
        # Test basic interaction
        result = agent("サクラメントの天気はどうですか？")
        print("MCPサーバー応答:")
        print(result.message)
        print("-" * 50)
        
        # Test RAG functionality with retrieve tool
        rag_query = "｢夢のみち｣路愛護ポスターコンクール の応募締切日を教えてください。"
        rag_result = agent(rag_query)
        print("RAG応答:")
        print(rag_result.message)
```

Strands Agents では、MCPクライアントを以下のように定義しています。今回はStandard I/Oタイプでの接続なので```stdio_client```を利用していますが、例えばStreamable HTTPでの接続の場合は```streamablehttp_client```を利用します。
```Python
mcp_client = MCPClient(lambda: stdio_client(
    # commandとargsは環境変数から読み込む
    StdioServerParameters(
        command=MCP_SERVER_COMMAND, 
        args=MCP_SERVER_ARGS.split(",")
    )
))
```

そして、エージェント作成の際には以下のように with文でMCPサーバーとのコネクションを張り、```list_tools_sync```でサーバーからツールを取得します。今回はすでに retrieve ツールがあるので、取得したツールを ```tools``` に追加する形になっています。

```Python
with create_mcp_client() as mcp_client:

    tools = [retrieve]

    mcp_tools = mcp_client.list_tools_sync()
    tools.extend(mcp_tools)

    agent =  Agent(
        model=bedrock_model,
        tools=tools,
    )
```

:::note
ちなみに、今回利用するMCPサーバーは、MCPの公式ドキュメントのQuickStartで実装したもので、**アメリカの気象予報や警報情報を調べるツール群**になります。詳しくは以下をご覧ください。

https://modelcontextprotocol.io/quickstart/server

:::


それでは test.py を実行して動作確認をしてみます。
まず最初にMCPサーバーの方の動作確認ですが、問題なくサクラメントの天気予報を回答してくれています。

![スクリーンショット 2025-06-01 15.59.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/0e0d51a8-f926-405b-a82e-00c9c80070bc.png)

次にRAGの動作確認ですが、こちらも問題なさそうですね！

![スクリーンショット 2025-06-01 15.59.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/6be040c5-8432-4286-b120-f9e9436091ed.png)



これで、Strands Agents の基本的な使い方に関して Claude Code と共に把握する事ができました！


:::note
**Tips:** MCPサーバーの利用にあたるコード修正に関して、Claude Codeが作成したコードを後から人手で直していますが、この後も同じファイルを Claude Code に操作させる場合は、一度**修正したコードを Claude Code に理解させておく**と、後続の処理の精度が体感上がったように感じたのでおすすめです。

```:Claude Codeへの指示例
test.pyをこちらで修正したので、その内容について理解しておいて下さい。必要があればweb検索を行っても構いません
```
:::



# Slack bot を Strands Agents 版にアップデートしよう

それではいよいよSlack botをアップデートしたいと思います。

元々のlangchainで実装したコードである app.py と、前節で作成したStrands Agentsのサンプルコードである test.py があるため、これらをマージするようなイメージで Claude Code に指示を出します。

```:Claude Codeへの指示
app.pyでSlack boltとLangChainを利用してRAG搭載のSlack botを実装しています。
先ほどまでtest.pyで実装してきたような Strands Agents を利用したRAG搭載かつMCPサーバを利用できるAIエージェントを搭載したslack botにアップグレードしたいです。
Slack boltによるSlack botの実装部分は基本的に踏襲しつつも、中身のエージェントはtest.pyで実装したものに置き換えるイメージです。
app.pyとtest.pyを比較しつつ、strands_app.pyというファイルでコードを作成してみて下さい。
```

:::note warn
**Tips:** 実はこのタイミングで Claude Code で以下のようなエラーが発生していました。これは、Anthropic APIのクレジットが切れたことが原因です。便利だからといって何も考えずに利用していると、結構お金が飛んでいきますので、ご利用は計画的に...

![スクリーンショット 2025-05-29 18.25.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/db3fcb39-b78b-45ad-9a46-ff72a7b100d9.png)

[MAXプラン](https://support.anthropic.com/en/articles/11145838-using-claude-code-with-your-max-plan)であれば、**$100/月**からという中々なお値段になりますが、Claude Codeが使い放題なので、よく利用する場合は検討の価値があります。
:::

app.pyのコードを読み込んで理解し、タスクを進めてくれています。

![スクリーンショット 2025-05-29 16.54.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/0c840e2c-61ad-4269-9766-089229dbbd8e.png)

一度コードが出来上がったのですが、レビューしてみると、MCPサーバーとのコネクション周り（with文）とストリーミング処理の実装が怪しかったので、参考になりそうなドキュメントを与えて再作成を依頼しました。

すると、一度の指摘のみで想定している動作をするコードを作成してくれました！
正直何往復かやり取りする覚悟でいたので、驚きです！

![スクリーンショット 2025-05-29 17.21.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/0937ed19-b62d-4bac-851a-cf520daa68f5.png)

作成されたコードが以下になります。

```Python:strands_app.py
import os
import asyncio
import logging
from typing import Dict, List, Optional
from dotenv import load_dotenv

# slack bolt周り
from slack_bolt import App
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from slack_bolt.async_app import AsyncApp

# Strands Agents周り
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.models.anthropic import AnthropicModel
from strands.tools.mcp import MCPClient
from strands_tools import retrieve
from mcp.client.stdio import stdio_client, StdioServerParameters

# Load environment variables from .env file
load_dotenv()

# 環境変数の設定
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_APP_TOKEN = os.environ["SLACK_APP_TOKEN"]  # Socket Modeに必要
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
KNOWLEDGE_BASE_ID = os.environ["KNOWLEDGE_BASE_ID"]  # Amazon Knowledge Base ID
MCP_SERVER_COMMAND = os.getenv("MCP_SERVER_COMMAND", "uvx")
MCP_SERVER_ARGS = os.getenv("MCP_SERVER_ARGS", "awslabs.aws-documentation-mcp-server@latest")
ANTHROPOC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "YOUR_ANTHROPIC_API_KEY")

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Slackアプリの初期化
app = AsyncApp(token=SLACK_BOT_TOKEN)

def create_mcp_client():
    """MCPクライアントを作成"""
    try:
        mcp_client = MCPClient(lambda: stdio_client(
            StdioServerParameters(
                command=MCP_SERVER_COMMAND, 
                args=MCP_SERVER_ARGS.split(",")
            )
        ))
        return mcp_client
    except Exception as e:
        logger.warning(f"Failed to connect to MCP server: {e}")
        return None

async def process_message_with_strands_streaming(text: str, channel_id: str, message_ts: str):
    """
    Strands Agentでメッセージを処理（ストリーミング対応）
    MCPクライアントのwith句制約を考慮した実装
    """
    # BedrockModel作成
    bedrock_model = BedrockModel(
        model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
        region_name=AWS_REGION
    )
    
    # ベースツール（RAG）
    tools = [retrieve]
    
    # MCPクライアントを作成
    mcp_client = create_mcp_client()
    
    if mcp_client:
        # MCPクライアントを介してMCPサーバーに接続し、処理を実行
        try:
            with mcp_client:
                # MCPツールを取得
                try:
                    mcp_tools = mcp_client.list_tools_sync()
                    tools.extend(mcp_tools)
                    logger.info(f"Added {len(mcp_tools)} MCP tools")
                except Exception as e:
                    logger.warning(f"Failed to get MCP tools: {e}")
                
                # エージェント作成
                agent = Agent(
                    # model=bedrock_model,
                    model=anthropic_model,
                    tools=tools
                )
                
                # ストリーミング処理
                full_response = ""
                update_interval = 1.0  # 更新間隔（秒）
                last_update = 0
                
                try:
                    # Strands Agentsの非同期ストリーミング
                    async for event in agent.stream_async(text):
                        if "data" in event:
                            # データイベントを追加
                            chunk = event["data"]
                            full_response += chunk
                            
                            # 一定間隔でSlackメッセージを更新
                            current_time = asyncio.get_event_loop().time()
                            if current_time - last_update >= update_interval and full_response.strip():
                                await app.client.chat_update(
                                    channel=channel_id,
                                    ts=message_ts,
                                    text=full_response
                                )
                                last_update = current_time
                        
                        elif "current_tool_use" in event:
                            # ツール使用イベントの表示
                            tool_name = event["current_tool_use"].get("name", "Unknown")
                            tool_status = f"\n🔧 Using tool: {tool_name}..."
                            
                            await app.client.chat_update(
                                channel=channel_id,
                                ts=message_ts,
                                text=full_response + tool_status
                            )
                    
                    # 最終応答を更新
                    if full_response.strip():
                        await app.client.chat_update(
                            channel=channel_id,
                            ts=message_ts,
                            text=full_response
                        )
                    else:
                        await app.client.chat_update(
                            channel=channel_id,
                            ts=message_ts,
                            text="応答を生成できませんでした。"
                        )
                    
                    return full_response
                    
                except Exception as e:
                    logger.error(f"Streaming error: {e}")
                    error_msg = f"ストリーミング処理中にエラーが発生しました: {str(e)}"
                    await app.client.chat_update(
                        channel=channel_id,
                        ts=message_ts,
                        text=error_msg
                    )
                    return error_msg
                    
        except Exception as e:
            logger.error(f"MCP connection error: {e}")
            # MCPなしで処理続行
            pass
    
    # MCPサーバーが利用できない場合はRAGのみで処理
    try:
        agent = Agent(
            model=bedrock_model,
            tools=[retrieve]
        )
        
        full_response = ""
        update_interval = 1.0
        last_update = 0
        
        async for event in agent.stream_async(text):
            if "data" in event:
                chunk = event["data"]
                full_response += chunk
                
                current_time = asyncio.get_event_loop().time()
                if current_time - last_update >= update_interval and full_response.strip():
                    await app.client.chat_update(
                        channel=channel_id,
                        ts=message_ts,
                        text=full_response
                    )
                    last_update = current_time
        
        if full_response.strip():
            await app.client.chat_update(
                channel=channel_id,
                ts=message_ts,
                text=full_response
            )
        else:
            await app.client.chat_update(
                channel=channel_id,
                ts=message_ts,
                text="応答を生成できませんでした。"
            )
        
        return full_response
        
    except Exception as e:
        logger.error(f"Agent processing error: {e}")
        error_msg = f"処理中にエラーが発生しました: {str(e)}"
        await app.client.chat_update(
            channel=channel_id,
            ts=message_ts,
            text=error_msg
        )
        return error_msg

# メンションされたときの処理
@app.event("app_mention")
async def handle_mention(event, say):
    """Botがメンションされたときの処理"""
    user_id = event["user"]
    channel_id = event["channel"]
    text = event["text"]
    
    # <@BOT_ID>の部分を削除してメッセージのみを取得
    message_text = text.split(">", 1)[1].strip() if ">" in text else text
    
    try:
        # スレッドで初期メッセージ送信
        response = await say(
            text="考え中...",
            thread_ts=event["ts"]
        )
        message_ts = response["ts"]
        
        # Strands Agentでストリーミング処理
        final_response = await process_message_with_strands_streaming(
            message_text, channel_id, message_ts
        )
        
    except Exception as e:
        logger.error(f"メンション処理エラー: {e}")
        await say(f"エラーが発生しました: {str(e)}")


# メイン関数
async def main():
    """アプリケーションのメイン関数"""
    logger.info("🚀 Starting Strands Agents Slack Bot...")
    logger.info(f"📚 Knowledge Base ID: {KNOWLEDGE_BASE_ID}")
    logger.info(f"🔌 MCP Server: {MCP_SERVER_COMMAND} {MCP_SERVER_ARGS}")
    logger.info(f"🌍 AWS Region: {AWS_REGION}")
    
    # AsyncSocketModeHandlerの初期化
    handler = AsyncSocketModeHandler(app, SLACK_APP_TOKEN)
    
    # アプリの起動
    await handler.start_async()

# スクリプトとして実行された場合
if __name__ == "__main__":
    asyncio.run(main())
```

それでは、Slack botを起動して実際に動作確認してみます。

コマンドラインを見る限り、起動自体は成功していますので、第一段階は突破です！

![スクリーンショット 2025-06-01 18.36.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/bdc721fc-2567-4613-99f5-683b19defe30.png)

Slack上でbotにメッセージを送ってみます。
とりあえず簡単な挨拶に対する返事は返ってきているのと、どのような機能を持っているかについての説明を見る限りは、正常にツールを読み込めていそうです。

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/24a6a240-6f6d-4a8c-8644-1f764a5e849a.png" width=600>

まずは、MCPサーバーの方の確認をします。先ほどと同様に、サクラメントの天気予報について聞いてみます。

ちなみにですが、ストリーミング表示もちゃんと出来ています。下画像はツール実行途中の出力になります。

![スクリーンショット 2025-06-01 18.33.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/9a0c9e12-f058-4b2d-99d2-5901ab868e5d.png)

サクラメントの天気予報が返ってきました！

![スクリーンショット 2025-06-01 18.34.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/5fcc8d13-91b1-45fe-b1dd-bd7f980ba2d1.png)

最後に、RAGの方の確認になります。今度は違う質問をしてみます。
旧芝離宮恩賜庭園で開催される「旧芝離宮夜会」の開催期間を聞いてみます。

![スクリーンショット 2025-06-02 1.17.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/dbb2c95c-9f40-42a3-84a4-29e135c22899.png)

正解は以下なのですが、開催期間に加え、他の付加情報に関しても誤りなく回答できていることが分かります！

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3026734/81a83273-1d3e-415c-a481-651a5b45cf52.png" width=350>


これで一通り動作確認は完了です！
無事にSlack botをアップデートすることが出来ました！！🎉








:::note
**Tips:** 今回はモデルに Claude 3.7 を利用していますが、もちろん **Claude Sonnet 4** を利用することも可能です。

```
model_id="us.anthropic.claude-sonnet-4-20250514-v1:0"
```

しかし、筆者が試した時点では BedrockModel 経由での呼び出しだとレスポンスが全然返ってこないという事象が発生していました。

もしそうなった場合は、以下のように**直接Anthropicのモデルを呼び出してみて下さい。** ただし、事前にAnthropicのAPIキーを払い出しておく必要があることに注意して下さい。

```Python
from strands.models.anthropic import AnthropicModel

anthropic_model = AnthropicModel(
    client_args={
        "api_key": ANTHROPOC_API_KEY,
    },
    # **model_config
    max_tokens=1028,
    model_id="claude-sonnet-4-20250514",
    params={
        "temperature": 1.0,
    }
)
```
:::



# 最後に

今回は、社内情報検索Slack botのアップデートの **Strands Agents** によるアップデート検証を、**Claude Code** を利用して実施した体験の記事でした。

**Strands Agents**は、とにかく手軽にエージェントを実装できるようになったのが驚きでした。AIエージェントへの入門時はもちろん、PoCなどで手早く動くものを作りたい時にも便利なのではと思います。もちろん、LangChain や Mastra AIといった他のフレームワークとは実装したいものの特性や状況によって使い分けになるかと思いますが、AWSの公式からSDKが提供されたというのはかなり大きいと思います。今後の展開に期待

**Claude Code** は実際に使ってみてその便利さと性能の高さに感動しました。こちらが出した指示を、人間と同じようにタスクに分解してを整理してくれますし、複雑なタスクでも、詳細に、段階に分けて指示すればかなり高精度なコードを生成してくれていました。Maxプランにしてこれからもガンガン使っていきたいです！

本記事はこれで以上になります。ご高覧ありがとうございました！！