Strands AgentsとAmazon Bedrockで構築したAIエージェントをMicrosoft 製 agent-sreでSRE運用してみる

Python 3.13.9
strands-agents 1.39.0
strands-agents-tools 0.5.2
agent-sre 3.5.0
boto3 1.42.10

$ uv pip install \
    "strands-agents==1.39.0" \
    "strands-agents-tools==0.5.2" \
    "agent-sre==3.5.0" \
    "boto3>=1.40.0"

$ export AWS_PROFILE="your-profile"
$ export AWS_DEFAULT_REGION="ap-northeast-1"

$ aws bedrock list-inference-profiles --region ap-northeast-1 \
    --query "inferenceProfileSummaries[?inferenceProfileId=='jp.anthropic.claude-sonnet-4-5-20250929-v1:0']"
[
    {
        "inferenceProfileName": "JP Anthropic Claude Sonnet 4.5",
        "inferenceProfileId": "jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
        "inferenceProfileArn": "arn:aws:bedrock:ap-northeast-1:111122223333:inference-profile/jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
        "status": "ACTIVE",
        "type": "SYSTEM_DEFINED"
    }
]

from strands import Agent
from strands.models import BedrockModel

model = BedrockModel(
    model_id="jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name="ap-northeast-1",
    temperature=0,
)
agent = Agent(model=model, system_prompt="短く一言で答えてください。")
print(agent("東京タワーの高さは何メートルですか？"))

333メートルです。

from strands import tool

@tool
def web_search(query: str) -> str:
    """Web で情報を検索する."""
    data = {
        "東京タワー": "東京タワーの高さは333メートルです。1958年に完成しました。",
        "富士山": "富士山の標高は3,776メートルで、日本最高峰です。",
    }
    for key, val in data.items():
        if key in query:
            return val
    return f"'{query}' の検索結果: 関連情報が見つかりました。"

@tool
def calculator(expression: str) -> str:
    """数式を計算する（四則演算のみ対応）."""
    allowed = set("0123456789+-*/.() ")
    if not all(c in allowed for c in expression):
        return "エラー: サポートされない式です"
    try:
        return str(eval(expression))  # noqa: S307 デモ用途のみ
    except Exception:
        return "計算エラー"

@tool
def get_weather(location: str) -> str:
    """指定した場所の現在の天気を取得する."""
    data = {
        "東京": "晴れ、気温 25°C、湿度 60%",
        "大阪": "曇り、気温 23°C、湿度 70%",
    }
    return f"{location}: {data.get(location, '晴れ、気温 20°C')}"

from agent_sre.adapters import BaseAdapter, TaskRecord

class StrandsAdapter(BaseAdapter):
    """agent-sre の BaseAdapter を Strands 用に拡張したアダプタ."""

    def __init__(self) -> None:
        super().__init__("strands")

    def on_invocation_start(self, agent_name: str = "") -> TaskRecord:
        return self._start_task({"agent_name": agent_name})

    def on_model_call(
        self,
        input_tokens: int = 0,
        output_tokens: int = 0,
        cost_usd: float = 0.0,
    ) -> None:
        if self._current:
            self._current.input_tokens += input_tokens
            self._current.output_tokens += output_tokens
            self._current.cost_usd += cost_usd

    def on_tool_call(self, tool_name: str, error: str = "") -> None:
        if self._current:
            self._current.tool_calls += 1
            if error:
                self._current.tool_errors += 1

    def on_invocation_end(self, success: bool = True, error: str = "") -> TaskRecord:
        return self._finish_task(success=success, error=error)

from strands.hooks import (
    AfterInvocationEvent,
    AfterModelCallEvent,
    AfterToolCallEvent,
    BeforeInvocationEvent,
    BeforeToolCallEvent,
    HookProvider,
    HookRegistry,
)

