AWSのエージェントフレームワーク「Strands Agents」 のコンセプト ②Tools

{
    "name": "Python 3",
    "image": "mcr.microsoft.com/devcontainers/python:1-3.12-bookworm",
    "features": {
        "ghcr.io/devcontainers/features/aws-cli:1": {}
    },
    "workspaceFolder": "/workspace", 
    "workspaceMount": "source=${localWorkspaceFolder},target=/workspace,type=bind,consistency=cached",
    "mounts": [
        "source=${localEnv:HOME}/.aws,target=/home/vscode/.aws,type=bind,consistency=cached"
    ]
}

pip install strands-agents strands-agents-tools strands-agents-builder
pip freeze | grep -i strands

strands-agents==1.0.1
strands-agents-builder==0.1.7
strands-agents-tools==0.2.1

aws bedrock-runtime converse \
    --model-id 'us.anthropic.claude-3-5-haiku-20241022-v1:0' \
    --messages '[{"role": "user", "content": [{"text": "こんにちは！"}]}]' \
    --region us-east-1 \
    | jq -r .output.message.content[0].text

こんにちは！お手伝いできることがありましたら、どうぞおっしゃってください。

export AWS_DEFAULT_REGION=us-east-1

from strands import Agent
from strands_tools import calculator, file_read, shell

# エージェントにツールを追加
agent = Agent(
    tools=[calculator, file_read, shell]
)

# エージェントが自動的にcalculatorツールを使用するタイミングを判断する
agent("42 ^ 9 は？")

print("\n\n")  # 改行

# エージェントが自動的にshellとfile_readツールを使用するタイミングを判断する
agent("このディレクトリ二どんなファイルがあるかを確認して、その中のファイルを1つ表示してください")

42の9乗を計算しますね。
Tool #1: calculator
42^9 = **406,671,383,849,472** です。

これは約40兆6,671億の値になります。


このディレクトリにあるファイルを確認して、1つを表示しますね。
Tool #2: file_read
では、この中から`tool.py`ファイルの内容を表示します。
Tool #3: file_read
`tool.py`ファイルの内容を表示しました。このファイルは、Strandsエージェントの使用例を示すPythonスクリプトです。

ファイルの概要：
- `strands`ライブラリからAgentクラスをインポート
- `strands_tools`から3つのツール（calculator、file_read、shell）をインポート
- これらのツールを持つエージェントを作成
- 2つのクエリ例を実行：
  1. 数学計算（42^9）
  2. ファイル表示（現在実行中のクエリ）

このスクリプトは、エージェントが適切なツールを自動的に選択して使用する能力を示しています。

# ツール名
print(agent.tool_names)  # -> ['calculator', 'file_read', 'shell']
# ツールの説明や入力値などの設定
print(agent.tool_config)  # -> AttributeError: 'Agent' object has no attribute 'tool_config'

agent = Agent(tools=["/path/to/my_tool.py"])

from strands import Agent

agent = Agent(load_tools_from_directory=True)

from strands import Agent
from strands_tools import calculator, file_read, shell

agent = Agent(
    tools=[file_read, shell]
)

# 自然言語でツールを呼び出し
agent("カレントディレクトリ内のtool.pyファイルの中身を読んで。")

print("\n" + "-" * 20 + "\n")  # 改行

# ツールをエージェントのメソッドで直接呼び出し
result = agent.tool.file_read(path="./tool.py", mode="view")
print(result["content"][0]["text"])

カレントディレクトリ内のtool.pyファイルの中身を読み取ります。
Tool #1: file_read
tool.pyファイルの中身を読み取りました。このファイルはStrandsというライブラリを使用したエージェントのデモンストレーションのようですね。

ファイルの内容は以下の通りです：

1. **インポート部分**：
   - `strands`から`Agent`をインポート
   - `strands_tools`から`calculator`、`file_read`、`shell`をインポート

2. **エージェントの作成**：
   - `file_read`と`shell`ツールを持つエージェントを作成

3. **使用例**：
   - 自然言語でのツール呼び出し（まさに今のリクエストと同じ内容）
   - 区切り線の出力
   - エージェントのメソッドを使った直接的なツール呼び出し

このコードは、Strandsライブラリでエージェントを作成し、ファイル読み取り機能を自然言語とプログラム的な方法の両方で使用する方法を示しているサンプルコードのようです。
--------------------

Content of ./tool.py:
from strands import Agent
from strands_tools import calculator, file_read, shell

agent = Agent(
    tools=[file_read, shell]
)

# 自然言語でツールを呼び出し
agent("カレントディレクトリ内のtool.pyファイルの中身を読んで。")

print("\n" + "-" * 20 + "\n")  # 改行

# ツールをエージェントのメソッドで直接呼び出し
result = agent.tool.file_read(path="./tool.py", mode="view")
print(result["content"][0]["text"])

import asyncio
from strands import Agent, tool


@tool
def get_user_location() -> str:
    """ユーザの位置情報を取得する。"""

    # 実際はここにユーザの位置情報を取得するロジックを実装する
    return "兵庫県神戸市"


