Python SDK - Niadra

The niadra package is the Python SDK: a synchronous client, Niadra, and an asynchronous twin, AsyncNiadra, with the same methods. It is open source under Apache 2.0, needs Python 3.10 or later, and depends only on httpx and pydantic. Every method on this page maps to a route of the API reference.

Install

pip install niadra

The client

from niadra import Niadra

niadra = Niadra()  # reads NIADRA_API_KEY

Niadra(
    api_key: str | None = None,
    *,
    base_url: str | None = None,
    channel: str | None = None,
    strict: bool = False,
    timeouts: Timeouts | None = None,
    cache: CacheOptions | None = None,
    queue: QueueOptions | None = None,
    http_client: httpx.Client | None = None,
)

Parameter	Type	Default	Description
`api_key`	`str`	`NIADRA_API_KEY`	A source key, `nia_sk_<live\|test>_<region>_<space>_<key_id>_<secret>`.
`base_url`	`str`	`NIADRA_BASE_URL`, then derived from the key	The key names its region and space, so the address is `https://<space>.<region>.api.niadra.com`. Override it for the local emulator.
`channel`	`str`	`None`	The default `channel` for events that do not name one, such as `"whatsapp"`.
`strict`	`bool`	`False`	Raise exceptions instead of logging and returning a safe value. Use it in tests.
`timeouts`	`Timeouts`	see below	Per-method time budgets.
`cache`	`CacheOptions`	see below	The per-conversation context cache.
`queue`	`QueueOptions`	see below	Batching of `track()`.
`http_client`	`httpx.Client`	`None`	Your own client, for proxies or custom transports.

AsyncNiadra takes the same arguments, with http_client: httpx.AsyncClient. One instance per process is enough, and it is thread-safe. The client also works as a context manager: leaving the block calls close(), and an exit hook flushes the queue for up to two seconds.

Properties

Property	Type	Description
`enabled`	`bool`	`False` when there is no usable key; every call is then a no-op.
`base_url`	`str`	The address the client talks to.
`mcp_url`	`str`	The MCP endpoint of the space, `<base_url>/mcp`.
`pending`	`int`	Items waiting in the local queue.
`dropped`	`int`	Items dropped so far: queue full, rejected by the API or not serializable.

Safe by default, strict on request

Without a key the client warns once and does nothing. Every public method catches and logs its own failures and returns a safe value: an empty Context (check context.error), an empty result, False or None. Logs carry method names, status codes, error codes and request ids, never handles or text. With strict=True the same failures raise the exceptions listed under Errors.

Time budgets

The SDK keeps its own time budget per method, whatever the platform around it allows.

from niadra import Timeouts

Timeouts(
    context=0.30,           # context(), every view except voice
    context_voice=0.15,     # context() with view="voice"
    navigation=0.60,        # search(), timeline(), open(), object reads
    navigation_voice=0.30,  # the same, with voice=True
    write=5.0,              # each attempt of a write sent at once
    upload=60.0,            # each attempt of sending media bytes to storage
)

Retries: 5xx answers, 429 and network errors are retried with backoff; 421 (the space is moving between cells) is retried at once on a fresh connection; other 4xx answers are final.

The context cache

Inside a conversation or task (a call with conversation_id or task_id), packs are cached in memory:

younger than 10 s: returned without a request;
up to 10 minutes older: returned at once while one background request refreshes it;
when a request fails: the last good pack, if it is less than 30 minutes old;
at most 1,000 packs, the least recently used evicted first.

Refreshes send the cached ETag, so an unchanged pack costs a not_modified answer instead of the full text, and only one refresh per pack runs at a time. A 401 or 403 is not an outage: the cached packs go (all of them on 401, the one requested on 403), so cutting a vendor’s access also cuts what it had cached. A plain read and a delta read of one conversation share one entry, and each delta is handed out once.

from niadra import CacheOptions

CacheOptions(enabled=True, ttl=10.0, stale_while_revalidate=600.0, max_stale=1800.0, max_entries=1000)

The write queue

track(), action() and handoff() only queue. A background thread (a task, with AsyncNiadra) sends a batch when 15 items are waiting or one second after the first arrived, with three attempts and backoff. When the queue is full, new items are dropped and counted in dropped.

