Agentic Data Contracts

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) 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, LangChain / deepagents, and Pydantic AI, plus a framework-free path (the tools are plain async functions). Optionally integrates with 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

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

With optional database adapters:

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

# 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

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):

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:

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:

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+

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 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.

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:

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+

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
• Source: 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.