文档中心 / SDK 教程 / 进阶篇

安全与控制

Cody Agent 能读写文件、跑终端命令、接 MCP——能力越大,越要把访问边界用量熔断持久化策略想清楚。本篇用一条「先跑通、再拆细节」的路线,把 allowed_rootsstrict_read_boundary、熔断器与 stateless() 串起来;模型仍默认使用 qwen3.5-plus(环境变量,见第 01 篇),不在示例里硬编码密钥。

请已配置:CODY_MODEL=qwen3.5-plusCODY_MODEL_API_KEYCODY_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 关闭)后,读文件同样只能发生在 workdirallowed_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_limitcost_limitstep_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 篇 · 事件与可观测性,把运行过程挂到日志与指标上。

← 上一篇 集成 MCP