from niadra import QueueOptions

QueueOptions(capacity=10_000, batch_size=15, interval=1.0, heartbeat_interval=60.0)

Handles

Handles identify a subject in a channel or system. The helpers are top-level functions that normalize the value and raise ValueError on input that cannot be valid, so a malformed phone fails where it enters your code.

Function	Handle	Example
`phone(number)`	`phone_e164`	`phone("+14155550123")`
`email(address)`	`email`, lowercased	`email("marina@example.com")`
`whatsapp(wa_id)`	`wa_id`	`whatsapp("14155550123")`
`whatsapp_bsuid(user_id, business_account)`	`wa_bsuid`, scoped to the business account	`whatsapp_bsuid("bsuid.8812", "waba-1")`
`system_id(namespace, id, *, kind=None)`	`system_id`; `kind="account"` or `"partner"` for organizations	`system_id("crm", "48213")`
`app_user(user_id)`	`app_user_id`	`app_user("u-7781")`
`anonymous(visitor_id)`	`anon_id`	`anonymous("visitor-31f")`

Every method that takes a handle also accepts a Handle model or a mapping with its fields ({"type": "phone_e164", "value": "+14155550123"}). Objects are accepted as "invoice:erp:0823" or as an ObjectRef.

context()

The context pack for a person, an organization or a business object. It maps to POST /v1/context.

niadra.context(
    subject: HandleLike | None = None,
    object: ObjectLike | None = None,
    *,
    about: HandleLike | None = None,
    view: str = "chat",
    verification: str = "V0",
    conversation_id: str | None = None,
    task_id: str | None = None,
    query: str | None = None,
    delta: bool = False,
    target: str | TargetModel | None = None,
    timeout: float | None = None,
    use_cache: bool = True,
) -> Context

Parameter	Type	Default	Description
`subject`	handle	`None`	Who the pack is about. Pass `subject` or `object`, exactly one.
`object`	`str` or `ObjectRef`	`None`	Center the pack on a business object, such as `"invoice:erp:0823"`.
`about`	handle	`None`	The account or partner the person acts for. Needs an active link.
`view`	`str`	`"chat"`	`voice`, `chat`, `brief`, `full`, `custom`, `account`, `partner` or a task view such as `task:billing`.
`verification`	`str`	`"V0"`	What the conversation proved, `V0` to `V4`, or `no_customer` for internal agents.
`conversation_id`	`str`	`None`	Pins the pack to the conversation and enables the cache.
`task_id`	`str`	`None`	The same, for an internal agent’s task.
`query`	`str`	`None`	What the turn is about. A read with a query is a one-off, never pinned.
`delta`	`bool`	`False`	Ask for what changed since this agent last read the subject. Never served from the cache.
`target`	`str` or `TargetModel`	`None`	The model that will read the pack, such as `"openai/gpt-4.1"`, so the pack aims at its prompt-cache floor.
`timeout`	`float`	from `Timeouts`	This call’s budget, in seconds.
`use_cache`	`bool`	`True`	Pass `False` to skip the cache for this call.

Returns a Context: every field of the API response plus what the SDK knows about the call.

Field	Description
`text`	The pack.
`system_block`	The pinned pack as a string, empty when there is none. Place it after your own instructions.
`turn_block`	The delta and the live turns from other channels, tagged for the model. Place it at the end of the prompt.
`withheld`	Items the policy held back at this verification level.
`verification`	`requested`, `effective` and the `reason` when effective is lower.
`etag`, `version`, `as_of`	Identity and freshness of the pack.
`live`, `delta`	Recent turns from other channels, and what changed.
`path`, `is_holdout`	Which read tier answered; `holdout` means the profile is in a control group and the pack is empty on purpose.
`origin`	`network`, `cache`, `stale`, `last_good` or `empty`.
`error`, `elapsed_ms`	Why the pack is empty, when it is, and how long the call took.

from niadra import Niadra, phone

niadra = Niadra()
ctx = niadra.context(phone("+14155550123"), view="voice", verification="V1", conversation_id="call-4471")

