Kb Strands Agentcore

Strands Agents ナレッジ

AWS が提供する AI エージェントフレームワーク「Strands Agents」に関する学びを記録する。 CDK/デプロイ/ランタイムは /kb-agentcore-cdk、Observabilityは /kb-agentcore-observability を参照。

基本情報

Strands Agents

• 公式: https://strandsagents.com/
• GitHub: https://github.com/strands-agents/strands-agents
• Python 3.10以上が必要

Bedrock AgentCore

• 15リージョンで利用可能（us-east-1, us-west-2, ap-northeast-1 等）
• Evaluations機能のみ一部リージョン限定（東京は非対応）

インストール

# pip
pip install strands-agents bedrock-agentcore

# uv
uv add strands-agents bedrock-agentcore

AWS CLI login 認証を使う場合

uv add 'botocore[crt]'

aws login で認証した場合、botocore[crt] が必要。これがないと認証エラーになる。

Agent作成

基本構造

from strands import Agent

agent = Agent(
    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
    system_prompt="あなたはアシスタントです",
)

利用可能なモデル（Bedrock）

クロスリージョン推論のプレフィックスはリージョンによって異なる：

# リージョンに応じてプレフィックスを自動判定
import os

def _get_model_id() -> str:
    region = os.environ.get("AWS_REGION", "us-east-1")
    prefix = "jp" if region == "ap-northeast-1" else "us"
    return f"{prefix}.anthropic.claude-sonnet-4-5-20250929-v1:0"

実行方法

同期実行

result = agent(prompt)
print(result)

非同期実行

result = await agent.invoke_async(prompt)

ストリーミング（同期）

for event in agent.stream(prompt):
    if "data" in event:
        print(event["data"], end="", flush=True)

ストリーミング（非同期）

async for event in agent.stream_async(prompt):
    if "data" in event:
        print(event["data"], end="", flush=True)

イベントタイプ

ストリーミング時に受け取るイベント：

async for event in agent.stream_async(prompt):
    if "data" in event:
        # テキストチャンク
        print(event["data"], end="")
    elif "current_tool_use" in event:
        # ツール使用中
        tool_info = event["current_tool_use"]
        print(f"Using tool: {tool_info['name']}")
    elif "result" in event:
        # 完了
        final_result = event["result"]

current_tool_use の input はストリーミング中は文字列型

current_tool_use イベントの input フィールドは、ストリーミング中は不完全なJSON文字列として徐々に構築される。辞書型を期待している場合はJSONパースが必要：

elif "current_tool_use" in event:
    tool_info = event["current_tool_use"]
    tool_name = tool_info.get("name", "unknown")
    tool_input = tool_info.get("input", {})

    # inputが文字列の場合はJSONパースを試みる
    if isinstance(tool_input, str):
        try:
            import json
            tool_input = json.loads(tool_input)
        except json.JSONDecodeError:
            pass  # パースできない場合はそのまま（不完全なJSON）

    # パース成功時のみ辞書として扱える
    if isinstance(tool_input, dict) and "query" in tool_input:
        print(f"Search query: {tool_input['query']}")

