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, shapes, quality, catalog, discovery, thesaurus, proposals, bundles, 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), or use from_workspace() after Node loxtep attach (writes streams into .loxtep/project.json).

Not yet ported from the Node.js SDK: skills / workflow authoring helpers. Codegen (loxtep generate) and workspace auto-config (from_workspace) are native Python. rstreams follow-ups: S3 byte-range fast-read, snapshot/archive queue transitions.

Install

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

Login from code

Mint an sdk_python session (isolated from CLI, web, Node SDK, and MCP):

from loxtep import LoxtepClient, login, browser_login

api_url = "https://api.loxtep.io"  # or apidev.loxtep.io

# Email / password (optional mfa_code=...). Defaults to client_channel="sdk_python".
tokens = login(api_url, "you@co.com", "secret")
# Or browser handoff → /auth/sdk?runtime=python:
# tokens = browser_login("https://app.loxtep.io")

client = LoxtepClient(
    api_url=api_url,
    auth={"type": "jwt", "token": tokens["access_token"]},
)

Refresh with from loxtep import refresh then refresh(api_url, tokens["refresh_token"]). For CLI credentials on disk, keep using Node loxtep login (that mints a CLI session) + LoxtepClient.from_workspace().

Quick start

Sync client (workspace auto-config)

After Node CLI login → init → attach (writes .loxtep/project.json + credentials):

from loxtep import LoxtepClient

client = LoxtepClient.from_workspace()
writer = client.get_writer("app-events")
writer.write({"user_id": "u_1", "action": "signup"})
writer.close()
client.close()

See Write to a data product for the full flow.

Sync client (explicit config)

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.build.workflows.list(project_id="project-id")
graph = client.build.workflows.get_graph("workflow-id", "project-id")
# Enrichment query_trigger (D14): preview_query_trigger / run_query_trigger

# --- 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.meaning.thesaurus

Canonical correlation keys + aliases (client.meaning.thesaurus).

client.define.shapes

Org / domain canonical schemas (distinct from client.define.schemas data-product versions).

client.meaning.proposals / client.meaning.bundles

client.instances

Python has no loxtep instances stream-config verb — use this method, or the Node CLI (npx loxtep instances stream-config).

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 loxtep you get a loxtep command:

• Native (Python): query, stream, replay, generate, workflows list|deploy, observe status, projects, templates, config export
• Delegated to Node (npx loxtep …): login, init, attach, ingest, deploy, and any other lifecycle command — requires Node.js/npx

Config: .loxtep/project.json (after attach) + project-local or ~/.loxtep/credentials.json, or env LOXTEP_*. Prefer LoxtepClient.from_workspace() in app code.

See docs/CLI.md and docs/sdk-first-ingest.md.

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 meaning/define Phase 2 (thesaurus CRUD, define.shapes, meaning.proposals, meaning.bundles.import_) are present. Python uses bundles.import_ because import is a keyword (Node: bundles.import). improvements/activity/process_intelligence exist and are internal/experimental (excluded from the documented surface), matching Node.

LoxtepClient.from_workspace() / AsyncLoxtepClient.from_workspace() match Node's fromWorkspace(): env > explicit kwargs > .loxtep/project.json + credentials, including region and streams from attach.

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.