用 SpendGuard 為 AWS Strands 做預算控管

你的 AWS Strands Agent 在 Bedrock 上跑 Claude Sonnet,結果撞進一個失控的 tool loop。每一個 turn 都再丟一個 InvokeModel HTTP request 出去。沒有 gate 的話,你要等到下個月的帳單 dashboard 才會知道花了多少錢。SpendGuard 透過 hooks=[provider] 把單一個 SpendGuardStrandsHookProvider 插進 Strands 的 typed event bus,讓每一次 model invocation 都在上游 call 真的送出去之前先對預算做 reserve——而且同一個 provider 在 OpenAI、Gemini、Ollama、LiteLLM backend 上完全不用改就能用。

為什麼你會想要這個

一個 provider,所有 backend 通吃。 Strands 的 Model 抽象本來就是多形態的(Bedrock / OpenAI / Anthropic / Gemini / Ollama / LiteLLM)。把 gating 卡在 agent-runtime 邊界,意思是同一個 provider instance 就涵蓋全部;你不用一家一家寫 per-vendor wrapper。
pre-call 直接拒絕,不是事後對帳。 DENY 會在 Strands 把 model HTTP 丟出去之前就 raise DecisionDenied。上游 call 完全不會送出—— 這點由 agent_real_strands_deny demo 驗證,它會 assert DENY 那個 turn 的 counting-stub 命中數維持不變。
Bedrock 優先,但本質上 model-agnostic。 直接讀 result.usage 的 field 形狀(Anthropic 的 input_tokens/output_tokens、OpenAI 的 prompt_tokens/completion_tokens、LiteLLM 正規化後的形狀)——不用去 parse model 字串。
稽核 + 核准 pipeline 跟其他 framework 共用。 這個 provider 寫進的是跟 LangChain、Pydantic-AI、OpenAI Agents、Google ADK 整合同一套 SpendGuard ledger。

安裝設定(60 秒)

pip install 'spendguard-sdk[strands]'

用 demo stack 把 sidecar 拉起來:

git clone https://github.com/m24927605/agentic-spendguard.git
cd agentic-spendguard && make demo-up

接起來

import asyncio

from strands import Agent
from strands.models.bedrock import BedrockModel

from spendguard import SpendGuardClient
from spendguard.integrations.strands import (
    SpendGuardStrandsHookProvider,
)
from spendguard._proto.spendguard.common.v1 import common_pb2


async def main() -> None:
    client = SpendGuardClient(
        socket_path="/var/run/spendguard/adapter.sock",
        tenant_id="00000000-0000-4000-8000-000000000001",
    )
    await client.connect()
    await client.handshake()

    unit = common_pb2.UnitRef(
        unit_id="usd_micros",
        token_kind="output_token",
        model_family="anthropic.claude-3-5-sonnet",
    )
    pricing = common_pb2.PricingFreeze(pricing_version="2026-q2")

    def estimate(invocation):
        # Conservative: reserve 500 atomic per call (above
        # Sonnet's typical ~30-token short response).
        return [common_pb2.BudgetClaim(
            budget_id="my-budget", unit=unit, amount_atomic="500",
            direction=common_pb2.BudgetClaim.DEBIT,
            window_instance_id="my-window",
        )]

    def reconcile(invocation, result):
        usage = result.usage
        total = (
            getattr(usage, "total_tokens", None)
            or ((getattr(usage, "input_tokens", 0) or 0)
                + (getattr(usage, "output_tokens", 0) or 0))
        )
        return [common_pb2.BudgetClaim(
            budget_id="my-budget", unit=unit, amount_atomic=str(total),
            direction=common_pb2.BudgetClaim.DEBIT,
            window_instance_id="my-window",
        )]

    guard = SpendGuardStrandsHookProvider(
        client=client,
        budget_id="my-budget",
        window_instance_id="my-window",
        unit=unit,
        pricing=pricing,
        claim_estimator=estimate,
        claim_reconciler=reconcile,
    )

    agent = Agent(
        model=BedrockModel(
            model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        ),
        hooks=[guard],
    )
    result = await agent.invoke_async(prompt="Say hello in three words.")
    print(result.message)


