Metadata-Version: 2.4
Name: agentic-data-contracts
Version: 0.33.0
Summary: YAML-first, domain-driven data governance for AI agents
Project-URL: Homepage, https://github.com/flyersworder/agentic-data-contracts
Project-URL: Repository, https://github.com/flyersworder/agentic-data-contracts
Project-URL: Issues, https://github.com/flyersworder/agentic-data-contracts/issues
Project-URL: Documentation, https://github.com/flyersworder/agentic-data-contracts/blob/main/docs/architecture.md
Author-email: Qing <qingye779@gmail.com>
License: MIT
License-File: LICENSE
Keywords: ai-agents,analytics,claude,data-contracts,data-governance,dbt,domain-driven,llm,semantic-layer,sql-validation
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Requires-Dist: pydantic>=2.11
Requires-Dist: pyyaml>=6.0.3
Requires-Dist: sqlglot>=28.0
Requires-Dist: thefuzz>=0.22.1
Provides-Extra: agent-contracts
Requires-Dist: ai-agent-contracts>=0.3.1; extra == 'agent-contracts'
Provides-Extra: agent-sdk
Requires-Dist: claude-agent-sdk>=0.2.96; extra == 'agent-sdk'
Requires-Dist: mcp>=1.23.0; extra == 'agent-sdk'
Provides-Extra: all
Requires-Dist: ai-agent-contracts>=0.3.1; extra == 'all'
Requires-Dist: claude-agent-sdk>=0.2.96; extra == 'all'
Requires-Dist: duckdb>=1.1.1; extra == 'all'
Requires-Dist: google-cloud-bigquery>=3.7.0; extra == 'all'
Requires-Dist: langchain-core>=1.3.3; extra == 'all'
Requires-Dist: langchain>=1.2.17; extra == 'all'
Requires-Dist: langgraph>=1.1.10; extra == 'all'
Requires-Dist: mcp>=1.23.0; extra == 'all'
Requires-Dist: psycopg2-binary>=2.9.10; extra == 'all'
Requires-Dist: pydantic-ai-slim[anthropic]>=2.0.0; extra == 'all'
Requires-Dist: snowflake-connector-python>=3.14.1; extra == 'all'
Provides-Extra: bigquery
Requires-Dist: google-cloud-bigquery>=3.7.0; extra == 'bigquery'
Provides-Extra: dev
Requires-Dist: duckdb>=1.1.1; extra == 'dev'
Requires-Dist: langchain-core>=1.3.3; extra == 'dev'
Requires-Dist: langchain>=1.2.17; extra == 'dev'
Requires-Dist: langgraph>=1.1.10; extra == 'dev'
Requires-Dist: pydantic-ai-slim[anthropic]>=2.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=1.3.0; extra == 'dev'
Requires-Dist: pytest-cov>=7.1.0; extra == 'dev'
Requires-Dist: pytest>=9.0.3; extra == 'dev'
Requires-Dist: ruff>=0.15.10; extra == 'dev'
Provides-Extra: duckdb
Requires-Dist: duckdb>=1.1.1; extra == 'duckdb'
Provides-Extra: langchain
Requires-Dist: langchain-core>=1.3.3; extra == 'langchain'
Requires-Dist: langchain>=1.2.17; extra == 'langchain'
Requires-Dist: langgraph>=1.1.10; extra == 'langchain'
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9.10; extra == 'postgres'
Provides-Extra: pydantic-ai
Requires-Dist: pydantic-ai-slim[anthropic]>=2.0.0; extra == 'pydantic-ai'
Provides-Extra: snowflake
Requires-Dist: snowflake-connector-python>=3.14.1; extra == 'snowflake'
Description-Content-Type: text/markdown

# agentic-data-contracts

