Introduction
Most “AI agent” proofs-of-concept stumble the moment they touch a real enterprise database. The problem isn’t SQL—it’s contracts, governance, concurrency, and auditability. This article lays down a practical blueprint for letting agents read and write safely against production-grade RDBMS (PostgreSQL, SQL Server, Oracle, MySQL), with patterns that survive compliance reviews and Friday night incident calls.
Why agents + databases is different
Enterprise databases carry regulated data, strict uptime targets, and shared ownership across apps, BI, and batch jobs. An agent that improvises queries or writes without receipts will break guarantees around RLS/OLS, PII masking, transactional integrity, and change management. The fix is to make the database a contracted surface: pre-authorized query shapes, typed tools, and observable side effects.
Reference architecture (at a glance)
Contract layer: A machine-readable spec that defines what the agent may ask (parameterized queries, allowed tables/views, max row counts), how it may write (stored procedures only, or approved upserts), and which roles it can impersonate.
Access layer: A thin server that exposes typed tools (read-only query, explain, metrics, mutation via stored procedures). It enforces RLS/OLS, data masking, and row limits, and injects idempotency keys and savepoints for writes.
Governance layer: Secrets management, role mapping, sensitivity labels, redaction rules, query allowlisting, and least-privilege policies.
Observability layer: Per-call logs with SQL fingerprints, bind parameters (hashed), latency, rows, result age, and receipts (e.g., order_id).
Change layer: CDC streams (Debezium/Oracle GoldenGate/SQL Server CDC) feeding caches, search indices, and vector stores; schema registry + contract tests in CI.
The contract: what the agent is allowed to do (and prove)
Use a concise YAML contract the agent must satisfy on every call.
# file: contracts/enterprise_db_v1.yaml
role: "EnterpriseDBAgent"
scope: >
Read via approved views/parameterized queries; write only via stored procedures with idempotency keys.
Never return raw PII; apply masking rules and row/column filters per user_role.
governance:
rls_impersonate_as: "${user_role}" # ex: 'support_tier2'
sensitivity_ceiling: "Confidential" # higher → summarize, don't show raw
max_rows: 2000
max_latency_ms: 5000
allow_reads_from: ["vw_customer_health", "vw_orders_rolling", "vw_tickets_open"]
allow_procedures: ["sp_create_support_case", "sp_update_order_status"]
pii_masking: ["email", "phone", "ssn_last4"]
output:
type: object
required: [summary, citations, sql_fingerprints, receipts, next_steps]
properties:
summary: {type: string, maxWords: 120}
citations: {type: array, items: string} # view/proc names used
sql_fingerprints: {type: array, items: string} # normalized SHA of SQL
receipts: {type: array, items: string} # e.g., case_id=..., order_id=...
next_steps: {type: array, items: string, maxItems: 5}
Read patterns that won’t page your DBA
Views over tables: Point agents at governed views (already RLS/OLS-safe) instead of raw tables.
Parameterized queries only: No free-form SQL; use named parameters with server-side binding.
Cardinality & freshness guards: Enforce row caps and compute max(updated_at) to report result staleness.
Explainers before answers: For complex results, provide a short EXPLAIN-based rationale (index used, filter selectivity) to help humans assess risk/latency.
Minimal-span citations: Report the exact views/metrics involved (e.g., vw_customer_health@v4).
Example: safe read tool (Python + SQLAlchemy)
# tools/db_read.py
from pydantic import BaseModel, Field
from sqlalchemy import text
from typing import Dict, Any, List
import hashlib, time
class ReadArgs(BaseModel):
view: str
params: Dict[str, Any] = {}
max_rows: int = 1000
user_role: str
ALLOWED_VIEWS = {"vw_customer_health", "vw_orders_rolling", "vw_tickets_open"}
def sql_fingerprint(sql: str) -> str:
return hashlib.sha256(sql.encode()).hexdigest()[:16]
def safe_read(conn, a: ReadArgs):
assert a.view in ALLOWED_VIEWS, "View not allowed"
sql = f"SELECT * FROM {a.view} WHERE 1=1"
# Add optional, allowlisted filters
for k in list(a.params.keys()):
if k not in {"customer_id", "region", "since"}:
a.params.pop(k) # strip unknown filters
if "customer_id" in a.params: sql += " AND customer_id = :customer_id"
if "region" in a.params: sql += " AND region = :region"
if "since" in a.params: sql += " AND updated_at >= :since"
sql += f" ORDER BY updated_at DESC LIMIT {min(a.max_rows, 2000)}"
start = time.time()
rows = conn.execute(text(sql), a.params).mappings().all()
latency_ms = int((time.time() - start) * 1000)
# Masking example
for r in rows:
if "email" in r: r["email"] = r["email"].split("@")[0][:3] + "***@***"
return {
"rows": rows,
"latency_ms": latency_ms,
"sql_fingerprint": sql_fingerprint(sql),
"citations": [a.view]
}
Write patterns you can defend in audit
Stored procedures only: Funnel writes through approved procedures with input validation and permission checks.
Idempotency keys: Pass a client-generated key; procedures must upsert or no-op on duplicates.
Transactional discipline: Wrap calls in a transaction with savepoints; roll back on partial failure.
Receipts: A successful write returns a business receipt (case_id, order_id) plus the change stamp (TX id).
CDC & GLUE: Ensure the write lands on your CDC stream for downstream caches/indices.
Example: mutation tool with receipts
# tools/db_write.py
from pydantic import BaseModel
from sqlalchemy import text
from typing import Dict, Any
class CreateSupportCaseArgs(BaseModel):
customer_id: int
subject: str
severity: str
idem_key: str
user_role: str
ALLOWED_PROCS = {"sp_create_support_case", "sp_update_order_status"}
def create_support_case(conn, a: CreateSupportCaseArgs):
assert a.user_role in {"support_tier1", "support_tier2"}
sql = text("""
SELECT * FROM sp_create_support_case(
:customer_id, :subject, :severity, :idem_key, :requested_by
)
""")
with conn.begin(): # transaction boundary
rec = conn.execute(sql, {
"customer_id": a.customer_id,
"subject": a.subject[:200],
"severity": a.severity,
"idem_key": a.idem_key,
"requested_by": a.user_role
}).mappings().first()
# Procedure returns {case_id, status, applied:boolean}
assert rec and rec.get("case_id")
return {
"receipt": f"case_id={rec['case_id']}",
"applied": rec["applied"],
"citations": ["sp_create_support_case"]
}
PostgreSQL procedure sketch
-- idempotent support case creation
CREATE OR REPLACE FUNCTION sp_create_support_case(
p_customer_id INT,
p_subject TEXT,
p_severity TEXT,
p_idem_key TEXT,
p_requested_by TEXT
) RETURNS TABLE(case_id BIGINT, applied BOOLEAN) AS $$
DECLARE
v_case_id BIGINT;
BEGIN
-- idempotency: return existing
SELECT id INTO v_case_id FROM support_cases WHERE idem_key = p_idem_key;
IF FOUND THEN
RETURN QUERY SELECT v_case_id, FALSE;
RETURN;
END IF;
-- role check (cheap example)
IF p_requested_by NOT IN ('support_tier1','support_tier2') THEN
RAISE EXCEPTION 'unauthorized role %', p_requested_by USING ERRCODE='42501';
END IF;
INSERT INTO support_cases(customer_id, subject, severity, idem_key)
VALUES (p_customer_id, left(p_subject,200), p_severity, p_idem_key)
RETURNING id INTO v_case_id;
RETURN QUERY SELECT v_case_id, TRUE;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;
Governance & security (non-negotiables)
RLS/OLS & role mapping: Map agent “user roles” to DB roles; every query runs as that role.
PII masking by default: Apply masks in views; never trust prompt logic to hide PII.
Sensitivity ceilings: If a result includes a column above the user’s label ceiling, return a qualitative summary instead of values.
Secrets & rotation: Use workload identity or vault-issued short-lived credentials; never embed secrets in prompts.
Schema drift policy: Contract tests fail the build if a view/proc signature changes without a corresponding contract update.
Performance & reliability
Bounded queries: LIMITs, timeouts, cancel tokens; reject Cartesian joins by policy.
Index-aware hints (sparingly): Prefer DBA-tuned indexes; only add hints via allowlisted patterns.
Read replicas: Point analytical reads at replicas; fence writes to primaries.
Result caching: Cache safe, small resultsets with TTL and user/role keys; include cache headers in responses.
Backpressure: If latency > SLO or queue grows, the access layer should throttle or degrade gracefully.
Observability & audit
Log per call:
Who/role, purpose string (user ask), citations (views/procs), SQL fingerprint, bind param hash, rows/latency, sensitivity level, and receipt IDs.
Emit metrics for p95 latency, error rate, top fingerprints, and blocked attempts.
End-to-end example: “Create a support case for any churn-risk customer in EMEA with < 3 logins last week”
Agent runs a read against vw_customer_health with filters (region='EMEA', logins_7d<3, churn_risk='HIGH') and cap 200 rows.
It summarizes top 5 customers (masked emails), cites vw_customer_health, includes result freshness (e.g., “as of 18 minutes ago”).
On operator approval, it calls sp_create_support_case per customer with unique idempotency keys; collects receipts case_id=….
It returns a final audit block: citations, SQL fingerprint(s), receipts, and next steps (assign owner, SLA).
CI/CD & safety rails
Golden traces: 10–20 canonical prompts with expected views/procs and result shapes.
Contract tests: Validate allowlists, RLS impersonation, masking, and sensitivity ceilings in CI.
Shadow mode: Log proposed SQL/procs and compare to allowlists before enabling writes.
Feature flags & rollback: Gate new procedures/columns; be able to flip the agent to read-only instantly.
Common failure modes (with fixes)
Free-form SQL hallucinations → force parameterized templates and view allowlists; reject unknown columns.
PII leakage via JOINs → push masking into views; ban table access; enforce column-level masking.
Write duplication on retries → require idempotency keys; procedures must be upsert/no-op safe.
Blocking long scans → add timeouts and row caps; precompute aggregates; route to replicas.
Schema drift breakage → contract tests in CI; versioned views (vw_*@vN), deprecate gracefully.
Conclusion
Agents can be first-class database citizens—but only when the database is treated as a governed API, not an open playground. Contracts, allowlists, RLS/OLS, masking, parameterized access, idempotent procedures, receipts, and observable traces convert model “intelligence” into operational reliability. Do that, and your AI agents stop being demos and start becoming dependable coworkers for your most critical data.