Off-Nadir Delta — Python SDK

Official Python client for the Off-Nadir Delta geospatial event-intelligence API and MCP server. Query geolocated, source-linked event signals, activity hotspots and statistics, search satellite imagery (STAC), and run AI assessments and ask the analyst agent — all from Python. The AI tools are available on every plan, including Free — gated only by your token balance.

• Sync (Client) and async (AsyncClient) — one dependency (httpx).
• Fully typed responses (Pydantic v2), forward-compatible with additive API fields.
• Automatic cursor pagination, retry with backoff, and a uniform error model.
• A thin MCP client (McpClient / AsyncMcpClient) for the JSON-RPC MCP endpoint.

Requires Python 3.9+. Licensed under Apache-2.0.

Install

pip install offnadir-delta

Authentication

Create an API key from your account's Developer API page. Keys start with ond_ and are shown once. Pass it explicitly or via the OFFNADIR_DELTA_API_KEY environment variable.

from offnadir_delta import Client

client = Client(api_key="ond_...")          # or: Client()  -> reads OFFNADIR_DELTA_API_KEY

The API meters usage in tokens against the same wallet as the web app. Every metered response reports what it cost under meta.tokens. Check your balance for free first:

usage = client.usage()
print(usage.tokens.remaining, "tokens left, LLM access:", usage.plan.api_llm_access)

Quickstart

from offnadir_delta import Client

with Client() as client:
    # One page of the highest-severity signals in an AOI (bbox = [min_lon, min_lat, max_lon, max_lat])
    page = client.signals.list(
        bbox=[22, 44, 40, 53],
        days=7,
        min_severity=5,
        escalating=True,
        sort="severity",
        limit=100,
    )
    print(page.meta.count, "signals,", page.meta.tokens.charged, "tokens charged")
    for s in page.signals:
        print(s.event_date, s.country_code, s.title, s.severity_score)

Auto-pagination

signals.iterate() follows the cursor for you (each page is a separately-metered request):

for signal in client.signals.iterate(bbox=[22, 44, 40, 53], days=30, min_severity=6):
    print(signal.title)

list() / iterate() also support differential-sync and observability filters: updated_since / created_since (only signals (re)enriched at/after an ISO 8601 timestamp — for incremental sync), observability (observable / not-observable), open_data (sufficient / commercial-recommended / not-applicable), and min_information_gain (0-1).

Fetch one signal by id

signal = client.signals.get(4123456789)   # global_event_id from a list() result
print(signal.title, signal.severity_score)

Aggregates & hotspots

stats = client.signals.stats(bbox=[22, 44, 40, 53], days=7)
for row in stats.stats.by_category:
    print(row.category, row.count)

hotspots = client.signals.hotspots(bbox=[22, 44, 40, 53], precision=0.5)
for h in hotspots.hotspots:
    print(h.lat, h.lng, h.count, h.max_severity)

Satellite imagery (STAC search)

Metadata only — no image bytes, no signed URLs. bbox is required.

scenes = client.imagery.search(
    bbox=[22, 44, 40, 53],
    collection="sentinel-2-l2a",
    cloud_cover_max=20,
    days=14,
)
for scene in scenes:
    print(scene.id, scene.datetime, scene.cloud_cover)

Optical-observability weather

Cloud cover is what gates a Sentinel-2 optical pass. Get a per-day outlook (cloud at the ~10:30 local overpass, best clear day) to decide optical vs SAR collection. Requires a deployment with a commercial Open-Meteo licence. Weather data by Open-Meteo.com (CC BY 4.0) — surface result.weather.attribution.

result = client.weather.observability(lat=48.85, lon=2.35, end_date="2026-07-22")
print("best optical day:", result.weather.best_optical_day)
for day in result:
    print(day.date, day.optical_verdict, day.overpass_cloud_cover_pct)

AI assessment & analyst

Available on every plan, including Free — gated only by your token balance (an insufficient balance raises InsufficientTokensError). Check client.usage() first.

assessment = client.intelligence.assess(event_id=123, kind="quick")   # 5 quick / 15 deep tokens
print(assessment.content)

