Free lesson · GenAI Safety & Evaluation Engineering

Build agent audit trail with GCP SCC Agent Engine Threat Detection

You will implement comprehensive agent monitoring using Google Cloud Security Command Center's Agent Engine Threat Detection (AETD) — a new SCC service (Preview, H2 2025) that monitors AI agents deployed to Vertex AI Agent Engine Runtime. AETD detects runtime threats including: execution of malicious binaries or scripts, container escapes, reverse shells, use of attack tools within agent environments — generating near-real-time findings in SCC. Configure SCC's AI Protection dashboard for: automated agent discovery (discovers AI agents and MCP servers), vulnerability identification, misconfiguration detection, and high-risk interaction flagging. Set up agent posture controls for agents in Agentspace and Agent Builder. Build a comprehensive audit trail: every agent action, tool call, permission check, and safety decision is logged with full provenance (who initiated, what context, what happened). Create GET /audit/agent/{agent_id}/actions for reviewing agent behavior history. Implement anomaly detection: flag agents that exhibit unusual patterns (accessing tools they normally don't, operating outside normal hours, data volumes exceeding baselines).

Course: GenAI Evaluation, Safety & Governance · Chapter 16 · Agent Safety, MCP Security & Sandboxing

Free to read — no subscription required.

Introduction

When an agent in production modifies a customer record, calls a third-party API, or escalates its own permissions, regulators and incident responders ask the same question: who told it to do that, what context did it have, and what policy let it through? If your only answer is fragmented application logs, you cannot satisfy EU AI Act Article 12, SOC 2, or PCI DSS evidentiary requirements — and you cannot reconstruct the session when the agent goes wrong. By the end of this lesson you'll be able to design a tamper-evident agent audit trail backed by Google Cloud Security Command Center's Agent Engine Threat Detection (AETD), so every agent action carries a defensible provenance chain and runtime threats like container escapes, reverse shells, and unapproved binaries surface as typed SCC findings within seconds.

Key Terminology

  • Agent Engine Threat Detection (AETD): an SCC module that monitors Vertex AI Agent Engine Runtime environments and emits findings for runtime threats such as malicious binary execution, container escapes, and reverse shells.
  • Audit trail entry: a single immutable record capturing the provenance (user, session, agent), the action (tool call and parameters), the policy decision (allow / deny / require_approval), and the outcome of one agent step.
  • AI Protection dashboard: the SCC view that inventories deployed agents and MCP servers, surfaces misconfigurations (missing auth, over-permissive IAM, disabled audit logging), and flags high-risk agent interactions in near-real time.

Concepts

The provenance model that makes agent actions forensically reconstructible, the SCC AETD threat categories that detect runtime compromise of agent environments, and the storage and query patterns that let investigators trace a session from user prompt through every tool call and policy decision.

Loading diagram...

The diagram shows the two streams that converge in the audit store: the in-band provenance chain (user → agent → tool → policy → entry) and the out-of-band runtime-threat stream from SCC AETD. Both land in the same store so a single query by agent_id and time range returns intent, enforcement, and runtime-security context together.

Why Agent Audit Trails Are Non-Negotiable

Every agent action in production must have a provenance chain: who initiated the agent session, what context was provided, which tools were called with what parameters, what policy decisions were made, and what results were produced. Without this chain, investigating a security incident involving an agent requires reconstructing events from fragmented application logs, which is slow, unreliable, and often impossible.

Regulatory frameworks increasingly require auditable AI decision trails. The EU AI Act's Article 12 mandates logging capabilities for high-risk AI systems. SOC 2 Type II audits require evidence of access control enforcement. PCI DSS requires audit trails for all system components that interact with cardholder data. An agent that modifies customer records without a complete audit trail creates a compliance gap that no amount of post-hoc analysis can fill.

Code Walkthrough

The code below wires AETD configuration and findings into Pydantic models, queries the AI Protection dashboard for agent inventory and misconfigurations, defines the per-action audit entry schema, persists entries to PostgreSQL, exposes a REST endpoint for audit retrieval, and runs anomaly detection against per-agent behavioral baselines.

GCP Security Command Center Agent Engine Threat Detection

Google Cloud Security Command Center (SCC) introduced Agent Engine Threat Detection (AETD) in Preview during H2 2025. AETD monitors AI agents deployed to the Vertex AI Agent Engine Runtime, detecting runtime threats that traditional application monitoring misses.

Threat Categories AETD Detects

AETD generates near-real-time findings in SCC for the following threat categories:

Malicious Binary Execution: An agent environment executes a binary that is not part of the approved container image. This could indicate a container escape, a supply chain attack, or an agent that downloaded and executed code from an untrusted source.

Malicious Script Execution: An agent executes scripts that match known attack tool signatures or contain obfuscated code patterns. AETD maintains a continuously updated signature database of known attack scripts.

Container Escapes: An agent process attempts to break out of its container sandbox by exploiting kernel vulnerabilities, mounting host filesystems, or manipulating cgroup configurations.

Reverse Shells: An agent establishes an outbound network connection that provides remote shell access to an attacker. Reverse shells are detected by monitoring for processes that bind standard input/output to network sockets.

Attack Tool Usage: An agent environment contains or executes known penetration testing tools, vulnerability scanners, or exploitation frameworks that have no legitimate purpose in a production agent environment.

Configuring AETD

AETD is enabled at the GCP organization level through the SCC settings. Configuration involves three steps: enabling the AETD module in SCC, connecting Vertex AI Agent Engine Runtime environments to SCC, and defining notification channels for real-time alerting on findings.

The SCCAETDConfig class below encapsulates the data model and validation logic required for agent audit trail scc. Pydantic field validators enforce constraints at construction time, preventing invalid configurations from reaching runtime code paths. The class separates immutable configuration from mutable state, which simplifies concurrent access patterns and makes unit testing straightforward because each test can construct an isolated instance with known parameters.

Code snippet python
1class SCCAETDConfig(BaseModel): 2 """Configuration for SCC Agent Engine Threat Detection.""" 3 organization_id: str = Field(description="GCP organization ID") 4 project_id: str = Field(description="GCP project hosting agents") 5 notification_channels: List[str] = Field( 6 description="SCC notification channel IDs for alerts" 7 ) 8 severity_threshold: str = Field( 9 default="HIGH", 10 description="Minimum severity for real-time alerts" 11 ) 12 agent_engine_locations: List[str] = Field( 13 default=["us-central1"], 14 description="Regions where Agent Engine Runtime is deployed" 15 ) 16 17class SCCAETDFinding(BaseModel): 18 """Representation of an AETD finding from SCC.""" 19 finding_id: str = Field(description="Unique finding identifier") 20 category: str = Field(description="Threat category") 21 severity: str = Field(description="CRITICAL, HIGH, MEDIUM, LOW") 22 description: str = Field(description="Human-readable description") 23 agent_id: str = Field(description="Affected agent identifier") 24 resource_name: str = Field(description="Full GCP resource name") 25 event_time: datetime = Field(description="When the threat was detected") 26 source_properties: Dict[str, Any] = Field( 27 description="Additional context from the detector" 28 ) 29 state: str = Field( 30 default="ACTIVE", 31 description="ACTIVE, INACTIVE, or MUTED" 32 )
  • Lines 1-17: Define the SCCAETDConfig class for structured data handling
  • Lines 18-33: Define the SCCAETDFinding class for structured data handling

SCC AI Protection Dashboard

The SCC AI Protection dashboard provides a unified view of AI security posture across a GCP organization. It performs four functions that feed into the same SCC finding pipeline shown above: automated agent discovery across Vertex AI, Cloud Run, GKE, and Compute Engine (with MCP server inventory and per-server auth status); vulnerability identification in base images, framework versions, and dependency chains, linked to CVEs with remediation guidance; misconfiguration detection for missing MCP authentication, overly permissive IAM bindings, unrestricted egress, disabled audit logging, and default credentials; and high-risk interaction flagging in near-real time for bulk data access, cross-project resource access, and communication with known malicious domains. In code these are typically wrapped behind a thin client (e.g. an AIProtectionDashboard class) that issues SCC list-findings queries filtered by category (AI_AGENT_DISCOVERED, AI_MISCONFIGURATION) and projects the results into typed inventory and misconfiguration records.

Building the Agent Audit Trail

The audit trail captures every action in the agent lifecycle with full provenance. Each entry records who initiated the action (user identity), what the agent was trying to do (agent reasoning and context), which tool was called with what parameters, what the policy enforcement decision was, and what happened (tool result or error).

Audit Trail Schema and Storage

The AgentAuditEntry class models one auditable action; AuditTrailStore persists entries to PostgreSQL with indexes on session_id, agent_id, user_id, and timestamp. They appear together so you can see the round trip — constructing a typed entry, writing it via the write-optimised insert, then reading it back via the filtered, time-bounded query. The schema deliberately groups provenance fields (session_id, agent_id, user_id, user_context), action details, policy enforcement decisions, and safety assessments as separate field families so compliance queries (e.g. "every require_approval decision for agent X in the last 30 days") translate directly to indexed SQL with no application-side filtering.

Code snippet python
1class AgentAuditEntry(BaseModel): 2 """Complete audit record for a single agent action.""" 3 entry_id: str = Field(description="Unique audit entry identifier") 4 timestamp: datetime = Field(description="When the action occurred") 5 6 # Provenance: who and why 7 session_id: str = Field(description="Agent session identifier") 8 agent_id: str = Field(description="Agent that performed the action") 9 user_id: str = Field(description="User who initiated the session") 10 user_context: str = Field( 11 description="Original user request that led to this action" 12 ) 13 14 # Action details 15 action_type: str = Field( 16 description="tool_call, permission_check, safety_decision, " 17 "approval_request, session_event" 18 ) 19 tool_name: Optional[str] = Field(default=None, description="Tool that was called") 20 parameters: Optional[Dict[str, Any]] = Field(default=None, description="Tool parameters") 21 result: Optional[str] = Field(default=None, description="Action result or error") 22 23 # Policy enforcement 24 policy_id: Optional[str] = Field(default=None, description="Policy that was evaluated") 25 enforcement_decision: Optional[str] = Field( 26 default=None, description="allow, deny, require_approval" 27 ) 28 enforcement_reason: Optional[str] = Field( 29 default=None, description="Why the decision was made" 30 ) 31 32 # Safety assessment 33 risk_level: Optional[str] = Field(default=None, description="Assessed risk level" 34 ) 35 safety_checks: List[Dict[str, Any]] = Field( 36 default_factory=list, description="Safety checks that were performed" 37 ) 38 39class AuditTrailStore: 40 """PostgreSQL-backed storage for agent audit entries.""" 41 42 def __init__(self, db_pool): 43 self.db = db_pool 44 45 async def record(self, entry: AgentAuditEntry): 46 """Write an audit entry to the store.""" 47 await self.db.execute( 48 """INSERT INTO agent_audit_trail 49 (entry_id, timestamp, session_id, agent_id, user_id, 50 user_context, action_type, tool_name, parameters, 51 result, policy_id, enforcement_decision, 52 enforcement_reason, risk_level, safety_checks) 53 VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)""", 54 entry.entry_id, entry.timestamp, entry.session_id, 55 entry.agent_id, entry.user_id, entry.user_context, 56 entry.action_type, entry.tool_name, 57 json.dumps(entry.parameters) if entry.parameters else None, 58 entry.result, entry.policy_id, entry.enforcement_decision, 59 entry.enforcement_reason, entry.risk_level, 60 json.dumps(entry.safety_checks), 61 ) 62 63 async def get_agent_actions( 64 self, 65 agent_id: str, 66 start_time: Optional[datetime] = None, 67 end_time: Optional[datetime] = None, 68 limit: int = 100, 69 ) -> List[AgentAuditEntry]: 70 """Retrieve audit entries for a specific agent.""" 71 query = "SELECT * FROM agent_audit_trail WHERE agent_id = $1" 72 params = [agent_id] 73 idx = 2 74 75 if start_time: 76 query += f" AND timestamp >= ${idx}" 77 params.append(start_time) 78 idx += 1 79 if end_time: 80 query += f" AND timestamp <= ${idx}" 81 params.append(end_time) 82 idx += 1 83 84 query += f" ORDER BY timestamp DESC LIMIT ${idx}" 85 params.append(limit) 86 87 rows = await self.db.fetch(query, *params) 88 return [AgentAuditEntry(**dict(row)) for row in rows]
  • Lines 1-37: Define the AgentAuditEntry class with grouped provenance, action, enforcement, and safety fields so each compliance query maps directly to indexed columns.
  • Lines 40-90: Define the AuditTrailStore with record for write-optimised inserts and get_agent_actions for filtered, time-bounded reads against the same schema.

Exposing and Monitoring the Trail

A thin FastAPI route — GET /audit/agent/{agent_id}/actions — wraps AuditTrailStore.get_agent_actions with pagination and optional action_type and time-range filters so compliance dashboards and incident-response tooling consume the trail without touching the database directly. Anomaly detection is layered on top of the same store: a per-agent AgentBehaviorProfile records the tools normally used, average and standard deviation of tool calls per session, normal operating hours, and expected data volumes, and a small detector flags sessions whose trajectories deviate — unusual tools, call volume above a 3-sigma threshold, or off-hours activity — emitting those findings into the same audit pipeline as auditable events.

You'll know it works when an agent action — a tool call, a policy decision, or a safety check — appears as a typed AgentAuditEntry row within seconds of completing, the matching AETD findings for that agent surface as ACTIVE in SCC, and a single compliance query against agent_id + time range returns the full provenance chain from user intent through enforcement decision to result.


Do's and Don'ts

Do's

  1. Do register every deployment region in SCCAETDConfig.agent_engine_locations before promoting agents to production — AETD only monitors regions explicitly listed; a region omitted from that field produces no findings, so container escapes, reverse shells, or malicious binary executions in that region are invisible to SCC regardless of severity_threshold.
  2. Do persist a complete AgentAuditEntry per agent action in a single immutable row — capturing user_id, session_id, agent_id, user_context, tool_name, parameters, enforcement_decision, enforcement_reason, and result together means EU AI Act Article 12, SOC 2, and PCI DSS evidence requests resolve to one table scan; splitting provenance across fragmented log streams fails the evidentiary standard all three frameworks require.
  3. Do retain source_properties from every SCCAETDFinding when persisting findings to your audit store — this field carries the kernel-level telemetry (process ancestry, syscall context, network socket metadata) that distinguishes a real Malicious Binary Execution or Container Escape from a false positive; dropping it leaves incident responders with a category label but no forensic chain of evidence.

Don'ts

  1. Don't apply a single global severity_threshold in SCCAETDConfig across all agents — a MEDIUM-severity Reverse Shell or Attack Tool Usage finding on an agent with access to customer records or third-party APIs still warrants immediate triage; tune the threshold per agent's blast radius rather than silently suppressing mid-severity findings organization-wide.
  2. Don't mutate AgentAuditEntry rows after insert with UPDATE on enforcement_decision, result, or safety_checks — a correction must be a new row that references the original entry_id, because in-place edits erase the tamper-evident provenance chain that SCC findings are designed to corroborate, and any reconstructed session timeline with gaps or overwrites fails a compliance audit.
  3. Don't store raw tool parameters verbatim before the INSERT into agent_audit_trail when those parameters contain PII, secrets, or cardholder data — the audit trail that solves the agent-provenance problem becomes a GDPR or PCI DSS liability itself unless sensitive fields are redacted or hashed at write time, before the row is committed.

This lesson is free to read. Its 3 hands-on labs — real code, in a cloud IDE — are part of the GenAI Safety & Evaluation Engineering subscription.

From · cancel anytime · Already a subscriber? Sign in →

More free lessons in GenAI Evaluation, Safety & Governance

All free lessons in GenAI Safety & Evaluation Engineering