@tool
def weather(location: str) -> str:
    """指定された場所の天気情報を取得する。

    Args:
        location: 都市名または場所名
    """

    # 実際はここに天気情報を取得するロジックを実装する
    return f"{location} の天気は晴れです。気温は25度です。"


@tool
async def call_api() -> str:
    """非同期でAPIを呼び出す。

    Strandsは非同期ツールを並行して呼び出すことができる。
    """

    await asyncio.sleep(5)  # APIコールをシミュレート
    return "APIを呼び出しました。結果は...です。"


def basic_example():
    agent = Agent(tools=[get_user_location, weather])
    agent("今の天気はどうですか？")


async def async_example():
    agent = Agent(tools=[call_api])
    await agent.invoke_async("APIを呼び出してください。")


def main():
    basic_example()
    asyncio.run(async_example())


if __name__ == "__main__":
    main()

現在の天気をお調べします。まず、あなたの位置情報を取得させていただきます。
Tool #1: get_user_location
位置情報が取得できました。神戸市の現在の天気をお調べします。
Tool #2: weather
神戸市の現在の天気は**晴れ**で、気温は**25度**です。とても過ごしやすい天気ですね！APIを呼び出します。
Tool #1: call_api
APIの呼び出しが完了しました。結果は「...です。」と返ってきました。何か他にお手伝いできることはありますか？

from typing import Any
from strands.types.tools import ToolResult, ToolUse

TOOL_SPEC = {
    "name": "get_user_location",
    "description": "ユーザの位置情報を取得する。",
    "inputSchema": {
        "json": {
            "type": "object",
            "properties": {},
            "required": []
        }
    }
}

def get_user_location(tool: ToolUse, **kwargs: Any) -> ToolResult:
    tool_use_id = tool["toolUseId"]

    # 実際はここに位置情報を取得するロジックを実装する
    location_info = "兵庫県神戸市"

    return {
        "toolUseId": tool_use_id,
        "status": "success",
        "content": [{"text": location_info}]
    }

from typing import Any
from strands.types.tools import ToolResult, ToolUse

TOOL_SPEC = {
    "name": "weather",
    "description": "指定された場所の天気情報を取得する。",
    "inputSchema": {
        "json": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "都市名または場所名"
                }
            },
            "required": ["location"]
        }
    }
}

# 関数名はツール名と一致している必要がある
# デコレーターを使用したツールと同様に、非同期で定義することもできる
def weather(tool: ToolUse, **kwargs: Any) -> ToolResult:
    tool_use_id = tool["toolUseId"]
    location = tool["input"]["location"]

    # 実際はここに天気情報を取得するロジックを実装する
    weather_info = f"{location} の天気は晴れです。気温は25度です。"

    return {
        "toolUseId": tool_use_id,
        "status": "success",
        "content": [{"text": weather_info}]
    }

from strands import Agent
import get_user_location
import weather

# ツールはPythonモジュールのインポートでエージェントに追加できる
agent = Agent(tools=[get_user_location, weather])

# エージェントを使用してツールを呼び出す
agent("今の天気はどうですか？")

現在の天気をお調べします。まず、あなたの現在地を取得させていただきますね。
Tool #1: get_user_location
神戸市の現在の天気を確認いたします。
Tool #2: weather
神戸市の現在の天気は**晴れ**で、気温は**25度**です。とても過ごしやすい気候ですね！

from strands import Agent

# ツールはファイルパス文字列でも追加できる
agent = Agent(tools=["./get_user_location.py", "./weather.py"])

agent("今の天気はどうですか？")

現在の天気をお調べするために、まずあなたの位置情報を取得させていただきます。
Tool #1: get_user_location
それでは、神戸市の現在の天気情報を取得いたします。
Tool #2: weather
神戸市の現在の天気は**晴れ**で、気温は**25度**です。とても過ごしやすい天気ですね！

.
├── agent.py
└── tools
    ├── get_user_location.py
    └── weather.py

from strands import Agent

agent = Agent(load_tools_from_directory=True)

agent("今の天気はどうですか？")

現在の天気をお調べするために、まずあなたの位置情報を取得させていただきます。
Tool #1: get_user_location
それでは神戸市の現在の天気情報を取得いたします。
Tool #2: weather
現在の神戸市の天気は**晴れ**で、気温は**25度**です。とても過ごしやすい天気ですね！

from mcp.client.sse import sse_client
from strands import Agent
from strands.tools.mcp import MCPClient

# MCPサーバーにSSEトランスポートを使用して接続
sse_mcp_client = MCPClient(lambda: sse_client("http://localhost:8000/sse"))

# MCPツールを使用してエージェントを作成
with sse_mcp_client:
    # MCPサーバーからツールを取得
    tools = sse_mcp_client.list_tools_sync()

    # MCPサーバーのツールを使用してエージェントを作成
    agent = Agent(tools=tools)

    # MCPツールを使用してエージェントを使用
    agent("144の平方根を計算して。")

