Python SDK
lvng is the 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 / 404 operations with full type hints.
Install
Not published yet: the client is not on PyPI, so pip install lvng does not work today. Until it is published, call the REST API with any HTTP client (for example requests or httpx), as in the Cookbook.
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
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
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.
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.
api_key—Required. 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.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
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
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:
# Already have an httpx.Response from a streaming request?
# Use iterate_sse / aiterate_sse directly.
from lvng import iterate_sse, aiterate_sse, SSEEventError handling
The SDK raises a typed exception hierarchy rooted at LvngError. Match on the subclass for actionable recovery.
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).from lvng import (
LvngClient,
LvngError,
AuthenticationError,
NotFoundError,
RateLimitError,
ValidationError,
)
client = LvngClient(api_key="YOUR_API_KEY")
try:
workflow = client.workflows.get_by_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.
from lvng import LvngClient
client = LvngClient(api_key="YOUR_API_KEY")
# Cursor-based pagination (a workspace's audit log, ...)
cursor: str | None = None
while True:
page = client.audit_log.audit_logs("YOUR_WORKSPACE_ID", limit=100, cursor=cursor)
for entry in page.get("logs", []):
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, generated from the API reference. Streaming chat is a method of the client itself: client.stream_chat.