asyncio.run(main())

你會得到什麼

每一次 Agent.invoke_async() turn 都會做 pre-call 預算 reservation, tool-loop 的每一輪迭代也算在內。
靠形狀涵蓋多家 vendor。 Bedrock / OpenAI / Anthropic / Gemini / Ollama / LiteLLM 全部從 result.usage 的 field 形狀去抽 usage——不用做 model 字串比對。
並行安全。 Strands 每次 attempt 都會配一個全新的 invocation_id; provider 的 stash 是用這個 id 當 key,所以對多個 agent.invoke_async() call 做 asyncio.gather 永遠不會撞在一起。
DEGRADE 預設 fail closed。 要設 SPENDGUARD_STRANDS_FAIL_OPEN=1 (環境變數)或 fail_closed=False(constructor)才會走 dev-only 的 fail-open 行為。

model-backend 涵蓋矩陣

| Backend | 涵蓋程度 | CI 有測 | 備註 | |---------|----------|--------------|-------| | BedrockModel | v1 已驗證 | 有 | Anthropic 形狀的 usage(input_tokens / output_tokens)。AWS 客戶端最吃重的主力。 | | OpenAIModel | v1 已驗證 | 有 | OpenAI 形狀的 usage(prompt_tokens / completion_tokens / total_tokens)。 | | AnthropicModel | v1 已驗證 | 有 | 跟走 Anthropic 的 Bedrock 同一個形狀。 | | LiteLLMModel | v1 已驗證 | 有 | LiteLLM 正規化後的形狀;涵蓋走 LiteLLM 的 Gemini / Cohere / Llama。 | | GeminiModel | v1 已涵蓋 | 間接 | 在錄製的 fixture 矩陣裡是走 LiteLLM。 | | OllamaModel | v1 已涵蓋 | 無 | 透過 Ollama 的相容層走 OpenAI 同一個形狀;不在 CI 矩陣裡。 |

要加一個新 backend = 在 tests/integrations/strands/test_hook_provider.py::test_I02_multi_backend_allow_path 加第四個 row。

常見用法

跨 framework 的 run_id 關聯

from spendguard.integrations.strands import (
    StrandsRunContext, run_context,
)

async with run_context(StrandsRunContext(run_id="my-parent-run-1")):
    result = await agent.invoke_async(prompt="hello")

Strands 的 event bus 本來就會把 invocation_id 一路帶到底,所以 StrandsRunContext 是選用的(不像 LangChain / MAF 你一定要 bind 一個)。只有在你要把一個 Strands run 橋接到另一個 framework 的 parent trace 時才用得到。

DENY 行為

當 request_decision 回傳 DENY,provider 會:

直接 raise DecisionDenied。

Strands 會把它包成 HookExecutionError;呼叫端從 __cause__ chain 去 catch:

try:
    await agent.invoke_async(prompt="...")
except Exception as exc:
    if isinstance(exc.__cause__, DecisionDenied):
        # handle the budget refusal
        ...

model HTTP 完全不會送出(由 agent_real_strands_deny demo 驗證)。
什麼都不會 stash——不會觸發 commit 也不會觸發 release。

DEGRADE 行為

DEGRADE 的意思是 sidecar 回了一個 non-CONTINUE 的 outcome,而 adapter 不應該因此擋下來(例如暫時性的 downgrade)。預設情況下 provider 會 fail closed 並 raise SpendGuardDegradeBlocked。設 fail_closed=False 或 SPENDGUARD_STRANDS_FAIL_OPEN=1 就會放行這次 invocation;這種情況不會產生 commit row(reservation 會被 TTL 掃掉)。

Tool callbacks

透過 before_tool / after_tool 做 per-tool 預算不在 v1 範圍內。Tool 的成本會被併進 parent invocation 的 estimator/reconciler 裡。per-tool gating 追蹤在 D20.1。

串流 `on_message`

v1 不支援 invocation 進行中的串流 token gating;commit 只會在 after_invocation 觸發。當 result.usage 是 None 時(用標準 Strands runtime 的話很少見),會由 estimator-snapshot fallback 來涵蓋這個情況。