@tool
def search_database(query: str, max_results: int = 10) -> list:
    """
    商品データベースから、クエリ文字列に一致する商品を検索する。

    キーワード、商品名、カテゴリなどに基づいて、詳細な商品情報を検索する必要がある場合にこのツールを
    使用してください。検索は、大文字小文字を区別せず、検索用語の誤字や変形を処理するために、あいまい
    検索をサポートしています。

    このツールは、企業の商品カタログデータベースに接続し、すべての商品フィールドにわたってセマンティック
    検索を実行し、すべての利用可能な商品メタデータを含む包括的な結果を提供します。

    例:
        [
            {
                "id": "P12345",
                "name": "ウルトラ・コンフォート・ランニング・シューズ",
                "description": "軽量なランニングシューズで、・・・",
                "price": 89.99,
                "category": ["靴", "アスレチック", "ランニング"]
            },
            ...
        ]

    注意:
        - このツールは、商品カタログのみを検索し、在庫や可用性の情報は提供しません。
        - 結果は、パフォーマンス向上のため、15分間キャッシュされます。
        - 検索インデックスは、6時間ごとに更新されるため、直近の新しい商品は表示されない場合があります。
        - 在庫状況の場合は、別の在庫確認ツールを使用してください。

    Args:
        query: 検索文字列 (商品名、カテゴリ、キーワードなど)
               例: "赤いランニングシューズ" または "スマートフォン充電器"
        max_results: 返される結果の最大数 (デフォルト: 10, 範囲: 1-100)
                     完全一致の場合は、より低い値を使用して、より迅速な応答を得ることができます。

    Returns:
        一致する商品レコードのリスト、各レコードには次のフィールドが含まれます:
        - id: 一意の商品識別子 (文字列)
        - name: 商品名 (文字列)
        - description: 詳細な商品説明 (文字列)
        - price: 現在の価格 (USD) (float)
        - category: 商品カテゴリ階層 (リスト)
    """

    # 実際の実装
    pass

from strands import tool

@tool
def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    # 実際の実装を書く
    return f"{city} の 今後 {days} 日間の天気予報は、..."

from strands import Agent, tool

@tool
def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    # 実際の実装を書く
    return f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

agent = Agent(
    tools=[weather_forecast]
)

agent("今週一週間の東京の天気予報を教えて。")

今週一週間の東京の天気予報をお調べします。
Tool #1: weather_forecast
今週一週間（7日間）の東京の天気予報をお調べしました。

**東京の7日間天気予報**
概ね晴れが続く見込みです。

比較的安定した晴天が期待できそうですね。外出予定を立てる際の参考にしてください。ただし、詳細な気温や時間帯別の変化については、お出かけ前に最新の天気予報もご確認いただくことをおすすめします。

from strands import Agent, tool

@tool(
    name="get_weather_forecast_for_city_with_days",
    description="指定された都市の天気予報を取得する。日数も指定可能。"
)
def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    return f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

agent = Agent(
    tools=[weather_forecast]    # 指定する際は関数名のまま
)

agent("今週一週間の東京の天気予報を教えて。")

東京の1週間の天気予報を取得いたします。
Tool #1: get_weather_forecast_for_city_with_days
東京の今週1週間（7日間）の天気予報は、**概ね晴れが続く**見込みです。

穏やかで過ごしやすい天気が続きそうですね。外出や洗濯などの予定も立てやすそうです！

from strands import Agent, tool

@tool
def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    try:
        # 実際の実装を書く
        data = {"city": city, "days": days, "weather_details": "概ね晴れ"}
        return {
            "status": "success",
            "content": [ {
                "json": data,
            }]
        }
    except Exception as e:
        return {
            "status": "error",
             "content": [
                {"text": f"Error:{e}"}
            ]
        }

agent = Agent(
    tools=[weather_forecast]
)

agent("今週一週間の東京の天気予報を教えて。")

from strands import Agent, tool
import asyncio

@tool
async def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    # 実際の実装を書く
    await asyncio.sleep(5)
    return f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

async def main():
    agent = Agent(
        tools=[weather_forecast]
    )
    await agent.invoke_async("今週一週間の東京の天気予報を教えて。")

if __name__ == "__main__":
    asyncio.run(main())

from typing import Any

# 1. ツール仕様
TOOL_SPEC = {
    "name": "weather_forecast",
    "description": "指定された都市の天気予報を取得する。",
    "inputSchema": {
        "json": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "都市名"
                },
                "days": {
                    "type": "integer",
                    "description": "予報を取得する日数",
                    "default": 3
                }
            },
            "required": ["city"]
        }
    }
}

# 2. Tool Function
def weather_forecast(tool, **kwargs: Any):
    # ツールパラメータを抽出
    tool_use_id = tool["toolUseId"]
    tool_input = tool["input"]

    # パラメータの値を取得
    city = tool_input.get("city", "")
    days = tool_input.get("days", 3)

    # ツールの実装
    result = f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

    # 構造化された応答を返す
    return {
        "toolUseId": tool_use_id,
        "status": "success",
        "content": [{"text": result}]
    }

from strands import Agent
# モジュールをインポート
import weather_forecast

# エージェントにツールを与えて初期化
agent = Agent(tools=[weather_forecast])

agent("今週一週間の東京の天気予報を教えて。")

東京の1週間（7日間）の天気予報を取得いたします。
Tool #1: weather_forecast
東京の今週1週間の天気予報は、概ね晴れが続く予報となっています。

全体的に良好な天気が期待できそうですね。お出かけやお仕事の予定を立てる際の参考にしていただければと思います。

