AWS製のOSS SDK「Strands Agents」で始めるAIエージェント開発

brew install uv

mkdir StrandsAgents
cd StrandsAgents

# 仮想環境を作成
uv venv

# 仮想環境を有効化
source .venv/bin/activate

from strands import Agent

agent = Agent()
agent("こんにちは！自己紹介してください。")

python main.py

(StrandsAgents) ~/work/StrandsAgents $ python main.py
こんにちは！私はClaude（クロード）です。Anthropic社によって開発されたAIアシスタントです。

日本語での会話はもちろん、様々な質問にお答えしたり、文章の作成、翻訳、問題解決のお手伝いなど、幅広くサポートできます。分からないことがあれば正直にお伝えしますし、できる限り丁寧で役立つ回答を心がけています。

何かお手伝いできることがあれば、お気軽にお声がけください。今日はどのようなことでお話ししましょうか？

from strands import Agent
from strands.models.litellm import LiteLLMModel

GEMINI_API_KEY = "YOUR_GEMINI_API_KEY_HERE"

gemini_model = LiteLLMModel(
    model_id="gemini/gemini-1.5-flash-latest",
    params={
        "api_key": GEMINI_API_KEY
    }
)

agent = Agent(model=gemini_model)
response = agent("あなたの名前と開発元を教えてください。")
print(response.message)

(StrandsAgents) ~/work/StrandsAgents $ python gemini_agent.py

Provider List: https://docs.litellm.ai/docs/providers

こんにちは！私はGoogleによってトレーニングされた、大規模言語モデルです。日本語で会話したり、さまざまな質問に答えたり、さまざまな種類の創造的なテキストを作成したりすることができます。  私の知識は広範囲に渡りますが、常に学習中であるため、完璧ではありません。

簡単に言うと、私はあなたと情報を共有し、あなたを助けるためにここにいます。  何か質問があれば、遠慮なく聞いてください！


{'role': 'assistant', 'content': [{'text': 'こんにちは！私はGoogleによってトレーニングされた、大規模言語モデルです。日本語で会話したり、さまざまな質問に答えたり、さまざまな種類の創造的なテキストを作成したりすることができます。  私の知識は広範囲に渡りますが、常に学習中であるため、完璧ではありません。\n\n簡単に言うと、私はあなたと情報を共有し、あなたを助けるためにここにいます。  何か質問があれば、遠慮なく聞いてください！\n'}]}

from strands import Agent
from strands.session.s3_session_manager import S3SessionManager

BUCKET_NAME = "strandsagents-s3"
SESSION_ID = "interactive-chat-session-001"

session_manager = S3SessionManager(
    session_id=SESSION_ID,
    bucket=BUCKET_NAME,
)

agent = Agent(session_manager=session_manager)
print("AIとの会話を開始します。終了するには 'exit' と入力してください。")

while True:
    user_input = input("あなた: ")
    if user_input.lower() == 'exit':
        print("会話を終了します。")
        break

    response = agent(user_input)
    print(f"AI: {response.message}")

(StrandsAgents) ~/work/StrandsAgents $ python chat.py
AIとの会話を開始します。終了するには 'exit' と入力してください。
あなた: 私の好きなAWSサービスはAmazon S3です。覚えておいてください。
Amazon S3がお気に入りのAWSサービスだということを覚えておきますね。S3は確かに便利なオブジェクトストレージサービスですね。

ただし、私は会話が終了すると記憶をリセットされるため、次回新しい会話を始められた際には、この情報を覚えていることができません。新しい会話では改めて教えていただく必要があります。

この会話の中では、S3についてのご質問や関連する話題があれば、お気に入りのサービスということを踏まえてお答えできます。S3について何かお聞きになりたいことはありますか？AI: {'role': 'assistant', 'content': [{'text': 'Amazon S3がお気に入りのAWSサービスだということを覚えておきますね。S3は確かに便利なオブジェクトストレージサービスですね。\n\nただし、私は会話が終了すると記憶をリセットされるため、次回新しい会話を始められた際には、この情報を覚えていることができません。新しい会話では改めて教えていただく必要があります。\n\nこの会話の中では、S3についてのご質問や関連する話題があれば、お気に入りのサービスということを踏まえてお答えできます。S3について何かお聞きになりたいことはありますか？'}]}
あなた: exit
会話を終了します。
(StrandsAgents) ~/work/StrandsAgents $

