安全与控制
Cody Agent 能读写文件、跑终端命令、接 MCP——能力越大,越要把访问边界、用量熔断和持久化策略想清楚。本篇用一条「先跑通、再拆细节」的路线,把 allowed_roots、strict_read_boundary、熔断器与 stateless() 串起来;模型仍默认使用 qwen3.5-plus(环境变量,见第 01 篇),不在示例里硬编码密钥。
请已配置:CODY_MODEL=qwen3.5-plus、CODY_MODEL_API_KEY、CODY_MODEL_BASE_URL=https://coding.dashscope.aliyuncs.com/v1。若要在代码里显式指定,可使用 Cody().model("qwen3.5-plus").base_url(...).api_key(...)。
从一整段脚本开始
下面把本篇要点写进同一个异步入口:多根目录、严格读边界、熔断、以及流式里识别熔断 chunk。跑通后再看各小节解释每个调用在防什么。
import asyncio from cody.sdk import Cody from cody.core.errors import CircuitBreakerError async def main(): # Monorepo:主仓在 app,另允许 shared / proto;strict 限制读也在这些根下 client = ( Cody() .workdir("/workspace/app") .allowed_root("/workspace/shared") .allowed_roots(["/workspace/shared", "/workspace/proto"]) .strict_read_boundary() .circuit_breaker( max_tokens=500_000, max_cost_usd=5.0, max_steps=50, loop_detect_turns=5, ) .build() ) async with client: # run():超限直接抛 CircuitBreakerError try: result = await client.run("列出 proto 目录下的 README 并总结") print(result.output) except CircuitBreakerError as e: print("熔断:", e.reason, e.tokens_used, e.cost_usd) # stream():熔断以 chunk 形式到达 async for chunk in client.stream("同上,流式输出"): if chunk.type == "circuit_breaker": print(chunk.content) break asyncio.run(main())
allowed_root 会往列表里追加路径;若再调用 allowed_roots([...]),会整体替换为传入列表。生产里通常只用 allowed_roots 一次性写全,或连续多次 allowed_root。
allowed_roots:Monorepo 多目录边界
workdir 是主工作区;当 Agent 需要读写到其他包(例如公共库、IDL 仓库)时,用 allowed_root / allowed_roots 声明额外允许的根路径。工具层会据此拒绝越界路径,避免「任务在 app,却顺手改了 /etc」这类事故。
client = (
Cody()
.workdir("/workspace/app")
.allowed_root("/workspace/shared")
.allowed_roots(["/workspace/shared", "/workspace/proto"])
.build()
)
路径请使用真实绝对路径;与第 01 篇一致,模型仍依赖环境变量中的 qwen3.5-plus 配置。
strict_read_boundary:读也锁在边界内
默认情况下,写操作受根边界约束更严;从 v1.9.2+ 起,链上 strict_read_boundary()(或传 enabled=False 关闭)后,读文件同样只能发生在 workdir 与 allowed_roots 并集内,减少敏感文件被「无意浏览」的风险。
client = Cody().workdir(".").strict_read_boundary().build()
熔断器:用量与死循环护栏
熔断在单次 run / stream 过程中累计 token、估算费用、工具步数,并在检测到循环模式时中止。可通过 Cody().circuit_breaker(...) 覆盖;也可传入 CircuitBreakerConfig 对象(见 cody.sdk.config)。
引擎内置默认值与 Builder 关键字默认值不一致:未调用 circuit_breaker() 时走配置合并后的 core 默认;一旦在 Builder 里调用 circuit_breaker() 且使用参数默认值,则采用 SDK 侧那一套。下表分两列便于你对照。
| 触发维度 | 含义 | Core 典型默认 | Cody().circuit_breaker() 参数默认 |
|---|---|---|---|
max_tokens |
单次运行累计 token 超过则熔断(多轮工具调用会快速堆高)。 | 约 1_000_000 |
200_000 |
max_cost_usd |
按 model_prices 估算费用,超过美元阈值则熔断。 |
10.0 |
5.0 |
max_steps |
工具调用步数上限;0 表示不限制。 |
0(无上限) |
0(无上限) |
loop_detect_turns |
连续若干轮工具输出高度相似时视为死循环并熔断。 | 6 |
6 |
run():捕获 CircuitBreakerError
同步语义下,熔断表现为异常。reason 取值包括 token_limit、cost_limit、step_limit,以及循环检测触发的 loop_detected。
from cody.core.errors import CircuitBreakerError client = Cody().circuit_breaker( max_tokens=500_000, max_cost_usd=5.0, max_steps=50, loop_detect_turns=5, ).build() try: result = await client.run("任务") except CircuitBreakerError as e: # e.reason: token_limit / cost_limit / step_limit / loop_detected e.reason e.tokens_used e.cost_usd
stream():处理 circuit_breaker chunk
流式路径下Runner 可能发出类型为 circuit_breaker 的 chunk,而不是让异常直接冒泡到应用层;读到后应结束消费或提示用户。
async for chunk in client.stream("任务"): if chunk.type == "circuit_breaker": print(chunk.content) break
stateless():CI 与无状态服务
v1.11.0+ 起,链式调用 stateless() 可启用无状态模式:不向磁盘写入会话库、审计、文件历史、项目记忆等持久化数据。适合 GitHub Actions、短时 Serverless、或「每次任务独立、不落盘」的网关形态。
client = Cody().workdir(".").stateless().build() # 不写入任何磁盘文件(可按需在 stateless 之后再挂自定义 audit 等)
若在无状态基础上仍要可观测性,可在 stateless() 之后用 Builder 再注入自定义 audit_logger 等(内存或远端),与「默认不写本地 sqlite」不冲突。
小结
本篇覆盖了:多根目录(allowed_roots)、严格读边界(strict_read_boundary)、熔断(表格中的四维阈值与 run/stream 两种处理面),以及 无状态(stateless())。下一步建议阅读 第 10 篇 · 事件与可观测性,把运行过程挂到日志与指标上。