onlymetrix 0.4.1

Governed data access for AI agents. Connect to Snowflake, Postgres, or any warehouse with curated metrics, PII masking, and audit trails.

OnlyMetrix Python SDK

Governed data access and reasoning layer for AI agents. Connect to any warehouse, query curated metrics, and run analytical reasoning — all through a governed API with PII masking and audit trails.

pip install onlymetrix

Quick Start

from onlymetrix import OnlyMetrix

om = OnlyMetrix("https://api.onlymetrix.com", api_key="omx_sk_...")

# Query a metric
result = om.metrics.query("total_revenue", filters={"time_start": "2025-01-01"})
print(f"Revenue: ${result.rows[0]['revenue_usd']:,.2f}")

# Search metrics by intent
metrics = om.metrics.list(search="churn")

# Describe a table
desc = om.tables.describe("customers")
for col in desc.columns:
    print(f"  {col.name} ({col.type}) {'🔒 PII' if col.is_pii else ''}")

Analysis — Reasoning Layer for Agents

Not just queries. Structured reasoning that agents can parse, chain, and explain.

# Why did revenue change?
om.analysis.root_cause(
    "quarterly_revenue",
    compare={"current": "2025-02", "previous": "2025-01"},
    dimensions=["country", "tier", "product"],
)
# → {primary_dimension: "country", driver: "Germany", contribution: 0.72,
#    explanation: "Germany accounts for 72% of the decline",
#    suggested_actions: ["Investigate DACH expansion strategy"]}

# Which customers are at risk and what do they have in common?
om.analysis.run_custom("at_risk_profile",
    metric="churn_risk_entities",
    compare_metric="high_spenders",
)
# → {at_risk_count: 2696, correlation: {jaccard: 0.049, interpretation: "independent"},
#    insights: ["Churn is not driven by spending level"]}

# What's our concentration risk?
om.analysis.sensitivity("revenue", "country", scenario="remove_top_3")
# → {impact_pct: 94, risk: "critical", herfindahl_index: 0.829,
#    suggested_actions: ["Diversify — 94% concentrated in 3 countries"]}

All Analysis Methods

Method	What it answers
root_cause(metric, compare, dimensions)	"Why did this metric change?"
correlate(metric_a, metric_b)	"Are these two populations related?"
threshold(metric)	"What's the optimal cutoff for this metric?"
sensitivity(metric, dimension, scenario)	"What's our concentration risk?"
segment_performance(metric, segments)	"How does this metric perform across segments?"
contribution(metric, compare, dimension)	"What drove the change between periods?"
drivers(metric, dimensions)	"Which dimension explains variance most?"
anomalies(metric, dimension)	"Which segments are behaving abnormally?"
pareto(metric)	"What's the precision-recall frontier?"
trends(metric)	"Is this accelerating or decelerating?"
forecast(metric, periods_ahead)	"Where is this heading?"
compare(metric, filter_a, filter_b)	"How do these two groups differ?"
health(metric)	"Can I trust this data?"

Every method returns:

{
    "value": {...},              # the structured finding
    "explanation": "...",        # one sentence, plain English
    "confidence": 0.85,          # 0.0-1.0
    "warnings": [...],           # data quality issues
    "suggested_actions": [...],  # what to do about it
}

Custom Analysis

Define domain-specific analytical workflows by composing primitives.

# Define
@om.analysis.custom("store_risk")
def store_risk(ctx, dimension="region"):
    sensitivity = ctx.sensitivity(dimension=dimension, scenario="remove_top_3")
    drivers = ctx.drivers(dimensions=[dimension])
    return {
        "risk": sensitivity["value"]["risk"],
        "top_driver_cv": drivers["dimensions"][0]["coefficient_of_variation"],
        "explanation": f"Concentration risk: {sensitivity['value']['risk']}",
    }

# Export as JSON DAG (storable, shareable)
dag = om.analysis.export_dag("store_risk", save_to_server=True)