system_prompt = f"{AGENT_INSTRUCTIONS}\n\n{ctx.system_block}"
if ctx.turn_block:
    messages.append({"role": "system", "content": ctx.turn_block})

search()

Searches the whole history of one subject by keyword and meaning. Maps to POST /v1/history/search.

niadra.search(
    subject: HandleLike,
    query: str,
    *,
    about: HandleLike | None = None,
    filters: HistoryFilters | Mapping | None = None,
    max_tokens: int = 800,
    verification: str = "V0",
    conversation_id: str | None = None,
    task_id: str | None = None,
    voice: bool = False,
    timeout: float | None = None,
) -> SearchResult

Parameter	Type	Default	Description
`subject`	handle	required	Whose history.
`query`	`str`	required	Natural language or keywords, up to 2,000 characters.
`about`	handle	`None`	The organization the person acts for.
`filters`	`HistoryFilters` or mapping	`None`	`since`, `until`, `channels`, `categories`, `item_kinds`, `outcome`, `object`.
`max_tokens`	`int`	`800`	Budget of the answer, 50 to 4,000. Use 300 for voice.
`verification`	`str`	`"V0"`	The same level the context uses.
`conversation_id`, `task_id`	`str`	`None`	Who is asking, for the receipt and the verification of the session.
`voice`	`bool`	`False`	Use the voice time budget.

Returns a SearchResult with items, recurrence (how many times the same category happened and how the last one ended), withheld, as_of, tokens_used, degraded and, on failure, error.

found = niadra.search(phone("+14155550123"), "credit for missed technician visit", max_tokens=300, voice=True)
if found.recurrence:
    print(found.recurrence.occurrences, found.recurrence.last_resolution)

timeline()

One page of the subject’s history, most recent first. Maps to POST /v1/history/timeline.

niadra.timeline(
    subject: HandleLike,
    *,
    about: HandleLike | None = None,
    filters: HistoryFilters | Mapping | None = None,
    cursor: str | None = None,
    limit: int = 20,
    verification: str = "V0",
    conversation_id: str | None = None,
    voice: bool = False,
    timeout: float | None = None,
) -> TimelinePage

limit goes from 1 to 100. Returns a TimelinePage with items, next_cursor, withheld, as_of and, on failure, error. Pass next_cursor as cursor to go on.

page = niadra.timeline(phone("+14155550123"), filters={"since": "2026-01-01T00:00:00Z"})
for item in page.items:
    print(item.at, item.kind, item.text)

open()

Opens one episode or object found by search() or timeline(). Maps to GET /v1/history/items/{item_id}.

niadra.open(
    item_id: str,
    *,
    verification: str = "V0",
    conversation_id: str | None = None,
    task_id: str | None = None,
    voice: bool = False,
    timeout: float | None = None,
) -> OpenedItem | None

Returns an OpenedItem (summary, requested, promises, outcome, resolution, derived, timeline) or None when it is unavailable. The literal excerpt only comes back to keys with an elevated scope.

item = niadra.open("ep_01J2", conversation_id="call-4471")
if item:
    print(item.summary, item.resolution)

object_state() and object_timeline()

Reads of a business object. They map to GET /v1/objects/{object_type}/{namespace}/{external_id} and its timeline.

niadra.object_state(object: ObjectLike, *, voice: bool = False, timeout: float | None = None) -> ObjectState | None

niadra.object_timeline(
    object: ObjectLike,
    *,
    cursor: str | None = None,
    limit: int = 20,
    voice: bool = False,
    timeout: float | None = None,
) -> ObjectTimeline | None

ObjectState carries ref, state, as_of, source_id, record_ref and the open_items. The state comes only from what the systems of record reported; an agent’s action counts once a system confirms it. ObjectTimeline carries ref, items, next_cursor and as_of: system events and agent actions, newest first, never what anyone said. Object ids are record ids, not personal data, so they go in the URL; an id that contains a slash cannot be addressed this way.

invoice = niadra.object_state("invoice:erp:0823")
page = niadra.object_timeline("invoice:erp:0823", limit=20)

track()

Queues a message, a system event or an action, or any other batch item, and returns at once. Maps to POST /v1/batch.