(StrandsAgents) ~/work/StrandsAgents $ python chat.py
AIとの会話を開始します。終了するには 'exit' と入力してください。
あなた: 私が好きなAWSサービスは何でしたか？
あなたがお気に入りのAWSサービスはAmazon S3でしたね。先ほど教えていただいたばかりです。

S3について何かご質問がございましたら、お聞かせください。AI: {'role': 'assistant', 'content': [{'text': 'あなたがお気に入りのAWSサービスはAmazon S3でしたね。先ほど教えていただいたばかりです。\n\nS3について何かご質問がございましたら、お聞かせください。'}]}
あなた: exit
会話を終了します。
(StrandsAgents) ~/work/StrandsAgents $

{
  "message": {
    "role": "user",
    "content": [
      {
        "text": "私の好きなAWSサービスはAmazon S3です。覚えておいてください。"
      }
    ]
  },
  "message_id": 0,
  "redact_message": null,
  "created_at": "2025-09-24T02:40:26.998554+00:00",
  "updated_at": "2025-09-24T02:40:26.998562+00:00"
}

{
  "message": {
    "role": "assistant",
    "content": [
      {
        "text": "Amazon S3がお気に入りのAWSサービスだということを覚えておきますね。S3は確かに便利なオブジェクトストレージサービスですね。\n\nただし、私は会話が終了すると記憶をリセットされるため、次回新しい会話を始められた際には、この情報を覚えていることができません。新しい会話では改めて教えていただく必要があります。\n\nこの会話の中では、S3についてのご質問や関連する話題があれば、お気に入りのサービスということを踏まえてお答えできます。S3について何かお聞きになりたいことはありますか？"
      }
    ]
  },
  "message_id": 1,
  "redact_message": null,
  "created_at": "2025-09-24T02:40:34.971803+00:00",
  "updated_at": "2025-09-24T02:40:34.971815+00:00"
}

{
  "message": {
    "role": "user",
    "content": [
      {
        "text": "私が好きなAWSサービスは何でしたか？"
      }
    ]
  },
  "message_id": 2,
  "redact_message": null,
  "created_at": "2025-09-24T02:41:00.722950+00:00",
  "updated_at": "2025-09-24T02:41:00.722973+00:00"
}

{
  "message": {
    "role": "assistant",
    "content": [
      {
        "text": "あなたがお気に入りのAWSサービスはAmazon S3でしたね。先ほど教えていただいたばかりです。\n\nS3について何かご質問がございましたら、お聞かせください。"
      }
    ]
  },
  "message_id": 3,
  "redact_message": null,
  "created_at": "2025-09-24T02:41:03.592281+00:00",
  "updated_at": "2025-09-24T02:41:03.592292+00:00"
}

uv pip install strands-agents-tools

from strands import Agent
from strands_tools import use_aws

aws_agent = Agent(
    system_prompt="あなたはAWSリソースを調査する専門家です。'use_aws'ツールを使って、ユーザーの指示に正確に従ってください。",
    tools=[use_aws]
)

response = aws_agent("東京リージョンにあるS3一覧を教えて下さい。")

print(f"AI: {response.message}")

from strands import Agent
from strands_tools import use_aws

def final_response_handler(event: AgentEvent) -> bool:
    return event.event_type == EventType.FINAL_RESPONSE

IMPROVED_SYSTEM_PROMPT = """
あなたは、AWSリソースを調査する非常に几帳面で正確な専門家です。'use_aws'ツールのみを使い、ユーザーの指示に完璧に従ってください。

## S3バケットのリージョンを特定する際の厳格なルール
S3の'list-buckets'はグローバルな結果を返すため、各バケットのリージョンは'get-bucket-location'で個別に確認する必要があります。
ユーザーから特定リージョンのバケット一覧を求められた場合、以下の思考プロセスと手順を**絶対に**守ってください。

### 思考プロセス
1.  まず'use_aws'ツールで's3 list-buckets'を実行し、アカウント内の**すべての**バケット名を取得する。
2.  取得した**すべての**バケット名のリストに対して、ループ処理を行う。省略は許されない。
3.  ループの中で、バケット名一つひとつに対して'use_aws'ツールで's3 get-bucket-location'を実行し、そのバケットのリージョンを特定する。
4.  ステップ3で特定したリージョンが、ユーザーの指定したリージョンと一致するものだけをリストとしてまとめる。
5.  最終的に、ステップ4で作成したリストが完全であることを確認し、ユーザーに報告する。
"""

