How to audit AI agent actions for security review

Most agents in production are black boxes until something breaks. To audit AI agent actions security, you need a deterministic record of every prompt sent, every tool invoked, and every external side effect—not just the conversational transcript. This guide walks through a concrete pipeline you can stand up in a day, with code you can copy into a Python service and adapt.

Step 1: Instrument the agent loop at three boundaries

Capture data at the model interface, the tool boundary, and the environment mutation point. If you only log the final answer, you cannot reconstruct a prompt-injection attack that silently redirected a file write or exfiltrated data through an allowed tool. The model interface tells you what the agent “thought”; the tool boundary tells you what it “did”; the environment diff tells you what changed.

Wrap every tool function with an audit decorator that serializes inputs and outputs. Keep payloads small; truncate large responses to avoid blowing up storage. Never log secrets—redact them in the wrapper.

import functools, time, json

audit_log = []

def audit_tool(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        record = {
            "type": "tool",
            "tool": fn.__name__,
            "args": kwargs,
            "ts": time.time(),
        }
        try:
            result = fn(*args, **kwargs)
            record["status"] = "ok"
            record["result"] = str(result)[:500]
        except Exception as e:
            record["status"] = "error"
            record["error"] = str(e)
        audit_log.append(record)
        return result
    return wrapper

For the LLM call, subclass or wrap your client. With the OpenAI Python SDK you can log the request body before sending:

import openai

class AuditingClient:
    def __init__(self, base):
        self.base = base
    def create(self, **kwargs):
        audit_log.append({"type": "llm_request", "kwargs": kwargs, "ts": time.time()})
        resp = self.base.chat.completions.create(**kwargs)
        audit_log.append({"type": "llm_response", "content": resp.choices[0].message.content, "ts": time.time()})
        return resp

client = AuditingClient(openai.OpenAI())

If you use an agent framework, hook its callback handler instead of hand-rolling, but the principle holds: nothing leaves the loop unlogged.

Verify success: Run a dummy agent loop that calls one tool and one model completion. Assert len(audit_log) >= 2 and that the tool name appears. Delete the log and re-run to confirm the wrapper is actually invoked.

Step 2: Attach identity and session context to every record

A log without correlation IDs is useless for security review. You must answer: which user triggered the agent, which agent instance ran, and what trace does this action belong to? Use contextvars to propagate a trace ID and actor identity across async calls and thread pools.

import contextvars, uuid

trace_id = contextvars.ContextVar("trace_id")
actor_id = contextvars.ContextVar("actor_id")
agent_id = contextvars.ContextVar("agent_id")

def start_session(user: str, agent: str):
    trace_id.set(uuid.uuid4().hex)
    actor_id.set(user)
    agent_id.set(agent)

def tag(record):
    record["trace_id"] = trace_id.get()
    record["actor"] = actor_id.get()
    record["agent"] = agent_id.get()
    return record

Call tag(record) inside your audit decorator and client wrapper. If your agent spawns sub-agents, derive child trace IDs but preserve a root ID so you can reconstruct the full tree. In multi-tenant systems, the actor field is what separates a compromised tenant from a noisy neighbor.

Verify success: Query audit_log filtered by a known trace_id. You should see every model call and tool invocation from that session in chronological order, each tagged with the same actor.

Step 3: Write logs to append-only storage with integrity

In-memory lists are fine for unit tests, but a security audit requires tamper-evidence. Implement a simple hash chain or use an existing immutable store like S3 Object Lock or a write-once database. The chain makes post-hoc edits detectable.

import hashlib, json

prev_hash = "0" * 64

def persist(entry):
    global prev_hash
    entry["prev"] = prev_hash
    raw = json.dumps(entry, sort_keys=True).encode()
    entry["hash"] = hashlib.sha256(raw).hexdigest()
    prev_hash = entry["hash"]
    with open("audit.log", "a") as f:
        f.write(json.dumps(entry) + "\n")

Any modification of a prior line breaks the chain because subsequent prev values no longer match. For regulated environments, add a signing key so the hash alone isn’t sufficient to forge entries.

Verify success: Alter a line in audit.log, then run a validator that recomputes hashes. It must report a mismatch at the first tampered entry and every entry after it.

Step 4: Define security policies as executable assertions

Auditing is passive; security review needs active guardrails. Encode allowed tools, argument constraints, and egress rules as code that runs before execution and logs violations. This shifts the audit AI agent actions security work from post-mortem to real-time prevention.

ALLOWED_TOOLS = {"search", "read_file", "send_slack"}
INTERNAL_HOSTS = {"intranet.example.com"}

def host_of(url):
    from urllib.parse import urlparse
    return urlparse(url).netloc

def policy_check(record):
    if record["tool"] not in ALLOWED_TOOLS:
        raise PermissionError(f"tool {record['tool']} not permitted")
    url = record.get("args", {}).get("url")
    if url and host_of(url) not in INTERNAL_HOSTS:
        raise PermissionError("external egress blocked")

Call policy_check inside audit_tool before fn(*args, **kwargs). Blocked calls still get logged with status: "denied". This gives reviewers a clear signal of attempted policy breaches—a core part of how you audit AI agent actions security at scale. Keep policies in version control; treat them like firewall rules.

Verify success: Attempt to invoke an unlisted tool. Confirm the call raises and a denied record exists in the log with the correct trace ID.

Step 5: Reconstruct and review sessions offline

A security reviewer needs a timeline, not a raw JSON dump. Write a small reader that groups by trace_id and renders a diff-friendly view. Include the model prompts and tool args side by side so injection attempts are visible.

def replay(trace_id):
    entries = [e for e in load_log() if e.get("trace_id") == trace_id]
    for e in sorted(entries, key=lambda x: x["ts"]):
        if e["type"] == "llm_request":
            print(f"[LLM] {e['kwargs']['messages'][-1]['content'][:80]}")
        elif e["type"] == "tool":
            print(f"[{e['status']}] {e['tool']}({e['args']})")

Reviewers can spot anomalous sequences: a sudden delete_row after an unrelated search with injected text. Store signed reviewer notes alongside the trace for compliance. Rotate reviewers so no single engineer approves their own agent’s actions.

Verify success: Pick a past incident trace and produce the replay output. A second engineer should independently confirm the timeline matches expectations and flag any unexplained tool call.

Step 6: Export for compliance and SIEM

Map your records to a standard schema (OCSF or your SIEM’s JSON). Strip verbose content if regulations require, but keep hashes and denial statuses. A security operations center cares about anomalies, not full prompt text.

{
  "class_uid": 1001,
  "trace_id": "abc123",
  "actor": "svc-agent",
  "tool": "send_slack",
  "status": "ok",
  "time": 1710000000
}

If your agent calls models through n4n.ai, the gateway’s per-token usage metering and automatic fallback logs provide an independent corroboration stream that you can join with your internal trace_id to confirm the agent’s reported model calls match billed usage. That cross-check closes a gap where an agent could lie about which model it used to dodge policy.

Verify success: Ship a sample export to a test SIEM index. Run a detection rule that alerts on status: "denied" and confirm it fires within minutes.

Step 7: Test the audit pipeline with chaos

Audit trails rot if untested. Inject a simulated prompt injection that attempts to call a forbidden tool, and verify the system captures, blocks, and reports it. Treat this like a game day.

@audit_tool
def rm(path): ...

# simulated attack
try:
    policy_check({"tool": "rm", "args": {"path": "/etc/passwd"}})
except PermissionError:
    pass

Run this in CI monthly. The goal is to prove that to audit AI agent actions security you actually have the data when incident response needs it. Also test log storage failure: if the WORM bucket is unreachable, the agent should fail closed, not silently drop audits.

Verify success: The CI job fails if any audit record is missing required fields or the hash chain validates incorrectly. A separate alert fires if audit write latency exceeds a threshold.

Operational notes

Keep audit retention aligned with your compliance window—12 months is common for SOC 2. Rotate signing keys for the hash chain quarterly. Treat the audit log as a production service; it deserves uptime alerts and on-call coverage. The moment you can’t answer “what did the agent do at 2am,” you’ve lost the ability to audit AI agent actions security at all. Build the pipeline before the incident, not during it.