from strands import Agent

agent = Agent(tools=["./weather_forecast.py"])

agent("今週一週間の東京の天気予報を教えて。")

from typing import Any
import asyncio

TOOL_SPEC = {
    "name": "weather_forecast",
    "description": "指定された都市の天気予報を取得する。",
    "inputSchema": {
        "json": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "都市名"
                },
                "days": {
                    "type": "integer",
                    "description": "予報を取得する日数",
                    "default": 3
                }
            },
            "required": ["city"]
        }
    }
}

async def weather_forecast(tool, **kwargs: Any):
    # ツールパラメータを抽出
    tool_use_id = tool["toolUseId"]
    tool_input = tool["input"]

    # パラメータの値を取得
    city = tool_input.get("city", "")
    days = tool_input.get("days", 3)

    # 擬似的な非同期処理
    await asyncio.sleep(5)

    # ツールの実装
    result = f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

    # 構造化された応答を返す
    return {
        "toolUseId": tool_use_id,
        "status": "success",
        "content": [{"text": result}]
    }

{
    "toolUseId": str,       # ツール使用リクエストのID (入力リクエストと一致する必要がある)。 オプション。
    "status": str,          # "success" または "error"
    "content": List[dict]   # 異なる形式のコンテンツ項目のリスト
}

{
    "toolUseId": "tool-123",
    "status": "success",
    "content": [
        {"text": "操作が正常に完了しました"},
        {"json": {"results": [1, 2, 3], "total": 3}}
    ]
}

{
    "toolUseId": "tool-123",
    "status": "error",
    "content": [
        {"text": "エラー: パラメータが無効です。リクエストを処理できませんでした。"}
    ]
}

(snip)

def weather_forecast(tool, **kwargs: Any):
    tool_use_id = tool["toolUseId"]
    tool_input = tool["input"]

    city = tool_input.get("city", "")
    days = tool_input.get("days", 3)

    result = f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

    return {
        "toolUseId": tool_use_id,
        "status": "success",
        "content": [{"text": result}]
    }

@tool
def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    # 実際の実装を書く
    return f"{city} の 今後 {days} 日間の天気予報は、概ね晴れが続くでしょう。"

@tool
def weather_forecast(city: str, days: int = 3) -> str:
    """指定された都市の天気予報を取得する。

    Args:
        city: 都市名
        days: 取得する予報の日数
    """
    try:
        # 実際の実装を書く
        data = {"city": city, "days": days, "weather_details": "概ね晴れ"}
        return {
            "status": "success",
            "content": [ {
                "json": data,
            }]
        }
    except Exception as e:
        return {
            "status": "error",
             "content": [
                {"text": f"Error:{e}"}
            ]
        }

curl -LsSf https://astral.sh/uv/install.sh | sh

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

# stdioトランスポートを使ってMCPサーバに接続する
# 注意:
# - uvxコマンドが必要
# - uvxコマンドの構文はプラットフォームによって異なる

# Linux/Macの場合。今回はLinuxコンテナなのでこちら
stdio_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(
        command="uvx", 
        args=["awslabs.aws-documentation-mcp-server@latest"]
    )
))

# Windowsの場合はこちら
#stdio_mcp_client = MCPClient(lambda: stdio_client(
#    StdioServerParameters(
#        command="uvx", 
#        args=[
#            "--from", 
#            "awslabs.aws-documentation-mcp-server@latest", 
#            "awslabs.aws-documentation-mcp-server.exe"
#        ]
#    )
#))

# MCPツールを使用したエージェントを作成
with stdio_mcp_client:
    # MCPサーバからツールを取得
    tools = stdio_mcp_client.list_tools_sync()

    # MCPツールを使用したエージェントを作成
    agent = Agent(tools=tools)
    agent("Amazon Bedrock AgentCoreについて教えて。")

Installed 38 packages in 16ms
[07/25/25 00:24:51] INFO     Processing request of type ListToolsRequest                                             server.py:625
Amazon Bedrock AgentCoreについて調べてみますね。まず関連する情報を検索してみます。
Tool #1: search_documentation
[07/25/25 00:24:54] INFO     Processing request of type CallToolRequest                                              server.py:625
[07/25/25 00:24:55] INFO     HTTP Request: POST                                                                    _client.py:1740
                             https://proxy.search.docs.aws.amazon.com/search?session=XXXXXXXX-XXXX-XXXX-XXXX-XXXX                
                             XXXXXXXX "HTTP/1.1 200 OK"                                                                            
メインのドキュメントページを確認してみましょう。
Tool #2: read_documentation
[07/25/25 00:25:08] INFO     Processing request of type CallToolRequest                                              server.py:625
[07/25/25 00:25:09] INFO     HTTP Request: GET                                                                     _client.py:1740
                             https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/what-is-bedrock-agentco                
                             re.html?session=XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX "HTTP/1.1 200 OK"                               
続いて、AgentCore Runtimeについてより詳しく確認してみましょう。
Tool #3: read_documentation
[07/25/25 00:25:25] INFO     Processing request of type CallToolRequest                                              server.py:625
[07/25/25 00:25:26] INFO     HTTP Request: GET                                                                     _client.py:1740
                             https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/agents-tools-runtime.ht                
                             ml?session=XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX "HTTP/1.1 200 OK"                                    
