drc-vantage

Thin HTTP client for the Vantage platform. Talks only to the Next.js API (VANTAGE_API_URL) with a developer account API key — never to Postgres or FastAPI directly.

Breaking change: the deployment-wide API_KEY and X-Organization-Id / organization_slug are no longer used by the SDK. Create a developer account and API key in the app (Developers page). The organization is bound to the key.

Install

pip install drc-vantage
# or
uv add drc-vantage

Configuration

Classes

Client, Dataset, DataSource, DataSourceCategory, DataStorageSystem

Orchestration helpers: vantage_flow, send_workflow_event, sync_workflow_configs, orchestration_hooks (see below).

Workflows (@vantage_flow)

Register runs on the platform when a Prefect deployment executes. Workflows appear in the org after the first event (no DataSource link required).

from vantage import Client
from vantage.orchestration import vantage_flow

client = Client()

@vantage_flow(client, name="ingest-anthropic")
def ingest_anthropic():
    ...

After prefect deploy, refresh deployment metadata (schedules, triggers, parameters) from Prefect:

from vantage.orchestration import sync_workflow_configs

sync_workflow_configs(client)  # all workflows registered for the org
# sync_workflow_configs(client, deployment_ids=["..."])

Prefect orchestration events (manual hooks)

from prefect import flow
from prefect.runtime import deployment
from vantage import Client, DataSource, DataSourceCategory
from vantage.orchestration import orchestration_hooks

client = Client()  # uses VANTAGE_API_KEY (developer key)
hooks = orchestration_hooks(client)

@flow(
    name="ingest-anthropic",
    on_running=hooks.on_running,
    on_completion=hooks.on_completion,
    on_failure=hooks.on_failure,
    on_crashed=hooks.on_crashed,
    on_cancellation=hooks.on_cancellation,
)
def ingest_anthropic():
    category = DataSourceCategory(client=client, slug="saas", name="SaaS", color="#3366FF")
    DataSource(
        client=client,
        slug="anthropic",
        name="Anthropic",
        category=category,
        prefect_deployment_id=deployment.id,
        logo_url="https://example.com/logo.svg",
    ).register()
    ...

Dataset lifecycle example

from vantage import Client, Dataset

client = Client()  # organization is inferred from the developer API key

dataset = Dataset(
    client=client,
    location={"schema": "marts", "table": "customer_success_nps"},
    name="Customer Success NPS",
    type="TABLE",
    description="Qualtrics survey NPS summaries enriched with HubSpot contacts.",
    chat_question_examples={
        "en": ["What is the average NPS score by survey?"],
        "fr": ["Quel est le score NPS moyen par enquête?"],
    },
    data_source_slugs=["qualtrics", "hubspot"],
    storage_system_slug="postgresql",
)

result = dataset.register()
dataset.assign_to_team("customer-success")
dataset.assign_owner("owner@example.com")
result = dataset.analyze_and_check_alerts()

Catalog register

from vantage import Client, DataSource, DataSourceCategory, DataStorageSystem

client = Client()