niadra.track(item: EventItem | Mapping) -> bool

Accepts an EventItem or a mapping of its fields. A mapping without channel gets the client’s default channel; idempotency_key (a UUIDv7) and occurred_at (now) are filled in when you leave them out. Use the provider’s message id as idempotency_key when there is one, so a retried batch never duplicates anything. Returns False when the item was dropped: invalid, not serializable, queue full or client disabled.

niadra.track({
    "channel": "whatsapp",
    "conversation_id": "wa-8812",
    "idempotency_key": "wamid.HBgLMTQxNTU1NTAxMjMVAgASGBQzQUQ",
    "handles": [phone("+14155550123")],
    "speaker": {"role": "customer"},
    "content": {"text": "The technician never showed up. I am calling you."},
})

action()

Records what an agent did in a system of record. Queued, like track().

niadra.action(
    operation: str,
    *,
    subject: HandleLike | None = None,
    object: ObjectLike | None = None,
    result: str | None = None,
    purpose: str | None = None,
    closes: str | Closes | Mapping | None = None,
    conversation_id: str | None = None,
    task_id: str | None = None,
    channel: str | None = None,
    speaker: str = "ai_agent",
    speaker_id: str | None = None,
    corrects_action_id: str | None = None,
    occurred_at: datetime | None = None,
    idempotency_key: str | None = None,
    context_stamp: ContextStamp | Mapping | None = None,
) -> bool

Parameter	Description
`operation`	The canonical operation, such as `credit` or `reschedule`.
`subject`, `object`	Who and what the action is about.
`result`	What happened, up to 2,000 characters.
`closes`	The open item the action fulfils: an open item id as a string, or `{"object": ..., "operation": ...}`.
`corrects_action_id`	An action is immutable; a correction is a new action that points to the old one.
`context_stamp`	Which context the agent acted on. Conversations and tasks set it for you after `mark_injected()`.

The action stays declared until the system of record confirms it with its own event. Recording actions needs the act scope on the key.

niadra.action(
    "credit",
    subject=system_id("crm", "48213"),
    object="invoice:erp:0823",
    result="$40 credit on the August bill",
    closes={"object": {"type": "invoice", "namespace": "erp", "id": "0823"}, "operation": "dispute"},
    channel="erp",
    task_id="billing-7741",
)

identify()

States that two or more handles belong to the same subject. Sent at once rather than queued, so a context() that follows sees it; if the request fails, the item is queued for the background sender and None is returned.

niadra.identify(
    handles: Sequence[HandleLike],
    *,
    method: str = "explicit_identify",
    subject_kind: str = "person",
    conversation_id: str | None = None,
) -> BatchResponse | None

niadra.identify([phone("+14155550123"), email("marina@example.com")], conversation_id="wa-8812")

verify()

Raises the verification level of one conversation or task, after you proved it. Niadra never infers it. Sent at once, and the cached pack of that conversation is dropped, so the next context() reflects the new level.

niadra.verify(
    method: str,
    level: str,
    *,
    handle: HandleLike,
    conversation_id: str | None = None,
    task_id: str | None = None,
    valid_until: datetime | None = None,
) -> BatchResponse | None

method is otp_whatsapp, otp_sms, login, kba, network_attestation or human_agent. A level above the ceiling of the source comes back as a verification_not_allowed item error.

niadra.verify("otp_whatsapp", "V3", handle=phone("+14155550123"), conversation_id="wa-8812")

feedback()

Corrects what Niadra derived about a subject. Sent at once and recorded as an event, so it is audited like any other. Maps to POST /v1/feedback.

niadra.feedback(
    action: str,
    subject: HandleLike,
    *,
    fact_id: str | None = None,
    open_item_id: str | None = None,
    conversation_id: str | None = None,
    value: str | None = None,
    reason: str | None = None,
    idempotency_key: str | None = None,
) -> BatchResponse | None

Each action takes its own fields:

`action`	Requires
`retract_fact`	`fact_id`. The fact leaves every pack.
`correct_fact`	`fact_id` and `value`, the right value.
`resolve_open_item`	`open_item_id`.
`conversation_outcome`	`conversation_id` and `value`.