**Amazon Bedrock AgentCore**について詳しく説明いたします。

## Amazon Bedrock AgentCoreとは

Amazon Bedrock AgentCoreは、AIエージェントを安全にスケールで構築、デプロイ、運用するためのエンタープライズグレードのサービスです。現在プレビュー版として提供されており、以下の特徴があります：

### 主な特徴
- **フレームワーク非依存**: 任意のフレームワークとモデルを使用可能
- **エンタープライズセキュリティ**: 本格的な運用に必要な信頼性とセキュリティを提供
- **モジュラー設計**: 各サービスを個別または組み合わせて使用可能
- **オープンソース対応**: LangGraph、CrewAI、Strands Agentsなどの人気フレームワークと連携

## 構成サービス

Amazon Bedrock AgentCoreは以下の7つのモジュラーサービスで構成されています：

### 1. **AgentCore Runtime**
- **目的**: AIエージェントとツールのセキュアなサーバーレス実行環境
- **特徴**:
  - 最大8時間の長時間実行をサポート
  - 100MBまでのペイロード処理（マルチモーダル対応）
  - セッション分離（各ユーザーセッションが専用microVMで実行）
  - 消費量ベースの料金体系
  - 高速なコールドスタート

### 2. **AgentCore Identity**
- **目的**: エージェント専用のIDとアクセス管理
- **特徴**:
  - 既存のIDプロバイダー（Okta、Microsoft Entra ID、Amazon Cognito）との統合
  - セキュアなトークンボルト
  - 最小権限アクセス
  - 第三者サービス（Slack、Zoom、GitHub）への安全なアクセス

### 3. **AgentCore Memory**
- **目的**: コンテキスト対応エージェント用のメモリ管理
- **特徴**:
  - 短期記憶（マルチターン会話用）
  - 長期記憶（エージェント・セッション間で共有可能）
  - 高精度な記憶機能
  - 複雑なメモリインフラの管理が不要

### 4. **AgentCore Code Interpreter**
- **目的**: 分離されたサンドボックス環境でのコード実行
- **特徴**:
  - セキュアな実行環境
  - 高度な設定サポート
  - 人気フレームワークとのシームレス統合
  - 複雑なワークフローとデータ分析に対応

### 5. **AgentCore Browser**
- **目的**: AIエージェント用の高速・セキュアなクラウドブラウザ
- **特徴**:
  - Webサイトとの大規模インタラクション
  - エンタープライズグレードセキュリティ
  - 包括的な可観測性
  - 自動スケーリング

### 6. **AgentCore Gateway**
- **目的**: エージェント用のツール発見・統合プラットフォーム
- **特徴**:
  - API、Lambda関数、既存サービスのエージェント対応ツールへの変換
  - セキュアなツール接続
  - カスタムコード開発の削減
  - インフラ管理不要

### 7. **AgentCore Observability**
- **目的**: エージェントのパフォーマンス監視・デバッグ
- **特徴**:
  - 統合運用ダッシュボード
  - OpenTelemetry対応
  - エージェントワークフローの詳細可視化
  - リアルタイム監視

## 主な使用例

1. **組み込みツールと機能の提供**
   - ブラウザ自動化とコード解釈のツール活用
   - 内部・外部ツールとの統合
   - ユーザーとのやり取りを記憶するエージェント作成

2. **セキュアな大規模デプロイメント**
   - フレームワーク、プロトコル、モデルに関係なくセキュアなデプロイ
   - 基盤リソースの管理不要
   - シームレスなIDとアクセス管理

3. **テストと監視**
   - エージェントの使用状況と運用メトリクス（トークン使用量、レイテンシ、セッション時間、エラー率）の可視化
   - リアルタイム運用インサイト

## 料金体系

Amazon Bedrock AgentCoreは、前払いや最低料金なしの柔軟な従量課金制を採用しています。

このサービスは、AI エージェントの開発から運用まで、企業レベルでの安全かつ効率的な実装を支援する包括的なプラットフォームとなっています。

from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp.mcp_client import MCPClient

streamable_http_mcp_client = MCPClient(
    lambda: streamablehttp_client("https://ramen-api.dev/mcp")
)

# MCPツールを使用したエージェントを作成
with streamable_http_mcp_client:
    # MCPサーバからツールを取得
    tools = streamable_http_mcp_client.list_tools_sync()

    # MCPツールを使用したエージェントを作成
    agent = Agent(tools=tools)
    agent("吉村家が気になる。")

吉村家について詳しく調べてみますね。まず、ラーメン店の一覧から吉村家を探してみます。
Tool #1: get_shops
吉村家が見つかりました！詳細情報を取得してみますね。
Tool #2: get_shop
写真も詳しく見てみましょう。
Tool #3: get_photos_with_data
吉村家について調べました！

**吉村家**は横浜家系ラーメンの源流とも言える有名店です。写真を見ると、家系ラーメンの特徴である：

