elan-sdk

Call the Elan platform's agents, approvals and billing from your own backend.

Async-first, with a synchronous facade for scripts and notebooks. It talks to the same API the CLI and the console do.

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

Requires Python 3.12 or newer.

The five-line version

from elan_sdk import Elan

elan = Elan.from_env()
reply = elan.runtime.run("invoice-triage", "What do I owe on INV-9?")
print(reply.text)

from_env() reads ELAN_CLIENT_ID, ELAN_CLIENT_SECRET and ELAN_PROJECT. ELAN_API_URL defaults to the production API, so you set it only to point somewhere else.

That one call subscribes to the event stream, waits for the subscription to be acknowledged, publishes the invocation, filters the stream to this run, joins the reply's parts, stops at a terminal event and closes the socket. Every one of those steps is something a hand-rolled client has got wrong.

AsyncElan is the same surface with await.

What it covers

Namespace	What it is for
runtime	start a run, watch its events, resume a paused one
runs	read a finished run back
sessions	conversations: create, list, read back and delete — from your agent's own database
actions	the approval inbox, and deciding a tool confirmation
approvals	approvals asked of somebody outside the conversation
tools	long-running tool calls your own backend owes an answer to
autonomy	what a tool may do without asking, per user and per tenant
wallets / credit	your end users' balances — read and write, split apart
usage / analytics	what your agents have consumed, and your own balance
artifacts	files an agent can read, uploaded without passing through us
tickets	a short-lived credential so a browser can watch a run
agents	what is deployed, and what it declares

Watching events yourself

The rest of this page uses AsyncElan, since streaming is where await earns its keep. Elan has every one of these methods too, without the await.

Subscribe before you invoke. That ordering is the whole reason this is a context manager — subscribing afterwards loses the entire reply on a warm agent.

async with elan.runtime.watching("sess-ada-1") as stream:
    await elan.runtime.invoke("hotel-concierge", "hi", session_id="sess-ada-1")
    async for event in stream.events():
        print(event.kind, event.text)

Running for one of your end users

invoke takes end_user_id; run deliberately does not. Naming a user addresses that run's events to them, so a backend that invoked and watched in one process would see nothing — the SDK raises EndUserEventsElsewhere rather than hanging.

await elan.runtime.invoke(
    "hotel-concierge", "refund my last night",
    session_id="sess-ada-1", end_user_id="ada")

Their browser cannot hold your client secret, so mint a ticket — it opens the event stream and nothing else:

ticket = await elan.tickets.mint(end_user_id="ada",
                                 origin="https://app.acme.example")

Seeding state and bounding a run

invoke, run and Conversation.say take two of ADK's own run_async options. Both are sent only when set, so an ordinary message is unchanged.

reply = await elan.runtime.run(
    "hotel-concierge", "Is late checkout available?",
    session_id="sess-ada-1",
    state_delta={"room": "412"},
    run_config={"max_llm_calls": 5,
                "custom_metadata": {"request_id": "req-81c2"}})

• state_delta writes keys into this conversation's state on this turn. app:, user:, temp: and agentkit: keys are refused with a BadRequest naming the key — they would reach other users or other conversations, or would not be kept. At most 64 KiB.
• run_config accepts max_llm_calls, which can only lower the agent's own limit, and custom_metadata, which ADK stamps on every event the run emits. Anything else is refused by name.

An agent deployed before these options existed must be redeployed once before it accepts an invocation carrying either.

Conversations

A conversation lives in your agent's own database, and elan.sessions reads it there, through ADK's own session service — so the history is exactly what the agent saw.

await elan.sessions.create("hotel-concierge", session_id="sess-ada-1",
                           user="ada", state={"plan": "pro"},
                           title="Late checkout for room 412")  # the agent never sees it

for s in await elan.sessions.list("hotel-concierge", user="ada"):
    print(s.id, s.title, s.state_keys)              # newest first

events = await elan.sessions.events("hotel-concierge", "sess-ada-1",
                                    user="ada", recent=200)
for turn in await elan.sessions.turns("hotel-concierge", "sess-ada-1", user="ada"):
    print(turn.role, turn.text)

await elan.sessions.delete("hotel-concierge", "sess-ada-1", user="ada")

• events() returns the same AgentEvents the stream delivers, so one fold renders a live run and a reopened conversation, and ends_the_run ends each turn in both. turns() is the messages alone; get() is the state without the events; run() is a workflow's steps and pause.
• user= names one of your end users by your own id and needs a tenant admin. Without it, every method acts on the caller's own conversations. Create and invoke for the same person — a conversation you create as yourself is not the one an end_user_id invocation runs in.
• recent and after page a long conversation. after is inclusive and seq is numbered per response, so dedupe on emitted_at.
• A Conflict saying "redeploy it once" means exactly that: the agent was deployed before its history could be read.
• Every method takes project=; without it, the client's own.

elan.runs answers a different question — how one run ended and what it cost.

Paused runs

A run that stops to ask a person something emits no terminal event. A client waiting for one waits out its whole timeout. Read the interrupt instead:

from elan_sdk.models.events import ends_the_run

async for event in stream.events():
    if ends_the_run(event):
        break

Money

Everything a wallet reports is micro-USD — divide by 1,000,000. Your tenant's own balance is in tokens, which is why the two use different field names (balance_micros against balance_units) rather than one whose unit a reader has to know.

Reads and writes are deliberately separate objects: elan.wallets issues no non-GET at all, so a read-only credential can be handed it and a future write cannot quietly land on it. elan.credit is the write half.

Errors

Every failure is an ElanError subclass carrying the status and the service's own message — BadRequest, Unauthorized, Forbidden, NotFound, Conflict, TransportError. Nothing raises a bare httpx exception.

Documentation

Full guides, the API reference and the TypeScript half are at https://openelan.ai/app/docs.

Licence

Apache-2.0. See LICENSE.