value goes up to 2,000 characters and reason up to 500. Returns None when the correction could not be delivered.

niadra.feedback("resolve_open_item", phone("+14155550123"), open_item_id="oi_01J8ZK", reason="Visit rescheduled")
niadra.feedback("conversation_outcome", phone("+14155550123"), conversation_id="call-4471", value="resolved")

upload_media()

Hands a file to Niadra, such as a call recording, and returns the reference for its event. Media never travels inside an event. Maps to POST /v1/media/uploads.

niadra.upload_media(
    data: bytes | bytearray | memoryview,
    content_type: str,
    *,
    subject: HandleLike | None = None,
) -> MediaUpload | None

The method hashes the bytes, reserves an upload, and sends the bytes straight to storage over the short-lived signed URL. It sends exactly the upload_headers the API returned, which are the headers the signature covers, and nothing else: never your key. The URL must be HTTPS, except against the local emulator. Storage checks the body against the declared size and digest. With subject, the file is stored under that person, so erasing them erases it even if no event ever references it. Returns a MediaUpload with media_ref, media_sha256, content_type, size_bytes and expires_at, or None when the upload failed. Each attempt of the transfer has the upload budget, 60 s by default.

upload = niadra.upload_media(recording, "audio/wav", subject=phone("+14155550123"))
if upload:
    niadra.track({
        "channel": "voice",
        "conversation_id": "call-4471",
        "handles": [phone("+14155550123")],
        "speaker": {"role": "customer"},
        "content": {
            "type": "audio",
            "media_ref": upload.media_ref,
            "media_sha256": upload.media_sha256,
            "transcript": transcript,
        },
    })

handoff()

Records a transfer to a human ("human") or another agent ("agent"). Queued. The context-use measurement reads it to count handoffs where the receiver never read the context.

niadra.handoff(
    conversation_id: str,
    target: str,
    *,
    target_source: str | None = None,
    reason: str | None = None,
    mode: str = "warm",
) -> bool

conversation()

A conversation with one customer, used as a context manager. Leaving the block emits conversation.ended, even when the block raised.

niadra.conversation(
    conversation_id: str | None = None,
    *,
    subject: HandleLike | None = None,
    object: ObjectLike | None = None,
    about: HandleLike | None = None,
    channel: str | None = None,
    view: str = "chat",
    verification: str = "V0",
    target: str | TargetModel | None = None,
    agent_id: str | None = None,
) -> Conversation

Without a conversation_id the SDK mints one. channel defaults to the client’s. agent_id identifies your agent within the source and is stamped on its turns and actions.

Method	Description
`context(**overrides)`	The pack for this turn: the pinned bytes, with every delta since the pin in `delta`.
`mark_injected(context=None, *, at=None)`	Records that the pack went into the prompt. The agent’s next turns and actions carry it as `context_stamp`.
`customer(text, **event)`	Records what the customer said.
`agent(text, **event)`	Records the agent’s answer, with the context stamp.
`human_agent(text, **event)`	Records a turn by a human attendant.
`action(operation, **options)`	`action()` with subject, object, channel, ids and stamp filled in.
`verify(method, level, *, handle=None, valid_until=None)`	Raises the level and reads at the new level from then on; `handle` defaults to the subject.
`handoff(target, **options)`	Records a transfer of this conversation.
`tools()`	The history kit bound to this conversation, at its current level; `None` without a subject.
`end()`	Emits `conversation.ended`. Later calls do nothing.

The first context() gets the pack the server pins for the conversation. Later reads also ask for the delta, and the conversation keeps every delta it receives, in order, so turn_block carries all the changes since the pin, followed by the live turns. When the server pins a new pack, after verify() for instance, the kept deltas are dropped: the new pack already has them. A read with query= is a one-off and leaves them alone. Call mark_injected() each time you put the pack in a prompt. It is how Niadra tells a context that arrived after the agent spoke from one the agent had and did not use. The session also keeps context_injected_at and first_agent_turn_at, the first of each, for your own checks.

