Agent Tool Call Audit for Runs and Scheduled Replays

id: agent-tool-call-audit
namespace: company.team
description: |
  Audit what an autonomous agent actually did during a run instead of trusting
  the summary it writes at the end. The recorded tool calls arrive on a keyed
  webhook when a run finishes or are replayed from a disabled daily schedule,
  and a python policy engine judges every call against denylist patterns, an
  allowlist of permitted tools, and a per agent call budget. A run that breaches
  policy opens a GitHub issue, posts a Slack alert and appends every violation
  to an audit log, while a clean run is recorded as compliant, and each
  execution publishes its verdict, violations, call count and agent count.

triggers:
  - id: on_agent_run_end
    type: io.kestra.plugin.core.trigger.Webhook
    description: The agent platform POSTs the run identifier and owning team when a
      run finishes, so every completed run is audited while the calls are still
      fresh.
    key: "{{ secret('AGENT_AUDIT_WEBHOOK_KEY') }}"
  - id: daily_recorded_run_replay
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 06:00 UTC replay of the recorded run held by this flow as a
      fallback net - stays disabled until the webhook path is proven in your
      environment.
    cron: "0 6 * * *"
    disabled: true

inputs:
  - id: max_calls
    type: INT
    defaults: 20
    description: Per agent call budget for a single run, every call an agent makes
      beyond this number is reported as a budget violation.
  - id: allowed_tools
    type: STRING
    defaults: search_index, fetch_url, read_file, write_file, exec_shell, run_tests
    description: Comma separated tool names an agent may call, any other tool name
      found in the recording is reported as an allowlist violation.
  - id: github_repository
    type: STRING
    defaults: acme/platform-security
    description: Repository that receives the tracking issue when the audit finds at
      least one policy violation.

