Claude Managed Agents 如何给会话设置花费上限budget 字段、usage 事件与 budget_reached 暂停【免费下载链接】claude-cookbooksA collection of notebooks/recipes showcasing some fun and effective ways of using Claude.项目地址: https://gitcode.com/GitHub_Trending/an/claude-cookbooks无人值守的 agent 没有天然的成本停止点带 web search 的调研任务可能多抓几百个页面写代码的循环可能反复重试一个不稳的测试。claude-cookbooks 仓库的 CMA_cap_session_spend.ipynb 演示了用 Managed Agents 的会话预算session budget解决这个问题在创建会话时设定一个强制的花费上限平台按公开列表价累计该会话所有 thread 的 token 成本累计值触顶时会话以budget_reached停止原因进入idle暂停文件、工具状态和对话都保持完整之后调高上限任务可以从暂停处继续。本文按照该 notebook 的顺序走通设置budget、读取session.usage快照、检查budget_reached暂停、用sessions.update调整或移除上限的完整路径。准备条件beta header 与客户端版本预算功能走标准的managed-agents-2026-04-01beta headerbudget参数和session.usage上的成本字段要求anthropic0.121.0。按 managed_agents/README.md 的 Getting started 说明先设置ANTHROPIC_API_KEY再在 Jupyter 中打开 notebook 从上到下运行每个 notebook 会自己安装依赖。该 notebook 的安装单元格内容是%%capture %pip install -qU anthropic0.121.0 python-dotenv%%capture与%pip是 notebook 内的写法在普通 Python 环境中等价于pip install -qU anthropic0.121.0 python-dotenv。客户端初始化如下import os import anthropic from dotenv import load_dotenv load_dotenv() BETAS [managed-agents-2026-04-01] MODEL os.environ.get(COOKBOOK_MODEL, claude-sonnet-5) client anthropic.Anthropic()COOKBOOK_MODEL是可选的环境变量用来覆盖默认模型未设置时使用 notebook 里的默认值claude-sonnet-5。创建演示用的 environment 与 agent演示 agent 用来写一份竞争格局简报配了 web search 和 web fetchprompt 里没有任何限制它会读多少来源的约束——notebook 原文说这种开放性正是预算要管的东西env client.beta.environments.create( namebudget-demo, config{type: anthropic_cloud, networking: {type: unrestricted}}, betasBETAS, ) analyst client.beta.agents.create( namemarket_analyst, descriptionWrites sourced competitive landscape briefs., model{id: MODEL}, systemYou write competitive landscape briefs for product teams. Given a market, use web search and web fetch to find the notable vendors, their positioning, and recent moves. Read primary sources rather than aggregators. Write the brief to /mnt/session/outputs/brief.md with a Sources section listing every URL you relied on. Keep researching until you are confident the brief is complete., tools[ { type: agent_toolset_20260401, configs: [{name: web_search}, {name: web_fetch}], } ], betasBETAS, ) print(f{analyst.name}: {analyst.id} v{analyst.version})在 sessions.create 上设置 budget 字段sessions.create的budget字段用来设上限形状是{type: limit, max_list_cost: {currency: USD, amount: 10}}选数字之前notebook 列出了四条规则amount是货币最小单位的整数字符串50是 50 美分2500是 $25.00。它始终是字符串小数美分会被拒绝因此不存在浮点舍入。USD是唯一接受的货币API 返回的所有成本金额包括usage.list_cost都用同一编码。上限按公开列表价累计模型 token 成本覆盖会话内的每个 thread包括 subagent thread。无论是否有协议折扣都按列表价计所以上限会早于或等于实际扣费触发且可以从session.usage加公开价目表复算出这个数字。会话可能运行的每个模型都需要有公开列表价否则创建失败错误为model_not_budgetable。不传budget即无上限。预算只能在创建时附上没有带预算创建的会话之后永远不能再加所以要一开始就决定。notebook 里把上限设为amount: 1010 美分故意设低以便几分钟后就能看到停止。创建会话的完整调用from decimal import Decimal def usd(money) - str: Render an integer minor-unit amount (50 fifty cents) as dollars. return f${Decimal(money.amount) / 100:.2f} session client.beta.sessions.create( agentanalyst.id, environment_idenv.id, titleLandscape brief: observability platforms, budget{ type: limit, max_list_cost: {currency: USD, amount: 10}, }, initial_events[ { type: user.message, content: [ { type: text, text: Write a competitive landscape brief on the observability platform market: the main vendors, how each positions itself, and any notable moves in the last year., } ], } ], betasBETAS, ) print(session.id, session.status) print(budget:, usd(session.budget.max_list_cost))usd辅助函数把最小单位金额渲染成美元后文所有打印都用它。创建成功后打印的budget: $0.10就是回读到的上限。从事件流读取 session.usage 快照除了常规的agent.message和agent.tool_use事件流在每次会话落回idle时都会交付一个session.usage快照包含累计 token 数、跟踪中的list_cost和配置好的budget——回合结束时不用第二次调用sessions.retrieve就知道花了多少。带预算时结束回合的停止原因有两种agent 正常完成是end_turn上限先触发是budget_reachedstop_reason None with client.beta.sessions.events.stream(session.id, betasBETAS) as stream: for ev in stream: if ev.type agent.tool_use: print(f[{ev.name}] {str(ev.input)[:80]}) elif ev.type session.usage: print(f usage: {usd(ev.usage.list_cost)} of {usd(ev.budget.max_list_cost)}) elif ev.type session.status_idle: stop_reason ev.stop_reason.type print(f[idle] stop_reason{stop_reason}) break assert stop_reason budget_reached, ( fexpected budget_reached, got {stop_reason}: lower the cap in step 3 or give the agent a bigger task )最后的 assert 是 notebook 自带的判定方式如果拿到的停止原因不是budget_reached比如 agent 在触顶前就结束了提示去调低上限或换更大的任务。notebook 中记录的一次运行输出如下文档示例[web_search] {query: Datadog positioning observability platform 2025} [web_search] {query: New Relic observability platform positioning 2025} ... usage: $0.12 of $0.10 [idle] stop_reasonbudget_reached检查 budget_reached 暂停后的会话budget_reached是暂停不是失败会话停在idleusage反映到目前为止的全部花费容器里 agent 在触顶前写下的内容都还在不需要重跑任何东西。注意上限是在模型请求之间强制的所以跨线的请求会先执行完再暂停记录下来的list_cost可能略高于max_list_cost示例中是 $0.12 对 $0.10 的上限——设上限时要把这“一个请求”的余量考虑进去。paused client.beta.sessions.retrieve(session.id, betasBETAS) print(status: , paused.status) print(list cost: , usd(paused.usage.list_cost)) print(cap: , usd(paused.budget.max_list_cost)) print(tokens: in, paused.usage.input_tokens, out, paused.usage.output_tokens) print(active secs: , round(paused.usage.active_seconds, 1))文档示例输出status: idle list cost: $0.12 cap: $0.10 tokens: in 471 out 606 active secs: 13.7再看事件日志能知道 agent 走多远它做过的工具调用以及如果它已经开始写最后一条消息。深度调研中的 agent 触顶时可能什么都没写所以消息要按可选项处理events list(client.beta.sessions.events.list(session.id, limit1000, betasBETAS)) tool_calls [ev for ev in events if ev.type agent.tool_use] messages [ev for ev in events if ev.type agent.message] print(ftool calls before the cap: {len(tool_calls)})示例输出文档示例tool calls before the cap: 11并且没有agent.message——agent 触顶时还在收集来源。用 sessions.update 调整上限sessions.update接受与创建时相同的budget形状。把max_list_cost提到已消耗成本之上暂停解除会话自动从停止时的状态接着被中断的那个回合往下跑不需要补发任何消息。反过来把上限降到已消耗成本含相等以下会被 400 拒绝max_list_cost必须保持在已消耗列表价之上这样更新永远不可能把会话卡死在它自己的历史里。try: client.beta.sessions.update( session.id, budget{type: limit, max_list_cost: {currency: USD, amount: 1}}, betasBETAS, ) except anthropic.BadRequestError as e: print(lower rejected:, e.message) client.beta.sessions.update( session.id, budget{type: limit, max_list_cost: {currency: USD, amount: 500}}, betasBETAS, ) print(cap raised to $5.00)第一个调用1 美分低于已消耗的 $0.12演示拒绝第二个调用把上限提到amount: 500即 $5.00。示例输出文档示例lower rejected: Error code: 400 - {type: error, error: {type: invalid_request_error, message: budget.max_list_cost must be greater than the sessions consumed list cost}, request_id: req_staging_011CdiPi8iPJKWdiuurZuDkB} cap raised to $5.00调高上限后再连事件流会话先回到running接着干直到正常结束with client.beta.sessions.events.stream(session.id, betasBETAS) as stream: for ev in stream: if ev.type session.status_running: print([resumed]) elif ev.type agent.tool_use: print(f[{ev.name}]) elif ev.type session.status_idle: print(f[idle] stop_reason{ev.stop_reason.type}) break done client.beta.sessions.retrieve(session.id, betasBETAS) print(final list cost:, usd(done.usage.list_cost), of cap, usd(done.budget.max_list_cost))示例输出文档示例一串[web_search]、[web_fetch]、[write]工具事件之后出现[idle] stop_reasonend_turn最后打印final list cost: $1.83 of cap $5.00——任务在被中断的位置上跑完了最终成本仍在上限之内。要确认产物还在可以像 notebook 一样从事件日志里把 agent 写的brief.md取出来取最后一次write防止有修订brief for ev in client.beta.sessions.events.list(session.id, limit1000, betasBETAS): if ( ev.type agent.tool_use and ev.name write and ev.input.get(file_path, ).endswith(brief.md) ): brief ev.input[content] # keep the last write, in case of revisions移除 budget 是单向操作更新时传budgetNone清除上限从此无上限而完全不传budget则保持当前上限不变。这是两个不同的请求None是显式移除缺省是保留。移除同样是单向的原因和“只能在创建时附上”一样会话一旦没有预算就再也加不回来。上限想调高调低多少次都行移除则是一扇门。uncapped client.beta.sessions.update(session.id, budgetNone, betasBETAS) print(budget:, uncapped.budget)示例输出文档示例budget: None。清理archive 会话与 environmentfrom utilities import wait_for_idle_status wait_for_idle_status(client, session.id) client.beta.sessions.archive(session.id, betasBETAS) client.beta.environments.archive(env.id, betasBETAS) print(archived)utilities即仓库中的 managed_agents/utilities.py。wait_for_idle_status的存在是因为一个竞态SSE 流里的session.status_idle事件可能比sessions.retrieve报出status idle略早紧跟在流结束后的archive()会 400报 cannot be archived while its status is running。这个辅助函数用短轮询吸收掉这段窗口在流式循环结束后、archive()之前调用。archive 保留记录用于审计与检索但会拆掉运行中的容器CMA_operate_in_production.ipynb 对 archive 与 delete 的区别有更完整的说明。可选分支不持流时用 session.budget_reached webhook如果你的会话队列规模很大、无法为每个会话持有连接notebook 给出的替代方式是订阅session.budget_reachedwebhook它在一个会话触顶时触发一次payload 携带会话 id监督进程据此决定是调高上限还是让会话保持暂停。webhook 在 Console 的 Settings → Webhooks 下做一次注册会得到一个只在创建时显示一次的whsec_...签名密钥应存入你的密钥管理器。CMA_operate_in_production.ipynb 里附了参考处理器校验签名后按event[event_type]分发session.budget_reached分支调review_budget(session_id)——要么调高预算让会话继续要么保持暂停并通知运维。该代码块在源 notebook 中明确标注是参考实现而非可运行单元依赖 FastAPI需要拷进你自己的服务。适用边界与限制把预算放进整体成本控制时notebook 的结论是它是每会话的控制约束的是单次运行而不是组织、workspace 或一天的流量因此与你在更高层设置的花费限制组合使用而不是替代。适用场景是没有人盯着的会话cron 驱动的部署、webhook 触发的 agent、coordinator 扇出量取决于数据的长多 agent 任务——CMA_plan_big_execute_small.ipynb 就有一个把会话budget当作 fan-out 护栏的用法。限制方面需要记住的有四点上限按公开列表价而非实际扣费计且强制发生在模型请求之间记录成本可能略超上限amount只接受货币最小单位的整数字符串USD是唯一支持的货币会话里可能运行的模型若没有公开列表价创建直接失败model_not_budgetable预算只能在创建时附加、移除不可逆这两个“单向性”决定了上限要在开跑前就定好。【免费下载链接】claude-cookbooksA collection of notebooks/recipes showcasing some fun and effective ways of using Claude.项目地址: https://gitcode.com/GitHub_Trending/an/claude-cookbooks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考