with niadra.conversation("wa-8812", subject=phone("+14155550123"), channel="whatsapp") as conv:
    conv.customer("The technician never showed up. I am calling you.")
    ctx = conv.context()
    conv.mark_injected(ctx)
    reply = llm(system=[AGENT_INSTRUCTIONS, ctx.system_block], turn=ctx.turn_block)
    conv.agent(reply)
    conv.verify("otp_whatsapp", "V3")
    conv.handoff("human", reason="asked for a person")

current_session() returns the conversation or task whose block is running in the current thread or task, if any.

task()

The same for an internal agent (billing, orders, tickets). It emits task.ended on exit.

niadra.task(
    task_id: str | None = None,
    *,
    subject: HandleLike | None = None,
    object: ObjectLike | None = None,
    about: HandleLike | None = None,
    channel: str | None = None,
    view: str = "brief",
    verification: str = "V0",
    target: str | TargetModel | None = None,
    agent_id: str | None = None,
) -> Task

With an object, the pack is centered on it; use a task view such as task:billing. A task has the same methods as a conversation except handoff().

with niadra.task("billing-7741", object="invoice:erp:0823", view="task:billing",
                 verification="no_customer", channel="erp") as task:
    ctx = task.context()
    task.mark_injected(ctx)
    task.action(
        "credit",
        result="$40 credit on the August bill",
        closes={"object": {"type": "invoice", "namespace": "erp", "id": "0823"}, "operation": "dispute"},
    )

tools()

The history kit as function-calling tools, bound to one customer. The customer is bound here, outside the model’s reach: the model picks the query, never the profile, which is what stops a prompt injection from switching customers.

niadra.tools(
    subject: HandleLike,
    *,
    about: HandleLike | None = None,
    conversation_id: str | None = None,
    task_id: str | None = None,
    verification: str = "V0",
    voice: bool = False,
) -> ToolKit

Member	Description
`definitions`	The three tools, `search_customer_history`, `get_customer_timeline` and `open_history_item`, as `{"type": "function", "function": {...}}`.
`names`	Their names.
`anthropic_definitions()`	The same tools as `name`, `description` and `input_schema`.
`call(name, arguments)`	Runs one tool call and returns the text for the model: compact JSON, or a short error that tells the model to carry on.

AsyncNiadra.tools() returns an AsyncToolKit, whose call() is awaited. The definitions are also served by GET /v1/history/tools.

kit = niadra.tools(phone("+14155550123"), conversation_id="call-4471")
reply = llm.chat.completions.create(model=MODEL, messages=messages, tools=kit.definitions)
for call in reply.choices[0].message.tool_calls or []:
    output = kit.call(call.function.name, call.function.arguments)
    messages.append({"role": "tool", "tool_call_id": call.id, "content": output})

subject_token()

Mints a signed 15-minute token that binds an MCP connection to one customer. Call it from your backend. Maps to POST /v1/subject-tokens.

niadra.subject_token(
    subject: HandleLike,
    *,
    about: HandleLike | None = None,
    conversation_id: str | None = None,
    task_id: str | None = None,
    verification: str = "V0",
) -> SubjectToken | None

Returns a SubjectToken with token, expires_at and headers, the Niadra-Subject-Token header to send with the source key when the agent opens the connection to mcp_url. With about, the organization is bound like the subject. See MCP with any LLM.

token = niadra.subject_token(phone("+14155550123"), conversation_id="call-4471", verification="V1")
# connect to niadra.mcp_url with the source key as Bearer and token.headers

wrap()

Wraps the OpenAI Python client, or any client with the same shape, so every call inside a conversation or task block gets the context and records the answer.

wrap(client, *, conversation: Conversation | Task | None = None) -> client

Inside a block (or for the session you pass), chat.completions.create and chat.completions.parse, sync or async, streaming or not, get the pinned pack as a system message right after your own leading system messages, and the turn block as a system message at the end. The injection is stamped with mark_injected(), and the model’s answer is recorded as the agent’s turn: when a stream ends or is closed and, through with_raw_response, when you call parse(). Outside a block, calls pass through untouched. The wrapper returns a proxy and never modifies your client. Nothing it does can fail your model call: a context it cannot fetch is left out, and a failure to record the answer is logged, without content.

