Minerva SDK

The official Python SDK for the Minerva Data API — identity resolution, person enrichment, and validation, with one client, typed responses, and input validation baked in.

from minerva import Minerva

mc = Minerva()  # reads MINERVA_API_KEY from the environment

mc.status.health()          # quick liveness check (unauthenticated) -> {"api": HealthStatus(ok=True, ...)}

results = mc.api.enrich([{"record_id": "1", "linkedin_url": "https://www.linkedin.com/in/example"}])
df = results.to_df()        # straight to a DataFrame

Early access. Data API surface — resolve, enrich, LinkedIn lookup, email validation, country inference, and usage tallies — is wired and tested.

Installation

pip install minerva-sdk

Optional extras:

pip install "minerva-sdk[pandas]"   # results.to_df()
pip install "minerva-sdk[table]"    # results.to_table()
pip install "minerva-sdk[gsheet]"   # mc.io.*_from_sheet / .to_sheet
pip install "minerva-sdk[excel]"    # mc.io.*_from_excel / .to_excel
pip install "minerva-sdk[storage]"  # mc.storage.* — bulk upload / download
pip install "minerva-sdk[all]"      # everything

Requires Python 3.11+ (tested through 3.14).

Authentication

Minerva is the single entry point. The only credential you need is your API key:

Find your key in your Minerva account (not available on the free tier) — full steps in the docs.

mc = Minerva()                          # api_key from MINERVA_API_KEY
mc = Minerva(api_key="mk_live_...")     # explicit

The SDK sends the API key as the x-api-key header on every request. The server-side authorizer decides what your key is entitled to.

Quickstart

from minerva import Minerva

mc = Minerva(api_key="mk_live_...")

# Resolve — match records to a Minerva PID
mc.api.resolve([{"record_id": "1", "first_name": "Jane", "last_name": "Doe",
                 "emails": ["jane.doe@example.com"]}])

# Enrich — full profiles (up to 500 records per call)
resp = mc.api.enrich(
    [{"record_id": "1", "linkedin_url": "https://www.linkedin.com/in/example"}],
    match_condition_fields=["linkedin_url"],   # only count it a match if a LinkedIn URL is on file
)
for r in resp.results:
    print(r.record_id, r.is_match, r.minerva_pid, r.match_score)

# Tabular output
resp.to_df()            # pandas DataFrame   (needs [pandas])
resp.to_csv("out.csv")  # write a CSV
resp.to_dicts()         # list[dict]

Validate before you call — dry_run

Every input-bearing method takes dry_run=True. It validates your input locally and returns the exact request that would be sent — without touching the network. Invalid input raises MinervaValidationError immediately, with a precise field message.

from minerva import Minerva, MinervaValidationError

mc = Minerva()

req = mc.api.enrich(records, match_condition_fields=["linkedin_url"], dry_run=True)
# -> EnrichRequest (nothing sent). Inspect req.model_dump() to see the payload.

try:
    mc.api.enrich([], dry_run=True)     # 0 records — must be at least 1
except MinervaValidationError as e:
    print("caught before any API call:", e)

Great for checking a batch is well-formed (record limits, allowed match_condition_fields, record shape, …) before spending a call.

What you can do

Every response is a typed model with IDE autocomplete, and the list-shaped ones support .to_df() / .to_csv() / .to_dicts() / .to_table().

Async — AsyncMinerva

Mirror of Minerva with async methods. Same constructor args, same exception types, same models. Use it whenever you want concurrency under one client (bulk pipelines, FastAPI/Starlette services, asyncio scripts).

import asyncio
from minerva import AsyncMinerva

async def main():
    async with AsyncMinerva() as mc:
        # All the same methods, just `await`ed
        resp = await mc.api.enrich([{"record_id": "1", "linkedin_url": "https://www.linkedin.com/in/x"}])
        print(resp.to_df())

asyncio.run(main())

Bulk enrich + auto-batching

enrich_many / resolve_many split a large list into per-request batches (up to 500 records each — the API max), fire concurrently, and aggregate results:

async with AsyncMinerva() as mc:
    results = await mc.api.enrich_many(
        my_10000_records,
        batch_size=250,           # default; ceiling is 500
        max_concurrency=20,       # optional cap; rate-limiter governs throughput anyway
    )
    # `results` is a flat list of EnrichResult, same order as input batches

