Lemma Python SDK

lemma-sdk is the Python client library for Lemma. It wraps a generated OpenAPI client with a pod-first, ergonomic surface for tables, files, functions, agents, workflows, schedules, surfaces, apps, and connectors.

It is the same SDK that runs inside Lemma functions (where the runtime injects auth automatically) and in standalone application code (where you supply a token). The CLI and TUI live in the sibling lemma-cli package.

• package name: lemma-sdk
• import root: lemma_sdk
• Python >=3.14,<3.15 (uv recommended)

Reading the source. In a Lemma sandbox the SDK is installed, and its source is readable where the interpreter found it: python -c "import lemma_sdk; print(lemma_sdk.__path__[0])". When you need an exact signature or response shape, read it rather than guessing — e.g. resources/data.py under that directory.

Install

The published package is lemma-sdk, not lemma-python — that is the directory it is built from.

uv add lemma-sdk            # or: pip install lemma-sdk

From a checkout, for working on the SDK itself:

uv pip install .            # or: uv pip install --editable .
python -c "from lemma_sdk import Pod, Lemma; print(Pod, Lemma)"

This package requires Python 3.14 (requires-python = ">=3.14,<3.15"), which is not what python3 is on current distributions or on macOS. Check before you install, because the failure is quiet: on an older interpreter neither pip nor uv errors — the resolver walks back to 0.6.2, the last release that allowed 3.11, and installs that instead. You get an SDK two minors behind the API you are calling and nothing says so. Confirm what you actually got with python -c "import importlib.metadata as m; print(m.version('lemma-sdk'))".

uv will provision 3.14 for you:

uv venv --python 3.14 && uv pip install lemma-sdk   # standalone environment

In a uv project, put requires-python = ">=3.14" in your pyproject.toml before uv add lemma-sdk; otherwise the project's own floor is what makes the resolver reach for the old release.

The lemma CLI is unaffected by any of this: uv tool install lemma-terminal provisions its own interpreter, so the CLI works whatever python3 on your machine happens to be.

Two entry points

from lemma_sdk import Pod, Lemma

pod = Pod.from_env()                    # token + pod id from env / CLI session
lemma = Lemma.from_env(org_id="org-id") # org-scoped client; lemma.pod("id") -> Pod

Pod is a context manager and owns an HTTP transport when constructed directly:

with Pod.from_env() as pod:
    pod.functions.run("triage_ticket", {"ticket_id": "rec-1"})

lemma.pod(...) and lemma.for_org(...) are views of the client they come from: same endpoint, same credential, same connection pool, so closing the parent closes everything. Only a directly constructed Pod/Lemma owns a transport of its own.

Every call is synchronous

There is no async client. Each method makes a blocking HTTP request, and a retried request sleeps (up to a few seconds) on the calling thread. That is what you want in a script, a function handler doing one thing at a time, or a worker thread — and it is not what you want on an event loop. From async def code that also serves other work, hand each call to a thread:

import asyncio

row = await asyncio.to_thread(pod.table("tickets").get, ticket_id)

# Fanning out? to_thread + gather runs the round trips concurrently; a plain
# loop of SDK calls serializes them and freezes the loop for the whole batch.
rows = await asyncio.gather(*(
    asyncio.to_thread(pod.table("tickets").get, tid) for tid in ticket_ids
))

Authentication & configuration

The SDK resolves settings from explicit arguments first, then environment, then the CLI config file (~/.lemma/config.json).

Environment variables:

export LEMMA_TOKEN="<access-token>"      # required if not using a CLI session
export LEMMA_POD_ID="<pod-id>"           # required for Pod.from_env()
export LEMMA_ORG_ID="<org-id>"           # required for org-scoped calls
export LEMMA_BASE_URL="https://api.lemma.work"
export LEMMA_AUTH_URL="https://lemma.work/auth"
export LEMMA_REFRESH_TOKEN="<refresh-token>"     # optional
export LEMMA_CONFIG_FILE="~/.lemma/config.json"  # optional override
export LEMMA_SSL_NO_VERIFY=1              # local/self-signed only