[![PyPI version](https://img.shields.io/pypi/v/agentic-data-contracts.svg)](https://pypi.org/project/agentic-data-contracts/)
[![PyPI downloads](https://img.shields.io/pepy/dt/agentic-data-contracts)](https://pypistats.org/packages/agentic-data-contracts)
[![CI](https://github.com/flyersworder/agentic-data-contracts/actions/workflows/ci.yml/badge.svg)](https://github.com/flyersworder/agentic-data-contracts/actions/workflows/ci.yml)
[![Python 3.12+](https://img.shields.io/badge/python-3.12%2B-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

**YAML-first, domain-driven data governance for AI agents.**

You teach agents your business domains, metrics, and governance rules upfront — in YAML — instead of letting them reverse-engineer your data landscape by trial and error. The agent learns *what* a domain means, discovers *which* metrics to use, then writes queries that are validated against your rules at query time (via [sqlglot](https://github.com/tobymao/sqlglot)) before anything reaches the database.

### Highlights

- **Governed, not guessed** — the agent uses *your* metric definitions (`SUM(amount) FILTER (WHERE status = 'completed')`), not an ad-hoc query it invented.
- **Bad SQL blocked before execution** — forbidden operations, disallowed tables, missing tenant filters, `SELECT *`, unbounded scans — caught by static analysis plus an optional EXPLAIN dry-run.
- **Validate a whole corpus, not just live queries** — re-check a verified-examples database (or a metric's arithmetic identity) against the contract in CI, and catch drift when the contract or the warehouse schema changes.
- **Business context first** — domain descriptions, metric ownership, freshness, and a metric graph (causal *and* arithmetic) guide the agent before it writes a line of SQL.
- **Resource governance built in** — per-session cost, retry, row, and token budgets, and wall-clock limits.
- **Per-caller row/column security** — allow/deny tables and filter values by principal, for multi-user bots.
- **Framework-agnostic** — plain-function tools for the Claude Agent SDK, LangChain/deepagents, Pydantic AI, or no framework at all.
- **Bring your own semantics** — read metrics from dbt, Cube, or inline YAML.

### Without a contract vs. with one

| A raw agent on your warehouse | With `agentic-data-contracts` |
|---|---|
| Invents `revenue = SUM(amount)` — silently wrong (counts refunds, cancelled orders) | Uses the governed definition: `SUM(amount) FILTER (WHERE status = 'completed')` |
| `SELECT *`, cross-tenant reads, unbounded scans | Blocked at query time — explicit columns, required `tenant_id`, row caps enforced |
| Loops on retries with no cost ceiling | Per-session retry / cost / token budgets |
| "Why did revenue drop?" → guesses | Walks the metric graph: arithmetic decomposition first, then causal drivers |

**Works with:** any Python agent framework — first-class helpers for the [Claude Agent SDK](https://github.com/anthropics/claude-agent-sdk-python), [LangChain](https://github.com/langchain-ai/langchain) / [deepagents](https://github.com/langchain-ai/deepagents), and [Pydantic AI](https://github.com/pydantic/pydantic-ai), plus a framework-free path (the tools are plain async functions). Optionally integrates with [ai-agent-contracts](https://pypi.org/project/ai-agent-contracts/) for formal resource governance.

> **See it running:** [three example agents](#examples) — `revenue_agent` (finance), `growth_agent` (experimentation), `ops_agent` (SRE) — each runs end-to-end in demo mode with no API key.

## Table of Contents

- [How It Works](#how-it-works)
- [Installation](#installation)
- [Quick Start](#quick-start)
- [The 9 Tools](#the-9-tools)
- [Domain-Driven Agent Workflow](#domain-driven-agent-workflow)
- [Contract Rules](#contract-rules)
- [Semantic Sources](#semantic-sources)
- [Table Relationships](#table-relationships)
- [Metric Impacts](#metric-impacts) (incl. [decomposition & drill dimensions](#metric-decomposition-and-drill-dimensions))
- [Custom Prompt Rendering](#custom-prompt-rendering)
- [Scaling to Large Organizations](#scaling-to-large-organizations)
- [Resource Limits](#resource-limits)
- [Optional Dependencies](#optional-dependencies)
- [Formal Governance with ai-agent-contracts](#optional-formal-governance-with-ai-agent-contracts)
- [Examples](#examples)
- [FAQ](#faq)
- [Architecture](#architecture)

## How It Works

The agent follows a domain-driven workflow — understanding business context before writing SQL:

```
1. Agent receives: "How is revenue trending?"
2. lookup_domain("revenue")     → "Revenue is recognized at fulfillment, not booking"
3. lookup_metric("total_revenue") → SUM(amount) FILTER (WHERE status = 'completed')
4. Agent writes SQL using the metric definition
5. inspect_query(sql)           → {"valid": true, "estimated_cost_usd": 0.0, ...}
6. run_query(sql)               → results returned
```

Governance rules are enforced automatically at query time:

```
Agent: "SELECT * FROM analytics.orders"
  -> BLOCKED (no SELECT * — specify explicit columns)

Agent: "SELECT order_id, amount FROM analytics.orders"
  -> BLOCKED (missing required filter: tenant_id)

Agent: "SELECT order_id, amount FROM analytics.orders WHERE tenant_id = 'acme'"
  -> PASSED + WARN (consider using semantic revenue definition)
```

**Two files, two responsibilities.** The **contract** (`contract.yml`) defines *governance* — allowed tables, rules, resource limits, and domain *catalog metadata* (what a domain means, who owns it, when it was last reviewed). The **semantic source** (`semantic.yml`, or dbt/Cube) defines the *metrics themselves* — their SQL, source tables, and which domain each belongs to.

Domain membership is **metric-first**: a metric declares the domains it belongs to (`domains: [...]`), and the contract's `Domain` block never lists its metrics. The grain that matters — the metric — owns the relationship; the domain is just a label it points at. `lookup_domain` reconstructs a domain's members at query time by reverse-looking-up the metrics that declare it, so the two files never drift. The library enforces all of this — before the query ever reaches the database.

## Installation

```bash
uv add agentic-data-contracts
# or
pip install agentic-data-contracts
```

With optional database adapters:

```bash
uv add "agentic-data-contracts[duckdb]"      # DuckDB
uv add "agentic-data-contracts[bigquery]"    # BigQuery
uv add "agentic-data-contracts[snowflake]"   # Snowflake
uv add "agentic-data-contracts[postgres]"    # PostgreSQL
uv add "agentic-data-contracts[agent-sdk]"   # Claude Agent SDK integration
uv add "agentic-data-contracts[langchain]"   # LangChain / deepagents integration
uv add "agentic-data-contracts[pydantic-ai]" # Pydantic AI integration
```

## Quick Start

### 1. Write a YAML contract

```yaml
# contract.yml
version: "1.0"
name: revenue-analysis

semantic:
  source:
    type: yaml
    path: "./semantic.yml"
  allowed_tables:
    - schema: analytics
      description: "Curated analytics tables — prefer for reporting"
      preferred: true
      tables: ["*"]          # all tables in schema (discovered from database)
    - schema: marketing
      tables: [campaigns]    # or list specific tables
      allowed_principals: [alice@co.com, bob@co.com]  # only these may query marketing.campaigns
  forbidden_operations:
    [DELETE, DROP, TRUNCATE, UPDATE, INSERT, CREATE, ALTER, MERGE, GRANT, REVOKE, COPY]
  domains:
    - name: revenue
      summary: "Financial metrics from completed orders"
      description: >
        Revenue is recognized at fulfillment, not at booking.
        Excludes refunds and chargebacks unless stated.
      # Membership is metric-first: metrics declare their domains in the
      # semantic source (see "Semantic Sources"), so no metric list here.
  rules:
    - name: tenant_isolation
      description: "All queries must filter by tenant_id"
      enforcement: block
      query_check:
        required_filter: tenant_id
    - name: no_select_star
      description: "Must specify explicit columns"
      enforcement: block
      query_check:
        no_select_star: true
    - name: pii_columns_redacted_for_juniors
      description: "Junior analysts may not select PII columns from analytics.users"
      enforcement: block
      table: analytics.users
      blocked_principals: [security_admin@co.com]   # everyone except security_admin
      query_check:
        blocked_columns: [ssn, dob, email]

resources:
  cost_limit_usd: 5.00
  max_retries: 3
  token_budget: 50000

temporal:
  max_duration_seconds: 300
```

### 2. Load the contract and create tools

```python
from agentic_data_contracts import DataContract, create_tools
from agentic_data_contracts.adapters.duckdb import DuckDBAdapter

dc = DataContract.from_yaml("contract.yml")
adapter = DuckDBAdapter("analytics.duckdb")

# Semantic source is auto-loaded from contract config (source.type + source.path)
tools = create_tools(dc, adapter=adapter)
```

### Per-Caller Access Control (Optional)

When different callers should see different subsets of a contract's tables, pass `caller_principal` to `create_tools`. Use a static string for single-user sessions (e.g. Chainlit), or a zero-arg callable when identity changes per request (e.g. a Webex room bot serving multiple users from one long-lived process):

```python
from agentic_data_contracts import DataContract, create_tools

dc = DataContract.from_yaml("contract.yml")

# Chainlit app (one user per session)
tools = create_tools(dc, adapter=adapter, caller_principal="alice@co.com")

# Webex bot (multiple users per bot instance, identity per message)
import contextvars
current_sender: contextvars.ContextVar[str | None] = contextvars.ContextVar("sender", default=None)
tools = create_tools(dc, adapter=adapter, caller_principal=lambda: current_sender.get())
# Handler sets current_sender before invoking the agent for each message.
```

The resolver is called per-query, not cached, so one long-lived `Validator` can serve different callers sequentially. Fail-closed: any `allowed_principals` or `blocked_principals` field on a table requires the caller to be identified — an anonymous caller is treated as unauthenticated and denied.

`Principal` and `resolve_principal` are available from the package root for integrators typing their own middleware:

```python
from agentic_data_contracts import Principal, resolve_principal
```

> **Known limitation:** `to_system_prompt()` lists all declared tables in the contract without filtering by principal. Query-time gating remains authoritative (denied queries never reach the database), but the agent may still be told about tables the current caller cannot access and can waste retry budget (`resources.max_retries`) on queries that will be blocked. Principal-aware prompt rendering is a candidate future feature — file an issue if your deployment needs it.

#### Per-Rule Principal Scoping

Individual `SemanticRule` entries accept the same `allowed_principals` / `blocked_principals` pair (mutually exclusive at load time). When a rule carries either field, it is skipped at validate-time for callers outside the scope. This works across every rule kind — `blocked_columns`, `required_filter`, `no_select_star`, `max_joins`, and `result_check`:

```yaml
rules:
  # Block selecting `ssn` for everyone except the security admin.
  - name: redact_ssn
    enforcement: block
    table: pii.users
    blocked_principals: [security_admin@co.com]
    query_check:
      blocked_columns: [ssn]

  # Only the on-call engineer is held to the 60-second timeout result-check.
  - name: oncall_query_budget
    enforcement: warn
    table: prod.events
    allowed_principals: [oncall@co.com]
    result_check:
      max_rows: 1_000_000
```

Same fail-closed contract as per-table scoping: a rule with `allowed_principals` or `blocked_principals` set requires the caller to be identified — anonymous callers are out of scope and the rule is skipped (it does not silently downgrade to "applies to everyone"). This lets you express things like "Alice may not select `ssn` from `pii.users`, but Bob may" directly in YAML, without splitting tables into per-principal views.

### 3. Framework integrations

Contract-aware tools are plain async functions, so they drop into any framework — expand the one you use.

<details>
<summary><b>Claude Agent SDK</b> — requires <code>claude-agent-sdk</code> 0.2.96+</summary>

```python
import asyncio
from agentic_data_contracts import create_sdk_mcp_server
from claude_agent_sdk import (
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
    query,
)

# One-liner: wraps all 9 tools and bundles into an SDK MCP server
server = create_sdk_mcp_server(dc, adapter=adapter)

options = ClaudeAgentOptions(
    model="claude-sonnet-4-6",
    system_prompt=f"You are a revenue analytics assistant.\n\n{dc.to_system_prompt()}",
    mcp_servers={"dc": server},
    **dc.to_sdk_config(),  # token_budget → task_budget, max_retries → max_turns
)

async def run(prompt: str) -> None:
    async for message in query(prompt=prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(run("What was total revenue by region in Q1 2025?"))
```

#### Layer Anthropic's `data` plugin on top (governed analyst skills)

The Agent SDK can load [knowledge-work plugins](https://github.com/anthropics/knowledge-work-plugins) alongside your governed tools, so the agent gets the `data` plugin's analyst *skills* (`validate-data`, `statistical-analysis`, `explore-data`, `sql-queries`, …) while every query it runs is still enforced by your contract. The skills are tool-agnostic — they drive "whatever warehouse tool is connected," which in your process is your governed in-process server.

The one rule that makes this safe: **suppress the plugin's bundled `.mcp.json` warehouse servers** so the agent can't bypass the contract. `strict_mcp_config=True` does exactly that — it uses *only* the servers you pass in `mcp_servers`.

```python
import dataclasses

server = create_sdk_mcp_server(dc, adapter=adapter)

opts_kwargs = {
    "model": "claude-sonnet-4-6",
    "mcp_servers": {"dc": server},                  # the ONLY data path
    "allowed_tools": [f"mcp__dc__{t.name}" for t in tools],
}

# Feature-detect SDK support, then overlay the plugin's skills.
fields = {f.name for f in dataclasses.fields(ClaudeAgentOptions)}
if {"plugins", "skills", "strict_mcp_config"} <= fields:
    opts_kwargs["plugins"] = [{"type": "local", "path": "/path/to/knowledge-work-plugins/data"}]
    opts_kwargs["skills"] = ["validate-data", "statistical-analysis", "explore-data", "sql-queries"]
    opts_kwargs["strict_mcp_config"] = True         # ← ignore the plugin's warehouse .mcp.json
    opts_kwargs["system_prompt"] = {                # skills need the claude_code harness
        "type": "preset", "preset": "claude_code",
        "append": dc.to_system_prompt(),            # your governance, appended
    }

options = ClaudeAgentOptions(**opts_kwargs)
```

Notes:

- **Curate the skill list — do not use `skills="all"`.** `data-context-extractor` is deliberately omitted: it generates a parallel semantic skill that competes with your contract as the source of metric truth. The viz/dashboard skills (`create-viz`, `build-dashboard`) need code-execution tools you may not want to grant.
- **Set metric precedence in your prompt** (e.g. "resolve metrics via `lookup_metric`/`lookup_domain` before writing SQL") so the plugin's "just write a query" instinct doesn't undercut your governed semantic layer.
- All three [examples](#examples) ship this wiring behind an opt-in `DATA_PLUGIN_PATH` env var — `growth_agent/agent.py` is the canonical, fully-commented template.

</details>

<details>
<summary><b>LangChain / deepagents</b> — requires <code>langchain</code> 1.2.17+</summary>

```python
from agentic_data_contracts import create_langchain_tools, ContractMiddleware
from deepagents import create_deep_agent

# `dc` and `adapter` are from the previous example.
# Returns list[BaseTool] — drop in anywhere LangChain accepts tools.
tools = create_langchain_tools(dc, adapter=adapter)

# Enforcement is auto-applied: session limits and BLOCKED envelopes from the
# underlying tools surface as ToolMessage(status="error"). Pair with
# ContractMiddleware (and apply_middleware=False) for graph-level interception.
agent = create_deep_agent(tools=tools)
```

Install: `pip install "agentic-data-contracts[langchain]"`. For graph-level enforcement instead of in-tool:

```python
tools = create_langchain_tools(dc, adapter=adapter, apply_middleware=False)
agent = create_deep_agent(tools=tools, middleware=[ContractMiddleware(dc, adapter=adapter)])
```

</details>

<details>
<summary><b>Pydantic AI</b> — requires <code>pydantic-ai-slim</code> 1.107.0+</summary>

```python
from agentic_data_contracts import create_pydantic_ai_tools
from pydantic_ai import Agent

# `dc` and `adapter` are from the previous example.
# Returns list[pydantic_ai.Tool] — drop into Agent(tools=...).
tools = create_pydantic_ai_tools(dc, adapter=adapter)
agent = Agent("anthropic:claude-sonnet-4-6", tools=tools)
```

Enforcement is auto-applied in-tool: a blocked query (bad SQL, forbidden
operation, missing required filter) raises `ModelRetry` so the model rewrites
and retries, while session-limit exhaustion raises a terminal
`ContractSessionLimitError` that ends the run. Install:
`pip install "agentic-data-contracts[pydantic-ai]"`.

**One shared agent for many users.** For a multi-user service, build the `Agent`
**once** and pass each user's state via `deps` — `create_pydantic_ai_toolset`
rebuilds the tools per run, bound to that user's session and principal, so you
don't construct a separate agent (or tools list) per user:

```python
from agentic_data_contracts import ContractDeps, create_pydantic_ai_toolset
from agentic_data_contracts.core.session import ContractSession
from pydantic_ai import Agent

# per_run_step=False: deps are stable within a run, so build the tools once
# per run instead of once per model step.
agent = Agent("anthropic:claude-sonnet-4-6", deps_type=ContractDeps)
agent.toolset(per_run_step=False)(create_pydantic_ai_toolset(dc, adapter=adapter))

# Per user: keep one ContractSession (cumulative limits) and pass it each turn.
result = await agent.run(
    "Revenue by region last quarter",
    deps=ContractDeps(session=user_session, caller_principal="alice@corp.com"),
    message_history=user_history,
)
```

The caller owns each user's `ContractSession` (created once per user, keyed by
user id). Per-user principals drive per-principal table/rule gating, and the same
`ModelRetry` / `ContractSessionLimitError` enforcement applies per user.

</details>

<details>
<summary><b>Directly</b> — no framework required</summary>

```python
import asyncio

async def demo() -> None:
    # Inspect a query without executing. Response is structured JSON.
    inspect = next(t for t in tools if t.name == "inspect_query")
    result = await inspect.callable(
        {"sql": "SELECT id, amount FROM analytics.orders WHERE tenant_id = 'acme'"}
    )
    print(result["content"][0]["text"])
    # {"valid": true, "violations": [], "warnings": [], "log_messages": [],
    #  "schema_valid": true, "explain_errors": [], "pending_result_checks": [...]}

    # Blocked query
    result = await inspect.callable({"sql": "SELECT * FROM analytics.orders"})
    print(result["content"][0]["text"])
    # {"valid": false,
    #  "violations": ["SELECT * is not allowed — specify explicit columns", ...],
    #  "warnings": [], ...}

asyncio.run(demo())
```

</details>

## The 9 Tools

| Tool | Description |
|------|-------------|
| `describe_table` | Get full column details for an allowed table |
| `preview_table` | Preview sample rows from an allowed table; returns `{schema, table, columns, rows}` |
| `list_metrics` | List metric definitions, optionally filtered by domain, tier, or indicator_kind; flags `stale` metrics |
| `lookup_metric` | Get a metric definition (SQL, tier, indicator_kind, impacts, impacted_by, decompositions, drill_by, owners, freshness); fuzzy search fallback when no exact match |
| `lookup_domain` | Get full domain context (description, metrics, tables, owners, freshness); fuzzy search fallback |
| `lookup_relationships` | Look up join paths for a table; finds multi-hop paths when given a target table |
| `trace_metric_impacts` | Walk the metric graph upstream (drivers) or downstream (affected) from a metric — across both causal impact edges and arithmetic decomposition edges, filtered by `kinds` |
| `inspect_query` | Validate a SQL query and estimate its cost via EXPLAIN without executing |
| `run_query` | Validate and execute a SQL query, returning results as `{columns, rows, row_count, session}` |

### Query protocol

`run_query` and `inspect_query` state the workflow rules in their own tool
descriptions, not only in the rendered system prompt:

> When computing a metric, you MUST call lookup_metric first and use its governed
> definition — never invent or adapt a metric formula.

plus, on `run_query`, *"Prefer this tool over any other SQL or data-access path."*

This is deliberate redundancy. The rendered contract prompt is opt-in — if you wire
`create_langchain_tools(...)` or `create_pydantic_ai_toolset(...)` into an agent with
your own system prompt, the contract may never be rendered — but tool descriptions
travel with the tools on every path. A governance rule the host can drop by writing
its own prompt isn't a governance rule. The precedence sentence matters when your
agent also has a generic SQL tool, a warehouse MCP server, or a shell: without it,
nothing tells the model which path is the governed one.

Each sentence appears only when the capability it names exists — ordering when the
semantic source actually has metrics, precedence when an adapter is configured (a
validation-only setup can't execute, so claiming precedence there would point the
agent at a tool that returns an error).

### Result encoding

`run_query` and `preview_table` both return a `columns` list alongside their
`rows`. By default `rows` is a list of **positional arrays** aligned to
`columns`:

```json
{"columns": ["region", "units"], "rows": [["EMEA", 412], ["APAC", 87]]}
```

This costs less than half the tokens of repeating every column name on every
row, and matters more than it looks: tool results stay in the message history
and are re-sent on every subsequent model request, so an oversized result is
paid for repeatedly rather than once.

Pass `row_format="records"` to `create_tools()` (or to any of the ecosystem
wrappers) to get one dict per row instead — the pre-0.31 shape:

```python
tools = create_tools(contract, adapter=adapter, row_format="records")
# {"columns": ["region", "units"],
#  "rows": [{"region": "EMEA", "units": 412}, {"region": "APAC", "units": 87}]}
```

Both renderings coerce values identically, and carry identical information for
distinctly-labelled columns; only the container differs. One case is not
symmetric: a query with duplicate column labels (e.g. `SELECT t.id, u.id FROM
t, t u`) collapses under `records`, since `dict(zip(columns, row))` is
last-value-wins and silently drops one column — `compact`'s positional arrays
keep both. An unrecognised value raises `ValueError` at `create_tools()` time,
not on the first query.

## Domain-Driven Agent Workflow

The core design principle: **agents should understand the business domain before writing SQL.** Instead of dumping table schemas and hoping for the best, the contract teaches the agent your business vocabulary through progressive disclosure:

```
1. Domain context     →  "What does 'revenue' mean here?"
2. Metric definitions →  "How is 'total_revenue' calculated?"
3. Query execution    →  "Run the validated SQL"
```

### Defining domains

A domain in the contract carries **catalog metadata** — a description that teaches the agent your business rules (things the SQL alone can't express), plus ownership and review cadence. It does **not** list its metrics: domain membership is *metric-first*, declared on each metric in the semantic source via `domains: [...]`. The contract describes a domain; the metrics decide which domain they belong to. `lookup_domain` stitches the two together at query time by reverse-looking-up metrics that declare the domain.

```yaml
# contract.yml — catalog metadata only
semantic:
  domains:
    - name: acquisition
      summary: "Customer acquisition costs and conversion metrics"
      description: >
        Acquisition metrics track the cost and efficiency of
        acquiring new customers across all channels.
        CAC is calculated using fully-loaded cost, not just ad spend.
      business_owner: growth-platform   # optional
      last_reviewed: 2026-05-15          # optional — drives staleness detection
    - name: retention
      summary: "Customer retention, churn, and lifetime value"
      description: >
        Retention metrics measure how well we keep customers.
        Churn is measured on a 30-day rolling window.
        A customer is "active" if they had at least one qualifying
        action in the window.
```

```yaml
# semantic.yml — metrics declare their domain membership
metrics:
  - name: CAC
    description: "Fully-loaded customer acquisition cost"
    sql_expression: "..."
    domains: [acquisition]
  - name: churn_rate
    description: "30-day rolling churn"
    sql_expression: "..."
    domains: [retention]
```

### How the agent uses domains

The system prompt gives the agent a compact domain index. When a user asks a domain-specific question, the agent explores progressively:

```
lookup_domain("acquisition")        → business context + metric descriptions
lookup_metric("CAC")                → SQL expression, source table, filters
lookup_metric("acquisition cost")   → fuzzy match, returns [CAC, CPA] as candidates
list_metrics(domain="retention")    → all metrics in the retention domain
```

**Navigation is bidirectional — two doors into the same graph.** Top-down: a user asks about a *domain*, the agent reads its context, then drills into the metrics. Bottom-up (the common case, since queries usually name a metric): the agent looks up a *metric*, and `lookup_metric` returns that metric's `domains`, which it can follow into `lookup_domain` for the business context. Because membership lives on the metric, the metric path is always a first-class on-ramp to the domain.

Either way, the agent knows that "revenue is recognized at fulfillment, not at booking" *before* it writes a single line of SQL — reducing hallucinated metrics and incorrect calculations.

### Why progressive disclosure works

This pattern — compact index in the prompt, detailed context on demand — is the same philosophy validated by agent skill systems, MCP tool servers, and RAG architectures. Instead of overloading the agent's context window with everything upfront, you give it just enough to know *where to look*, then let it pull details when needed. The result is better token efficiency, more focused reasoning, and fewer hallucinations from context overload.

## Contract Rules

Rules are enforced at three levels:

- **`block`** — query is rejected and an error is returned to the agent
- **`warn`** — query proceeds and a `WARNINGS:` preamble is prepended to the `run_query` response (also in `inspect_query` under `warnings`)
- **`log`** — query proceeds and a `LOG:` preamble is prepended to the `run_query` response (also in `inspect_query` under `log_messages`); rules at this level are omitted from the system prompt so the agent can't adapt behavior to avoid triggering them

Each rule carries a `query_check` (pre-execution) or `result_check` (post-execution) block. Rules with neither are advisory — they appear in the system prompt but don't enforce anything. Every rule can be scoped to a specific table or applied globally.

**Built-in query checks** (pre-execution, validated against SQL AST):

| Check | Description |
|-------|-------------|
| `required_filter` | Require a column in WHERE clause (e.g., `tenant_id`) |
| `no_select_star` | Forbid `SELECT *` — require explicit columns |
| `blocked_columns` | Forbid specific columns in SELECT (e.g., PII) |
| `require_limit` | Require a LIMIT clause |
| `max_joins` | Cap the number of JOINs |

**Built-in result checks** (post-execution, validated against query output):

| Check | Description |
|-------|-------------|
| `min_value` / `max_value` | Numeric bounds on a column's values |
| `not_null` | Column must not contain nulls |
| `min_rows` / `max_rows` | Row count bounds on the result set |

Example with table scoping and both check types:

```yaml
rules:
  - name: tenant_isolation
    description: "Orders must filter by tenant_id"
    enforcement: block
    table: "analytics.orders"      # only applies to this table
    query_check:
      required_filter: tenant_id

  - name: hide_pii
    description: "Do not select PII columns from customers"
    enforcement: block
    table: "analytics.customers"
    query_check:
      blocked_columns: [ssn, email, phone]

  - name: wau_sanity
    description: "WAU should not exceed world population"
    enforcement: warn
    table: "analytics.user_metrics"
    result_check:
      column: wau
      max_value: 8_000_000_000

  - name: no_negative_revenue
    description: "Revenue must not be negative"
    enforcement: block
    result_check:
      column: revenue
      min_value: 0
```

## Semantic Sources

A semantic source provides metric, table schema, and relationship metadata to the agent. Paths are resolved relative to the contract file's directory (not the process CWD).

**YAML** (built-in):
```yaml
# semantic.yml
metrics:
  - name: total_revenue
    description: "Total revenue from completed orders"
    sql_expression: "SUM(amount) FILTER (WHERE status = 'completed')"
    source_model: analytics.orders
    domains: [revenue]                 # optional — see "Metric Impacts" below
    tier: [north_star, department_kpi] # optional — north_star / department_kpi / team_kpi
    indicator_kind: lagging            # optional — leading | lagging
    business_owner: revenue-platform   # optional — team that owns the definition
    operational_owner: data-eng-finance # optional — team that owns data health
    last_reviewed: 2026-05-15          # optional — drives staleness detection

tables:
  - schema: analytics
    table: orders
    columns:
      - name: id
        type: INTEGER
      - name: amount
        type: DECIMAL
      - name: tenant_id
        type: VARCHAR
```

`tier`, `indicator_kind`, and `domains` are all optional. For dbt and Cube sources, these fields live under the metric's `meta:` block and are read through the same field names.

**Ownership & review cadence (optional).** `business_owner` / `operational_owner` (always *teams*, not individuals — owners outlive any one person) and `last_reviewed` declare who owns a metric's definition vs. its data health, and when it was last vetted. `last_reviewed` feeds `DataContract.find_stale()` (see [Governance Staleness](docs/architecture.md#governance-staleness)) and surfaces in `lookup_metric` / `lookup_domain` as a `stale` flag so the agent can disclose drift at query time. The same three fields are accepted on a `domain`. They are read from the **YAML source** today; dbt/Cube metrics default to unset.

**dbt** — point to a `manifest.json`:
```yaml
semantic:
  source:
    type: dbt
    path: "./dbt/manifest.json"
```

dbt's built-in `relationships` schema test compiles into the manifest as a test node — `DbtSource` projects each one into a `Relationship`, resolving the owner via `attached_node` (manifest v12+) and the referenced model via `depends_on.nodes`. Tests with non-`relationships` types (`not_null`, `unique`, custom tests) and tests that can't be resolved are silently ignored. Three optional knobs read from the test's `meta:` block (matching how `tier` / `domains` are read on metrics):

```yaml
# In your dbt schema.yml
models:
  - name: orders
    columns:
      - name: customer_id
        tests:
          - relationships:
              to: ref('customers')
              field: id
              meta:
                preferred: true
                required_filter: "status != 'cancelled'"
                relationship_type: many_to_one  # default; one_to_one / many_to_many also accepted
```

**Cube** — point to a Cube schema file:
```yaml
semantic:
  source:
    type: cube
    path: "./cube/schema.yml"
```

Each cube's `joins:` block projects into `Relationship` instances. The parser handles the single-equality form `{CUBE}.col1 = {Other}.col2` (in either direction); the `from` side is always the column on the cube declaring the join, regardless of how the SQL was written. Cube's `relationship` enum (`belongsTo`, `hasOne`, `hasMany`, plus the snake_case aliases `many_to_one` / `one_to_one` / `one_to_many`) maps to the canonical `Relationship.type`. Reads from each join's `meta:` block:

```yaml
# In your Cube schema
cubes:
  - name: Orders
    sql_table: analytics.orders
    joins:
      - name: Users
        sql: "{CUBE}.customer_id = {Users}.id"
        relationship: belongsTo
        meta:
          preferred: true
          required_filter: "status != 'cancelled'"
          relationship_type: many_to_one  # optional override
```

Joins whose SQL doesn't match the single-equality pattern (composite keys with `AND`-chained equalities) or whose target cube can't be resolved by name are skipped silently — fall back to declaring those in your contract YAML via `YamlSource`.

## Table Relationships

Define join paths so the agent knows how to combine tables correctly:

```yaml
# semantic.yml
relationships:
  - from: analytics.orders.customer_id
    to: analytics.customers.id
    type: many_to_one
    description: >
      Join orders to customers for region-level breakdowns.
      Every order has exactly one customer.

  - from: analytics.bdg_attribution.contact_id
    to: analytics.contacts.contact_id
    type: many_to_one
    description: "Bridge table — filter to avoid fan-out from multiple attribution records."
    required_filter: "attribution_model = 'last_touch_attribution'"

  # When multiple parallel join paths exist between the same pair of tables
  # (role-playing dimensions, multi-role FKs), mark the canonical one
  # `preferred: true`. The agent sees `preferred="true"` in the prompt and
  # `lookup_relationships` returns preferred edges first.
  - from: analytics.orders.customer_id
    to: analytics.users.id
    type: many_to_one
    description: "Customer who placed the order — canonical user join."
    preferred: true
  - from: analytics.orders.sales_rep_id
    to: analytics.users.id
    type: many_to_one
    description: "Salesperson who closed the order."
```

| Field | Required | Description |
|-------|----------|-------------|
| `from` / `to` | Yes | Fully qualified column references (`schema.table.column`) |
| `type` | No | Cardinality: `many_to_one` (default), `one_to_one`, `many_to_many` |
| `description` | No | Free-text context for the agent (join guidance, caveats, data quality notes) |
| `required_filter` | No | SQL condition that **must** be applied when using this join (e.g., bridge table disambiguation) |
| `preferred` | No | Mark the canonical join when alternatives exist (defaults to `false`). Surfaces as `preferred="true"` in the prompt, floats the edge to the front of `lookup_relationships` direct-lookup output, and biases multi-hop BFS path-finding toward it. Leave unset for role-playing peers (e.g. `order_date` vs `ship_date`) where no single path is canonical. |

The agent sees these in its system prompt and uses them to write correct JOINs instead of guessing from column names.

### Relationship Validation

When a `SemanticSource` is passed to the `Validator`, declared relationships are actively validated against the agent's SQL:

| Check | Trigger | Warning |
|-------|---------|---------|
| **Join-key correctness** | Agent joins on wrong columns for a declared relationship | "uses `email` but declared relationship specifies `customer_id → id`" |
| **Required-filter missing** | Join has `required_filter` but WHERE clause doesn't include it | "has required filter `status != 'cancelled'` but query does not filter on: status" |
| **Fan-out risk** | Aggregation (SUM, COUNT, etc.) across a `one_to_many` join | "Results may be inflated by row multiplication" |

All relationship checks are **advisory only** (warnings, never blocks). Undeclared joins are silently ignored — the checker only validates relationships you've explicitly defined.

## Metric Impacts

Table relationships tell the agent *how to join*. Metric impacts tell the agent *what drives what* — the causal / economic graph between KPIs. When an agent is asked "why did revenue drop?", an impact graph lets it walk upstream to the drivers (conversion rate, active customers, traffic) rather than blindly querying revenue again. When it's asked to recommend an action, it can cite verified evidence rather than hand-waving.

Declare impacts at the top level of the semantic YAML, alongside `metrics:` and `relationships:`:

```yaml
# semantic.yml
metric_impacts:
  - from: active_customers
    to: total_revenue
    direction: positive           # positive | negative
    confidence: verified          # verified | correlated | hypothesized
    evidence: "A/B test exp-042 (Q3 2025), +3.2% revenue lift, p<0.01"
    description: "Retained customers drive repeat purchases."
```

| Field | Required | Description |
|-------|----------|-------------|
| `from` / `to` | Yes | Metric names (must match a metric declared in the same contract) |
| `direction` | No | `positive` (default) or `negative` |
| `confidence` | No | `hypothesized` (default), `correlated`, or `verified` — lets the agent prioritize backed-up drivers over hunches |
| `evidence` | No | Free text — study reference, A/B test ID, anything the agent should quote when making a recommendation |
| `description` | No | Optional elaboration |

Edges are directional. There's no `domains` field on the edge itself: an impact surfaces whenever either endpoint is in the agent's active domain, so cross-domain drivers (Checkout → Revenue) get discovered for free.

### How the agent uses impacts

`lookup_metric` surfaces an enriched response: each metric carries `impacts` (outgoing edges) and `impacted_by` (incoming edges), each rendered as a one-line citation string:

```
"positive impact on total_revenue (verified): A/B test exp-042 (Q3 2025), +3.2% revenue lift, p<0.01"
```

The agent can quote this verbatim in its answer — structured enough to reason over, readable enough to paste.

`trace_metric_impacts` walks the graph via BFS:

```python
await trace.callable({
    "metric_name": "total_revenue",
    "direction": "upstream",     # upstream = drivers, downstream = affected
    "max_depth": 2,
})
# Returns: {"edges": [{"depth": 1, "from": "active_customers", "to": "total_revenue",
#                       "direction": "positive", "confidence": "verified",
#                       "evidence": "A/B test exp-042..."}]}
```

Impacts declared in contract YAML reference metric names regardless of where the metric itself is defined, so this works even for dbt and Cube-sourced metrics — neither semantic layer has a native causal-graph concept. Unknown metric references in `metric_impacts` emit a warning at tool-creation time (same pattern as domain validation).

### Metric decomposition and drill dimensions

Impact edges are *causal* — evidential and non-exhaustive. A metric can also declare its *arithmetic identity*: how its value is exactly reconstructed from other metrics. Declare `decompositions` on the metric itself:

```yaml
metrics:
  - name: total_revenue
    sql_expression: "SUM(amount)"
    decompositions:
      - operator: product            # sum | product | ratio | difference
        operands: [paying_customers, arpu]
    drill_by:
      - {dimension: region, column: analytics.dim_customer.region}
```

An identity decomposition is *exhaustive and exact* — nothing else can move `total_revenue` except `paying_customers` and `arpu` — so an agent diagnosing a change can localize it deterministically before reaching for the speculative impact edges. `ratio` and `difference` are binary; `sum` and `product` take two or more operands. Decompositions are validated at load time: every operand must resolve to a declared metric, and the identity edges must form a DAG (a metric cannot transitively decompose into itself). A metric with no decomposition is a valid leaf.

`drill_by` lists the dimensional cuts a metric can be sliced by, in priority order (`schema.table.column`) — the exhaustive `GROUP BY` axes ("revenue by region") that dominate weekly-review diagnosis. Its column reference is soft-validated: a malformed shape raises, but a reference to a table the contract hasn't declared is allowed (schemas are optional).

Decomposition operands become identity edges in the same graph as impacts, so `trace_metric_impacts` walks both. Its `kinds` argument selects which — `identity` (arithmetic), `influence` (causal), or `all` (default) — and every returned edge is tagged with its `kind`:

```python
await trace.callable({
    "metric_name": "total_revenue",
    "direction": "downstream",
    "kinds": "identity",           # walk the arithmetic skeleton first
})
# Returns edges like: {"depth": 1, "from": "total_revenue", "to": "arpu",
#                      "kind": "identity", "operator": "product"}
```

Today `decompositions` / `drill_by` are declared directly in YAML contracts; dbt/Cube extraction and a variance-diagnosis tool are deferred.

## Validating a verified-examples corpus

If you keep a corpus of known-good `question → SQL` examples — the kind an analytics agent accumulates from real sessions and promotes through review — `validate_examples` re-checks each example's SQL against a contract using the *same* two-layer `Validator` that gates live queries. The corpus stays entirely yours (your repo, your format, your review flow); the library never stores, loads, or executes it — it contributes exactly one verb, *validate*.

```python
from agentic_data_contracts import DataContract
from agentic_data_contracts.validation import VerifiedExample, validate_examples

contract = DataContract.from_yaml("contract.yml")
examples = [VerifiedExample.from_dict(row) for row in load_your_yaml()]  # you own the load step

report = validate_examples(examples, contract, explain_adapter=adapter)  # adapter → live EXPLAIN
if not report.ok:
    print(report.summary())   # markdown, ready to post as an MR comment
    # in CI: sys.exit(1)
```

Each example lands in exactly one `status` — `valid` (statically contract-checked and passed), `violation` (a check rejected it), `unverified` (the engine planned it but policy couldn't be statically checked — see below), or `unchecked` (no verdict possible) — with two flags, `contract_checked` and `engine_checked`, recording *what* was verified. `report.ok` is a **safe gate**: it is True only when *every* example is `valid`, so `if not report.ok: sys.exit(1)` fails on violations, unchecked, *and* unverified rows (test `report.violations` directly for a laxer gate). Two uses of the same call:

- **MR gate** — validate the corpus in CI *before* a human reviews it; fail on `not report.ok`, so the human is no longer the only check.
- **Drift sweep** — re-run against a *changed* contract; `report.violations` are the examples the change just broke. With an `explain_adapter`, the live EXPLAIN also catches a dropped or renamed column that static checks can't see.

It confirms an example is still *allowed, well-formed, and plannable against the current schema* — never that it still returns the right answer, because it **never executes** the SQL (result correctness stays with your review). For SQL an engine parses but sqlglot cannot (e.g. Denodo/VDP), a parse failure falls back to the engine's own planner; those pass as plannable but policy-unverified, flagged in `report.unverified_compliance`. See [`examples/revenue_agent/verify_examples.py`](examples/revenue_agent/verify_examples.py) for a runnable, DuckDB-backed demo.

Its sibling `reconcile_decomposition(...)` applies the same CI-first, contract-relative spirit to a metric's declared arithmetic identity, executing the `decompositions` above against live data to assert the identity still holds within tolerance.

## Custom Prompt Rendering

The system prompt is generated by a `PromptRenderer`. The default `ClaudePromptRenderer` produces XML-structured output optimized for Claude models:

```python
dc = DataContract.from_yaml("contract.yml")
print(dc.to_system_prompt())  # XML output, optimized for Claude
```

For other models (GPT-4, Gemini, Llama), implement the `PromptRenderer` protocol:

```python
from agentic_data_contracts import PromptRenderer, DataContract

class MarkdownRenderer:
    def render(self, contract, semantic_source=None):
        tables = "\n".join(f"- {t}" for t in contract.allowed_table_names())
        return f"## {contract.name}\n\nAllowed tables:\n{tables}"

dc = DataContract.from_yaml("contract.yml")
print(dc.to_system_prompt(renderer=MarkdownRenderer()))
```

## Scaling to Large Organizations

Tested for 200+ tables, 300+ metrics, 50+ relationships across multiple schemas.

| Concern | How it scales |
|---|---|
| **System prompt size** | With domains: compact index (name + summary + count). Without domains: >20 metrics auto-switches to count. >30 relationships: per-table join counts with `lookup_relationships` hint |
| **Relationship lookup** | `lookup_relationships(table=...)` returns joins for a table on demand. With `target_table`, finds shortest multi-hop join path via BFS (up to 3 hops) |
| **Wildcard schemas** | `tables: ["*"]` discovers tables from the database. Resolution is cached — no repeated queries |
| **Metric lookup** | Fuzzy search via `thefuzz` (C++ backed) — sub-millisecond even with 1000+ metrics |
| **SQL validation** | Set-based allowlist check — O(1) per table reference regardless of allowlist size |

## Resource Limits

```yaml
resources:
  cost_limit_usd: 5.00          # max estimated query cost
  max_retries: 3                 # max blocked queries per session
  token_budget: 50000            # max tokens consumed
  max_query_time_seconds: 30     # max wall-clock query time
  max_rows_scanned: 1000000      # max rows an EXPLAIN may estimate
```

## Optional Dependencies

| Extra | Package | Purpose |
|-------|---------|---------|
| `duckdb` | `duckdb>=1.1.1` | DuckDB adapter (the one adapter that ships) |
| `bigquery` | `google-cloud-bigquery>=3.7.0` | Driver only — for a `DatabaseAdapter` you write |
| `snowflake` | `snowflake-connector-python>=3.14.1` | Driver only — for a `DatabaseAdapter` you write |
| `postgres` | `psycopg2-binary>=2.9.10` | Driver only — for a `DatabaseAdapter` you write |
| `agent-sdk` | `claude-agent-sdk>=0.2.96`, `mcp>=1.23.0` | Claude Agent SDK integration |
| `langchain` | `langchain-core>=1.3.3`, `langchain>=1.2.17`, `langgraph>=1.1.10` | LangChain / deepagents integration |
| `pydantic-ai` | `pydantic-ai-slim[anthropic]>=2.0.0` | Pydantic AI integration |
| `agent-contracts` | `ai-agent-contracts>=0.3.1` | ai-agent-contracts bridge |

## Optional: Formal Governance with ai-agent-contracts

The library works standalone with lightweight enforcement. Install [`ai-agent-contracts`](https://pypi.org/project/ai-agent-contracts/) to upgrade to the formal governance framework:

```bash
pip install "agentic-data-contracts[agent-contracts]"
```

```python
from agentic_data_contracts.bridge.compiler import compile_to_contract

contract = compile_to_contract(dc)  # YAML → formal 7-tuple Contract
```

**What you get with the bridge:**

| Concern | Standalone | With ai-agent-contracts |
|---|---|---|
| Resource tracking | Manual counters | Formal `ResourceConstraints` with auto-enforcement |
| Rule violations | Exception + retry | `TerminationCondition` with contract state machine |
| Success evaluation | Log-based | Weighted `SuccessCriterion` scoring, LLM judge support |
| Contract lifecycle | None | `DRAFTED → ACTIVE → FULFILLED / VIOLATED / TERMINATED` |
| Framework support | Claude Agent SDK | + LiteLLM, LangChain, LangGraph, Google ADK |
| Multi-agent | Single agent | Coordination patterns (sequential, parallel, hierarchical) |

**When to use it:** formal audit trails, success scoring, multi-agent coordination, or integration with non-Claude agent frameworks.

## Examples

Three end-to-end working examples, each demonstrating a different governance archetype. All three run in demo mode without the Claude Agent SDK installed — DuckDB is used for the sample data and the tools are exercised directly.

| Example | Archetype | Governance patterns it teaches |
|---|---|---|
| [`examples/revenue_agent/`](examples/revenue_agent/) | Finance / lagging KPIs / audit-strict | Tenant isolation, `hypothesized` impact edges, north-star metric tier, undefined-metric policy recipe |
| [`examples/growth_agent/`](examples/growth_agent/) | Experimentation / leading indicators | `verified` / `correlated` / `hypothesized` metric impacts with real-ish A/B evidence, time-bounded events rule, `log`-level PII audit invisible to the agent, stale-review detection, **`preferred: true` on the canonical `events.user_id → users.id` join** (alongside a non-preferred `events.referrer_user_id → users.id` for referral-mechanics questions) |
| [`examples/ops_agent/`](examples/ops_agent/) | SRE reliability / real-time dashboards | `blocked_columns` for PII, two `log`-level audit rules (governance trail), `require_limit` + `max_joins` caps, **negative-direction** metric impact (DORA pattern), aggressive resource limits, **`blocked_principals` on `sre.deploys`** (try `--caller intern@co.com` to see a per-table principal denial) |

Run any of them:

```bash
uv run python examples/revenue_agent/agent.py "What was Q1 revenue by region?"
uv run python examples/growth_agent/agent.py  "Which onboarding variant lifted activation?"
uv run python examples/ops_agent/agent.py     "What's our MTTR by severity this week?"
```

Each example directory contains four files:
- `contract.yml` — governance rules, allowed tables, resource limits
- `semantic.yml` — metrics, relationships, metric impacts
- `setup_db.py` — sample DuckDB data (auto-created on first run)
- `agent.py` — runnable demo with a Claude Agent SDK path plus a fallback that exercises the tools directly

`revenue_agent` additionally ships a **verified-examples validation** demo — `verified_examples.yml` (an external corpus) and `verify_examples.py`, which re-checks it against the contract with a live DuckDB EXPLAIN (valid, static violations, a schema-drift catch only the dry-run finds, and the same SQL diverging by principal):

```bash
uv run python examples/revenue_agent/verify_examples.py
```

Reading all three gives you a complete tour of the library's design space: different enforcement levels (`block` / `warn` / `log`), different impact confidences and directions, and resource profiles tuned for very different user-latency expectations.

All three `agent.py` files also carry the [`data`-plugin skills overlay](#layer-anthropics-data-plugin-on-top-governed-analyst-skills) behind an opt-in `DATA_PLUGIN_PATH` env var (off by default, so the examples run with zero external setup):

```bash
git clone https://github.com/anthropics/knowledge-work-plugins /tmp/kwp
DATA_PLUGIN_PATH=/tmp/kwp/data \
    uv run python examples/growth_agent/agent.py "Which onboarding variant lifted activation?"
```

## FAQ

**Do I need `ai-agent-contracts`?** No. The library works standalone with lightweight enforcement (session counters, cost/retry/token budgets). Install [`ai-agent-contracts`](#optional-formal-governance-with-ai-agent-contracts) only if you want the formal 7-tuple contract, weighted success scoring, or multi-agent coordination.

**Which databases are supported?** Any, via the `DatabaseAdapter` protocol. A **DuckDB** adapter ships in the box; the `bigquery`, `snowflake`, and `postgres` [extras](#optional-dependencies) install the driver only, so you implement the (small) adapter against your warehouse's client. Layer 1 validation (static SQL analysis) runs even with no adapter configured; a database adapter adds the EXPLAIN dry-run and query execution.

**Does it execute my SQL?** Only `run_query` does, and only after validation passes (plus an optional EXPLAIN dry-run). `inspect_query` validates without executing, and forbidden operations (DELETE/DROP/UPDATE/…) are blocked before they ever reach the database.

**Do I have to use dbt or Cube?** No. Author metrics inline in a `semantic.yml` with `YamlSource`. dbt (`manifest.json`) and Cube are supported if you already have them — the agent-facing behavior is identical regardless of source.

**Does it work without the Claude Agent SDK?** Yes. The tools are plain async functions usable from LangChain/deepagents, Pydantic AI, or directly; the example agents fall back to a no-SDK demo mode.

**Is it production-ready?** It's pre-1.0 and actively evolving. Every breaking change is documented in [`CHANGELOG.md`](CHANGELOG.md) with migration notes.

## Architecture

See [`docs/architecture.md`](docs/architecture.md) for the full design spec covering the layered architecture, YAML schema, validation pipeline, tool design, semantic sources, database adapters, and the optional `ai-agent-contracts` bridge.

## License

MIT