Built-in rate limiting

AsyncMinerva (and Minerva) self-throttle to whatever your plan allows. On first call, the SDK GETs /v2/usage/limits and configures an in-process token bucket sized to your rps_limit / burst_limit. Concurrent calls under one client share the bucket, so you can fan out enrich_many without writing your own throttle.

Opt out via Minerva(rate_limit=False) / AsyncMinerva(rate_limit=False) — useful in tests or when you've got your own client-side throttler. Older servers without the limits endpoint degrade gracefully (no throttling, normal calls keep working).

Inspect the current policy after the first call (or via the explicit probe):

async with AsyncMinerva() as mc:
    limits = await mc.ensure_limits()
    print(limits.rps_limit, limits.burst_limit, limits.daily_quota)

Recommended workflows

The shortest path between "I have a list of names/emails" and "I have full profiles" isn't a single call — it's Resolve → Enrich by PID. The second call carries only minerva_pid per record, which lets our backend skip the match pipeline entirely. Cheaper, faster, and idempotent for re-runs (cache the PIDs once; refresh enrichment on a schedule against just those PIDs).

mc.workflows.resolve_then_enrich is the encoded version:

result = mc.workflows.resolve_then_enrich(
    records=[
        {"record_id": "u1", "emails": ["a@example.com"]},
        {"record_id": "u2", "emails": ["b@example.com"]},
        {"record_id": "u3", "full_name": "Some Person"},
    ],
)
result.resolved              # ResolveResult[] — PIDs (or None for unmatched)
result.enriched              # EnrichResult[]  — full profiles, keyed by PID
result.unmatched_record_ids  # list[str]        — record_ids that didn't match

The async sibling await mc.workflows.resolve_then_enrich(...) has the same shape.

Liveness & metrics — mc.status.health()

Never raises — failure is surfaced as ok=False with the cause, so it's safe to call in a polling loop. The same method does two things depending on whether you've configured an API key:

mc = Minerva()                            # MINERVA_API_KEY from env

report = mc.status.health()               # auto-picks the right endpoint

for name, h in report.items():
    if not h.ok:
        print(f"{name} {h.status}: {h.message} ({h.status_code})")
    else:
        print(f"{name} ok — {h.message} ({h.latency_ms:.0f} ms)")

# Override the auto-detect:
mc.status.health(detailed=False)          # force basic /health (skip the metrics overhead)
mc.status.health(detailed=True)           # force /health/metrics (raises MinervaAuthError if no key)

# Probe a specific endpoint:
mc.status.health(endpoints="api")
mc.status.health(endpoints=["api"])

When (and when NOT) to call it

mc.status.health() is a starter check, not a per-call gate.

✅ Polling once every 30-60s from a dashboard / monitor. ✅ A one-shot call at process startup to confirm reachability. ✅ Debugging "is the API down, or is it my key?" (works without a key).

❌ Don't wrap every mc.api.enrich(...) / mc.api.resolve(...) in a health check. Each call is still a real HTTP round-trip; gating every API call on it doubles your latency and contributes no useful signal — the next call itself will already raise MinervaAPIError if something's wrong.

The basic /health call (no API key) hits an API Gateway mock and returns instantly. The authed /health/metrics call (API key set) goes through a lambda backed by DataDog APM — the data is up to ~60 s stale and the call has the usual Lambda invocation overhead. The "don't gate every API call" rule matters more for /health/metrics than for /health, but applies to both: a health probe is a checkpoint, not a per-request precondition.

Spreadsheets — mc.io

Run the Data API directly off a Google Sheet or Excel workbook, and write results back the same way. Each format is gated on an optional extra so the base wheel stays small.

pip install "minerva-sdk[gsheet]"   # Google Sheets
pip install "minerva-sdk[excel]"    # .xlsx

# The sheet id is the long string between `/d/` and `/edit` in the URL
SHEET_ID = "<paste-your-google-sheet-id-here>"
CREDS    = "/path/to/service-account.json"   # or a gspread.Client / dict / google Credentials

# Read sheet → enrich → write results to a different tab
resp = mc.io.enrich_from_sheet(
    SHEET_ID,
    credentials=CREDS,
    sheet_name="Customers",
    match_condition_fields=["linkedin_url"],
)

