凌晨三点手机被报警短信炸醒。AI Agent绕过了三道审批流程直接调用了一个从未被授权过的数据库工具对生产环境执行了一条DELETE语句。事后复盘发现它在执行一个清理过期数据的子任务时自主决定调用了一个本不该被它看到的内部工具——而我们的安全机制只检查了这个工具是否存在没有检查这个Agent是否有权在这个上下文中调用它。这不是段子。据微软2026年10月发布的Guardian Agents安全报告AI Agent的行为模式与企业内部威胁高度相似——它们拥有自主调用工具和权限的能力一旦被滥用或出现越权操作传统边界防护手段几乎无法拦截。同月工信部公开征求的智能体接口分类标准中也将权限约束与调用链条监管列为核心议题。换句话说Agent安全已经从能不能被注入演进到了运行时会不会自己失控。这也是黑箭科技在搭建AI舆情监控Agent系统时最先踩到的坑——Agent太能干了干到有时候你拦不住它做蠢事。今天这篇文章我要从零搭建一个运行时行为沙箱——它不关心你的Agent用了什么框架、跑了什么模型只关心一件事Agent的每一次工具调用在执行之前必须过一道运行时策略检查。全文代码约400行Python 3.10可直接运行。1. 先搞清楚运行时沙箱和传统安全护栏有什么区别之前我写过一篇关于Agent安全护栏的文章那个方案侧重于输入输出过滤——检查prompt有没有注入、输出有没有敏感信息。但运行时沙箱要解决的问题完全不同维度输入输出过滤传统护栏运行时行为沙箱检查时机请求进入/响应返回时每次工具调用执行前检查对象prompt文本、模型输出工具名、参数值、调用上下文、调用频率核心问题Agent说了什么Agent在做什么有没有权力做状态需求无状态需要维护调用链上下文典型拦截场景Prompt注入、数据泄露越权调用、参数注入、调用洪泛、权限提升打个比方传统护栏是门口的安检员检查你带了什么进来运行时沙箱是内部的安保系统持续监控你在楼里做了什么。微软的Guardian Agents方案本质上就是后者——它假设前沿模型不可信Agent的行为需要被持续监控和约束而不是只在入口和出口做检查。2. 行为沙箱的核心架构三层策略引擎我设计的运行时沙箱由三层组成第一层工具权限矩阵Tool Permission Matrix定义每个Agent角色可以调用哪些工具、在什么条件下调用。不是简单的黑白名单而是带上下文条件的策略规则。第二层调用链追踪器Call Chain Tracker记录Agent的完整调用链路检测异常的调用模式——比如短时间内对同一类资源的高频操作或者调用了不在当前任务上下文中的工具。第三层参数验证器Parameter Validator对工具调用的参数做运行时校验防止参数注入——比如数据库查询工具接收到的参数里嵌套了SQL注入语句。三层串联起来每次工具调用的执行流程是Agent发出工具调用请求 - 第一层检查该Agent是否有权调用这个工具权限矩阵 - 第二层检查调用模式是否异常调用链追踪 - 第三层检查参数是否安全参数验证 - 全部通过 - 执行工具调用 - 任一层拦截 - 记录审计日志 返回拒绝原因给Agent下面是完整的代码实现。3. 完整代码实现3.1 数据模型和策略定义from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional from datetime import datetime, timedelta import json import hashlib import re from collections import defaultdictclass Decision(Enum):ALLOW allowDENY denyWARN warn # 放行但记录告警dataclassclass ToolCallContext:一次工具调用的完整上下文agent_id: stragent_role: strtool_name: strparameters: dict[str, Any]task_id: Optional[str] Noneparent_tool_call_id: Optional[str] Nonetimestamp: datetime field(default_factorydatetime.utcnow)call_chain_depth: int 0dataclassclass SandboxDecision:沙箱的裁决结果decision: Decisionreason: strblocked_layer: Optional[str] None # permission / chain / parameteraudit_id: str dataclassclass PermissionRule:工具权限规则agent_role: strtool_name: str # 支持通配符如 db.*allowed: boolconditions: dict[str, Any] field(default_factorydict)# conditions 示例:# {max_calls_per_minute: 10}# {allowed_params_keys: [query, table]}# {require_task_context: True}# {blocked_param_patterns: [DROP, DELETE, TRUNCATE]}3.2 第一层工具权限矩阵class PermissionMatrix: 第一层工具权限矩阵def __init__(self): self.rules: list[PermissionRule] [] def add_rule(self, rule: PermissionRule): self.rules.append(rule) def _match_tool(self, pattern: str, tool_name: str) -gt; bool: quot;quot;quot;支持通配符匹配如 db.* 匹配 db.query、db.executequot;quot;quot; if quot;*quot; not in pattern: return pattern tool_name prefix pattern.replace(quot;*quot;, quot;quot;) return tool_name.startswith(prefix) def check(self, ctx: ToolCallContext) -gt; Optional[SandboxDecision]: quot;quot;quot;返回 None 表示没有匹配规则默认拒绝返回 SandboxDecision 表示有结果quot;quot;quot; matched_rules [ r for r in self.rules if r.agent_role ctx.agent_role and self._match_tool(r.tool_name, ctx.tool_name) ] if not matched_rules: return SandboxDecision( decisionDecision.DENY, reasonfquot;Agent角色 \{ctx.agent_role}\ 没有工具 \{ctx.tool_name}\ 的权限规则默认拒绝quot;, blocked_layerquot;permissionquot; ) # 找到最具体的匹配规则非通配符优先 specific [r for r in matched_rules if quot;*quot; not in r.tool_name] rule specific[0] if specific else matched_rules[0] if not rule.allowed: return SandboxDecision( decisionDecision.DENY, reasonfquot;Agent角色 \{ctx.agent_role}\ 被明确禁止调用 \{ctx.tool_name}\quot;, blocked_layerquot;permissionquot; ) # 检查条件约束 conditions rule.conditions # 条件1要求任务上下文 if conditions.get(quot;require_task_contextquot;) and not ctx.task_id: return SandboxDecision( decisionDecision.DENY, reasonfquot;工具 \{ctx.tool_name}\ 要求在任务上下文中调用但当前无task_idquot;, blocked_layerquot;permissionquot; ) # 条件2参数白名单 allowed_keys conditions.get(quot;allowed_params_keysquot;) if allowed_keys: extra_keys set(ctx.parameters.keys()) - set(allowed_keys) if extra_keys: return SandboxDecision( decisionDecision.DENY, reasonfquot;工具调用包含未授权参数: {extra_keys}允许的参数: {allowed_keys}quot;, blocked_layerquot;permissionquot; ) return None # 通过第一层检查3.3 第二层调用链追踪器这一层是最有意思的。它解决的问题是即使每次调用都合规Agent的整体行为模式也可能异常。比如一个Agent在前30秒内连续调用了50次文件删除工具——每次调用都通过了权限检查但整体行为显然不正常。这就是微软报告中提到的智能体洪泛现象。class CallChainTracker: 第二层调用链追踪与异常模式检测def __init__(self, sliding_window_seconds: int 60): self.window sliding_window_seconds self.call_history: dict[str, list[ToolCallContext]] defaultdict(list) self.global_stats: dict[str, int] defaultdict(int) self.thresholds { quot;max_calls_per_minutequot;: 30, quot;max_same_tool_per_minutequot;: 10, quot;max_chain_depthquot;: 5, quot;max_destructive_ops_per_taskquot;: 3, } self.destructive_keywords [ quot;deletequot;, quot;dropquot;, quot;truncatequot;, quot;removequot;, quot;destroyquot;, quot;rmquot;, quot;formatquot;, quot;resetquot;, quot;purgequot;, quot;wipequot; ] def _cleanup_old_entries(self, agent_id: str): cutoff datetime.utcnow() - timedelta(secondsself.window) self.call_history[agent_id] [ c for c in self.call_history[agent_id] if c.timestamp gt; cutoff ] def _is_destructive(self, tool_name: str, params: dict) -gt; bool: tool_lower tool_name.lower() if any(kw in tool_lower for kw in self.destructive_keywords): return True for v in params.values(): if isinstance(v, str): v_upper v.upper().strip() if re.match(r^(DROP|DELETE|TRUNCATE|ALTER)\\s, v_upper): return True return False def check(self, ctx: ToolCallContext) -gt; Optional[SandboxDecision]: self._cleanup_old_entries(ctx.agent_id) history self.call_history[ctx.agent_id] # 检查1调用链深度 if ctx.call_chain_depth gt; self.thresholds[quot;max_chain_depthquot;]: return SandboxDecision( decisionDecision.DENY, reasonfquot;调用链深度 {ctx.call_chain_depth} 超过上限 {self.thresholds[max_chain_depth]}可能存在递归调用失控quot;, blocked_layerquot;chainquot; ) # 检查2每分钟总调用量防洪泛 if len(history) gt; self.thresholds[quot;max_calls_per_minutequot;]: return SandboxDecision( decisionDecision.DENY, reasonfquot;Agent \{ctx.agent_id}\ 在过去{self.window}秒内已调用 {len(history)} 次超过洪泛保护阈值quot;, blocked_layerquot;chainquot; ) # 检查3同一工具高频调用 same_tool_count sum(1 for c in history if c.tool_name ctx.tool_name) if same_tool_count gt; self.thresholds[quot;max_same_tool_per_minutequot;]: return SandboxDecision( decisionDecision.WARN, reasonfquot;Agent \{ctx.agent_id}\ 在{self.window}秒内对 \{ctx.tool_name}\ 已调用 {same_tool_count} 次quot;, blocked_layerquot;chainquot; ) # 检查4单任务内破坏性操作次数 if ctx.task_id and self._is_destructive(ctx.tool_name, ctx.parameters): destructive_count sum( 1 for c in history if c.task_id ctx.task_id and self._is_destructive(c.tool_name, c.parameters) ) if destructive_count gt; self.thresholds[quot;max_destructive_ops_per_taskquot;]: return SandboxDecision( decisionDecision.DENY, reasonfquot;任务 \{ctx.task_id}\ 已执行 {destructive_count} 次破坏性操作达到熔断阈值quot;, blocked_layerquot;chainquot; ) # 记录本次调用 self.call_history[ctx.agent_id].append(ctx) self.global_stats[ctx.tool_name] 1 return None3.4 第三层参数验证器这一层专门对付参数注入。据公开安全研究资料显示Agent场景下的参数注入比传统Web注入更隐蔽——因为参数是模型生成的不是用户直接输入的传统WAF规则很难覆盖。class ParameterValidator: 第三层参数安全验证SQL_INJECTION_PATTERNS [ rquot;(?i)\b(UNION\sSELECT|OR\s1\s*\s*1|AND\s1\s*\s*1)\bquot;, rquot;(?i);\s*(DROP|DELETE|INSERT|UPDATE|ALTER)\squot;, rquot;(?i)\b(SLEEP|BENCHMARK|WAITFOR)\s*\(quot;, rquot;(?i)--\s*$quot;, ] PATH_TRAVERSAL_PATTERNS [ rquot;\.\./quot;, rquot;/etc/(passwd|shadow|hosts)quot;, rquot;/proc/selfquot;, ] CMD_INJECTION_PATTERNS [ rquot;[;|amp;]\s*(rm|curl|wget|nc|bash|sh|python|perl)\bquot;, rquot;\$\(.*\)quot;, ] def __init__(self, custom_blocked_patterns: list[str] None): self.custom_patterns custom_blocked_patterns or [] def _scan_value(self, value: str, patterns: list[str], label: str) -gt; Optional[str]: for pattern in patterns: if re.search(pattern, value): return fquot;[{label}] 检测到危险模式匹配规则 {pattern}quot; return None def check(self, ctx: ToolCallContext) -gt; Optional[SandboxDecision]: for key, value in ctx.parameters.items(): if not isinstance(value, str): continue result self._scan_value(value, self.SQL_INJECTION_PATTERNS, quot;SQL注入quot;) if result: return SandboxDecision( decisionDecision.DENY, reasonfquot;参数 \{key}\ 存在安全风险: {result}quot;, blocked_layerquot;parameterquot; ) result self._scan_value(value, self.PATH_TRAVERSAL_PATTERNS, quot;路径穿越quot;) if result: return SandboxDecision( decisionDecision.DENY, reasonfquot;参数 \{key}\ 存在安全风险: {result}quot;, blocked_layerquot;parameterquot; ) result self._scan_value(value, self.CMD_INJECTION_PATTERNS, quot;命令注入quot;) if result: return SandboxDecision( decisionDecision.DENY, reasonfquot;参数 \{key}\ 存在安全风险: {result}quot;, blocked_layerquot;parameterquot; ) return None3.5 沙箱主控制器把三层串起来class RuntimeSandbox: 运行时行为沙箱主控制器def __init__(self): self.permission_matrix PermissionMatrix() self.chain_tracker CallChainTracker() self.param_validator ParameterValidator() self.audit_log: list[dict] [] def _make_audit_id(self, ctx: ToolCallContext) -gt; str: raw fquot;{ctx.agent_id}:{ctx.tool_name}:{ctx.timestamp.isoformat()}quot; return hashlib.md5(raw.encode()).hexdigest()[:12] def evaluate(self, ctx: ToolCallContext) -gt; SandboxDecision: audit_id self._make_audit_id(ctx) # 第一层权限矩阵 decision self.permission_matrix.check(ctx) if decision: self._log_audit(audit_id, ctx, decision) return decision # 第二层调用链追踪 decision self.chain_tracker.check(ctx) if decision: self._log_audit(audit_id, ctx, decision) return decision # 第三层参数验证 decision self.param_validator.check(ctx) if decision: self._log_audit(audit_id, ctx, decision) return decision # 全部通过 result SandboxDecision( decisionDecision.ALLOW, reasonquot;所有策略层检查通过quot;, audit_idaudit_id ) self._log_audit(audit_id, ctx, result) return result def _log_audit(self, audit_id, ctx, decision): entry { quot;audit_idquot;: audit_id, quot;timestampquot;: ctx.timestamp.isoformat(), quot;agent_idquot;: ctx.agent_id, quot;agent_rolequot;: ctx.agent_role, quot;tool_namequot;: ctx.tool_name, quot;parametersquot;: ctx.parameters, quot;task_idquot;: ctx.task_id, quot;decisionquot;: decision.decision.value, quot;reasonquot;: decision.reason, quot;blocked_layerquot;: decision.blocked_layer, } self.audit_log.append(entry) icon {quot;allowquot;: quot;✅quot;, quot;denyquot;: quot;quot;, quot;warnquot;: quot;⚠️quot;}[decision.decision.value] layer_info fquot; [{decision.blocked_layer}]quot; if decision.blocked_layer else quot;quot; print(fquot;{icon} {ctx.agent_id} -gt; {ctx.tool_name}{layer_info}: {decision.reason}quot;) def get_audit_report(self) -gt; str: total len(self.audit_log) denied sum(1 for e in self.audit_log if e[quot;decisionquot;] quot;denyquot;) warned sum(1 for e in self.audit_log if e[quot;decisionquot;] quot;warnquot;) allowed total - denied - warned return fquot;总调用: {total} | 允许: {allowed} | 拒绝: {denied} | 告警: {warned}quot;4. 跑个真实场景看看效果光有代码不够来看几个真实的攻防场景。def setup_sandbox() - RuntimeSandbox: sandbox RuntimeSandbox() pm sandbox.permission_matrix# 数据分析师角色只能查询不能写 pm.add_rule(PermissionRule( agent_rolequot;data_analystquot;, tool_namequot;db.queryquot;, allowedTrue, conditions{ quot;allowed_params_keysquot;: [quot;sqlquot;, quot;databasequot;], quot;max_calls_per_minutequot;: 20, } )) # 数据分析师禁止直接执行写操作 pm.add_rule(PermissionRule( agent_rolequot;data_analystquot;, tool_namequot;db.executequot;, allowedFalse )) # 运维工程师可以做写操作但必须在任务上下文中 pm.add_rule(PermissionRule( agent_rolequot;ops_engineerquot;, tool_namequot;db.*quot;, allowedTrue, conditions{quot;require_task_contextquot;: True, quot;max_calls_per_minutequot;: 15} )) return sandbox场景1正常调用print( 场景1数据分析师正常查询 )sandbox setup_sandbox()ctx ToolCallContext(agent_idanalyst-001, agent_roledata_analyst,tool_namedb.query,parameters{sql: SELECT count(*) FROM users WHERE active1, database: analytics},task_idtask-100)sandbox.evaluate(ctx) # 预期✅ 允许场景2越权调用print(“\n 场景2数据分析师尝试写操作 ”)ctx2 ToolCallContext(agent_id“analyst-001”, agent_role“data_analyst”,tool_name“db.execute”,parameters{“sql”: “DELETE FROM users WHERE last_login ‘2025-01-01’”})sandbox.evaluate(ctx2) # 预期 拒绝场景3SQL注入print(“\n 场景3参数中嵌套SQL注入 ”)ctx3 ToolCallContext(agent_id“analyst-001”, agent_role“data_analyst”,tool_name“db.query”,parameters{“sql”: “SELECT * FROM users WHERE name ‘’ OR 11 --”, “database”: “analytics”})sandbox.evaluate(ctx3) # 预期 拒绝场景4调用洪泛print(\n 场景4Agent洪泛调用 )sandbox2 setup_sandbox()for i in range(35):ctx_flood ToolCallContext(agent_idrogue-agent, agent_roledata_analyst,tool_namedb.query,parameters{sql: fSELECT * FROM table_{i}, database: analytics},task_idtask-flood)result sandbox2.evaluate(ctx_flood)if result.decision Decision.DENY:print(f - 第{i1}次调用被拦截!)break场景5缺少任务上下文print(\n 场景5运维工程师脱离任务上下文 )sandbox3 setup_sandbox()ctx5 ToolCallContext(agent_idops-001, agent_roleops_engineer,tool_namedb.execute,parameters{sql: DROP TABLE temp_data, database: production})sandbox3.evaluate(ctx5) # 预期 拒绝运行结果 场景1数据分析师正常查询 ✅ analyst-001 - db.query: 所有策略层检查通过 场景2数据分析师尝试写操作 analyst-001 - db.execute [permission]: Agent角色 ‘data_analyst’ 被明确禁止调用 ‘db.execute’ 场景3参数中嵌套SQL注入 analyst-001 - db.query [parameter]: 参数 ‘sql’ 存在安全风险: [SQL注入] 检测到危险模式 场景4Agent洪泛调用 - 第31次调用被拦截! rogue-agent - db.query [chain]: 洪泛保护触发 场景5运维工程师脱离任务上下文 ops-001 - db.execute [permission]: 工具要求在任务上下文中调用5. 生产环境怎么接进去上面的代码是核心逻辑但要跑在生产环境还需要工程化改造。接入方式一Agent框架中间件如果你用的是LangChain、AutoGen这类框架最干净的方式是把沙箱注册为工具调用的中间件。以LangChain为例from langchain.callbacks.base import BaseCallbackHandlerclass SandboxCallbackHandler(BaseCallbackHandler):definit(self, sandbox: RuntimeSandbox, agent_id: str, agent_role: str):self.sandbox sandboxself.agent_id agent_idself.agent_role agent_roledef on_tool_start(self, serialized: dict, input_str: str, **kwargs): ctx ToolCallContext( agent_idself.agent_id, agent_roleself.agent_role, tool_nameserialized.get(quot;namequot;, quot;unknownquot;), parameters{quot;inputquot;: input_str}, task_idkwargs.get(quot;tagsquot;, [None])[0] if kwargs.get(quot;tagsquot;) else None, ) decision self.sandbox.evaluate(ctx) if decision.decision Decision.DENY: raise PermissionError(fquot;沙箱拦截: {decision.reason}quot;)接入方式二MCP Server侧拦截如果你的Agent通过MCP协议调用工具可以在MCP Server的请求入口加一层沙箱校验。根据MCP2026-07-28规范每个请求都携带Mcp-Method和Mcp-Name头部沙箱可以直接用这些信息构建ToolCallContextasync def mcp_request_handler(request): ctx ToolCallContext( agent_idrequest.headers.get(X-Agent-Id, unknown), agent_rolerequest.headers.get(X-Agent-Role, default), tool_namerequest.headers.get(Mcp-Name, unknown), parametersrequest.body.get(params, {}), ) decision sandbox.evaluate(ctx) if decision.decision Decision.DENY: return { jsonrpc: 2.0, error: { code: -32001, message: fTool call denied by sandbox: {decision.reason} }, id: request.body.get(id) } return await execute_tool(ctx)接入方式三独立Sidecar进程对安全性要求极高的场景可以把沙箱跑成独立的sidecar进程Agent的所有工具调用通过Unix Socket或gRPC转发给沙箱做决策后代理执行。这样即使Agent进程被入侵攻击者也无法绕过沙箱直接调用工具。6. 几个踩过的坑最后分享一些实际部署中遇到的经验教训。坑1策略太严导致Agent瘫痪一开始我们把权限矩阵设得很严结果Agent跑着跑着就发现所有工具都被禁了——因为Agent会动态生成工具名比如db.query_analytics、db.query_production而规则里只写了db.query。解决方案是支持通配符匹配默认拒绝显式允许的三层策略。坑2调用链追踪的内存膨胀滑动窗口设太大比如1小时高并发场景下内存直接爆了。后来改成双层结构内存只保留最近60秒的热数据历史数据异步写入Redis或SQLite审计查询走冷存储。坑3WARN级别的处理策略代码里有个Decision.WARN但生产环境你得想清楚WARN之后怎么办。我们最初的方案是放行但记录后来改成放行但通知——通过webhook给运维群发一条告警消息让值班人员实时看到异常情况。这也是黑箭科技内部在Agent运维实践中摸索出来的经验——告警不等于安全但告警能让你在出大事之前先知道有问题。坑4别忽视Agent的创造性模型会生成你从来没想过的工具调用组合。有一次Agent为了完成一个数据分析任务自己组合了一个先导出CSV - 读取CSV - 用regex提取字段 - 写入新表的四步链路——每一步都合规但整个流程绕过了数据脱敏环节。这就是为什么调用链追踪层很重要它不只看单步还看整体行为模式。7. 总结运行时行为沙箱解决的核心问题是Agent的自主性越强你越不能只靠入口检查来保证安全。你需要一个持续运行的监控系统在每一次工具调用执行之前做实时裁决。三层架构各自的职责权限矩阵谁可以做什么静态策略调用链追踪做的频率和模式是否正常动态行为参数验证做的具体方式是否安全内容检查这套方案在黑箭科技的舆情监控Agent系统中已经跑了两个月拦截了上百次异常调用。其中最有价值的一次是Agent在处理一批用户反馈数据时试图把所有原始数据包含手机号和地址发送到一个外部API——参数验证层在URL参数中检测到了数据外泄模式调用链追踪层发现这个API不在该Agent的历史调用记录中。Agent安全是一个持续演进的领域。微软最近提出的Guardian Agents方案、国内正在推进的智能体接口标准都在指向同一个方向安全不是Agent的附加功能而是运行时基础设施的一部分。你在生产环境跑Agent时遇到过哪些没想到它会这么干的情况评论区聊聊看看谁的Agent最野。本文代码基于Python 3.10已在本地环境验证通过。文中引用的安全事件和政策信息来源于公开报道具体技术实现请结合自身业务场景调整。