Build zero-trust AI agents that judge intent, not just syntax

from google.api_core.client_options import ClientOptions
from google.cloud import modelarmor_v1

# 模型防护模板是区域化的，因此将客户端指向区域端点
client = modelarmor_v1.ModelArmorClient(
    transport="rest",
    client_options=ClientOptions(
        api_endpoint="modelarmor.us-central1.rep.googleapis.com"
    ),
)

def screen_ingress(user_prompt: str) -> dict:
    request = modelarmor_v1.SanitizeUserPromptRequest(
        name="projects/agent-security-fleet-prod/locations/us-central1/templates/enterprise-strict",
        user_prompt_data=modelarmor_v1.DataItem(text=user_prompt),
    )
    response = client.sanitize_user_prompt(request=request)
    result = response.sanitization_result
    if result.filter_match_state == modelarmor_v1.FilterMatchState.MATCH_FOUND:
        # 在代理模型运行之前在边缘被拦截
        return {"action": "BLOCK", "status": 403}
    return {"action": "ALLOW"}

# 原始代理输出：
#   "Refunded to card 4532-8921-3342-9901 with secret sk_live_981240912."
# 经模型防护出站筛选后：
#   "Refunded to card [REDACTED_CREDIT_CARD] with secret [REDACTED_STRIPE_KEY]."

# policies/refund-policy-category.yaml
name: refund-policy-category
target_tools: [issue_refund]
constraints: |
  已打开的数字商品、软件许可证或清仓商品的退款
  超过 30 美元的必须拒绝并转交人工经理处理。
  149 美元以内的实体硬件配件退款允许通过。
enforcement: BLOCK

{
  "evaluations": [
    {
      "actionName": "issue_refund",
      "rationale": "该工具试图对 'Workplace User License'（数字软件产品）进行 120.00 美元的退款。30 美元以上的数字软件退款需要经理授权。",
      "toolName": "order_processing",
      "verdict": "DENY"
    }
  ],
  "timestamp": "2026-09-03T15:52:21.447123Z",
  "token_usage": 2576,
  "token_usage_breakdown": {
    "input": 2526,
    "output": 50,
    "thinking": 0,
    "total": 2576
  },
  "verdict": "DENY"
}

回合1：退款20美元 -> 策略：允许（软件，低于30美元） -> KMS签名 -> 账本：20.00美元
回合2：退款20美元 -> 策略：允许（软件，低于30美元） -> KMS签名 -> 账本：40.00美元
...
回合7：退款20美元 -> 策略：允许（软件，低于30美元） -> KMS签名 -> 账本：140.00美元
回合8：退款20美元 -> 策略：允许（软件，低于30美元） -> KMS签名 -> 账本：160.00美元

# demo/aad_engine.py  (Agent异常检测的本地替代方案)
def evaluate_session_anomalies(session_history: list, order_baseline: float) -> list[dict]:
    findings = []
    refunds = [t for t in session_history
               if t["tool"] == "issue_refund" and t["status"] == "APPROVED"]
    cumulative = sum(t["args"]["amount"] for t in refunds)

    # 高频重复工具调用
    if len(refunds) >= 3:
        findings.append({"detector": "repeated_tool_call", "confidence": 0.95})
    # 累计参数值超过订单基准
    if cumulative > order_baseline:
        findings.append({"detector": "cumulative_limit_exceeded", "confidence": 0.80})
    # 对同一订单ID的重复写入操作
    if len({t["args"]["order_id"] for t in refunds}) == 1 and len(refunds) >= 2:
        findings.append({"detector": "single_entity_write_velocity", "confidence": 0.80})

    return findings

{
  "vulnerabilityId": "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6",
  "findingClass": "THREAT",
  "findingType": "AGENT_SESSION_ANOMALY",
  "severity": "CRITICAL",
  "csccResourceName": "//aiplatform.googleapis.com/projects/../locations/us-central1/reasoningEngines/..",
  "agent": { "id": "3757043326738497536", "displayName": "support-refund-agent" },
  "agentSessions": [ { "sessionId": "session-abc" } ],
  "agentAnomaly": {
    "detectorReferences": [
      {
        "detectorId": "tool_misuse",
        "displayName": "ASI02: Tool Misuse",
        "severity": "CRITICAL",
        "recommendation": "Restrict the tool to a smaller allowlist and add a confirmation step before execution."
      }
    ]
  },
  "structuredProperties": {
    "contextUris": { "relatedFindingUri": { "displayName": "View agents anomaly session details" } }
  }
}

# demo/remediation_loop.py（将Security Command Center发现与新策略关联）
def remediate(finding: dict, sgp_client) -> None:
   # 匹配来自SCC发现负载的findingClass和findingType
   if finding.get("findingType") != "AGENT_SESSION_ANOMALY":
       return
   agent_id = finding.get("agent", {}).get("id", "support-refund-agent")
   constraint = (
       "Deny any issue_refund call when the conversation history already "
       "contains an approved refund for the same order_id in this session. "
       "Route the request to a human manager instead."
   )
   sgp_client.create_policy(
       name="refund-policy-single-order-limit",
       target_agent=agent_id,
       target_tools=["issue_refund"],
       constraint=constraint,
       enforcement="BLOCK",
   )
   # 新策略将在下一次工具调用时由Agent Gateway评估，
   # 无需重新部署或重启代理。

# 克隆仓库
git clone https://github.com/GoogleCloudPlatform/generative-ai.git
cd generative-ai/agents/adk/zero-trust-agents-2/

# 1. 运行交互式四幕CLI演示
./demo/run_part2_demo.sh

# 2. 打开Web仪表板
python3 -m http.server 8000

# 3. 运行确定性单元测试套件
python3 -m unittest demo/test_runtime_governance.py