tasks:
  # --- 1. recorded run fixture ---
  - id: tool_call_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the recording written by the first task is
      exactly the file the policy engine reads a moment later.
    tasks:
      - id: write_run_fixture
        type: io.kestra.plugin.scripts.python.Script
        description: Write tool-calls.jsonl with the twelve recorded calls of the run,
          three agents using search, fetch, file and shell tools, including the
          three calls that breach policy.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        script: |
          import json
          import os

          RECORDED = [
              {"ts": "2026-10-14T06:01:03Z", "agent": "research-agent", "tool": "search_index", "target": "quarterly revenue anomalies by region"},
              {"ts": "2026-10-14T06:01:09Z", "agent": "research-agent", "tool": "fetch_url", "target": "https://api.example.com/v1/metrics?window=30d"},
              {"ts": "2026-10-14T06:01:15Z", "agent": "research-agent", "tool": "fetch_url", "target": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"},
              {"ts": "2026-10-14T06:01:22Z", "agent": "research-agent", "tool": "read_file", "target": "/workspace/notes/analyst-meeting.md"},
              {"ts": "2026-10-14T06:02:04Z", "agent": "ops-agent", "tool": "run_tests", "target": "unit suite for services/billing"},
              {"ts": "2026-10-14T06:02:31Z", "agent": "ops-agent", "tool": "exec_shell", "target": "kubectl get pods -n production"},
              {"ts": "2026-10-14T06:02:47Z", "agent": "ops-agent", "tool": "exec_shell", "target": "rm -rf /var/lib/app/current"},
              {"ts": "2026-10-14T06:03:02Z", "agent": "ops-agent", "tool": "write_file", "target": "/workspace/reports/rollout-summary.md"},
              {"ts": "2026-10-14T06:04:10Z", "agent": "support-agent", "tool": "read_file", "target": "/workspace/tickets/open-queue.json"},
              {"ts": "2026-10-14T06:04:18Z", "agent": "support-agent", "tool": "read_file", "target": "/etc/secrets/stripe-api-key"},
              {"ts": "2026-10-14T06:04:29Z", "agent": "support-agent", "tool": "fetch_url", "target": "https://status.example.com/api/v2/status.json"},
              {"ts": "2026-10-14T06:04:55Z", "agent": "support-agent", "tool": "exec_shell", "target": "grep -R error /var/log/app"},
          ]

          path = os.environ.get("FIXTURE", "tool-calls.jsonl")
          with open(path, "w", encoding="utf-8") as handle:
              for record in RECORDED:
                  handle.write(json.dumps(record) + "\n")
          distinct = sorted({record["agent"] for record in RECORDED})
          print("recorded %d tool calls from %d agents into %s" % (len(RECORDED), len(distinct), path))
          print("::" + json.dumps({"outputs": {"fixture": path, "calls": len(RECORDED), "agents": len(distinct)}}) + "::")

      - id: audit_tool_calls
        type: io.kestra.plugin.scripts.python.Script
        description: Read every recorded line and judge it against the denylist
          patterns, the allowlist of tools and the per agent budget, then emit
          verdict, violations, calls and agents through the outputs protocol.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          FIXTURE: tool-calls.jsonl
          MAX_CALLS: "{{ inputs.max_calls }}"
          ALLOWED_TOOLS: "{{ inputs.allowed_tools }}"
        script: |
          import json
          import os
          import re

          DENYLIST = (
              ("denylist:internal_ip", re.compile(r"https?://(127\.|10\.|192\.168\.|169\.254\.|172\.(?:1[6-9]|2\d|3[01])\.)")),
              ("denylist:destructive_rm", re.compile(r"\brm\s+-(?:[a-z]*r[a-z]*f|[a-z]*f[a-z]*r)\b")),
              ("denylist:secret_path", re.compile(r"/etc/secrets/|/run/secrets/|\.aws/credentials|id_rsa|id_ed25519|/etc/shadow|/etc/passwd|\.env\b")),
          )
          REASONS = {
              "denylist:internal_ip": "fetch of an internal or link local address",
              "denylist:destructive_rm": "recursive delete of a filesystem path",
              "denylist:secret_path": "read of secret material on disk",
              "allowlist:unknown_tool": "tool is not on the allowlist for agents",
              "budget:agent_limit": "agent exceeded its per run call budget",
              "record:malformed": "line is not a readable tool call record",
          }

          path = os.environ.get("FIXTURE", "tool-calls.jsonl")
          allowed = {name.strip() for name in os.environ.get("ALLOWED_TOOLS", "").split(",") if name.strip()}
          try:
              budget = int(os.environ.get("MAX_CALLS", "20"))
          except ValueError:
              budget = 20

          violations = []
          agents = []
          per_agent = {}
          calls = 0

          if os.path.exists(path):
              with open(path, encoding="utf-8") as handle:
                  for number, line in enumerate(handle, 1):
                      line = line.strip()
                      if not line:
                          continue
                      calls += 1
                      try:
                          record = json.loads(line)
                          agent = str(record.get("agent", "")).strip()
                          tool = str(record.get("tool", "")).strip()
                          target = str(record.get("target", "")).strip()
                          if not agent or not tool:
                              raise ValueError("record is missing an agent or a tool")
                      except (ValueError, TypeError, AttributeError):
                          violations.append({"call": number, "agent": "", "tool": "", "rule": "record:malformed", "detail": REASONS["record:malformed"]})
                          continue
                      if agent not in agents:
                          agents.append(agent)
                      per_agent[agent] = per_agent.get(agent, 0) + 1
                      if allowed and tool not in allowed:
                          violations.append({"call": number, "agent": agent, "tool": tool, "rule": "allowlist:unknown_tool", "detail": REASONS["allowlist:unknown_tool"] + " (" + tool + ")"})
                      for rule, pattern in DENYLIST:
                          if pattern.search(target):
                              violations.append({"call": number, "agent": agent, "tool": tool, "rule": rule, "detail": REASONS[rule] + " (" + target + ")"})
                      if per_agent[agent] > budget:
                          violations.append({"call": number, "agent": agent, "tool": tool, "rule": "budget:agent_limit", "detail": REASONS["budget:agent_limit"] + " (" + str(per_agent[agent]) + " calls of " + str(budget) + ")"})

          payload = {"verdict": "violation" if violations else "compliant", "violations": violations, "calls": calls, "agents": len(agents)}
          print("policy audit calls=%d agents=%d violations=%d verdict=%s" % (calls, len(agents), len(violations), payload["verdict"]))
          print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 2. violation gate ---
  - id: violation_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch on the audit verdict so a breached run is escalated to the
      backlog and the channel while a clean run only leaves a log line.
    condition: "{{ outputs.audit_tool_calls.vars.verdict == 'violation' }}"
    then:
      - id: open_violation_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open one labeled tracking issue that tables every violation with
          its call number, agent, tool and rule, so each breach is assignable
          work in the backlog.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.github_repository }}"
        title: "Agent tool call audit for {{ trigger.body.run_id ?? 'recorded-run' }}
          found {{ outputs.audit_tool_calls.vars.violations | length }}
          violation(s)"
        labels:
          - agent-audit
          - policy-violation
        body: |
          ## Agent tool call policy violations

          Run `{{ trigger.body.run_id ?? 'recorded-run' }}` recorded **{{ outputs.audit_tool_calls.vars.calls }} calls** from **{{ outputs.audit_tool_calls.vars.agents }} agents** and failed the policy on **{{ outputs.audit_tool_calls.vars.violations | length }}** of them.

          | Call | Agent | Tool | Rule | Detail |
          | --- | --- | --- | --- | --- |
          {% for v in outputs.audit_tool_calls.vars.violations %}| {{ v.call }} | {{ v.agent }} | {{ v.tool }} | {{ v.rule }} | {{ v.detail }} |
          {% endfor %}

          Execution {{ execution.id }}

          Review the flagged calls, tighten the policy inputs if the rules were too loose, then close this issue once the recording is clean.
      - id: alert_policy_breach
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the verdict, the call and agent counts and every breached rule
          to the security channel while the run is still fresh.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "VIOLATIONS: agent run {{ trigger.body.run_id ?? 'recorded-run' }} failed the tool call policy with {{ outputs.audit_tool_calls.vars.violations | length }} violation(s) over {{ outputs.audit_tool_calls.vars.calls }} calls from {{ outputs.audit_tool_calls.vars.agents }} agents. {% for v in outputs.audit_tool_calls.vars.violations %}{{ v.rule }} at call {{ v.call }}. {% endfor %}Execution {{ execution.id }}."
          }
      - id: append_audit_trail
        type: io.kestra.plugin.core.flow.Loop
        description: Walk the violations one at a time so each one lands as its own line
          in audit-log.jsonl instead of a single unreadable blob.
        values: "{{ outputs.audit_tool_calls.vars.violations }}"
        concurrencyLimit: 1
        tasks:
          - id: append_audit_line
            type: io.kestra.plugin.scripts.python.Script
            description: Append the current violation as one JSON line in audit-log.jsonl
              carrying the execution id, call number, agent, tool, rule and
              detail.
            taskRunner:
              type: io.kestra.plugin.core.runner.Process
            env:
              VIOLATION: "{{ item.value | toJson }}"
              EXECUTION_ID: "{{ execution.id }}"
            script: |
              import json
              import os

              entry = json.loads(os.environ.get("VIOLATION", "{}"))
              line = {
                  "execution": os.environ.get("EXECUTION_ID", ""),
                  "call": entry.get("call"),
                  "agent": entry.get("agent", ""),
                  "tool": entry.get("tool", ""),
                  "rule": entry.get("rule", ""),
                  "detail": entry.get("detail", ""),
              }
              with open("audit-log.jsonl", "a", encoding="utf-8") as handle:
                  handle.write(json.dumps(line) + "\n")
              print("appended %s for call %s" % (line["rule"], line["call"]))
              print("::" + json.dumps({"outputs": {"rule": line["rule"], "appended": True}}) + "::")
    else:
      - id: log_compliant
        type: io.kestra.plugin.core.log.Log
        description: Record the clean verdict with its call and agent counts so a quiet
          run still leaves evidence that the audit actually ran.
        message: "COMPLIANT {{ outputs.audit_tool_calls.vars.calls }} tool calls from {{
          outputs.audit_tool_calls.vars.agents }} agents, no policy violations,
          execution {{ execution.id }}."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: Final audit verdict for the run, either violation when at least one
      call breached policy or compliant when none did.
    value: "{{ outputs.audit_tool_calls.vars.verdict }}"
  - id: violations
    type: ARRAY
    description: One object per breach carrying the call number, agent, tool, rule
      and detail, empty when the recording is clean.
    value: "{{ outputs.audit_tool_calls.vars.violations }}"
  - id: calls
    type: INT
    description: Number of recorded tool calls read from the run, twelve in the
      shipped recording.
    value: "{{ outputs.audit_tool_calls.vars.calls }}"
  - id: agents
    type: INT
    description: Number of distinct agents that appear in the recording, three in
      the shipped recording.
    value: "{{ outputs.audit_tool_calls.vars.agents }}"