class SREHook(HookProvider):
    """Strands Hook を agent-sre の各コンポーネントに橋渡しする HookProvider."""

    def __init__(
        self,
        adapter: StrandsAdapter,
        detector: RogueAgentDetector,
    ) -> None:
        self.adapter = adapter
        self.detector = detector
        self._tool_names: list[str] = []

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeInvocationEvent, self._on_before_invocation)
        registry.add_callback(BeforeToolCallEvent, self._on_before_tool)
        registry.add_callback(AfterToolCallEvent, self._on_after_tool)
        registry.add_callback(AfterModelCallEvent, self._on_after_model)
        registry.add_callback(AfterInvocationEvent, self._on_after_invocation)

    def _on_before_invocation(self, event: BeforeInvocationEvent) -> None:
        self._tool_names.clear()
        self.adapter.on_invocation_start(agent_name=event.agent.name)

    def _on_before_tool(self, event: BeforeToolCallEvent) -> None:
        name = event.tool_use["name"]
        self._tool_names.append(name)
        # action にツール名を渡すことで ActionEntropyScorer がツール多様性を評価する
        # （"tool_call" 等の固定文字列だとエントロピーが常に最小値に張り付くため）
        self.detector.record_action(
            agent_id=AGENT_ID,
            action=name,
            tool_name=name,
            timestamp=time.time(),
        )

    def _on_after_tool(self, event: AfterToolCallEvent) -> None:
        name = event.tool_use["name"]
        error = ""
        # event.exception は Strands がツール実行で捕捉した例外（None なら正常）
        if event.exception is not None:
            error = str(event.exception)
        elif event.result and event.result.get("status") == "error":
            error = "tool_error"
        self.adapter.on_tool_call(name, error=error)

    def _on_after_model(self, event: AfterModelCallEvent) -> None:
        # トークン使用量は最終的に AgentResult.metrics に集約されるので
        # ここでは個別記録はせず、ラッパー側でまとめて処理する
        pass

    def _on_after_invocation(self, event: AfterInvocationEvent) -> None:
        # AfterInvocationEvent は try/finally の finally 側で発火するため例外時にも来る。
        # event.result が None なら途中で失敗したとみなす。最終的な成功/失敗判定は
        # ラッパー (run_with_sre) で行うので、ここではアダプタ側 TaskRecord に
        # 暫定ステータスを残す扱い。
        success = event.result is not None
        self.adapter.on_invocation_end(success=success)

    @property
    def last_tool_names(self) -> list[str]:
        return list(self._tool_names)

from strands import Agent
from strands.models import BedrockModel

AGENT_ID = "research-agent"
MODEL_ID = "jp.anthropic.claude-sonnet-4-5-20250929-v1:0"
REGION = "ap-northeast-1"

def build_agent(hook: SREHook) -> Agent:
    model = BedrockModel(
        model_id=MODEL_ID,
        region_name=REGION,
        temperature=0,
        # 本番では guardrail_id を指定して Bedrock Guardrails を有効化する
        # guardrail_id="xxxxxxxx",
        # guardrail_version="DRAFT",
        # guardrail_trace="enabled",
    )
    return Agent(
        name=AGENT_ID,
        model=model,
        tools=[web_search, calculator, get_weather],
        system_prompt=(
            "あなたはツールを使ってユーザーの質問に回答するアシスタントです。"
            "必要に応じてツールを呼び出して、簡潔に答えてください。"
        ),
        hooks=[hook],
    )

from agent_sre.anomaly import RiskLevel, RogueAgentDetector, RogueDetectorConfig
from agent_sre.cascade.circuit_breaker import (
    CircuitBreaker,
    CircuitBreakerConfig,
    CircuitOpenError,
)
from agent_sre.cost import CostGuard
from agent_sre.slo import SLI, SLO, ErrorBudget, SLIValue
from agent_sre.slo.objectives import ExhaustionAction

class TaskSuccessRateSLI(SLI):
    def collect(self) -> SLIValue:
        values = self.values_in_window()
        if not values:
            return self.record(1.0)
        good = sum(1 for v in values if v.is_good)
        return self.record(good / len(values))

def init_sre_components(
    failure_threshold: int = 3,
    recovery_timeout: float = 10.0,
) -> tuple[CircuitBreaker, SLO, RogueAgentDetector, CostGuard]:
    breaker = CircuitBreaker(
        agent_id=AGENT_ID,
        config=CircuitBreakerConfig(
            failure_threshold=failure_threshold,
            recovery_timeout_seconds=recovery_timeout,
            half_open_max_calls=1,
        ),
    )
    slo = SLO(
        name=f"{AGENT_ID}-reliability",
        indicators=[
            TaskSuccessRateSLI(name="task_success_rate", target=0.95, window="1h"),
        ],
        error_budget=ErrorBudget(
            total=0.05,
            window_seconds=3600,
            burn_rate_alert=2.0,
            burn_rate_critical=10.0,
            exhaustion_action=ExhaustionAction.THROTTLE,
        ),
        agent_id=AGENT_ID,
    )
    detector = RogueAgentDetector(
        config=RogueDetectorConfig(
            frequency_z_threshold=2.0,
            quarantine_risk_level=RiskLevel.HIGH,
        ),
    )
    detector.register_capability_profile(
        agent_id=AGENT_ID,
        allowed_tools=["web_search", "calculator", "get_weather"],
    )
    cost_guard = CostGuard(
        per_task_limit=0.10,
        per_agent_daily_limit=1.00,
        org_monthly_budget=100.00,
        auto_throttle=True,
        kill_switch_threshold=0.95,
    )
    return breaker, slo, detector, cost_guard