ポイント: ストリーミング中はイベントが複数回発火し、{"query" -> {"query": "検索 -> {"query": "検索ワード"} のように徐々に完成する。完全なJSONになったタイミングでのみパースが成功する。

重要: バックエンドで重複スキップしてはいけない

current_tool_use の重複イベントをバックエンドで continue してはいけない。理由：

• 最初のチャンクの input は不完全なJSON文字列（例: "{\"qu"）
• JSONパースが失敗し、query 等の必要なパラメータが取得できない
• 後続チャンク（パラメータが完成したもの）がスキップされ、イベントが一切フロントに送信されなくなる

重複の吸収はフロントエンド側（hasInProgress チェック等）で行うのが正しい。

# NG: バックエンドで重複スキップ -> 最初のチャンク（input不完全）のみ処理される
if tool_name == last_tool_name:
    continue  # 2回目以降のチャンク（inputが完全）がスキップされる！
last_tool_name = tool_name

# OK: 重複スキップせず、条件に合うときだけyield（フロント側で重複吸収）
if tool_name == "web_search":
    if isinstance(tool_input, dict) and "query" in tool_input:
        yield {"type": "tool_use", "data": tool_name, "query": tool_input["query"]}
    # queryが不完全なチャンクではyieldしない -> 完成したチャンクでyieldされる
else:
    yield {"type": "tool_use", "data": tool_name}

ツールの定義

関数デコレータ方式

from strands import Agent, tool

@tool
def get_weather(city: str) -> str:
    """指定した都市の天気を取得します。

    Args:
        city: 都市名

    Returns:
        天気情報
    """
    return f"{city}の天気は晴れです"

agent = Agent(
    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
    tools=[get_weather],
)

クラス方式

from strands import Agent, Tool

class WeatherTool(Tool):
    name = "get_weather"
    description = "指定した都市の天気を取得します"

    def run(self, city: str) -> str:
        return f"{city}の天気は晴れです"

agent = Agent(
    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
    tools=[WeatherTool()],
)

ツール駆動型の出力パターン

LLMの出力をフロントエンドでフィルタリングするのが難しい場合、出力専用のツールを作成してツール経由で出力させる方式が有効。

# グローバル変数で出力を保持
_generated_markdown: str | None = None

@tool
def output_slide(markdown: str) -> str:
    """生成したスライドのマークダウンを出力します。

    Args:
        markdown: Marp形式のマークダウン全文

    Returns:
        出力完了メッセージ
    """
    global _generated_markdown
    _generated_markdown = markdown
    return "スライドを出力しました。"

agent = Agent(
    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
    system_prompt="スライドを作成したら、必ず output_slide ツールを使って出力してください。",
    tools=[output_slide],
)

メリット:

• フロントエンドでのテキスト除去処理が不要
• ツール使用中のステータス表示が容易
• マークダウンがテキストストリームに混入しない

外部APIカスタムツール（追加パッケージ不要）

外部REST APIを呼ぶカスタムツールは urllib.request（標準ライブラリ）で実装すると、requirements.txtに追加パッケージ不要で済む。

import json
import os
import urllib.request

from strands import tool

TAVILY_API_KEY = os.environ.get("TAVILY_API_KEY", "")

@tool
def web_search(query: str) -> str:
    """ウェブ検索を行い、最新の情報を取得します。

    Args:
        query: 検索クエリ

    Returns:
        検索結果のテキスト
    """
    req = urllib.request.Request(
        "https://api.tavily.com/search",
        data=json.dumps({
            "query": query,
            "max_results": 5,
            "search_depth": "basic",
            "include_answer": True,
        }).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {TAVILY_API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read().decode("utf-8"))

    parts = []
    if result.get("answer"):
        parts.append(f"【要約】\n{result['answer']}")
    for item in result.get("results", []):
        title = item.get("title", "")
        url = item.get("url", "")
        content = item.get("content", "")
        parts.append(f"■ {title}\n{url}\n{content}")

    return "\n\n".join(parts) if parts else "検索結果が見つかりませんでした。"

ポイント: tavily-python パッケージを使う方法もあるが、urllib.request なら追加依存なし。Docker/AgentCoreのビルド時間短縮にも有効。

会話履歴の管理

from strands import Agent

agent = Agent(model="us.anthropic.claude-sonnet-4-5-20250929-v1:0")

# 会話を継続
response1 = agent("私の名前は太郎です")
response2 = agent("私の名前は何ですか？")  # 「太郎」と答える

# 履歴をクリア
agent.clear_history()

SlidingWindowConversationManager（履歴トリミング）

トークンコスト削減のため、古いメッセージを自動削除する組み込み機能。

from strands import Agent
from strands.agent.conversation_manager import SlidingWindowConversationManager

agent = Agent(
    model=model,
    system_prompt=SYSTEM_PROMPT,
    tools=tools,
    conversation_manager=SlidingWindowConversationManager(window_size=6),
)

パラメータ

window_size の意味

window_size は Bedrock APIのメッセージ配列の要素数。1つのメッセージが複数の content block（text, toolUse, toolResult）を持ちうるため、実際のテキスト量は window_size だけでは決まらない。

典型的な1リクエストのメッセージ数（ツール2つ使用時）:

[user]      ユーザーメッセージ          ... 1
[assistant] toolUse: web_search         ... 2
[user]      toolResult: 検索結果        ... 3
[assistant] toolUse: output_slide       ... 4
[user]      toolResult: スライド出力    ... 5
[assistant] 最終テキスト応答           ... 6

-> 1リクエスト = 6メッセージ -> window_size=6 で約1リクエスト分を保持

トリミングアルゴリズム（2段階）

1. フェーズ1: ツール結果の圧縮（優先）

   • 古い toolResult の内容を "The tool result was too large!" に置換
   • メッセージ数は減らさない
   • 同じ toolResult を2回圧縮しない

2. フェーズ2: メッセージの削除（最終手段）

   • フェーズ1で不十分な場合、古いメッセージを削除
   • toolUse/toolResult ペアの整合性を保持（ペアが壊れない位置で削除）

per_turn パラメータの使い分け

# デフォルト: エージェントループ完了後にのみトリミング
# -> 処理中は全メッセージが利用可能（品質重視）
SlidingWindowConversationManager(window_size=6)

# 毎回のモデル呼び出し前にトリミング
# -> ループが多いユースケースでトークン爆発を防止
SlidingWindowConversationManager(window_size=10, per_turn=True)

# N回ごとにトリミング（パフォーマンスバランス）
SlidingWindowConversationManager(window_size=10, per_turn=5)

per_turn=True と並列ツール実行の非互換性

per_turn=True は、エージェントが複数ツールを並列発行するユースケースでは使用禁止。

Strands Agents は並列ツール（例: web_search x2）の結果を1件ずつ個別にセッション履歴に追加する。per_turn=True だと各LLMコール前にトリミングが走り、以下の正のフィードバックループが発生する：

1. LLMが web_search x2 を並列発行
2. ツール結果1件目が履歴に追加 -> トリミング発生
3. 「検索結果が1件しかない」状態でLLMが呼ばれる
4. LLMは「情報不足」と判断して追加の web_search を発行
5. 1-4が連鎖 -> 1セッションで web_search が16-20回に増殖

さらに output_slide 等のツール結果が "The tool result was too large!" に圧縮され、ツールの正常動作も阻害される。

推奨: per_turn=False（デフォルト）のまま使用し、コスト削減はツール結果のサイズ制限（要約等）で対応する。

window_size チューニングの考え方

• フロントエンドが毎回最新コンテキスト（例: 生成済みMarkdown）を送信する設計なら、古い履歴は不要 -> 小さい window_size でOK
• per_turn=False（デフォルト）の場合、処理中のメッセージはトリミングされないため品質に影響しない
• トリミングの効果は 多ターンセッション で最大化される

Bedrock AgentCore との統合

基本構造

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp()
agent = Agent(model="us.anthropic.claude-sonnet-4-5-20250929-v1:0")

@app.entrypoint
async def invoke(payload):
    prompt = payload.get("prompt", "")
    stream = agent.stream_async(prompt)
    async for event in stream:
        yield event

if __name__ == "__main__":
    app.run()  # ポート8080でリッスン

エンドポイント

• POST /invocations - エージェント実行
• GET /ping - ヘルスチェック

必要な依存関係

# requirements.txt
bedrock-agentcore
strands-agents
tavily-python  # Web検索が必要な場合

注意: fastapi/uvicorn は不要（bedrock-agentcore SDKに内包）

セッションIDでAgentを管理（複数ユーザー対応）

AgentCoreで複数ユーザーの会話履歴を保持する場合、セッションIDごとにAgentインスタンスを管理する：

from strands import Agent

# セッションごとのAgentインスタンスを管理
_agent_sessions: dict[str, Agent] = {}

def get_or_create_agent(session_id: str | None) -> Agent:
    """セッションIDに対応するAgentを取得または作成"""
    # セッションIDがない場合は新規Agentを作成（履歴なし）
    if not session_id:
        return Agent(
            model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
            system_prompt="...",
            tools=[...],
        )

    # 既存のセッションがあればそのAgentを返す
    if session_id in _agent_sessions:
        return _agent_sessions[session_id]

    # 新規セッションの場合はAgentを作成して保存
    agent = Agent(
        model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
        system_prompt="...",
        tools=[...],
    )
    _agent_sessions[session_id] = agent
    return agent

@app.entrypoint
async def invoke(payload):
    session_id = payload.get("session_id")
    agent = get_or_create_agent(session_id)
    # ...

注意: コンテナ再起動でセッションは消える（メモリ管理）。永続化が必要な場合はDynamoDB等を検討。

SSE keep-alive パターン（長時間処理のコネクション維持）

同期的な重い処理（ファイル変換、外部API呼び出し等）をSSEで返す場合、処理中にkeep-aliveイベントを送信してコネクションを維持する。asyncio.run_in_executor + asyncio.shield + タイムアウトの組み合わせ。

import asyncio

async def _wait_with_keepalive(task, format_name):
    """タスク完了を待ちつつ、5秒ごとにSSE keep-aliveイベントをyield"""
    while not task.done():
        try:
            await asyncio.wait_for(asyncio.shield(task), timeout=5.0)
        except asyncio.TimeoutError:
            yield {"type": "progress", "message": f"{format_name}変換中..."}

@app.entrypoint
async def invoke(payload, context=None):
    if action == "export_pptx" and markdown:
        try:
            print(f"[INFO] PPTX export started")
            loop = asyncio.get_event_loop()
            task = loop.run_in_executor(None, generate_pptx, markdown, theme)
            async for event in _wait_with_keepalive(task, "PPTX"):
                yield event  # 5秒ごとにprogressイベント送信
            result_bytes = task.result()
            yield {"type": "pptx", "data": base64.b64encode(result_bytes).decode()}
        except Exception as e:
            print(f"[ERROR] PPTX export failed: {e}")
            yield {"type": "error", "message": str(e)}
        return

ポイント:

• asyncio.shield(task) で TimeoutError 時もタスクがキャンセルされない
• task.done() でループ脱出を判定、task.result() で結果取得
• フロントエンドのSSEパーサーは未知の type を無視するため、既存コードの変更不要

ツール使用イベント送信

@app.entrypoint
async def invoke(payload):
    global _generated_markdown
    _generated_markdown = None

    stream = agent.stream_async(payload.get("prompt", ""))
    async for event in stream:
        if "data" in event:
            yield {"type": "text", "data": event["data"]}
        elif "current_tool_use" in event:
            tool_name = event["current_tool_use"].get("name", "unknown")
            yield {"type": "tool_use", "data": tool_name}

    if _generated_markdown:
        yield {"type": "markdown", "data": _generated_markdown}

Bedrockプロンプトキャッシュが突然停止する

症状: Cost ExplorerのCacheWrite/CacheRead費用が特定のコミット以降ずっと0になる

原因: Bedrockのprompt cachingはツール定義の合計トークンが1024以上必要。ツールのdocstringを短くしたり、リッチなビルトインツール（strands_tools.http_request 等）をシンプルなカスタムツールに置き換えると、合計が1024を割り込んでキャッシュが機能停止する。

診断方法:

1. Cost Explorerでモデル別にCacheWrite/CacheReadを確認し、0になり始めた日付を特定
2. その日付のコミットで @tool 付き関数のdocstringが短くなっていないか確認

解決策: docstringを拡充して合計1024トークン以上に戻す（パラメータの詳細説明・使用例・注意事項を追加すると大幅に増やせる）

@tool デコレータとトークン数の関係:

• Strands Agentsの @tool デコレータはPython関数のdocstringをBedrockへのツール説明（tool spec）として送信する
• パラメータ数・docstringの文字数がそのまま毎リクエストのトークンコストになる
• docstringが増えてもキャッシュリード時は90%オフになるため、キャッシュが有効な状態では長いdocstringでもコスト増にはならない

Bedrockキャッシュの課金構造（Sonnet 4.6）:

トラブルシューティング

AWS認証エラー

aws login で認証した場合、botocore[crt] が必要：

uv add 'botocore[crt]'

モデルが見つからない

クロスリージョン推論のモデルID（us. プレフィックス）を使用しているか確認。 リージョンによって利用可能なモデルが異なる。

ストリーミングが動かない

stream() と stream_async() を環境に合わせて使い分ける：

• 同期コンテキスト -> stream()
• 非同期コンテキスト（async/await） -> stream_async()

Kimi K2関連

Kimi K2（Moonshot AI）特有の問題は /kb-kimi スキルを参照してください。

関連スキル

• /kb-agentcore-cdk - AgentCore CDK、デプロイ、ランタイム統合、コンテナ構成
• /kb-agentcore-observability - OpenTelemetry、ログ、メトリクス、トレース
• /kb-agentcore-identity - アウトバウンド認証（3LO/M2M/デコレータ分離/callback等）
• /kb-bidi-agent - BidiAgent（双方向ストリーミング / 音声対話 / Nova Sonic）
• /kb-kimi - Kimi K2（Moonshot AI）特有の問題・ワークアラウンド

参考リンク

• Strands Agents 公式ドキュメント
• GitHub リポジトリ
• Bedrock AgentCore 統合ガイド

Converted and distributed by TomeVault — claim your Tome and manage your conversions.