Loxtep Python SDK

Python client for the Loxtep API, organized around one journey: ingest → define → deliver. Reach for the namespace that matches the stage you're in:

• Ingest — triggers, connectors, workflows, data_products.get_writer
• Define — data_products, schemas, quality, catalog, discovery, thesaurus, domains, standards, data_contracts
• Deliver — data_products (get_reader/stream/replay/query), targets
• Advanced / platform — projects, templates, instances, observe, queues, metrics*, process_intelligence

_{* metrics is a no-op reporter — matching the Node.js SDK, where it is also a no-op stub.}

The namespace names, casing, method names, and journey grouping match the Node.js SDK — see Parity with the Node.js SDK at the bottom.

Stream bus: loxtep.rstreams is a native Leo data-plane client (Kinesis write + DynamoDB/S3 read). Enable it with pip install loxtep[streams] and pass streams= (or set LEO_* env).

Not yet ported from the Node.js SDK: the config, auth, codegen, skills, authoring, http, checkpoint modules (client namespaces are at parity). rstreams perf/edge follow-ups: S3 byte-range fast-read, and snapshot/archive queue transitions.

Install

pip install -e .
# or from repo root: pip install -e sdks/python

Quick start

Sync client

from loxtep import LoxtepClient

client = LoxtepClient(
    api_url="https://api.loxtep.com",
    auth={"type": "jwt", "token": "YOUR_JWT"},
    organization_id="org-id",
    project_id="project-id",
)

# --- Ingest: connect a source and write events ---
triggers = client.triggers.list()
# With stream config (streams= / LEO_* env + pip install loxtep[streams]) this
# writer produces to the Kinesis bus; otherwise it uses the HTTP data path.
writer = client.data_products.get_writer("my-data-product")  # resolves name→id
writer.write({"id": "e1", "name": "Alice"})
writer.close()

# Workflows (the ingestion → transformation → export DAG)
workflows = client.workflows.list(project_id="project-id")
graph = client.workflows.get_graph("workflow-id", "project-id")

# --- Define: data products, schema, quality, discovery ---
asset = client.data_products.get("data-product-id")
items, pagination = client.data_products.list(page=1, page_size=20)
results = client.data_products.query("data-product-id", "SELECT * FROM t LIMIT 10")
client.quality.list()
client.catalog.search("query")
# Discovery (access-filtered when user context present)
client.discovery.search("events", include_evidence=True)
client.schemas.get("data-product-id")

# --- Deliver: consume events and route data out ---
for event in client.data_products.stream("data-product-id"):
    print(event)

# Targets — how a data product delivers data to external systems
client.targets.list("data-product-id")
client.targets.create(
    "data-product-id",
    target_type="webhook",
    endpoint_url="https://example.com/hook",
    method="POST",
)

client.close()

Async client

import asyncio
from loxtep import AsyncLoxtepClient

async def main():
    async with AsyncLoxtepClient(
        api_url="https://api.loxtep.com",
        auth={"type": "jwt", "token": "YOUR_JWT"},
    ) as client:
        asset = await client.data_products.get("data-product-id")
        workflows = await client.workflows.list(project_id="project-id")
        async for event in client.data_products.stream("data-product-id"):
            print(event)

asyncio.run(main())

API surface

Grouped by journey stage; kind labels: Resource (full CRUD), Reference (read-only), Runtime (live stream I/O).

Ingest

Define

Deliver

Advanced / platform

client.workflows

client.triggers

Ingest-side source bindings (external systems that feed a workflow).

client.targets

Delivery sink bindings — how a data product delivers data to external systems (webhooks, API endpoints, exports, database syncs, BI connections, event streams).

Target model fields: consumption_id, data_product_id, organization_id, target_type, delivery_method, status, is_active, endpoint_url, method, name, description, headers, filters, configuration, metadata, created_at, updated_at.

target_type values: webhook, api_endpoint, export, database_sync, bi_connect, event_stream.

client.connectors

client.projects

client.templates

client.data_products

Writer/reader transport: when the client is configured with stream-bus config (streams= or LEO_* env) and boto3 is installed (pip install loxtep[streams]), data_products.get_writer returns a native Kinesis producer and get_reader a native DynamoDB/S3 consumer — the performant path, matching the Node.js SDK — on both the sync and async clients. Without stream config both fall back to the HTTP data path. Pass auto_checkpoint=True (or call reader.checkpoint()) to persist read position to LeoCron. See the loxtep.rstreams module.

client.thesaurus

Canonical correlation keys + aliases.

client.instances

client.procedures

client.observe

client.metrics

metrics.log / metrics.get_reporter are no-ops — matching the Node.js SDK, where the metrics surface is also a no-op stub (it POSTs nowhere).

client.process_intelligence (experimental)

Returns unified context from connected systems for a given entity. Experimental and excluded from the core surface (parity with the Node.js SDK).

context = client.process_intelligence.get_entity_context("org-id", "order", "4821")

Type hints

Types are hand-written; all public methods and responses use type hints. Run mypy src/loxtep for static checking.

CLI

After pip install -e . you get a loxtep command:

• login – runs Node.js loxtep login (requires Node/npx)
• query / stream / replay – data-product operations (Python SDK)
• workflows list / workflows deploy – (Python SDK)
• observe status – bot/queue status (Python SDK)
• projects list / projects get – (Python SDK)
• templates list / templates get – (Python SDK)
• config export – export SDK bootstrap config from a connector (Python SDK)

Config: ~/.loxtep/config.json or env LOXTEP_API_URL, LOXTEP_ORGANIZATION_ID, LOXTEP_PROJECT_ID. Auth: run loxtep login once or set LOXTEP_TOKEN.

See docs/CLI.md for full CLI usage.

Parity with the Node.js SDK

The client surface matches: same namespace names, same snake_case method names, same ingest → define → deliver grouping. flows merged into workflows, connections→triggers, delivery→targets, data_products writer/reader + get_lexicon/readiness/promote/invalidate_cache, schemas.list/ tag_pii_fields, quality.create, projects.repository, instances.get_stream_config, and the thesaurus namespace are all present. improvements/activity/process_intelligence exist and are internal/ experimental (excluded from the documented surface), matching Node.

Remaining, intentional differences:

domains, standards, and data_contracts are real HTTP-backed namespaces. The loxtep.rstreams module is a native Leo data-plane client — sync and async write + read, LeoCron checkpointing, and >600 KB S3 write-offload — with no dependency on any external Leo SDK. Remaining follow-ups are perf/edge only: S3 byte-range fast-read, and snapshot/archive queue transitions.

The low-level writer escape hatch (workflows.get_writer, formerly flows.get_writer) exists in both and is intentionally undocumented for customers.