aws_agent = Agent(
    system_prompt=IMPROVED_SYSTEM_PROMPT,
    tools=[use_aws]
)

response = aws_agent("東京リージョンにあるS3バケットを一覧表示して")

print(f"AI: {response.message}")

uv pip install uvx

from mcp import stdio_client, StdioServerParameters
from strands import Agent
from strands.tools.mcp import MCPClient

aws_doc_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(
        command="uvx",
        args=["awslabs.aws-documentation-mcp-server@latest"]
    )
))

with aws_doc_mcp_client:
    print("MCPサーバーからツールを取得しています...")
    tools = aws_doc_mcp_client.list_tools_sync()
    
    print(f"取得したツール: {[tool.spec.name for tool in tools]}")

    agent = Agent(
        system_prompt="あなたはAWSの専門家です。AWS公式ドキュメントを検索できるツールを使い、質問に正確に答えてください。",
        tools=tools
    )
    
    response = agent("Amazon S3の耐久性について、公式ドキュメントに基づいて教えて")
    print(f"\nAI: {response.message}")

(StrandsAgents) ~/work/StrandsAgents $ python mcp_agent.py
MCPサーバーからツールを取得しています...
[09/24/25 13:15:13] INFO     Processing request of type ListToolsRequest                                                                                                                                                                                           server.py:623
取得したツール: ['read_documentation', 'search_documentation', 'recommend']
Amazon S3の耐久性について、公式ドキュメントを検索して詳細な情報をお調べします。
Tool #1: search_documentation
[09/24/25 13:15:16] INFO     Processing request of type CallToolRequest                                                                                                                                                                                            server.py:623
[09/24/25 13:15:18] INFO     HTTP Request: POST https://proxy.search.docs.aws.amazon.com/search?session=92b5748a-fd14-4a40-92b3-bc0a0aee3082 "HTTP/1.1 200 OK"                                                                                                   _client.py:1740
「Data protection in Amazon S3」が耐久性に関する情報を含んでいそうです。詳しく確認してみましょう。
Tool #2: read_documentation
[09/24/25 13:15:21] INFO     Processing request of type CallToolRequest                                                                                                                                                                                            server.py:623
                    INFO     HTTP Request: GET https://docs.aws.amazon.com/AmazonS3/latest/userguide/DataDurability.html?session=92b5748a-fd14-4a40-92b3-bc0a0aee3082 "HTTP/1.1 200 OK"                                                                          _client.py:1740
さらに詳しい情報を得るために、ストレージクラスに関するページも確認してみましょう。
Tool #3: read_documentation
[09/24/25 13:15:24] INFO     Processing request of type CallToolRequest                                                                                                                                                                                            server.py:623
                    INFO     HTTP Request: GET https://docs.aws.amazon.com/AmazonS3/latest/userguide/storage-class-intro.html?session=92b5748a-fd14-4a40-92b3-bc0a0aee3082 "HTTP/1.1 200 OK"                                                                     _client.py:1740
それでは、Amazon S3の耐久性について、公式ドキュメントから得られた情報をまとめてお答えします。

## Amazon S3の耐久性（Durability）について

### 高い耐久性の設計
Amazon S3は、**99.999999999%（11個の9）の耐久性**を提供するように設計されています。これは、年間で1,000万個のオブジェクトを保存した場合に、平均して1個のオブジェクトを失う可能性があるという意味です。

### 冗長化による耐久性の確保

#### 1. **マルチAZ冗長化**
- S3 Standard、S3 Intelligent-Tiering、S3 Standard-IA、S3 Glacier Instant Retrieval、S3 Glacier Flexible Retrieval、S3 Glacier Deep Archiveは、**最低3つのアベイラビリティゾーン（AZ）**にまたがって冗長的にデータを保存
- 各アベイラビリティゾーンは物理的に数キロメートル離れた独立したデータセンター
- **1つのAZ全体が失われても**データを保持可能

#### 2. **デバイスレベルの冗長化**
- 各AZ内で**複数のデバイス**にデータを冗長保存
- デバイス障害を迅速に検出し、失われた冗長性を素早く修復
- 定期的にチェックサムを使用してデータの整合性を検証

### ストレージクラスによる耐久性の違い