answer = client.intelligence.analyst(
    "What is escalating in the Black Sea this week?",
    bbox=[22, 44, 40, 53],
)                                                                      # metered 5-45 tokens
print(answer.brief)

assess is cached per (account, event, kind), so re-assessing the same event is free. A signal with no satellite-imageable footprint raises NotObservableError (422) before any token is charged — pre-filter with signals.list(observability="observable").

Long analyst runs (async jobs)

The analyst is hybrid-async: a run that finishes within ~95s returns the AnalystBrief directly (as above); a longer run returns an AnalystJobPending and completes in the background — poll it for free, and pay only when it completes. Pass an idempotency_key so a re-send resolves to the same run instead of being charged again (without one, analyst is not idempotent and is never retried automatically):

from offnadir_delta import AnalystBrief

result = client.intelligence.analyst(
    "Compare naval activity across the Black Sea and Baltic this month",
    idempotency_key="black-sea-baltic-2026-07-17",
)
if not isinstance(result, AnalystBrief):            # AnalystJobPending (HTTP 202)
    job = client.intelligence.get_job(result.job_id)  # free; status: running|done|error
    print(job.status)

Or let the SDK poll to completion for you (sync and async):

answer = client.intelligence.analyst_and_wait(
    "Compare naval activity across the Black Sea and Baltic this month",
    idempotency_key="black-sea-baltic-2026-07-17",
    poll_interval=5.0,
    timeout=300.0,
)
print(answer.brief)

On timeout this raises APITimeoutError, but the run keeps going server-side — re-fetch the brief later via get_job(job_id) (or the same idempotency_key) at no extra charge.

Daily World Brief (free)

brief = client.brief.get()          # latest; or client.brief.get("2026-07-11")

Data freshness / pipeline status (free)

Pre-flight whether the underlying data is fresh enough before spending on a metered query.

status = client.status.get()        # or client.status.current()
print(status.pipeline_status, "— data current through", status.data_current_through)

Async

Every call has an await-able equivalent on AsyncClient:

import asyncio
from offnadir_delta import AsyncClient

async def main():
    async with AsyncClient() as client:
        async for signal in client.signals.iterate(bbox=[22, 44, 40, 53], days=7):
            print(signal.title)

asyncio.run(main())

Error handling

All failures raise a subclass of OffnadirError carrying .status_code, .code, and .request_id (quote it in support requests):

from offnadir_delta.errors import (
    AuthenticationError, PermissionDeniedError, InvalidRequestError,
    NotObservableError, InsufficientTokensError, RateLimitError, APIError,
)

try:
    client.signals.list(bbox=[22, 44, 40, 53])
except InsufficientTokensError as e:
    print("Need", e.required, "have", e.available)
except RateLimitError as e:
    print("Retry after", e.retry_after, "seconds")

Transient failures (429, 5xx, connection errors) are retried automatically with backoff (honoring Retry-After); configure with Client(max_retries=...). The last response's rate-limit headers are available on client.last_rate_limit.

MCP

The API also exposes a stateless MCP server at /api/v1/mcp (JSON-RPC 2.0). MCP hosts such as Claude connect to it directly over OAuth 2.1 — see the docs. For programmatic use with a static API key, this SDK ships a thin client:

from offnadir_delta import McpClient

with McpClient() as mcp:
    mcp.initialize()
    print([t["name"] for t in mcp.list_tools()])
    result = mcp.call_tool("query_signals", {"bbox": [22, 44, 40, 53], "days": 7})
    brief = mcp.read_resource("brief://latest")

Tools: query_signals, query_stats, query_hotspots, search_imagery, assess_weather_observability (optical outlook; commercial weather licence only), get_world_brief, get_usage, assess_signal (metered), ask_analyst (metered) — all available on every plan, gated only by token balance.

Development

pip install -e ".[dev]"
ruff check src tests
mypy src
pytest

Tests run fully offline (HTTP mocked with respx) — they never call the live API or spend tokens.

License

Apache-2.0. See LICENSE and NOTICE.