from openai import OpenAI
from niadra import Niadra, phone, wrap

niadra = Niadra(channel="whatsapp")
openai = wrap(OpenAI())

with niadra.conversation("wa-8812", subject=phone("+14155550123")) as conv:
    conv.customer(incoming_text)
    reply = openai.chat.completions.create(model="gpt-4.1", messages=messages)

flush() and close()

niadra.flush(timeout: float | None = None) -> bool
niadra.close(timeout: float | None = 5.0) -> None

flush() sends everything queued from the calling thread and returns True when nothing is left. close() flushes for up to timeout seconds and releases connections. With AsyncNiadra, both are awaited.

AsyncNiadra

AsyncNiadra has the same methods. The reads, identify(), verify(), feedback(), upload_media(), subject_token(), flush() and close() are awaited; track(), action(), handoff(), conversation(), task() and tools() are not. Conversations and tasks are async context managers.

from niadra import AsyncNiadra, phone

niadra = AsyncNiadra(channel="whatsapp")

async with niadra.conversation("wa-8812", subject=phone("+14155550123")) as conv:
    ctx = await conv.context()
    conv.mark_injected(ctx)

Errors

With strict=True, or from helpers you call directly such as ApiKey.parse() and the handle functions, the SDK raises:

Exception	When
`NiadraError`	Base class of all of them.
`ConfigurationError`	Missing or malformed key, bad base URL.
`APIConnectionError`	No HTTP answer: DNS, TCP, TLS or a dropped connection.
`APITimeoutError`	The method’s own budget ran out.
`APIError`	Any error status. `status_code`, `code`, `request_id` and `problem` (the RFC 9457 body).
`BadRequestError`	400.
`AuthenticationError`	401: the key is unknown, revoked or malformed.
`PermissionDeniedError`	403: the key lacks the scope, or the source was cut.
`NotFoundError`	404.
`ConflictError`	409: the same `Idempotency-Key` with a different body.
`WrongCellError`	421: the space is moving between cells.
`UnprocessableEntityError`	422.
`RateLimitError`	429, with `retry_after`.
`ServerError`	5xx.

The full code catalog is in Errors. Quote request_id when you contact support.

Local emulator

niadra-mock, installed with the package, is a local in-memory emulator of the same API. In tests, run the SDK against it in-process:

import httpx
from niadra import Niadra
from niadra_mock import MOCK_KEY, MockApp

mock = MockApp()
http = httpx.Client(transport=httpx.WSGITransport(app=mock.wsgi))  # ASGITransport(app=mock.asgi) for AsyncNiadra
niadra = Niadra(MOCK_KEY, base_url="http://mock", http_client=http, strict=True)

Recent turns become a small pack, deltas are sent once per change, search matches keywords, verification only rises through verify(), objects take their state from system events, feedback becomes a feedback.* event and media uploads land in mock.cell.media. mock.cell also lets you inspect events and inject failures (fail_next, revoke, cut, put_in_holdout). From a shell, niadra-mock --port 8765 serves it over HTTP.

Next steps

TypeScript SDK

The same surface for Node and edge runtimes.

Quickstart

From a key to the first delivered context.

Context and views

What goes into the pack and why.

Errors

The code catalog and what to do with each code.

​Install

​The client

​Properties

​Safe by default, strict on request

​Time budgets

​The context cache

​The write queue

​Handles

​context()

​search()

​timeline()

​open()

​object_state() and object_timeline()

​track()

​action()

​identify()

​verify()

​feedback()

​upload_media()

​handoff()

​conversation()

​task()

​tools()

​subject_token()

​wrap()

​flush() and close()

​AsyncNiadra

​Errors

​Local emulator

​Next steps

TypeScript SDK

Quickstart

Context and views

Errors

Install

The client

Properties

Safe by default, strict on request

Time budgets

The context cache

The write queue

Handles

context()

search()

timeline()

open()

object_state() and object_timeline()

track()

action()

identify()

verify()

feedback()

upload_media()

handoff()

conversation()

task()

tools()

subject_token()

wrap()

flush() and close()

AsyncNiadra

Errors

Local emulator

Next steps