Agent-Safe Governed Views

An agent with credentials to raw tables can read every column, including the ones with names and email addresses, and it will write its own joins. Some of those joins will be wrong. The fix is old and boring: publish views that hold the curated columns and the correct joins, and give the agent access to those views and nothing else. This page builds that, step by step, over Apache Iceberg tables.

The shape of the pattern

Raw schema. The Iceberg tables as written by your pipelines. Engineers and jobs read these. Agents never do.
Governed schema. Views over the raw tables that select allowed columns only, apply the joins the business agreed on, and apply the row policy for the agent's role (covered in row and column policies).
A locked engine. The connection the agent's queries run on cannot read files, call URLs, or change its own settings.

In the demo the engine is DuckDB and the views live in a governed schema. In a production lakehouse engine the same shape is a folder or schema of views plus grants that give the agent's service identity SELECT on the views and no privilege on the tables underneath.

Step 1: Keep the raw tables where agents cannot name them

The demo writes two Iceberg tables with PyIceberg: sales.customers, which deliberately includes customer_name and email, and sales.orders. The catalog is a SQLite file and the warehouse is a local folder, so there is nothing to install beyond Python packages.

from pyiceberg.catalog import Catalog, load_catalog

def get_catalog(home=None) -> Catalog:
    """Load (or create) the SQL catalog backed by SQLite and a local warehouse."""
    root = demo_home(home)
    warehouse = root / "warehouse"
    warehouse.mkdir(parents=True, exist_ok=True)
    return load_catalog(
        "demo",
        **{
            "type": "sql",
            "uri": f"sqlite:///{root / 'catalog.db'}",
            "warehouse": warehouse.as_uri(),
        },
    )

The agent never receives catalog credentials. The MCP server holds them, and the only things it exposes are metrics and dimensions. That is the first boundary: the agent cannot ask for a table it has no way to name.

Step 2: Declare the governed view in the semantic model

The view definition belongs next to the metrics that use it, in version control, reviewed like code. The column list is the column policy: anything not selected here cannot be reached by any metric or dimension.

sources:
  orders:
    table: sales.orders
  customers:
    table: sales.customers

governed_views:
  orders_enriched:
    description: One row per order with the customer's segment. No personal data.
    sql: |
      SELECT
        o.order_id,
        o.order_date,
        o.customer_id,
        o.region,
        o.channel,
        o.product_category,
        o.status,
        o.amount,
        o.discount,
        c.segment
      FROM {orders} AS o
      JOIN {customers} AS c ON o.customer_id = c.customer_id
    row_policy_column: region
    time_column: order_date
    cost_source: orders

Three details matter. The join from orders to customers is written once, by someone who knows the keys, so no agent reinvents it. customer_id stays because counting distinct customers is a legitimate metric, while the name and email columns stay behind. And {orders} and {customers} are placeholders, so the view text never hard-codes where raw data lives.

Step 3: Build raw and governed schemas in the engine

At startup the server reads each source table from Iceberg into a private raw schema, then creates each governed view in the governed schema.

RAW_SCHEMA = "raw"
GOVERNED_SCHEMA = "governed"

def build_connection(catalog, model, role) -> duckdb.DuckDBPyConnection:
    """Load sources from Iceberg and return a locked DuckDB connection."""
    con = duckdb.connect(database=":memory:")
    con.execute(f"CREATE SCHEMA {RAW_SCHEMA}")
    con.execute(f"CREATE SCHEMA {GOVERNED_SCHEMA}")

    for key, identifier in model.sources.items():
        try:
            arrow = catalog.load_table(identifier).scan().to_arrow()
        except NoSuchTableError as exc:
            raise RuntimeError(f"Iceberg table {identifier} not found. Run `slh-seed` first.") from exc
        con.register("_incoming", arrow)
        con.execute(f"CREATE TABLE {RAW_SCHEMA}.{key} AS SELECT * FROM _incoming")
        con.unregister("_incoming")

    for view_name in model.views:
        con.execute(governed_view_sql(model, view_name, role))

    # Defense in depth: no file, HTTP, or extension access from here on.
    con.execute("SET enable_external_access = false")
    con.execute("SET lock_configuration = true")
    return con

governed_view_sql fills in the placeholders and, when the role has one, wraps the view in its row filter. The next pattern covers that part.

Step 4: Lock the engine

The last two statements above are easy to skip and worth keeping. enable_external_access = false stops DuckDB from reading files, URLs, or extensions, so a query that somehow got past every other check still could not call read_csv('/etc/passwd'). lock_configuration = true stops anyone from turning it back on. Every query then runs on a cursor of this connection, and cursors share the same locked database.

In a server-based engine the equivalent is a service account for the agent that holds SELECT on the governed views and nothing else: no table privileges, no ability to create objects, no external data sources.

Step 5: Only compile SQL against governed views

The views only help if nothing ever queries around them. In the demo the agent cannot send SQL at all. The compiler turns metric names into SQL, and it only knows one FROM clause:

sql = f"  SELECT {', '.join(select)}\n  FROM governed.{view.name}"

A second, independent check parses the compiled statement with DuckDB's own parser and refuses anything that reads a relation outside the governed views. That check is described in query guardrails.

Step 6: Test the boundary, not just the happy path

These tests run against real Iceberg tables in a temporary directory. They fail if a PII column appears in a governed view, or if the engine can read a file.

def test_governed_view_has_no_pii_columns(analyst):
    cols = [r[0] for r in analyst.con.execute("DESCRIBE governed.orders_enriched").fetchall()]
    assert "email" not in cols and "customer_name" not in cols
    assert {"region", "segment", "amount"} <= set(cols)


def test_external_access_is_off_and_locked(analyst):
    with pytest.raises(duckdb.Error):
        analyst.con.execute("SELECT * FROM read_csv('/etc/hosts')").fetchall()
    with pytest.raises(duckdb.Error):
        analyst.con.execute("SET enable_external_access = true")
    # Cursors (used for every query) share the same locked database.
    with pytest.raises(duckdb.Error):
        analyst.con.cursor().execute("SELECT * FROM read_csv('/etc/hosts')").fetchall()

Done when

The agent's identity has no privilege on any raw table, only on governed views.
No governed view selects a column you would not show in a meeting. Check with DESCRIBE, not by reading the SQL.
Joins between facts and dimensions happen inside views, never in agent-generated SQL.
The engine connection used for agent queries cannot read files or URLs, and that setting cannot be changed from a query.
A test fails if any of the above stops being true.

Taking it to production

The demo loads Iceberg tables into memory because it has to run anywhere in a few seconds. In production the views sit in the engine that queries Iceberg in place, and the engine enforces the grants. Keep the view definitions in the same repository as the semantic model so a pull request that adds a column to a view is reviewed by someone who knows what the column holds. Iceberg's own view specification lets several engines share a view definition; see Iceberg view specification for how far that support goes today.