# Agentic Data Contracts

> YAML-first, domain-driven data governance for AI agents — teach agents your business domains, metrics, and rules before they write SQL

- **Type:** MCP server
- **Install:** `agentstack add mcp-flyersworder-agentic-data-contracts`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [flyersworder](https://agentstack.voostack.com/s/flyersworder)
- **Installs:** 0
- **Category:** [Databases](https://agentstack.voostack.com/c/databases)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [flyersworder](https://github.com/flyersworder)
- **Source:** https://github.com/flyersworder/agentic-data-contracts
- **Website:** https://pypi.org/project/agentic-data-contracts/

## Install

```sh
agentstack add mcp-flyersworder-agentic-data-contracts
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# agentic-data-contracts

[](https://pypi.org/project/agentic-data-contracts/)
[](https://pypistats.org/packages/agentic-data-contracts)
[](https://github.com/flyersworder/agentic-data-contracts/actions/workflows/ci.yml)
[](https://www.python.org/downloads/)
[](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.

Claude Agent SDK — requires claude-agent-sdk 0.2.96+

```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"} 

LangChain / deepagents — requires langchain 1.2.17+

```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 — note the **shared session**, without which the middleware enforces against one budget while `run_query` reports `tokens_remaining` from another:

```python
from agentic_data_contracts.core.session import ContractSession

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

Pydantic AI — requires pydantic-ai-slim 2.0.0+

```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 filte

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [flyersworder](https://github.com/flyersworder)
- **Source:** [flyersworder/agentic-data-contracts](https://github.com/flyersworder/agentic-data-contracts)
- **License:** MIT
- **Homepage:** https://pypi.org/project/agentic-data-contracts/

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/mcp-flyersworder-agentic-data-contracts
- Seller: https://agentstack.voostack.com/s/flyersworder
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.