resp.to_sheet(
    SHEET_ID,
    credentials=CREDS,
    sheet_name="Enriched",
)

Excel works the same way:

resp = mc.io.enrich_from_excel("customers.xlsx", sheet_name="Sheet1")
resp.to_excel("enriched.xlsx")

Conventions:

• The first row of the sheet is the header. Column names map 1:1 to enrich fields (record_id, linkedin_url, first_name, emails, …). Use field_mapping={"customer_id": "record_id"} for renames.
• Rows above the 500-per-request limit auto-chunk; results are merged into one EnrichResponse / ResolveResponse.
• Google auth: pass a path to a service-account JSON, a dict, a built google.oauth2 Credentials, or a gspread.Client. Or set GOOGLE_APPLICATION_CREDENTIALS and skip the kwarg.
• Errors mirror the rest of the SDK: 401/403 → MinervaAuthError, 404 → MinervaAPIError(status_code=404), malformed sheet → MinervaValidationError.

Storage — mc.storage

For workflows that need to hand off larger files than fit in a request body — input batches, exports, point-in-time snapshots — mc.storage is a thin wrapper around your private storage area with two well-known directories:

• Incoming/ — files you send.
• Outgoing/ — files placed for you to read.

upload* defaults to Incoming/; download*, list*, and delete default to Outgoing/ so the natural list → download / delete loop targets the same direction. Pass direction="outgoing" / direction="incoming" to override.

pip install "minerva-sdk[storage]"

from minerva import Minerva

mc = Minerva()

# Single file — upload defaults to Incoming/
mc.storage.upload("data/customers.csv")                   # → Incoming/customers.csv
mc.storage.upload("data/customers.csv", subdir="2026-06-29")
                                                          # → Incoming/2026-06-29/customers.csv

# Many files — parallel, order preserved (max_concurrency default 8)
mc.storage.upload_many(["a.csv", "b.csv", "c.csv"])
mc.storage.upload_directory("./batch-2026-06-29")         # walks recursively

# Browse what's waiting
for obj in mc.storage.list():                             # defaults to Outgoing/
    print(obj.key, obj.size, obj.last_modified)

# Pull results
mc.storage.download("results.csv")                        # → ./results.csv
mc.storage.download_many(["a.csv", "sub/b.csv"], local_dir="./out")
mc.storage.download_all(local_dir="./out")                # list-then-download everything

Credentials are fetched on first use and cached, auto-refreshing as they near expiry — construct one Minerva() per process and reuse it. Errors mirror the rest of the SDK:

from minerva import (
    MinervaStorageError,                  # base for everything below
    MinervaStorageObjectNotFoundError,    # 404 — has .key
    MinervaStorageAccessDeniedError,      # 403 — has .key
    MinervaStorageCredentialError,        # couldn't fetch / parse credentials
    MinervaStorageTransferError,          # transient S3 failure — .original
    MinervaStorageExtraNotInstalledError, # boto3 not installed
)

Custom / tailored endpoints — mc.api.call

For client-specific routes (preview endpoints, partner integrations, paths Minerva built just for your org) on the Data API, use the generic mc.api.call:

result = mc.api.call("POST", "/v2/acme/lookup", json={"record_id": "abc"})

Same x-api-key auth, same error mapping, same rate-limit handling as the typed methods — the difference is the SDK doesn't know the response schema, so you get back the raw parsed JSON. The server enforces entitlement; callers without it see a 403 → MinervaAuthError.

Error handling

All errors derive from MinervaError:

from minerva import (
    MinervaValidationError,  # bad input — raised locally, before the call
    MinervaAuthError,        # 401/403 — bad key, not entitled
    MinervaRateLimitError,   # 429 — has .retry_after
    MinervaAPIError,         # other 4xx/5xx — has .status_code and .api_request_id
)

try:
    mc.api.enrich(records)
except MinervaRateLimitError as e:
    time.sleep(e.retry_after or 1)
except MinervaAPIError as e:
    print("request failed:", e.api_request_id)   # quote this to support

Responses are forward-compatible: new fields the API adds won't break an older client.

Python support

3.11, 3.12, 3.13, 3.14. Fully type-annotated (ships py.typed).

License

MIT © Minerva Data.