- **濃厚な豚骨醤油スープ** - 白濁したクリーミーなスープ
- **太麺** - コシのあるストレート麺
- **チャーシュー** - 厚切りのジューシーなチャーシュー
- **ほうれん草** - 新鮮な緑の葉野菜
- **海苔** - 大判の焼き海苔
- **半熟玉子** - とろっとした黄身の煮玉子

これらの定番トッピングが美しく盛り付けられています。

吉村家は1974年に横浜で創業された家系ラーメンの元祖的存在で、多くの弟子店を輩出したことでも知られています。濃厚だけど飲みやすいスープと、しっかりとした麺のバランスが絶妙で、家系ラーメンファンなら一度は訪れたい名店ですね！

from mcp.client.sse import sse_client
from strands import Agent
from strands.tools.mcp import MCPClient

# SSEトランスポートを使ってMCPサーバに接続する
sse_mcp_client = MCPClient(lambda: sse_client("http://localhost:8000/sse"))

# MCPツールを使用したエージェントを作成
with sse_mcp_client:
    # MCPサーバからツールを取得
    tools = sse_mcp_client.list_tools_sync()

    # MCPツールを使用したエージェントを作成
    agent = Agent(tools=tools)

from typing import Callable
from strands import Agent
from strands.tools.mcp.mcp_client import MCPClient
from strands.tools.mcp.mcp_types import MCPTransport

# カスタムなトランスポートを返す関数を定義
def custom_transport_factory() -> MCPTransport:
    # カスタムなトランスポートを実装
    # タプル (read_stream, write_stream) を返す必要がある
    # 両方とも AsyncIterable と AsyncIterator プロトコルを実装する必要がある
    ...
    return read_stream, write_stream

# カスタムなトランスポートを使用したMCPClientを作成
custom_mcp_client = MCPClient(transport_callable=custom_transport_factory)

# コンテキストマネージャを使用してサーバを使用
with custom_mcp_client:
    # MCPサーバからツールを取得
    tools = custom_mcp_client.list_tools_sync()

    # MCPツールを使用したエージェントを作成
    agent = Agent(tools=tools)

from mcp import stdio_client, StdioServerParameters
from mcp.client.sse import sse_client
from strands import Agent
from strands.tools.mcp import MCPClient

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

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

# 両方のサーバを使用
with document_mcp_client, diagram_mcp_client:
    # 両方のサーバからツールを取得
    tools = document_mcp_client.list_tools_sync() + diagram_mcp_client.list_tools_sync()

    # MCPツールを使用したエージェントを作成
    agent = Agent(tools=tools)
    agent("Bedrock AgentCoreについて概要を教えて。あと、その場合の一般的なインフラ構成図も書いて。")

Installed 38 packages in 13ms
Installed 57 packages in 111ms
[07/25/25 01:49:02] INFO     Processing request of type ListToolsRequest                                             server.py:625
Bedrock AgentCoreについて調べて、概要をお伝えし、インフラ構成図も作成いたします。
Tool #1: search_documentation
[07/25/25 01:49:06] INFO     Processing request of type CallToolRequest                                              server.py:625
[07/25/25 01:49:07] INFO     HTTP Request: POST                                                                    _client.py:1740
                             https://proxy.search.docs.aws.amazon.com/search?session=XXXXXXXX-XXXX-XXXX-XXXX-XXXXX                
                             XXXXXXX "HTTP/1.1 200 OK"                                                                            
メインのドキュメントを確認してみましょう。
Tool #2: read_documentation
[07/25/25 01:49:09] INFO     Processing request of type CallToolRequest                                              server.py:625
                    INFO     HTTP Request: GET                                                                     _client.py:1740
                             https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/what-is-bedrock-agentco                
                             re.html?session=XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXXX "HTTP/1.1 200 OK"                               
続いて、より詳しい構成について確認しましょう。
Tool #3: read_documentation
[07/25/25 01:49:42] INFO     Processing request of type CallToolRequest                                              server.py:625
[07/25/25 01:49:43] INFO     HTTP Request: GET                                                                     _client.py:1740
                             https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/agents-tools-runtime.ht                
                             ml?session=XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXXX "HTTP/1.1 200 OK"                                    
さらに詳しい情報とサンプルコードについても確認しましょう。
Tool #4: recommend
[07/25/25 01:50:07] INFO     Processing request of type CallToolRequest                                              server.py:625
[07/25/25 01:50:08] INFO     HTTP Request: GET                                                                     _client.py:1740
                             https://contentrecs-api.docs.aws.amazon.com/v1/recommendations?path=https://docs.aws.                
                             amazon.com/bedrock-agentcore/latest/devguide/what-is-bedrock-agentcore.html&session=b                
                             XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXXX "HTTP/1.1 200 OK"                                                
では、構成図を作成するために、diagram exampleを確認しましょう。
Tool #5: get_diagram_examples
次に利用可能なAWSアイコンを確認します。
Tool #6: list_icons
MLサービスを確認して、Bedrockアイコンがあることを確認しました。次に、一般的なBedrockとAgentCoreの構成図を作成しましょう。
Tool #7: generate_diagram
[07/25/25 01:51:48] ERROR    Could not load 'sarif': No module named 'sarif_om'                                   extension.py:227
グラフビズの問題のようです。より簡単な構成図を作成しましょう。
Tool #8: generate_diagram
Graphvizが利用できないようですので、テキストベースでインフラ構成図を説明いたします。

