> ## Documentation Index > Fetch the complete documentation index at: https://docs.niadra.com/llms.txt > Use this file to discover all available pages before exploring further. # Python SDK > Every method of the niadra package, with parameters, return value and an example. 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](/en/api). ## Install ```sh theme={null} pip install niadra ``` ## The client ```python theme={null} from niadra import Niadra niadra = Niadra() # reads NIADRA_API_KEY ``` ```python theme={null} 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_____`. | | `base_url` | `str` | `NIADRA_BASE_URL`, then derived from the key | The key names its region and space, so the address is `https://..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, `/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](#errors). ### Time budgets The SDK keeps its own time budget per method, whatever the platform around it allows. ```python theme={null} 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. ```python theme={null} 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`. ```python theme={null} 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`](/en/api/context). ```python theme={null} 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. | ```python theme={null} 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`](/en/api/history-search). ```python theme={null} 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`. ```python theme={null} 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`](/en/api/history-timeline). ```python theme={null} 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. ```python theme={null} 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}`](/en/api/history-item). ```python theme={null} 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. ```python theme={null} 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}`](/en/api/object) and its [timeline](/en/api/object-timeline). ```python theme={null} 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. ```python theme={null} 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`](/en/api/batch). ```python theme={null} 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. ```python theme={null} 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()`. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} niadra.identify( handles: Sequence[HandleLike], *, method: str = "explicit_identify", subject_kind: str = "person", conversation_id: str | None = None, ) -> BatchResponse | None ``` ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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`](/en/api/feedback). ```python theme={null} 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. ```python theme={null} 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`](/en/api/media-uploads). ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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()`. ```python theme={null} 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. ```python theme={null} 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`](/en/api/history-tools). ```python theme={null} 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`](/en/api/subject-tokens). ```python theme={null} 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](/en/guides/mcp). ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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() ```python theme={null} 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. ```python theme={null} 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](/en/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: ```python theme={null} 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 The same surface for Node and edge runtimes. From a key to the first delivered context. What goes into the pack and why. The code catalog and what to do with each code.