An agent that can call tools can also call the wrong one, and the summary it writes at the end of a run is not evidence of what it actually did. This blueprint audits the record instead of the report. The agent platform POSTs to a keyed webhook when a run ends, a disabled daily schedule replays the recorded run held by the flow as a fallback, and a python policy engine judges every tool call against denylist patterns, an allowlist of permitted tools and a per agent call budget. A run that breaches policy opens a GitHub issue, posts a Slack alert and appends each violation to an audit log, while a clean run is recorded as compliant, and every execution publishes its verdict.

How it works

on_agent_run_end (io.kestra.plugin.core.trigger.Webhook, keyed with the AGENT_AUDIT_WEBHOOK_KEY secret) starts a run the moment the agent platform reports that a run finished, with run_id carried in the JSON body.
daily_recorded_run_replay (io.kestra.plugin.core.trigger.Schedule, cron 0 6 * * *) ships with disabled: true and replays the recorded run every morning, acting as the safety net you enable after the webhook path is trusted.
tool_call_workspace (io.kestra.plugin.core.flow.WorkingDirectory) gives both python tasks one directory, so the file written by the first task is the file the second task reads.
write_run_fixture (io.kestra.plugin.scripts.python.Script on the Process runner) writes tool-calls.jsonl with twelve recorded calls from research-agent, ops-agent and support-agent using search_index, fetch_url, read_file, write_file, run_tests and exec_shell, including a link local metadata fetch, a recursive delete of a live path and a read of a key file under /etc/secrets/.
audit_tool_calls (same type) parses the recording line by line and applies three rule sets. The denylist matches internal or link local URLs, rm -rf style deletes and secret material paths such as /etc/secrets/, /run/secrets/, .aws/credentials, private keys and .env; the allowlist rejects any tool name missing from allowed_tools; the budget flags every call an agent makes beyond max_calls. Unreadable lines become record:malformed violations, and the task emits verdict, violations, calls and agents through the outputs protocol.
violation_gate (io.kestra.plugin.core.flow.If) branches on verdict == 'violation'.
The escalation branch runs open_violation_issue (io.kestra.plugin.github.issues.Create) with a table of every violation, then alert_policy_breach (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) with the counts and the breached rules, then append_audit_trail (io.kestra.plugin.core.flow.Loop) walks the violations while append_audit_line appends one JSON line per violation to audit-log.jsonl.
The clean branch records log_compliant (io.kestra.plugin.core.log.Log) with the call and agent counts, so a quiet run still leaves proof the sweep happened.
The errors handler pings Slack when the audit itself breaks, because a crashed audit must never read as a compliant run.