## Amazon Bedrock AgentCore概要

Amazon Bedrock AgentCoreは、AIエージェントを安全かつスケーラブルに構築、デプロイ、運用するためのエンタープライズグレードのプラットフォームです。現在プレビュー段階で、以下の特徴があります：

### 主要な特徴

1. **フレームワーク非依存**: LangGraph、CrewAI、Strands Agentsなど、任意のオープンソースフレームワークに対応
2. **モデル柔軟性**: AWS Bedrock、Anthropic Claude、Google Gemini、OpenAIなど任意のLLMで動作
3. **プロトコルサポート**: Model Context Protocol (MCP) による他のエージェントやツールとの連携
4. **拡張実行時間**: 最大8時間の長時間実行をサポート
5. **セッション分離**: 各ユーザーセッションが独立したmicroVMで実行される
6. **従量課金**: 実際に消費されたリソースに対してのみ課金

### AgentCoreのコンポーネント

**Amazon Bedrock AgentCore Runtime**
- AIエージェントとツールをデプロイ・スケールするためのサーバーレスランタイム
- 高速なコールドスタート、セッション分離、組み込み認証を提供

**Amazon Bedrock AgentCore Identity**
- AIエージェント専用の認証・認可管理サービス
- 既存のIDプロバイダー（Okta、Microsoft Entra ID、Amazon Cognito）と統合

**Amazon Bedrock AgentCore Memory**
- コンテキスト対応エージェント構築のためのメモリ管理サービス
- 短期（会話）メモリと長期メモリの両方をサポート

**Amazon Bedrock AgentCore Code Interpreter**
- 分離されたサンドボックス環境でのコード実行ツール

**Amazon Bedrock AgentCore Browser**
- AIエージェントがWebサイトと相互作用するためのクラウドベースブラウザランタイム

**Amazon Bedrock AgentCore Gateway**
- エージェントがツールを発見・使用するための安全なゲートウェイ
- API、Lambda関数、既存サービスをエージェント互換ツールに変換

**Amazon Bedrock AgentCore Observability**
- エージェントのトレース、デバッグ、監視機能
- OpenTelemetry対応のテレメトリサポート

## 一般的なインフラ構成（テキスト図）

```
┌─────────────────────────────────────────────────────────────────┐
│                        End Users                                │
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      v
┌─────────────────────────────────────────────────────────────────┐
│                    API Gateway                                  │
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      v
┌─────────────────────────────────────────────────────────────────┐
│                Authentication Layer                             │
│  ┌─────────────────┐  ┌─────────────────────────────────────────┐│
│  │Amazon Cognito   │  │   AgentCore Identity                    ││
│  │or Other IdP     │  │   - Agent authentication               ││
│  └─────────────────┘  │   - OAuth/API key management           ││
│                       └─────────────────────────────────────────┘│
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      v
┌─────────────────────────────────────────────────────────────────┐
│               Amazon Bedrock AgentCore                          │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │                AgentCore Runtime                            ││
│  │  - Serverless execution environment                        ││
│  │  - Session isolation with microVMs                         ││
│  │  - Multi-framework support (LangGraph, CrewAI, etc.)       ││
│  │  - Extended runtime (up to 8 hours)                        ││
│  └─────────────────────────────────────────────────────────────┘│
│                                                                 │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────────┐│
│  │   Memory    │ │Code Interp. │ │        Browser Tool         ││
│  │ - Short-term│ │- Sandboxed  │ │ - Web interaction           ││
│  │ - Long-term │ │  execution  │ │ - Cloud browser runtime     ││
│  └─────────────┘ └─────────────┘ └─────────────────────────────┘│
│                                                                 │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │                AgentCore Gateway                            ││
│  │  - Tool discovery and integration                          ││
│  │  - API transformation                                      ││
│  └─────────────────────────────────────────────────────────────┘│
│                                                                 │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │              Observability                                 ││
│  │  - OpenTelemetry compatible                                ││
│  │  - Agent workflow visualization                            ││
│  └─────────────────────────────────────────────────────────────┘│
└─────────┬───────────────────────────┬───────────────────────────┘
          │                           │
          v                           v
┌─────────────────────┐    ┌─────────────────────────────────────┐
│   AI Models         │    │        Storage & Data               │
│                     │    │                                     │
│ ┌─────────────────┐ │    │ ┌─────────────┐ ┌─────────────────┐ │
│ │Amazon Bedrock   │ │    │ │Amazon S3    │ │   DynamoDB      │ │
│ │- Claude         │ │    │ │- Documents  │ │ - Session data  │ │
│ │- Other models   │ │    │ │- Assets     │ │ - Memory store  │ │
│ └─────────────────┘ │    │ └─────────────┘ └─────────────────┘ │
│                     │    │                                     │
│ ┌─────────────────┐ │    │ ┌─────────────────────────────────┐ │
│ │External Models  │ │    │ │           RDS                   │ │
│ │- OpenAI         │ │    │ │    - Structured data            │ │
│ │- Google Gemini  │ │    │ │    - Application data           │ │
│ └─────────────────┘ │    │ └─────────────────────────────────┘ │
└─────────────────────┘    └─────────────────────────────────────┘
          │
          v
┌─────────────────────────────────────────────────────────────────┐
│                    External Systems                             │
│                                                                 │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────────┐│
│  │ Slack API   │ │ GitHub API  │ │       Other APIs            ││
│  └─────────────┘ └─────────────┘ └─────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
```

