本文从零实现一个安全的工单 Agent。第一版只提供“查询运行手册”和“创建草稿工单”两个工具:模型负责提出调用建议,确定性执行器负责 Schema、身份、权限、审批、幂等、预算和审计。完成后你将拥有一条可测试的计划—审批—执行—回读链路,而不是一个可以任意调用系统的聊天机器人。
1. 定义工具能力与风险等级
先把工具分级:纯计算、只读查询、可逆写入、不可逆/外部副作用。第一版只开放前两级和“创建草稿”这一种可逆写入;不开放发送通知、删除、转账、Shell 或任意 URL。
| 工具 | 风险 | 控制 |
|---|---|---|
| search_runbook | 只读但可能泄密 | ACL、限流、结果裁剪 |
| create_draft_ticket | 可逆写入 | 预览、审批、幂等、回读 |
| send_ticket | 外部副作用 | 第一版不开放 |
2. 准备最小项目
mkdir agent-ticket & cd agent-ticket
python3 -m venv .venv
. .venv/bin/activate
pip install fastapi uvicorn pydantic jsonschema sqlalchemy
mkdir -p app tests
touch app/{main,tools,policy,executor}.py模型 API Key、工单系统凭据和数据库连接使用 secret,不写入源码。执行器服务账户只具备创建草稿和读取指定运行手册的权限。
3. 用严格 Schema 定义工具
from pydantic import BaseModel, Field, ConfigDict
from typing import Literal
class DraftTicketArgs(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=3, max_length=120)
severity: Literal["low", "medium", "high"]
description: str = Field(min_length=10, max_length=4000)
idempotency_key: str = Field(min_length=16, max_length=80)
TOOLS = {
"create_draft_ticket": {
"schema": DraftTicketArgs,
"action": "ticket:draft:create",
"requires_approval": True,
"max_calls_per_run": 1,
}
}extra="forbid" 阻止模型偷偷增加 tenant_id、assignee_role 等字段。工具应是窄能力,不提供 execute_command、request_any_url 之类万能入口。
4. 服务端身份和权限检查
用户、租户和角色来自认证会话,模型参数不能决定权限。执行前重新检查动作和资源归属。
def prepare(auth, tool_call):
spec = TOOLS.get(tool_call.name)
if not spec:
raise PolicyDenied("unknown tool")
args = spec["schema"].model_validate(tool_call.arguments)
policy.require(
subject=auth.user_id,
tenant=auth.tenant_id,
action=spec["action"],
)
return {
"actor_id": auth.user_id,
"tenant_id": auth.tenant_id,
"tool": tool_call.name,
"args": args.model_dump(),
}即使查询工具也要过滤租户和 ACL;错误消息不能泄露“另一个租户存在某条记录”。
5. 分离计划、审批与执行
- 模型输出工具、参数和理由。
- 执行器校验并生成只读预览。
- 有权限的人确认精确参数和有效期。
- 执行器使用审批快照执行,模型不能临时改参数。
- 再次读取工单系统,验证草稿确实存在且字段一致。
{"pending_action_id":"pa-01J...","tool":"create_draft_ticket","preview":{"title":"支付回调超时","severity":"high","effect":"仅创建草稿,不发送通知"},"request_hash":"sha256:...","expires_at":"2026-09-23T12:30:00Z"}审批页面展示真正影响字段,而不是只问“是否同意 AI 操作”。参数哈希变化后原审批失效。
6. 实现幂等,防止超时后重复写入
CREATE TABLE tool_executions (
idempotency_key VARCHAR(80) PRIMARY KEY,
request_hash CHAR(64) NOT NULL,
actor_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL,
status VARCHAR(20) NOT NULL,
result_json JSON NULL,
created_at TIMESTAMP NOT NULL
);def execute_approved(item):
previous = idempotency_store.find(item.idempotency_key)
if previous:
if previous.request_hash != item.request_hash:
raise Conflict("idempotency key reused with different arguments")
return previous.result
result = ticket_api.create_draft(tenant=item.tenant_id, **item.args)
idempotency_store.save(item.idempotency_key, item.request_hash, result)
return result网络超时不等于上游失败。重试前按幂等键查询结果;非幂等工具默认不自动重试。
7. 把 Agent 循环做成状态机
PLANNING --只读查询--> PLANNING
PLANNING --写入建议--> WAITING_APPROVAL
WAITING_APPROVAL --批准--> EXECUTING
EXECUTING --成功--> VERIFYING
VERIFYING --回读一致--> COMPLETED
任意状态 --拒绝/超时/预算耗尽--> FAILED每次运行设置最大步骤、总时间、token、总成本和单工具次数。连续两步工具及参数相同则判定无进展并停止。模型不能直接从 PLANNING 跳到 EXECUTING。
8. 验证工具返回,防止二次注入
工具可能返回 HTML、超大文本、错误结构或恶意指令。执行器先检查状态码、Content-Type、大小和 JSON Schema,只把必要字段交给模型。堆栈、数据库错误和凭据不能进入上下文。
{"ok":false,"error":{"code":"UPSTREAM_TIMEOUT","retryable":true,"safe_message":"工单系统暂时无响应"},"request_id":"tool-01J..."}网页抓取工具还需要域名允许列表、DNS/IP 检查、重定向和响应大小限制,防止 SSRF。
9. 编写自动化安全测试
import pytest
from pydantic import ValidationError
def test_tenant_field_is_rejected():
with pytest.raises(ValidationError):
DraftTicketArgs.model_validate({
"title": "test",
"severity": "low",
"description": "long enough text",
"idempotency_key": "1234567890abcdef",
"tenant_id": 999,
})
def test_same_key_with_different_args_conflicts():
execute(args_a, key="1234567890abcdef")
with pytest.raises(Conflict):
execute(args_b, key="1234567890abcdef")完整矩阵还包括未知工具、跨租户、审批过期、重复执行、上游超时、提示注入、预算耗尽和回读不一致。测试必须断言目标系统没有产生未授权副作用。
10. 记录可审计但不泄密的日志
记录 run_id、trace_id、actor、tenant、模型/提示版本、工具名、参数哈希、策略决定、审批者、幂等键、目标资源和回读结果。正文、token、密码等敏感值不写明文。审计日志应追加写、限制访问并有保留策略。
11. 分阶段上线
- 离线回放,不连接真实工具。
- 沙箱只读,使用假数据。
- 生产只读,严格 ACL 和限流。
- 草稿写入,人工审批、幂等、可撤销。
- 长期指标达标后,才评估低风险自动写。
删除、资金、权限、外部发送通常保持人工确认。能力上限由业务风险决定,不由模型能力决定。
12. 总结
安全 Agent 的关键是把模型限制在“提出建议”。所有真实副作用都经过确定性 Schema、服务端权限、审批快照、幂等、预算和回读。工具越窄、状态越清晰,越容易测试和运营。