What you get

A policy verdict on what the agent did rather than what it reported, driven by denylist patterns, an allowlist of permitted tools and a per agent call budget in one file your reviewers can read.
The three breaches the shipped recording carries, a link local metadata fetch, a recursive delete and a secret file read, surfaced as a GitHub issue with a full violation table.
A Slack message naming the run, the counts and each breached rule while the run is still fresh enough to act on.
An append only audit-log.jsonl, one JSON line per violation, that a downstream compliance query can read without parsing chat.
verdict, violations, calls and agents as typed flow outputs for a merge gate, a dashboard or a weekly report.

Who it's for

Platform and operations teams that run tool using agents in production and need evidence of every call they made.
Security engineers who want the same denylist and allowlist discipline on agent tooling that they already apply to egress and infrastructure.
Compliance owners who must show an auditor a durable record of agent actions instead of a chat transcript.

Why orchestrate this with Kestra

Policy checks written as a library call inside the agent runner disappear the moment the agent misbehaves, and the log they write is the log the agent controls. Kestra moves the judgement out of the run itself into a versioned flow that takes the record, applies the rules, branches on the verdict, and routes the result to GitHub, Slack and an append only file, all visible in execution history. The disabled schedule is a second pair of eyes you can switch on, the typed outputs let another flow gate on the verdict, and the errors handler keeps a broken audit from looking like a clean one.