### 主要なユースケース

1. **組み込みツール活用**: ブラウザ自動化やコード実行機能を持つエージェントの構築
2. **安全なスケール展開**: フレームワークやプロトコル、モデルを問わず、動的AIエージェントの安全なデプロイ
3. **エージェント監視**: リアルタイムでの使用状況やパフォーマンスメトリクス（トークン使用量、レイテンシ、セッション期間、エラー率）の監視

このプラットフォームにより、開発者は複雑なインフラ管理から解放され、AIエージェントのイノベーションに集中できるようになります。

{
    "toolUseId": str,       # ツール使用リクエストのID (入力リクエストと一致する必要がある)。 オプション。
    "status": str,          # "success" または "error"
    "content": List[dict]   # 異なる形式のコンテンツ項目のリスト
}

import asyncio
from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp.mcp_client import MCPClient
import json

streamable_http_mcp_client = MCPClient(
    lambda: streamablehttp_client("https://ramen-api.dev/mcp")
)

async def main():
    with streamable_http_mcp_client:
        result = await streamable_http_mcp_client.call_tool_async(
        tool_use_id="tool-123",
        name="get_shop",
        arguments={"shopId": "yoshimuraya"},
    )
    print(json.dumps(result, indent=2, ensure_ascii=False))

asyncio.run(main())

{
  "status": "success",
  "toolUseId": "tool-123",
  "content": [
    {
      "text": "{\"id\":\"yoshimuraya\",\"name\":\"吉村家\",\"photos\":[{\"name\":\"yoshimuraya-001.jpg\",\"width\":1200,\"height\":900,\"authorId\":\"yusukebe\",\"url\":\"https://ramen-api.dev/images/yoshimuraya/yoshimuraya-001.jpg\"}]}"
    }
  ]
}

from strands import Agent
from strands_tools import python_repl

agent = Agent(tools=[python_repl])

agent("""\
Pythonで1〜100までの数字の合計を計算するプログラムを作成して。
そしてそれを実行して。\
""")

1〜100までの数字の合計を計算するPythonプログラムを作成し、実行します。
Tool #1: python_repl
Do you want to proceed with Python code execution? [y/*]

方法1 (forループ): 1〜100の合計 = 5050
方法2 (sum関数): 1〜100の合計 = 5050
方法3 (数学公式): 1〜100の合計 = 5050

全ての方法で同じ結果が得られたか: True
1〜100までの数字の合計を計算するプログラムを作成し、実行しました。

**結果: 5050**

3つの異なる方法でプログラムを作成しました：

1. **forループを使用**: 1から100まで順番に足し算
2. **sum()関数を使用**: Pythonの組み込み関数を活用
3. **数学公式を使用**: ガウスの公式 n×(n+1)÷2 を適用

全ての方法で同じ結果（5050）が得られることを確認できました。この値は、1から100までの連続する整数の合計の正しい答えです。

from strands import Agent
from strands_tools import python_repl
import os

# ツールの同意をバイパスする
os.environ["BYPASS_TOOL_CONSENT"] = "true"

agent = Agent(tools=[python_repl])

agent("""\
Pythonで1〜100までの数字の合計を計算するプログラムを作成して。
そしてそれを実行して。\
""")

1〜100までの数字の合計を計算するPythonプログラムを作成して実行しますね。
Tool #1: python_repl
方法1 (forループ): 1〜100の合計 = 5050
方法2 (sum関数): 1〜100の合計 = 5050
方法3 (数学公式): 1〜100の合計 = 5050

全ての方法で同じ結果が得られました: True
1〜100までの数字の合計を計算するプログラムを3つの異なる方法で作成し、実行しました。

**結果**: 1〜100までの数字の合計は **5050** です。

**3つの方法の説明**:

1. **forループを使った方法**: 最も基本的な方法で、1から100まで順番に数字を足していきます
2. **sum()関数を使った方法**: Pythonの組み込み関数を使ったより簡潔な書き方です
3. **数学公式を使った方法**: 等差数列の和の公式 n(n+1)/2 を使った最も効率的な方法です

どの方法でも同じ結果（5050）が得られることを確認できました。用途に応じて適切な方法を選択できます。

from strands import Agent
from strands_tools import handoff_to_user

agent = Agent(tools=[handoff_to_user])

response = agent.tool.handoff_to_user(
    message="処理を続行するにはあなたの承認が必要です。承認する場合は'yes'と入力してください。",
    breakout_of_loop=False,
)

# ユーザにハンドオフする（エージェント実行を停止する）
agent.tool.handoff_to_user(
    message="処理が完了しました。結果を確認してください。",
    breakout_of_loop=True
)
print("エージェントの実行は終了しました。")

Your response:  yes
エージェントの実行は終了しました。

Your response:  no
エージェントの実行は終了しました。