#### 高耐久性クラス
- **S3 Standard**: 99.999999999%の耐久性
- **S3 Intelligent-Tiering**: 99.999999999%の耐久性
- **その他のGlacierクラス**: すべて99.999999999%の耐久性

#### 特別なケース
- **S3 One Zone-IA**: 単一のAZ内でのみ冗長化（コスト削減のため）
- **S3 Express One Zone**: 単一のAZ内で冗長化、最高性能を提供
- **Reduced Redundancy Storage (RRS)**: 年間0.01%のオブジェクト損失率（**推奨されません**）

### 追加の保護機能

#### 1. **バージョニング**
- オブジェクトのすべてのバージョンを保持
- 意図しない削除や変更からの保護
- アプリケーション障害からの復旧支援

#### 2. **その他の保護機能**
- **S3 Object Lock**: オブジェクトの変更・削除を防止
- **S3 Replication**: 別のリージョンへの複製
- **データ整合性の監視**: チェックサムによる定期検証

### まとめ
Amazon S3は、複数のAZにまたがる冗長化、デバイスレベルの冗長化、継続的な監視・修復により、**99.999999999%という極めて高い耐久性**を実現しています。これにより、ミッションクリティカルなデータの長期保存に適した信頼性の高いストレージサービスとなっています。

import boto3
from mcp import stdio_client, StdioServerParameters
from strands import Agent, tool
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient
from strands_tools import use_aws

# --- 小エージェント1: AWSリソース操作担当 ---
@tool
def aws_operator(query: str) -> str:
    """
    ユーザーの実際のアカウントを操作し、AWSリソースの現在の状態を取得する場合に使用します。
    (例: 「S3の一覧を教えて」)
    """
    IMPROVED_SYSTEM_PROMPT = """
    あなたは、AWSリソースを調査する非常に几帳面で正確な専門家です。'use_aws'ツールのみを使い、ユーザーの指示に完璧に従ってください。

    ## S3バケットのリージョンを特定する際の厳格なルール
    S3の'list-buckets'はグローバルな結果を返すため、各バケットのリージョンは'get-bucket-location'で個別に確認する必要があります。
    ユーザーから特定リージョンのバケット一覧を求められた場合、以下の思考プロセスと手順を**絶対に**守ってください。

    ### 思考プロセス
    1.  まず'use_aws'ツールで's3 list-buckets'を実行し、アカウント内の**すべての**バケット名を取得する。
    2.  取得した**すべての**バケット名のリストに対して、ループ処理を行う。省略は許されない。
    3.  ループの中で、バケット名一つひとつに対して'use_aws'ツールで's3 get-bucket-location'を実行し、そのバケットのリージョンを特定する。
    4.  ステップ3で特定したリージョンが、ユーザーの指定したリージョンと一致するものだけをリストとしてまとめる。
    5.  最終的に、ステップ4で作成したリストが完全であることを確認し、ユーザーに報告する。
    """
    agent = Agent(
        system_prompt=IMPROVED_SYSTEM_PROMPT,
        tools=[use_aws]
    )
    return agent(query).message

# --- 小エージェント2: AWSドキュメント調査担当 ---
aws_doc_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(command="uvx", args=["awslabs.aws-documentation-mcp-server@latest"])
))
@tool
def aws_investigator(query: str) -> str:
    """AWSのサービス仕様やベストプラク-ティス、料金など、一般的な知識について調べる場合に使用します。"""
    with aws_doc_mcp_client:
        tools = aws_doc_mcp_client.list_tools_sync()
        agent = Agent(
            system_prompt="あなたはAWSドキュメント調査の専門家です。取得したツールを使い、質問に正確に答えてください。",
            tools=tools
        )
        return agent(query).message

# --- スーパーバイザーエージェントの定義---
supervisor = Agent(
    system_prompt="""
    あなたはAWS調査を監督するスーパーバイザーです。ユーザーの質問を分析し、最も適切なツール（aws_operatorかaws_investigator）を一つだけ選択して実行してください。
    """,
    tools=[aws_operator, aws_investigator]
)

# --- 対話ループの実行部分---
if __name__ == "__main__":
    print("AWS統合調査エージェントを起動しました。終了するには 'exit' と入力してください。")
    while True:
        user_input = input("あなた: ")
        if user_input.lower() == 'exit':
            print("エージェントを終了します。")
            break
        
        response = supervisor(user_input)
        print(f"スーパーバイザー:\n{response.message}\n")