Agent安全基线的内容较多，可以放到下一部分讲讲，这里只是简单实现Agent的身份审批验证相关思路
在Agent 在执行 file.read、exec、message.send、http.request、db.query 等工具时，已经不只是生成文本，而是在真实环境中执行操作。这些操作可能读取本地文件、修改代码、调用外部服务、发送消息，甚至触发部署或审批流程。
例如企业Dify平台，运行一个Claw类型的Agent，都可能存在权限问题，以root运行或者任意文件可写
因此，Agent运行时必须明确回答几个问题：
- 后续审计、追踪、审批和权限判断应该基于哪一组身份信息？
- 谁最初触发了这次 Agent 执行？
- 当前实际执行工具调用的是哪个 Agent 或子 Agent？
- Agent 是否是在代表某个用户执行操作？
- 工具调用发生在哪个会话、哪次执行，以及哪个宿主运行时中？
为了便于统一管理，需要一个统一的控制面，功能包括
- 接入不同 Agent：让 LangChain、OpenClaw、MCP、自研 Agent 都能把事件上报进来。
- 收集身份上下文：知道是谁触发、哪个 Agent 执行、在哪个会话、哪次执行里调用了工具。
- 记录工具调用：记录调用了什么工具、访问了什么资源、做了什么动作、结果是什么。
- 生成审计日志：把不同框架的调用记录统一成可查询、可追踪的审计日志。
- 为后续管控预留能力：以后可以加授权判断、人工审批、风险告警、策略管理。
控制面主要在Agent注册、请求进入，工具调用前后采集事件并生成审计日志，授权，审批和风险判断可作为后续拓展，授权由企业同意认证系统来控制
实现方案-框架映射
不同框架不需要使用完全相同的字段名，但需要能映射到本文定义的身份实体、执行上下文和审计事件。
- LangChain / LangGraph：将 Runtime context、thread、run、tool call 和 callbacks / middleware 中的信息映射到运行时身份上下文。
- Claude Tool Use：将请求方应用作为宿主运行时，将 tool_use.id 映射为工具调用标识；会话、执行实例和用户身份需要由应用层补齐。
- MCP：将 MCP client、server、transport、OAuth token 或 HTTP Header 中的身份信息映射到调用方、执行环境和工具调用上下文。
- Cursor-like IDE Agent：将 IDE 登录用户、workspace、session、run、tool dispatcher 和 approval 状态映射到身份上下文和审计事件。
- OpenAI Agents SDK：将 RunContext、tool call、session / thread 等运行信息映射到执行上下文。
- 自研 Agent：不要求固定 envelope 结构，但必须在工具边界传递运行时身份上下文，并生成审计事件。
实现方案-身份管理
一次 Agent 执行中至少需要区分以下角色：
- 最初发起者：最早触发这次执行的一方，可以是用户、系统任务、Webhook、Cron 或另一个服务。
- 实际执行者：当前真正执行工具调用的一方，通常是主 Agent 或某个子 Agent。
- 被代表用户：Agent 代表其执行操作的用户，可选。例如企业应用中由 Agent 代用户调用 API。
- 直接委托者：把任务委托给当前执行者的一方，可选。例如主 Agent 调用子 Agent。
实现方案-审计事件
每次工具调用至少需要记录以下信息：
- 发生时间：工具调用发生的时间，用于排序、排查和合规审计。
- 执行环境：工具调用发生在哪个宿主运行时中。
- 工具信息：被调用的工具名称和本次工具调用标识。
- 身份信息：最初发起者、实际执行者、被代表用户和直接委托者。
- 执行上下文：会话、执行实例和追踪链路。
- 操作对象：被访问或修改的资源，以及执行的动作。
- 决策结果：本次调用是允许、拒绝，还是需要人工审批。
- 决策原因：策略命中、审批结果或系统判断的说明。
审计日志不应保存完整敏感内容、token 或 secret。
LangChain
首期只需控制面提供两个能力：
- 注册 Agent：记录 Agent 是谁、归谁管、默认能使用哪些工具。
- 记录审计事件：记录谁在什么上下文中调用了什么工具、访问了什么资源、结果是什么。
创建 Agent
-> 向控制面注册 Agent 身份、owner、workspace 和可用工具
调用 Agent
-> 宿主应用认证用户
-> 生成身份上下文
-> 通过 CallbackHandler 或 contextvars 传入 LangChain / LangGraph
调用工具
-> CallbackHandler 拦截工具调用
-> 工具调用前上报 tool_call_started
-> 执行真实工具
-> 工具调用后上报 tool_call_finished 或 tool_call_failedAgent 创建与注册
Agent 创建完成后，向控制面注册一次。注册接口建议幂等，避免服务重启或配置重载时重复写入。
tools = [read_file, run_shell]
agent = create_agent(
model="gpt-5-nano",
tools=tools,
)
# 也可以在注册的时候，由控制面提供 agent_id
control_plane.register_agent({
"agent_id": "agent-code-review",
"runtime": "langchain",
"owner": "team-a",
"workspace": "repo-frontend",
"tools": [tool.name for tool in tools],
})CallbackHandler 采集审计事件
CallbackHandler 适合记录 tool start / end / error，不侵入工具实现。
class GovernanceCallbackHandler(BaseCallbackHandler):
def __init__(self, identity_context):
self.context = identity_context
self.tool_names = {}
def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, **kwargs):
tool_call_id = str(run_id)
tool_name = serialized.get("name")
self.tool_names[tool_call_id] = tool_name
control_plane.audit_event({
"type": "tool_call_started",
"tool": tool_name,
"tool_call_id": tool_call_id,
"parent_run_id": str(parent_run_id) if parent_run_id else None,
"context": self.context,
"input_summary": input_str[:200],
})
def on_tool_end(self, output, *, run_id, **kwargs):
tool_call_id = str(run_id)
control_plane.audit_event({
"type": "tool_call_finished",
"tool": self.tool_names.get(tool_call_id),
"tool_call_id": tool_call_id,
"context": self.context,
"status": "success",
})
def on_tool_error(self, error, *, run_id, **kwargs):
tool_call_id = str(run_id)
control_plane.audit_event({
"type": "tool_call_failed",
"tool": self.tool_names.get(tool_call_id),
"tool_call_id": tool_call_id,
"context": self.context,
"status": "error",
"error_type": type(error).__name__,
})
# 调用 Agent 时，由宿主应用生成身份上下文，并通过 callback 带入本次执行：
identity_context = {
"user_id": "user-123",
"agent_id": "agent-code-review",
"runtime": "langchain",
"session_id": "session-1",
"run_id": "run-1",
"trace_id": "trace-1",
"workspace": "repo-frontend",
"trust_level": "authenticated_user",
}
agent.invoke(
{"messages": messages},
config={"callbacks": [GovernanceCallbackHandler(identity_context)]},
)Tool Wrapper / Middleware 进行权限控制
如果后续需要在工具执行前做强授权或人工审批，建议放在 tool wrapper、middleware 或 LangGraph node 边界。
def governed_tool(tool_name, real_tool):
def wrapper(*args, runtime_context, **kwargs):
decision = control_plane.authorize({
"context": runtime_context,
"tool": tool_name,
"params_summary": summarize(args, kwargs),
})
if decision["result"] == "deny":
raise PermissionError(decision["reason"])
return real_tool(*args, **kwargs)
return wrapperOpenClaw
创建 / 加载 Agent
-> 注册 Agent 身份、owner、workspace 和可用工具
请求进入
-> OpenClaw runtime 生成身份上下文
-> hook event 携带上下文
调用工具
-> before_tool_call 上报 tool_call_started
-> 执行真实工具
-> after_tool_call 上报 tool_call_finished 或 tool_call_failedAgent注册
优先在插件启动或 runtime 启动时注册。
async function registerAgent(agentInfo) {
await controlPlane.registerAgent({
agent_id: agentInfo.agentId,
runtime: "openclaw",
owner: agentInfo.owner,
workspace: agentInfo.workspace,
tools: agentInfo.tools,
});
}Plugin Hook 采集审计事件
export default definePluginEntry({
id: "governance",
name: "Governance",
register(api) {
api.on("before_tool_call", async (event) => {
auditQueue.enqueue({
type: "tool_call_started",
tool: event.toolName,
tool_call_id: event.toolCallId,
context: event.context,
params_summary: summarizeParams(event.params),
});
});
api.on("after_tool_call", async (event) => {
auditQueue.enqueue({
type: event.error ? "tool_call_failed" : "tool_call_finished",
tool: event.toolName,
tool_call_id: event.toolCallId,
context: event.context,
status: event.error ? "error" : "success",
});
});
},
});
Comments NOTHING