PATH 01 · AGENT TOOL · GOVERNED WRITE
The write tool asks first
Agents on Databricks act through tools — Unity Catalog functions or Python functions in the agent’s own code. Wrap the write in one governed tool that accepts only allowlisted tables and columns and a single key, posts to Veridect, and performs the SQL only on a greenlight, returning the verdict and the record id to the agent either way. The agent holds EXECUTE on this tool and no write privilege of its own; humans keep theirs.
~35 lines · one tool function · allowlisted tables, agents hold no write grants
Fail-closed by construction: a network error, a timeout or a malformed response never reaches the SQL branch. The Veridect key lives in a Databricks secret scope, never in the notebook or the agent’s prompt.
# the agent's only write tool — asks Veridect, then runs the SQL on greenlight import os, requests from databricks.sdk import WorkspaceClient from databricks.sdk.service.sql import StatementState WRITABLE = { # identifiers can't be parameterized — allowlist them "crm.gold.accounts": {"account_tier", "owner_team"}, "crm.silver.customers": {"phone_number", "email_address"}, } KEYS = {"crm.gold.accounts": "account_id", "crm.silver.customers": "customer_id"} def governed_update(table: str, set_fields: dict, key_value: str, agent_id: str, agent_scope: str) -> dict: """Update one row in an allowlisted Unity Catalog table. The change runs only on a Veridect greenlight.""" if table not in WRITABLE or not set_fields or not set(set_fields) <= WRITABLE[table] or not key_value: return {"executed": False, "verdict": "rejected_by_tool", "record_id": ""} # unknown table/column or no key: never asked, never run verdict, record_id = "unavailable", "" # closed until proven open try: r = requests.post("https://<your-veridect-host>/api/ai/consensus/pre-action", timeout=60, headers={"Authorization": "Bearer " + os.environ["VERIDECT_API_KEY"]}, # from a secret scope json={"action": "update_record", "target": "unity_catalog:" + table, "parameters": {"set": set_fields, "where": {KEYS[table]: key_value}}, "callingAgent": agent_id, "callingAgentScope": agent_scope, # its configured authority, in words "department": "data_operations"}) d = r.json() if r.status_code == 200 else {} verdict, record_id = d.get("verdict", "unavailable"), d.get("auditBundleId", "") except Exception: pass # network, timeout, parse: all stay closed if verdict != "greenlight": return {"executed": False, "verdict": verdict, "record_id": record_id} sets = ", ".join(f"`{c}` = :{c}" for c in set_fields) # identifiers from the allowlist, values as parameters resp = WorkspaceClient().statement_execution.execute_statement( warehouse_id=os.environ["WAREHOUSE_ID"], wait_timeout="50s", statement=f"UPDATE {table} SET {sets}, veridect_record = :rec WHERE `{KEYS[table]}` = :key", parameters=[*({"name": c, "value": str(v)} for c, v in set_fields.items()), {"name": "key", "value": key_value}, {"name": "rec", "value": record_id}]) ok = resp.status is not None and resp.status.state == StatementState.SUCCEEDED # pending or failed is not "executed" return {"executed": ok, "verdict": "greenlight", "record_id": record_id, "statement_state": str(resp.status.state if resp.status else "unknown")}