import time
from typing import Any

from strands.types.exceptions import ModelThrottledException

# Claude Sonnet 4.5 の単価（$/token）
INPUT_COST_PER_TOKEN = 3.0 / 1_000_000
OUTPUT_COST_PER_TOKEN = 15.0 / 1_000_000

def run_with_sre(
    question: str,
    *,
    breaker: CircuitBreaker,
    slo: SLO,
    detector: RogueAgentDetector,
    cost_guard: CostGuard,
    adapter: StrandsAdapter,
    hook: SREHook,
    agent: Agent,
    estimated_cost: float = 0.005,
    simulate_failure: bool = False,
) -> dict[str, Any]:
    # 1. コストチェック（プリフライト）
    allowed, reason = cost_guard.check_task(AGENT_ID, estimated_cost=estimated_cost)
    if not allowed:
        return {"status": "cost_blocked", "reason": reason}

    # 2. サーキットブレーカーチェック + エージェント実行
    def _invoke():
        if simulate_failure:
            raise RuntimeError("simulated_fault")
        return agent(question)

    start = time.time()
    try:
        result = breaker.call(_invoke)
    except CircuitOpenError as e:
        slo.indicators[0].record(0.0)
        slo.record_event(good=False)
        return {"status": "circuit_open", "retry_after": round(e.retry_after, 1),
                "cb_state": breaker.state}
    except ModelThrottledException as e:
        slo.indicators[0].record(0.0)
        slo.record_event(good=False)
        return {"status": "throttled", "error": str(e), "cb_state": breaker.state}
    except Exception as e:  # CircuitBreaker.call() は任意の例外を failure として再送出する
        slo.indicators[0].record(0.0)
        slo.record_event(good=False)
        return {"status": "failed", "error": str(e), "cb_state": breaker.state}

    latency = time.time() - start

    # 3. 不正エージェント検出（Hookで record_action 済みなので assess のみ）
    assessment = detector.assess(AGENT_ID)

    # 4. SLO 記録
    slo.indicators[0].record(1.0)
    slo.record_event(good=True)

    # 5. コスト記録（直近の Agent 呼び出し分のトークン使用量を取得）
    # ※ accumulated_usage は同一 Agent インスタンスの「全リクエスト累積」のため、
    #   個別タスクのコスト計上には latest_agent_invocation.usage を使う
    latest = result.metrics.latest_agent_invocation
    usage = latest.usage if latest else {"inputTokens": 0, "outputTokens": 0}
    input_tokens = usage.get("inputTokens", 0)
    output_tokens = usage.get("outputTokens", 0)
    prompt_cost = input_tokens * INPUT_COST_PER_TOKEN
    completion_cost = output_tokens * OUTPUT_COST_PER_TOKEN
    actual_cost = prompt_cost + completion_cost
    cost_guard.record_cost(
        agent_id=AGENT_ID,
        task_id=f"task-{int(time.time() * 1000)}",
        cost_usd=actual_cost,
        breakdown={"input": prompt_cost, "output": completion_cost},
    )

    return {
        "status": "success",
        "response": str(result),
        "latency": round(latency, 2),
        "tool_calls": hook.last_tool_names,
        "risk_level": assessment.risk_level.value,
        "cost": round(actual_cost, 6),
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
    }

$ python demo.py

============================================================
デモ 1: Strands エージェント実行と Agent SRE 監視
============================================================

  質問 1: 東京タワーの高さは何メートルですか？
  回答: 東京タワーの高さは**333メートル**です。1958年に完成した日本を代表する電波塔で、現在も東京のシンボルとして親しまれています。
  レイテンシ=4.6s  ツール=['web_search']  リスク=low  トークン=1939/121
  SLO バジェット残=100.0%  累積コスト=$0.0076

  質問 2: 123 * 456 を計算してください。
  回答: 123 × 456 = **56,088** です。
  レイテンシ=2.73s  ツール=['calculator']  リスク=low  トークン=2256/72
  SLO バジェット残=100.0%  累積コスト=$0.0155

  質問 3: 東京の天気を教えてください。
  回答: 東京の現在の天気は晴れで、気温は25°C、湿度は60%です。
  レイテンシ=3.32s  ツール=['get_weather']  リスク=low  トークン=2479/104
  SLO バジェット残=100.0%  累積コスト=$0.0245