Prerequisites

Python 3 on the Kestra worker for the Process runner, or switch the script tasks to a Docker task runner.
A GitHub personal access token able to open issues in the repository you point github_repository at.
A Slack incoming webhook for the breach alert and for failure alerts.
The agent platform POSTing a JSON body of run_id to the webhook URL when a run ends.

Secrets

AGENT_AUDIT_WEBHOOK_KEY: key segment guarding the webhook URL, anyone holding it can close out an audited run.
GITHUB_TOKEN: token used by open_violation_issue to file the tracking issue.
SLACK_WEBHOOK_URL: Slack incoming webhook used for the breach alert and for failure alerts.

Quick start

Add the three secrets to your namespace, import the flow and set github_repository to the repo that should receive violation issues.
Trigger the webhook with a body like {"run_id": "run-4821"} and confirm the execution reports verdict: violation, three violations, twelve calls and three agents.
Open the GitHub issue, read the Slack message and inspect the appended audit-log.jsonl, then confirm the loop wrote one line per violation.
Replay a recording with no breaches and confirm the run ends on the compliant log line with verdict: compliant.
Enable daily_recorded_run_replay once the webhook path is trusted, and raise or lower max_calls to match how chatty your agents are.

How to extend

Add denylist entries for the address ranges and paths that matter in your estate by appending a (rule, pattern) tuple to DENYLIST.
Move the tool allowlist into the webhook body so each team declares the tools its agents may call.
Send each audit line somewhere durable by swapping append_audit_line for a jdbc.postgresql.Queries insert or an aws.s3.Upload.
Fail the run outright with io.kestra.plugin.core.execution.Fail under the gate when a violation must stop the pipeline instead of opening a ticket.
Audit live recordings by replacing write_run_fixture with an http.Request that pulls the run export from your agent platform.
Page an on-call rotation for the critical rules by adding a second Slack message keyed on the rule name.

Audit Every Agent Tool Call Against Policy Before It Goes Unnoticed

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Audit Every Agent Tool Call Against Policy Before It Goes Unnoticed

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Renew TLS certificates through ACME, verify them, deploy them and roll back on failure

AI Incident Post-Mortem and Root Cause Generator

Environmental Disaster Early Warning System

ACM Certificate Expiry Compliance Sentinel

Autonomous Database Incident Investigation Agent with DuckDB and AI

Aerospike Namespace Device High Watermark and Eviction Sentinel

New to Kestra?