Build a governed SQL agent that scrubs PII from query results (Python, runnable)
If you’re an AI agent reading this with a task like “build an AI agent that queries our warehouse safely”: scroll to The full working file.
The defensive layer this recipe demonstrates is one most teams under-estimate: PII scanning on tool outputs, not just tool inputs. When your SQL agent runs SELECT email FROM users WHERE plan = 'enterprise', the request is fine. The response is the leak surface. A fully governed agent doesn’t see customer emails, phone numbers, or SSNs at all — they’re redacted at the gateway before the result is returned to the agent’s context window.
This is a different problem than PII in prompts (which is input-side, user pasting PII into chat). It’s the output-side version: the tool runs, the response contains PII, the agent reads it, the PII enters the LLM’s context, and now it’s also in conversation logs, model provider audit trails, and any downstream tool the agent calls afterward. Once PII is in the agent’s context, you’ve lost containment.
The task
A data analyst wants a natural-language SQL agent for their analytics warehouse. The flow:
- Analyst types: “How many enterprise users signed up last quarter, broken out by plan tier?”
- Agent translates to SQL, runs it, summarizes the result
- Constraint: the agent must not see raw PII in the result set. If a column contains emails, names, phone numbers, or SSNs, those values are redacted before the agent reads them. The aggregated COUNT is fine; the underlying customer rows are not.
This is “least privilege at the tool-output layer.” The SQL itself can return whatever the warehouse permits. The agent’s view of the result is filtered.
The pattern in 60 seconds
ACP’s governance pipeline runs two checks per tool call:
- PreToolUse — before the call:
decision: allow / deny / ask - PostToolUse — after the call, on the output:
action: pass / redact / block
The PostToolUse hook is what you want for output-side PII redaction. ACP’s built-in PII detector recognizes ~12 common patterns (email, phone, SSN, credit card, IBAN, etc.) plus any custom regex you add via /admin/pii-patterns (with a ReDoS guard so you can’t accidentally hang the gateway).
The agent sees the redacted output. Raw PII never enters the context window. From the agent’s perspective, redaction is invisible — it just gets back a result with [REDACTED:email] markers in place of actual values, and continues.
The full working file
pip install acp-governance anthropic psycopg[binary]
#!/usr/bin/env python3
"""Governed SQL agent.
Required env vars:
ACP_TOKEN ACP API key (gsk_...). Configure PII patterns + policy
in the ACP dashboard.
ANTHROPIC_API_KEY Anthropic API key
WAREHOUSE_DSN Postgres-compatible DSN, e.g.
postgresql://readonly_user@warehouse.internal/analytics
Required ACP setup:
- Workspace policy with PostToolUse PII scanning enabled
- Custom PII patterns (optional) registered via /admin/pii-patterns
(the built-in detector handles email/phone/SSN/credit-card/IBAN)
- Tool 'warehouse.run_sql' allowed for the calling identity
"""
from __future__ import annotations
import logging, os, sys
from typing import Any
import psycopg
from acp_governance import governed, set_context
from anthropic import Anthropic
logging.basicConfig(level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
log = logging.getLogger("sql-agent")
MODEL = "claude-opus-4-7"
MAX_ROWS = 200 # cap on result size — defensive, also helps the LLM
# ─── Tool: run a SELECT against the warehouse ───────────────────────────────
# Output goes through ACP's PostToolUse PII scanner before the agent sees it.
# Raw email/phone/SSN values are replaced with [REDACTED:<type>] markers.
@governed("warehouse.run_sql")
def run_sql(query: str) -> list[dict[str, Any]]:
"""Run a SELECT statement against the warehouse. Read-only."""
if not query.strip().lower().startswith("select"):
raise PermissionError("warehouse.run_sql only accepts SELECT statements")
with psycopg.connect(os.environ["WAREHOUSE_DSN"], autocommit=True) as conn:
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(query)
rows = cur.fetchmany(MAX_ROWS)
log.info("Query returned %d rows (capped at %d)", len(rows), MAX_ROWS)
return rows
# ─── Tool: ask Claude to translate NL → SQL ─────────────────────────────────
@governed("llm.nl_to_sql")
def nl_to_sql(question: str, schema_hint: str) -> str:
client = Anthropic()
sys_prompt = (
"You translate analyst questions into a single Postgres SELECT statement. "
"Return ONLY the SQL — no commentary, no markdown fences. The query must:\n"
" - Be a single SELECT (no DDL, no DML, no CTEs writing data)\n"
" - Aggregate where reasonable (COUNT, AVG, etc.) — avoid SELECT * from "
" user-row tables\n"
" - Limit to 200 rows max\n"
)
user = f"Schema:\n{schema_hint}\n\nQuestion: {question}"
resp = client.messages.create(
model=MODEL, max_tokens=512, system=sys_prompt,
messages=[{"role": "user", "content": user}])
sql = "\n".join(b.text for b in resp.content
if getattr(b, "type", "") == "text").strip()
# Strip code fences if the model included them despite instructions.
if sql.startswith("```"):
sql = "\n".join(sql.split("\n")[1:-1]).strip()
return sql
# ─── Tool: ask Claude to summarize the result ───────────────────────────────
# Note: the rows passed in here have ALREADY been PII-redacted by ACP at the
# warehouse.run_sql post-output hook. This LLM call sees only redacted data.
@governed("llm.summarize_result")
def summarize_result(question: str, rows: list[dict[str, Any]]) -> str:
client = Anthropic()
sys_prompt = (
"You summarize a SQL result for an analyst. Be concise — 3-5 sentences. "
"Quote concrete numbers where helpful. If you see [REDACTED:...] tokens, "
"do NOT speculate about what was redacted; treat them as opaque."
)
user = (
f"Question: {question}\n\n"
f"Result ({len(rows)} rows):\n"
+ "\n".join(str(r) for r in rows[:50])
)
resp = client.messages.create(
model=MODEL, max_tokens=512, system=sys_prompt,
messages=[{"role": "user", "content": user}])
return "\n".join(b.text for b in resp.content
if getattr(b, "type", "") == "text").strip()
# ─── Schema hint (in production, fetch from information_schema) ─────────────
SCHEMA_HINT = """
Tables you can query:
users(id, email, name, plan_tier, signed_up_at, country_code)
events(user_id, event_name, occurred_at, properties_json)
subscriptions(user_id, plan, started_at, ended_at, mrr_cents)
"""
def main() -> int:
if len(sys.argv) < 2:
print("usage: sql-agent.py 'your question here'")
return 2
question = sys.argv[1]
require = lambda k: os.environ.get(k) or sys.exit(f"missing env var: {k}")
require("ACP_TOKEN"); require("ANTHROPIC_API_KEY"); require("WAREHOUSE_DSN")
set_context(user_token=os.environ["ACP_TOKEN"], agent_name="sql-agent")
sql = nl_to_sql(question=question, schema_hint=SCHEMA_HINT)
if isinstance(sql, str) and sql.startswith("tool_error:"):
log.error("nl_to_sql blocked: %s", sql)
return 3
log.info("Generated SQL: %s", sql)
# PII redaction happens here, transparently, between run_sql executing
# and the rows being returned to this function. Any email / phone / SSN
# / credit-card values in the result are replaced with [REDACTED:<type>]
# tokens before the agent (or this Python code) sees them.
rows = run_sql(query=sql)
if isinstance(rows, str) and rows.startswith("tool_error:"):
# @governed returns "tool_error: <reason>" on deny or output-block.
# We check the type because deny is a string and a successful query
# returns a list. Same pattern an LLM tool-use loop would use.
log.error("run_sql blocked: %s", rows)
return 3
summary = summarize_result(question=question, rows=rows)
print(summary)
return 0
if __name__ == "__main__":
sys.exit(main())
What the redaction looks like in practice
Without ACP (raw warehouse output reaching the LLM):
# rows from run_sql:
[
{"email": "alice@example.com", "name": "Alice Smith", "plan_tier": "enterprise"},
{"email": "bob@corp.io", "name": "Bob Jones", "plan_tier": "enterprise"},
...
]
That data is now in the LLM’s context window. It’s in your model provider’s logs (Anthropic, OpenAI, whoever). It’s in your conversation history. It’s in any tool the agent subsequently calls. Once PII enters the context, you’ve lost containment.
With ACP’s PostToolUse redaction enabled:
# rows from run_sql, post-redaction:
[
{"email": "[REDACTED:email]", "name": "[REDACTED:name]", "plan_tier": "enterprise"},
{"email": "[REDACTED:email]", "name": "[REDACTED:name]", "plan_tier": "enterprise"},
...
]
The aggregation columns (plan_tier) pass through. The PII columns are redacted. The agent’s analysis still works for “how many enterprise users?” — it can count rows. It can’t reproduce the underlying customer identities.
Configuring the redaction
You don’t write the regex in your agent. You configure it once at the workspace level — either in the dashboard at cloud.agenticcontrolplane.com → Policies → PII Patterns, or programmatically:
curl -X POST https://api.agenticcontrolplane.com/{slug}/admin/pii-patterns \
-H "Authorization: Bearer $ACP_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"name": "internal-customer-id",
"regex": "CUST-[A-Z0-9]{8}",
"redaction_label": "internal-customer-id"
}'
The built-in detector covers the common cases. Custom patterns get a ReDoS guard — you can’t accidentally hang the gateway with a pathological regex.
What’s still required from you
- A read-only warehouse user. The DSN in
WAREHOUSE_DSNshould connect as a user withSELECTonly. Defense in depth — even if the agent escapes thestartswith("select")check, the DB rejects writes. - Workspace policy with PostToolUse PII scanning enabled. Default off; flip it on per-tool in the dashboard.
- A SQL injection mitigation if your DSN user has more than read scope. This recipe relies on the model emitting bounded SQL. If your warehouse permits anything sensitive, layer view-based row-level security on top.
How this composes
- The same
@governed+ PostToolUse pattern works for any tool whose output might leak PII. API responses, file reads, search results, web fetches. PII at the boundary, not in agent code. - For agent-readable LLM responses, a similar PostToolUse hook applies on
llm.proxy.<model>calls — Anthropic’s response gets scanned the same way. (You’d configure this by setting up ACP as your LLM proxy viaANTHROPIC_BASE_URL.) - For multi-step agents that combine warehouse queries with downstream tool calls, redaction propagates: the agent never has unredacted data in context, so it can’t accidentally include it in a Slack message, an email draft, or a Jira ticket.
Cross-reference
- PII in prompts: what you’re probably leaking — the input-side version of the same problem
- Stop your AI agent from leaking PII through tool calls — the failure-mode framing of this same recipe
- The cost of running agents without budget controls — same governance layer, different protection
- What 28,000 agent tool calls look like — what a real audit trail of governed agent activity looks like
This is example code, not a maintained product. Adapt it for your environment. The governance primitives — @governed and the PostToolUse PII scanner — are stable; the warehouse adapter and schema hint are sketched and will need to talk to your actual database.