LEMMA_REFRESH_TOKEN is what keeps a long-running process working past its access token's lifetime: on a 401 the client exchanges it once, replaces the token in memory, and replays the request. Nothing is written to ~/.lemma/config.json — the refreshed token lives for the process. A refresh token found in a CLI session is used the same way, unless you passed token= yourself, in which case the credential stays exactly what you supplied.

Inside a Lemma function, LEMMA_TOKEN (a workload token scoped to the function's grants) and LEMMA_POD_ID are injected for you — just call Pod.from_env().

Explicit construction (no env needed):

pod = Pod(pod_id="pod-id", org_id="org-id", token="token",
          base_url="https://api.lemma.work")

When LEMMA_TOKEN is unset, settings fall back to the selected server in the CLI config:

{
  "active_server": "cloud",
  "servers": {
    "cloud": {
      "base_url": "https://api.lemma.work",
      "auth_url": "https://lemma.work/auth",
      "defaults": { "org_id": "org-id", "pod_id": "pod-id" }
    }
  }
}

(Legacy active_context / contexts keys are still accepted and translated.) Install and manage the CLI from lemma-cli; see lemma-cli/SETUP.md.

Response shapes — read this first

Every method returns a typed response object. Call .to_dict() for plain data, then unwrap:

Single-record create/get/update return the record directly — there is no {"data": {...}} envelope. Call .to_dict() and use the result as the row. The pod.records helpers already unwrap to a plain dict for you. The bulk helpers return the integer count directly.

Pod facades

pod.tables · pod.records · pod.queries · pod.files · pod.functions · pod.agents · pod.workflows · pod.schedules · pod.conversations · pod.members · pod.apps · pod.surfaces · pod.connectors

Plus helpers: pod.table(name) (bound single-table helper), pod.query(sql), pod.generated (raw OpenAPI client escape hatch).

Tables & records — full CRUD

t = pod.table("tickets")

row = t.create({"title": "Refund", "status": "new"})   # already a plain dict
ticket_id = row["id"]

row = t.get(ticket_id)                            # bare record dict, no envelope
t.update(ticket_id, {"status": "resolved"})       # only passed fields change
t.delete(ticket_id)

# Writing more than a row or two? Use the batch form -- one round trip instead
# of N. A loop of t.create(...) pays a full request per row.
t.bulk_create([{"title": f"Refund {i}", "status": "new"} for i in range(50)])

rows = pod.records.list(
    "tickets", limit=50,
    filter=[
        {"field": "status", "op": "eq", "value": "new"},
        {"field": "priority", "op": "ne", "value": "low"},
    ],
    sort=[{"field": "created_at", "direction": "desc"}],
).to_dict()["items"]

# `list` returns one page — `limit` defaults to 20 and the rest is behind
# `next_page_token`. When the answer has to be complete, page to exhaustion:
every_open = pod.records.list_all(
    "tickets", filter=[{"field": "status", "op": "eq", "value": "new"}]
)   # -> list of plain row dicts; t.list_all() is the same walk for one table

totals = pod.query(
    "select status, count(*) as total from tickets group by status"
).to_dict()["items"]

The pod.records / pod.table(...) create/get/update helpers return the bare record as a plain dict (no .to_dict(), no ["data"] unwrap). list and query return response objects; call .to_dict() and read ["items"]; list_all returns the rows themselves.

Record data is dynamic because table schemas are user-defined.

RLS vs shared tables

Tables carry an enable_rls flag that defaults to true (row-level security on). With RLS on, each row is owned by its creator: non-admin members read/update/delete only their own rows (other users' rows are invisible — cross-user access returns 404), while pod admins see and manage every row. This is the right default for per-user/personal data.

Set enable_rls: false for SHARED/reference/team tables that all members should see and mutate. RLS only scopes which rows a non-admin can touch — it does not change the permission a write needs: writing any table requires the DATASTORE_RECORD_WRITE permission (POD_USER and above), RLS or not. The read-only query endpoint can join across tables only when they are non-RLS.

Bulk record operations

Reach for these whenever you write more than a couple of rows: each one is a single request, where a loop of create is one request per row. The same three methods exist on the bound helper — pod.table("tickets").bulk_create(rows) — so you never have to leave the table handle to get a batch.

# create: row dicts (ids generated)
created_count = pod.records.bulk_create("ticket_events", [
    {"ticket_id": ticket_id, "kind": "created"},
    {"ticket_id": ticket_id, "kind": "triaged"},
])

# update: FLAT dicts that MUST include the primary key
updated_count = pod.records.bulk_update("tickets", [
    {"id": id_a, "status": "resolved"},
    {"id": id_b, "status": "waiting", "priority": "urgent"},
])

# delete: list of primary-key values
deleted_count = pod.records.bulk_delete("tickets", [id_a, id_b])

Files — searchable documents (built-in RAG)

Files uploaded to a pod are automatically indexed: text is extracted, chunked, and embedded, so they become searchable with no separate vector DB or infra. That makes files the pod's built-in retrieval-augmented generation store.

Only document formats are indexed: PDF, DOC/DOCX, ODT, RTF, Markdown, plain text, HTML, EPUB. Data/binary formats (CSV, TSV, JSON, YAML, XLSX, images, email) are stored but not indexed (status NOT_REQUIRED) and never appear in search — so keep structured data in tables and prose/documents in files. search_enabled toggles indexing per file; status flows PENDING → PROCESSING → COMPLETED (searchable) / NOT_REQUIRED / FAILED, and only COMPLETED documents are searchable. Documents are also converted to markdown.

/me is each user's private per-user tree (only the owner sees their /me files); all other paths are pod-shared, and folder grants cascade to every descendant.

pod.files.create_folder("/reports", description="Generated reports")
pod.files.upload("/tmp/summary.md", directory_path="/reports")

# Plain search (defaults to the whole pod):
hits = pod.files.search("refund policy").to_dict()

# Directory-scoped RAG + method selection:
hits = pod.files.search(
    "refund policy",
    scope_path="/knowledge",     # restrict to a folder
    scope_mode="SUBTREE",        # SUBTREE = folder + all descendants (default); DIRECT = immediate children only
    search_method="HYBRID",      # TEXT (full-text), VECTOR (semantic), or HYBRID
).to_dict()

md  = pod.files.download_markdown("/knowledge/policy.pdf")           # converted markdown bytes
kids = pod.files.list_children("/knowledge/policy.pdf")              # derived child files (md, figures, pages)
raw = pod.files.download("/knowledge/policy.pdf")                     # bytes

Functions, agents, workflows

run = pod.functions.run("triage_ticket", {"ticket_id": "rec-1"}).to_dict()
# run["status"], run["output_data"], run["logs"]

agent = pod.agents.get("triage").to_dict()
conv = pod.conversations.create_for_agent("triage", title="Triage")
pod.conversations.send(str(conv.to_dict()["id"]), "Classify ticket rec-1")

wf_run = pod.workflows.create_run("nightly_review").to_dict()
# Workflow inputs are collected by FORM nodes mid-run, not at start; submit them with
# pod.workflows.submit_form(wf_run["id"], node_id="<form_node>", inputs={"limit": 10})

Connectors (calling external apps)

pod.connectors.execute(auth_config, operation, payload) runs a third-party operation. The first argument is the auth config name (often the app id), the operation id and payload come from discovery, and the response is under ["result"].

sent = pod.connectors.execute(
    "workspace-gmail",          # auth config name
    "GMAIL_SEND_EMAIL",         # operation id from discovery
    {"recipient_email": "a@example.com", "subject": "Hi", "body": "..."},
).to_dict()["result"]

# discover before you call:
matches = pod.connectors.operations.search("workspace-gmail", "send email")
schema  = pod.connectors.operations.get("workspace-gmail", "GMAIL_SEND_EMAIL")

Operation ids and payload keys differ between the lemma and composio providers — confirm with discovery for the provider your org installed. Don't pass account_id unless pinning a specific account; the backend resolves the fixed or invoking-user account from the token.

Org & global usage (Lemma)

lemma = Lemma.from_env(org_id="org-id")

org   = lemma.org.get()
pods  = lemma.pods.list()
pod   = lemma.pod("pod-id")          # -> Pod sharing this transport
me    = lemma.user.profile()

# org-level connector setup
auth_configs = lemma.connectors.auth_configs.list()
accounts     = lemma.connectors.accounts.list(app="gmail")

# first-party tools
results = lemma.tools.web_search("vendor SLA policy", max_results=5)

Facades: lemma.orgs · lemma.org · lemma.pods · lemma.user · lemma.connectors · lemma.tools · lemma.agent_hosts · lemma.web_logins · lemma.org_runtime.

Writing a function

A Lemma function is a Python file with header comments declaring its types, plus a handler (ctx, data) -> output:

#input_type_name: TriageInput
#output_type_name: TriageResult
#function_name: triage_ticket

from pydantic import BaseModel
from lemma_sdk import FunctionContext, Pod

class TriageInput(BaseModel):
    ticket_id: str

class TriageResult(BaseModel):
    status: str

async def triage_ticket(ctx: FunctionContext, data: TriageInput) -> TriageResult:
    pod = Pod.from_env()    # authenticated as this function's workload principal
    pod.table("tickets").update(data.ticket_id, {"status": "triaged"})
    return TriageResult(status="triaged")

FunctionContext fields: pod_id, function_id, user_id, user_email, config. The function runs with zero default access — grant it the tables, folders, and apps it touches (see the lemma-builder skill / lemma functions permissions).

Typed models

lemma_sdk.models re-exports the common response types with friendly names:

from lemma_sdk.models import Record, FunctionRun, OperationExecution, Agent, Function

record: Record = pod.records.create("tickets", {"title": "Typed"})

Generated request models live under lemma_sdk.openapi_client.models (e.g. CreateTableRequest) for endpoints you build payloads for by hand.

Errors

from lemma_sdk import LemmaAPIError, LemmaConfigError

try:
    pod.records.get("tickets", "missing")
except LemmaAPIError as e:
    print(e.status_code, e.code, e.message)
except LemmaConfigError:
    ...   # missing token / pod id / unreadable config

Generated client escape hatch

For endpoints not yet wrapped by the ergonomic SDK:

generated = pod.generated   # authenticated; same base URL/token/timeout/SSL

Testing against a real Lemma API

Unit tests cover wrapper behavior; an opt-in integration scenario runs real end-to-end work against a running API. Start the local stack from the repo root:

make dev

Then from lemma-python:

export LEMMA_TOKEN="<access-token>"
LEMMA_RUN_CONNECTOR=1 uv run --with pytest --with pytest-asyncio \
  pytest tests/integration -m integration -s

This development-only scenario defaults to the fixed API port used by make dev; a packaged Desktop installation instead discovers its dynamic endpoint through locald. The scenario falls back to the CLI auth session if LEMMA_TOKEN is unset. Point elsewhere with LEMMA_CONNECTOR_BASE_URL / LEMMA_CONNECTOR_TOKEN. The scenario creates a fresh org and pod, exercises tables/records/query/files/functions/agents/ workflows/connectors, prints a summary, and deletes the pod.

Regenerate the SDK

Run the backend locally, then regenerate from its OpenAPI spec:

bash scripts/generate_openapi_client.sh
OPENAPI_URL=http://127.0.0.1:8000/openapi.json OPENAPI_INSECURE=1 \
  bash scripts/generate_openapi_client.sh

Generator env vars: LEMMA_API_URL, OPENAPI_URL, OPENAPI_INSECURE, LEMMA_SSL_NO_VERIFY.

Development checks

uv run ruff check lemma_sdk tests
uv run --with pytest python -m pytest tests