breaker, slo, detector, cost_guard = init_sre_components(
    failure_threshold=3, recovery_timeout=2.0,
)

============================================================
デモ 2: サーキットブレーカー (失敗シミュレーション)
============================================================
  リクエスト 1: failed          CB=CLOSED
  リクエスト 2: failed          CB=CLOSED
  リクエスト 3: failed          CB=OPEN
  リクエスト 4: circuit_open    CB=OPEN
  リクエスト 5: circuit_open    CB=OPEN

  (recovery_timeout=2秒 待機...)
  リクエスト 6 (復帰): success         CB=CLOSED

============================================================
デモ 3: SLO トラッキングとエラーバジェット
============================================================
  SLO: タスク成功率 >= 95%  エラーバジェット: 5%

  [1] success         バジェット残=100.0%  バーンレート= 0.00  SLO=healthy    アラート=[]
  [2] success         バジェット残=100.0%  バーンレート= 0.00  SLO=healthy    アラート=[]
  [3] failed          バジェット残=  0.0%  バーンレート=24000.00  SLO=exhausted  アラート=['burn_rate_warning', 'burn_rate_critical']
  [4] success         バジェット残=  0.0%  バーンレート=18000.00  SLO=exhausted  アラート=['burn_rate_warning', 'burn_rate_critical']
  [5] failed          バジェット残=  0.0%  バーンレート=28800.00  SLO=exhausted  アラート=['burn_rate_warning', 'burn_rate_critical']
  [6] success         バジェット残=  0.0%  バーンレート=24000.00  SLO=exhausted  アラート=['burn_rate_warning', 'burn_rate_critical']
  [7] success         バジェット残=  0.0%  バーンレート=20571.43  SLO=exhausted  アラート=['burn_rate_warning', 'burn_rate_critical']
  [8] success         バジェット残=  0.0%  バーンレート=18000.00  SLO=exhausted  アラート=['burn_rate_warning', 'burn_rate_critical']

============================================================
デモ 4: 不正エージェント検出（3シグナル別）
============================================================
  A) 正常              リスク=low       スコア=0.00  隔離推奨=False
    capability=0.00  frequency=0.00  entropy=0.00
  B) 許可外ツール多用  リスク=low       スコア=0.50  隔離推奨=False
    capability=0.50  frequency=0.00  entropy=0.00
  C) ループ (低entropy)  リスク=medium    スコア=1.00  隔離推奨=False
    capability=0.00  frequency=0.00  entropy=1.00
  D) バースト攻撃     リスク=critical  スコア=18.30  隔離推奨=True
    capability=0.00  frequency=18.30  entropy=0.00

============================================================
デモ 5: コストガード
============================================================
  タスク1(通常): allowed=True  支出=$0.05/0.50  利用率=10%
  タスク2(通常): allowed=True  支出=$0.10/0.50  利用率=20%
  タスク3(通常): allowed=True  支出=$0.15/0.50  利用率=30%

  大型タスク: allowed=False  reason=Estimated cost $0.50 exceeds per-task limit $0.10
  タスク1: 支出=$0.20  利用率= 40%  状態=ok  アラート=[]
  タスク2: 支出=$0.25  利用率= 50%  状態=ok  アラート=['warning']
  タスク3: 支出=$0.30  利用率= 60%  状態=ok  アラート=[]
  タスク4: 支出=$0.35  利用率= 70%  状態=ok  アラート=[]
  タスク5: 支出=$0.40  利用率= 80%  状態=ok  アラート=['warning']
  タスク6: 支出=$0.45  利用率= 90%  状態=throttled  アラート=['warning']
  タスク7: 支出=$0.50  利用率=100%  状態=throttled,killed  アラート=['critical', 'critical', 'critical']
  タスク8: 支出=$0.55  利用率=110%  状態=throttled,killed  アラート=['critical']

model = BedrockModel(
    model_id=MODEL_ID,
    region_name=REGION,
    guardrail_id="your-guardrail-id",
    guardrail_version="DRAFT",
    guardrail_trace="enabled",          # トレース情報を有効化
    guardrail_stream_processing_mode="sync",
    guardrail_redact_input=True,
    guardrail_redact_output=False,
    guardrail_latest_message=True,      # 直近のユーザーメッセージのみ評価
)