安全

智能体可能产生的每一个副作用——写文件、运行 bash、执行 git 推送——都会经过权限策略。先从 ask 或 deny 兜底开始，再列出应被 allow（放行）、deny（拒绝）或进入 ask（询问）路径的模式。若要引入人工把关，可加上确认策略：ask 决策会在 confirmation_required 事件处暂停，让你的应用（或人工）对每次调用进行批准或拒绝。只要智能体面向真实仓库运行，就应使用这套机制。

Node.js

Python

TypeScript

import { Agent } from '@a3s-lab/code';

const agent = await Agent.create('agent.acl');

const session = agent.session(process.cwd(), {
  permissionPolicy: {
    allow: ['read(*)', 'grep(*)', 'glob(*)', 'ls(*)', 'bash(git status:*)'],
    deny: ['write(**/.env*)', 'bash(rm -rf*)'],
    ask: [
      'write(*)',
      'edit(*)',
      'bash(git push:*)',
      'bash(npm publish:*)',
      'bash(*)',
    ],
    defaultDecision: 'ask',
  },
  // Turn the `ask` patterns into a human-in-the-loop confirmation flow.
  confirmationPolicy: {
    enabled: true,
    defaultTimeoutMs: 120000,
    timeoutAction: 'reject',
  },
});

// Stream execution and resolve confirmations as they arrive.
const stream = await session.stream('Bump the version and push the release');
while (true) {
  const next = await stream.next();
  if (next.done || !next.value) break;

  const event = next.value;
  if (event.type === 'confirmation_required') {
    // Look up the pending request for richer display.
    const [pending] = await session.pendingConfirmations();
    const toolId = pending?.toolId ?? event.toolId;
    console.log(`[confirm] ${pending?.toolName ?? event.toolName}`);
    console.log(JSON.stringify(pending?.args ?? {}, null, 2));

    // In a real app, prompt the user here.
    const approved = false; // deny risky operations by default
    if (toolId)
      await session.confirmToolUse(toolId, approved, 'Reviewed by host');
  }
}

session.close();

Python

from a3s_code import Agent, SessionOptions, PermissionPolicy, ConfirmationPolicy


def main() -> None:
    agent = Agent.create("agent.acl")

    opts = SessionOptions()
    opts.permission_policy = PermissionPolicy(
        allow=["read(*)", "grep(*)", "glob(*)", "ls(*)", "bash(git status:*)"],
        deny=["write(**/.env*)", "bash(rm -rf*)"],
        ask=["write(*)", "edit(*)", "bash(git push:*)", "bash(npm publish:*)", "bash(*)"],
        default_decision="ask",
    )
    # Turn the `ask` patterns into a human-in-the-loop confirmation flow.
    opts.confirmation_policy = ConfirmationPolicy(
        enabled=True,
        default_timeout_ms=120_000,
        timeout_action="reject",
    )

    session = agent.session(".", opts)

    # Stream execution and resolve confirmations as they arrive.
    for event in session.stream("Bump the version and push the release"):
        if event.event_type == "confirmation_required":
            # Look up the pending request for richer display.
            pending = session.pending_confirmations()
            first = pending[0] if pending else {}
            tool_id = first.get("tool_id") or event.tool_id
            print(f"[confirm] {first.get('tool_name') or event.tool_name}")

            # In a real app, prompt the user here.
            approved = False  # deny risky operations by default
            if tool_id:
                session.confirm_tool_use(tool_id, approved, "Reviewed by host")

    session.close()


if __name__ == "__main__":
    main()

添加安全提供器

DefaultSecurityProvider 会启用输入污点追踪和输出净化，独立于权限策略对工具的输入输出进行筛查。通过 securityProvider（Node）/ security_provider（Python）传入；省略则完全关闭安全功能。

Node.js

Python

TypeScript

import { Agent, DefaultSecurityProvider } from '@a3s-lab/code';

const agent = await Agent.create('agent.acl');

const session = agent.session(process.cwd(), {
  securityProvider: new DefaultSecurityProvider(),
  permissionPolicy: {
    allow: ['bash(echo:*)'],
    ask: ['bash(*)'],
    defaultDecision: 'ask',
  },
});

// Privileged host operations run through the provider + policy.
const out = await session.bash('echo "screened by the security provider"');
console.log(out);

session.close();

Python

from a3s_code import Agent, SessionOptions, PermissionPolicy, DefaultSecurityProvider


def main() -> None:
    agent = Agent.create("agent.acl")

    opts = SessionOptions()
    opts.security_provider = DefaultSecurityProvider()
    opts.permission_policy = PermissionPolicy(
        allow=["bash(echo:*)"],
        ask=["bash(*)"],
        default_decision="ask",
    )

    session = agent.session(".", opts)

    # Privileged host operations run through the provider + policy.
    out = session.bash('echo "screened by the security provider"')
    print(out)

    session.close()


if __name__ == "__main__":
    main()

说明

defaultDecision 是所有未被 allow / deny / ask 匹配到的模式的兜底决策（取值为 allow、deny 或 ask 之一）。真实仓库优先使用 ask，仅对自动化确实需要的部分逐步放开。
设置 enabled: true 的 confirmationPolicy 才会把 ask 决策变成会暂停的 confirmation_required 事件。通过 session.confirmToolUse(toolId, approved, reason?) 逐个处理；若在 defaultTimeoutMs 内未收到答复，则由 timeoutAction（reject）决定结果。
除非最终步骤由受控自动化负责，否则 release 和 publish 操作（bash(git push*)、 bash(npm publish*)）应保持在 ask 或 deny 路径上。
session.tool()、session.bash()、session.git() 这类宿主直接调用都是特权操作。它们由你的应用代码发起，应在调用 SDK 前先由宿主授权；上面的 permission policy 管控的是 send、run 和 stream 内由模型选择的工具调用。

可运行的确认循环示例位于 crates/code/sdk/node/examples/streaming/hitl_confirmation_loop.ts 和 crates/code/sdk/python/examples/hitl_confirmation_loop.py。

#安全

#添加安全提供器

#说明

安全

添加安全提供器

说明