Using Lamatic SDK with Python

This guide demonstrates how to integrate the Lamatic Python SDK into your Python applications to execute flows.

Installation

Requires Python 3.10+. Install with pip:

pip install lamatic

Getting Started

To use the Lamatic Python SDK, you’ll need:

• An API key or access token
• Your project ID
• The Lamatic API endpoint

Basic Setup

from lamatic import Lamatic
 
lamatic = Lamatic(
    endpoint="https://yourproject.lamatic.dev/graphql",
    project_id="your-project-id",
    api_key="your-api-key",
)

Authentication

The SDK supports two authentication methods:

from lamatic import Lamatic
 
lamatic = Lamatic(
    endpoint="https://yourproject.lamatic.dev/graphql",
    project_id="your-project-id",
    api_key="your-api-key",
)

from lamatic import Lamatic
 
lamatic = Lamatic(
    endpoint="https://yourproject.lamatic.dev/graphql",
    project_id="your-project-id",
    access_token="your-access-token",
)

You must provide either api_key or access_token, but not both, the constructor raises ValueError if neither is set.

Executing Flows

Execute a workflow synchronously with execute_flow:

response = lamatic.execute_flow(
    flow_id="your-flow-id",
    payload={"prompt": "Hello, Lamatic!"},
)
 
print(response.status)   # "success"
print(response.result)
print(response.message)

Async Usage

For async applications, use async_execute_flow instead:

import asyncio
from lamatic import Lamatic
 
lamatic = Lamatic(
    endpoint="https://yourproject.lamatic.dev/graphql",
    project_id="your-project-id",
    api_key="your-api-key",
)
 
async def main():
    response = await lamatic.async_execute_flow(
        flow_id="your-flow-id",
        payload={"prompt": "Hello from async!"},
    )
    print(response.status)
    print(response.result)
 
asyncio.run(main())

Streaming Responses

Use execute_flow_token_stream to show LLM output as it is generated, instead of waiting for the whole flow to finish. It runs the flow and yields one event at a time. Streaming requires lamatic v0.2.0 or later.

for event in lamatic.execute_flow_token_stream("your-flow-id", {"prompt": "Write a short story"}):
    if event.type == "token":
        print(event.token, end="", flush=True)  # one piece of text
    elif event.type == "final":
        print("\nFull text:", event.text)
    elif event.type == "error":
        print("\nStream failed:", event.message)

In async code, use async_execute_flow_token_stream with async for:

async for event in lamatic.async_execute_flow_token_stream("your-flow-id", {"prompt": "Write a short story"}):
    if event.type == "token":
        print(event.token, end="", flush=True)

Events

• final.text is every token joined together. Use it instead of reading generatedResponse from a node’s output, because the flow’s result only holds what its response node returns.
• final.text_by_node splits the text by node, for flows with more than one LLM or RAG node.
• raw is the original server frame. On the node event of an LLM node, raw["data"]["_meta"] holds token counts and cost.

Stopping a stream

In sync code, break out of the loop to stop generation; the connection is closed for you.

In async code, wrap the stream in contextlib.aclosing. Python does not close an async generator straight away when you break out of an async for loop, so without it the connection can stay open until the generator is garbage-collected:

from contextlib import aclosing
 
async with aclosing(lamatic.async_execute_flow_token_stream("your-flow-id", {"prompt": "Write a long story"})) as stream:
    async for event in stream:
        if event.type == "token":
            print(event.token, end="", flush=True)
        if user_pressed_stop():
            break  # aclosing closes the stream and its connection here

Cancelling the task that consumes the stream also stops it.

Handling stream errors

Failures arrive as an error event, not as a raised exception. That covers invalid credentials, network errors and a connection that drops before the flow finishes.

Checking Request Status

For long-running flows, poll for the result with check_status (sync) or async_check_status (async). Both accept a polling interval and a timeout, in seconds, and return as soon as the request reaches a terminal state (success, error, or failed):

response = lamatic.check_status(
    request_id="your-request-id",
    poll_interval=5,    # seconds between polls, default 15
    poll_timeout=120,   # max wait time in seconds, default 900
)
 
print(response.status)
print(response.result)

async def poll():
    response = await lamatic.async_check_status(
        request_id="your-request-id",
        poll_interval=5,
        poll_timeout=120,
    )
    print(response.status)
    print(response.result)

Updating Access Token

If you’re using access token authentication, update it at runtime with update_access_token:

lamatic.update_access_token("new-access-token")

Response Structure

execute_flow, async_execute_flow, check_status, and async_check_status all return a LamaticResponse:

@dataclass
class LamaticResponse:
    status: str                        # "success" | "error" | "failed" | in-progress while running
    result: dict | None
    message: str | None = None
    status_code: int | None = None

Example

response = lamatic.execute_flow("your-flow-id", {"prompt": "hey, how are you?"})
 
if response.status == "success":
    print("Result:", response.result)
else:
    print("Failed:", response.message)

Error Handling

execute_flow and async_execute_flow raise an exception on network failures (they don’t catch and swallow errors), so wrap calls in a try/except block:

try:
    response = lamatic.execute_flow(flow_id, payload)
    if response.status == "success":
        print("Success:", response.result)
    else:
        print("Execution failed:", response.message)
except Exception as e:
    print("SDK error:", e)

check_status and async_check_status behave differently: they catch errors internally and return a LamaticResponse with status="error" instead of raising, so you can check response.status without a try/except.

Environment Variables

For better security, keep credentials out of source code:

import os
from lamatic import Lamatic
 
lamatic = Lamatic(
    endpoint=os.environ["LAMATIC_PROJECT_ENDPOINT"],
    project_id=os.environ["LAMATIC_PROJECT_ID"],
    api_key=os.environ["LAMATIC_PROJECT_API_KEY"],
)

Resources

• PyPI package: pypi.org/project/lamatic
• GitHub repository: github.com/Lamatic/Lamatic-Python-SDK
• Examples: see examples/basic_usage.py in the repository

Support

For help and support with the Python SDK:

• Open an issue on GitHub
• Contact [email protected]