Python SDK

lvng==1.0.0 is the official Python client for the LVNG API. It runs on Python 3.11+, ships sync and async clients with identical surfaces, and covers all 51 OpenAPI tags / 386 operations with full type hints.

Install

Install
pip install lvng
# or
uv pip install lvng
poetry add lvng

Quickstart

Construct a client with your API key and start calling resources. The SDK ships two clients: LvngClient for blocking code and AsyncLvngClient for asyncio.

Sync

quickstart_sync.py
from lvng import LvngClient

client = LvngClient(api_key="YOUR_API_KEY")

response = client.chat.send(
    message="Summarize the latest sales report",
    platform="api",
)
print(response["reply"])

workflows = client.workflows.list(limit=20)
for workflow in workflows.get("data", []):
    print(workflow["id"], workflow["name"])

Async

quickstart_async.py
import asyncio
from lvng import AsyncLvngClient

async def main() -> None:
    async with AsyncLvngClient(api_key="YOUR_API_KEY") as client:
        response = await client.chat.send(
            message="Summarize the latest sales report",
            platform="api",
        )
        print(response["reply"])

        workflows = await client.workflows.list(limit=20)
        for workflow in workflows.get("data", []):
            print(workflow["id"], workflow["name"])

asyncio.run(main())

Authentication

The SDK authenticates every request with Authorization: Bearer <api_key>. Generate a key at Settings → Developer and store it in the LVNG_API_KEY environment variable — never commit it to source control.

auth.py
import os
from lvng import LvngClient

# Recommended — read from the environment
client = LvngClient(api_key=os.environ["LVNG_API_KEY"])

# Override the base URL (staging or self-hosted)
staging = LvngClient(
    api_key=os.environ["LVNG_STAGING_API_KEY"],
    base_url="https://staging.api.lvng.ai",
)

Configuration

Every option besides api_key is optional and has a sensible default.

OptionDefaultDescription
api_keyRequired. Per-user API key.
base_urlhttps://api.lvng.aiOverride for staging or self-hosted deployments.
timeout30.0Per-request timeout in seconds.
max_retries3Retries on 429 / 503 / transport errors.
retry_base_delay0.2Base for exponential backoff in seconds.
default_headersNoneExtra headers attached to every request.
config.py
from lvng import LvngClient

client = LvngClient(
    api_key="YOUR_API_KEY",
    base_url="https://api.lvng.ai",   # default
    timeout=30.0,                      # seconds; default 30
    max_retries=3,                     # 429 / 503 retries; default 3
    retry_base_delay=0.2,              # exponential backoff base in seconds
    default_headers={"X-App": "demo"}, # extra headers on every request
)

Streaming chat

Both clients expose a stream_chat() helper that yields parsed SSEEvent objects. Internally they call iterate_sse() (sync) and aiterate_sse() (async) on the streaming HTTP response.

Sync

stream_sync.py
from lvng import LvngClient

client = LvngClient(api_key="YOUR_API_KEY")

for event in client.stream_chat(message="Walk me through the Q4 numbers"):
    if event.event == "text_delta":
        delta = (event.data or {}).get("delta", "")
        print(delta, end="", flush=True)
    elif event.event == "tool_use":
        print("\n[tool]", event.data)
    elif event.event == "done":
        print("\n[done]")
        break

client.close()

Async

stream_async.py
import asyncio
from lvng import AsyncLvngClient

async def main() -> None:
    async with AsyncLvngClient(api_key="YOUR_API_KEY") as client:
        async for event in client.stream_chat(
            message="Walk me through the Q4 numbers",
        ):
            if event.event == "text_delta":
                delta = (event.data or {}).get("delta", "")
                print(delta, end="", flush=True)
            elif event.event == "done":
                print("\n[done]")
                break

asyncio.run(main())

Need a lower-level entry point? The SSE iterators are exported from the package:

low-level SSE
# Already have an httpx.Response from a streaming request?
# Use iterate_sse / aiterate_sse directly.
from lvng import iterate_sse, aiterate_sse, SSEEvent

Error handling

The SDK raises a typed exception hierarchy rooted at LvngError. Match on the subclass for actionable recovery.

ExceptionStatusWhen it fires
LvngErroranyBase class. Has message, status_code, code, body.
AuthenticationError401API key is missing, expired, or revoked.
ValidationError400 / 422Request body or query parameters were rejected.
NotFoundError404Resource does not exist or is not visible to your workspace.
RateLimitError429Throttled. Inspect retry_after (seconds).
errors.py
from lvng import (
    LvngClient,
    LvngError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

client = LvngClient(api_key="YOUR_API_KEY")

try:
    workflow = client.workflows.get(id="wf_does_not_exist")
except AuthenticationError as err:
    print("API key is invalid or revoked:", err.message)
except RateLimitError as err:
    print(f"Rate limited. Retry after {err.retry_after:.1f}s")
except NotFoundError as err:
    print(f"Workflow {err.identifier!r} does not exist")
except ValidationError as err:
    print("Bad request:", err.message, err.body)
except LvngError as err:
    print(f"LVNG {err.status_code} {err.code}: {err.message}")

Pagination

List endpoints come in two flavours: cursor-based (audit log, conversations, developer events) and offset-based (workflows, agents, channels, ...). Both keep their server-side wire shape so the SDK doesn't paper over the difference.

pagination.py
from lvng import LvngClient

client = LvngClient(api_key="YOUR_API_KEY")

# Cursor-based pagination (audit log, conversations, developer events, ...)
cursor: str | None = None
while True:
    page = client.audit_log.list(limit=100, cursor=cursor)
    for entry in page.get("data", []):
        print(entry)
    cursor = page.get("nextCursor")
    if not cursor:
        break

# Offset-based pagination (workflows, agents, channels, ...)
first = client.workflows.list(limit=50, offset=0)
second = client.workflows.list(limit=50, offset=50)

Resources reference

Every OpenAPI tag is exposed as a snake-cased attribute on the client. The hand-tuned resources (chat, workflows, agents, knowledge, api_keys) extend their generated counterparts with stronger types and convenience helpers.

ResourceExampleAPI docs
chatclient.chat.send / stream_chatReference →
messagesclient.messages.list / createReference →
channelsclient.channels.list / createReference →
workspacesclient.workspaces.listReference →
agentsclient.agents.create / messageReference →
toolsclient.tools.listReference →
oodaclient.ooda.executeReference →
workflowsclient.workflows.execute / parseReference →
twinsclient.twins.listReference →
artifactsclient.artifacts.listReference →
knowledgeclient.knowledge.search / ingestReference →
canvasclient.canvas.listReference →
calendarclient.calendar.listReference →
voice_notesclient.voice_notes.listReference →
searchclient.search.queryReference →
settingsclient.settings.getReference →
integrationsclient.integrations.listReference →
conversationsclient.conversations.listReference →
api_keysclient.api_keys.create / rotateReference →
invitationsclient.invitations.listReference →
emailsclient.emails.sendReference →
billingclient.billing.balanceReference →
contactsclient.contacts.listReference →
analyticsclient.analytics.costsReference →
developerclient.developer.listReference →
platform_connectionsclient.platform_connections.listReference →
mcpsclient.mcps.listReference →
activityclient.activity.listReference →
onboardingclient.onboarding.startReference →
preferencesclient.preferences.getReference →

Next steps