# Run — from any session, any machine
result = om.analysis.run_custom("store_risk", metric="revenue")

# Share — another user loads from server
om.analysis.load_from_server("store_risk")

Custom analyses are governed:

• Can only call OM analysis primitives (no raw SQL)
• Health check runs automatically before execution
• Stored as JSON DAGs — auditable, version-tracked
• Shareable across teams and tenants

Setup & Configuration

# Connect a warehouse
om.setup.connect_warehouse(type="postgres", host="db.example.com",
                           database="analytics", user="readonly", password="...")

# Import metrics from dbt
with open("metrics.yml") as f:
    om.compiler.import_format("dbt", f.read())

# Import from LookML
with open("views.lkml") as f:
    om.compiler.import_format("lookml", f.read())

# Create a metric
om.setup.create_metric(
    name="churn_risk",
    sql="SELECT customer_ref FROM customers WHERE last_seen < NOW() - INTERVAL '90 days'",
    description="Customers at risk of churning",
    ground_truth_sql="SELECT customer_ref, is_churned FROM customers",
)

# Run autoresearch — find optimal metric definition
om.autoresearch.run("churn_risk", max_variations=30)
# → {baseline: {f1: 0.660}, improvements: 3, pareto_frontier: [...]}

CLI

# Health
omx health

# Metrics
omx metrics list --search revenue
omx metrics query total_revenue --filter time_start=2025-01-01

# Analysis
omx analysis root-cause quarterly_revenue --current 2025-02 --previous 2025-01 --dimension country
omx analysis sensitivity churn_by_country -d country --scenario remove_top_3
omx analysis at-risk-profile churn_risk_entities --compare high_spenders

# Custom analysis
omx analysis list-custom
omx analysis export store_risk
omx analysis save store_risk dag.json
omx analysis load store_risk
omx analysis run store_risk revenue

# Setup
omx setup status
omx compiler status
omx auth login --email [e-mail] --password ...

Environment variables: OMX_API_URL (default http://localhost:8080), OMX_API_KEY.

Agent Integrations

LangChain

from onlymetrix.integrations.langchain import onlymetrix_tools

tools = onlymetrix_tools("https://api.onlymetrix.com", api_key="omx_sk_...")
# → [search_metrics, query_metric, request_metric]

CrewAI

from onlymetrix.integrations.crewai import onlymetrix_tools

tools = onlymetrix_tools("https://api.onlymetrix.com", api_key="omx_sk_...")

Async Client

from onlymetrix import AsyncOnlyMetrix

async with AsyncOnlyMetrix("https://api.onlymetrix.com", api_key="...") as om:
    metrics = await om.metrics.list(search="revenue")
    result = await om.metrics.query("total_revenue")

API Reference

Client Resources

Resource	Methods
om.metrics	list(tag, search), query(name, filters, dimension, limit), get(name)
om.tables	list(), describe(table)
om.metric_requests	list(status), create(description), resolve(id, status)
om.setup	connect_warehouse(), configure_access(), status(), create_metric(), delete_metric(), import_metrics(), dbt_sync(), list_datasources(), generate_key(), list_keys(), revoke_key()
om.auth	signup(), login(), demo(), me(), change_password()
om.compiler	status(), import_format(format, content)
om.autoresearch	run(metric, ground_truth_sql, max_variations, filters)
om.admin	invalidate_cache(metric), sync_catalog()
om.custom_analyses	register(name, definition), list(), get(name), delete(name)
om.analysis	13 built-in primitives + custom analysis engine

Error Handling

from onlymetrix import OnlyMetrix, OnlyMetrixError

try:
    result = om.metrics.query("nonexistent")
except OnlyMetrixError as e:
    print(f"Error: {e.message} (HTTP {e.status_code})")

Installation

# Core
pip install onlymetrix

# With LangChain integration
pip install onlymetrix[langchain]

# With CrewAI integration
pip install onlymetrix[crewai]

# Everything
pip install onlymetrix[all]

Requires Python 3.9+.

License

MIT