Niadra Python SDK

Niadra is the omnichannel memory layer for a company's AI agents. It takes the events from every channel, platform and system, builds one memory of each customer and delivers it to any AI agent, from any vendor, under governance. The WhatsApp agent, the voice agent, the billing agent inside the ERP and the human team read the same memory before they act and write back what they said and did. This package connects a Python agent to it.

Website · Documentation · Talk to us · TypeScript SDK

pip install niadra

The problem it solves

A customer tells your WhatsApp agent that order 4471 arrived with a broken lid and that she needs a replacement by Friday. An hour later she calls, and the voice agent runs on another vendor's platform. Without a memory that spans both, the voice agent asks her to explain everything again, and nobody remembers the Friday promise. With Niadra, the voice agent starts the call knowing about the open replacement and its deadline, and when the billing agent credits her invoice in the ERP, the other agents see it within seconds.

Niadra does the remembering for you:

• it takes in every event, whatever its source: what the customer said on any channel, what the CRM, the ERP or the help desk recorded, what an agent did, through the SDK, a webhook, a batch or a file, and through the ready-made adapters below;
• it ties every event to the right person across phone numbers, e-mails, WhatsApp ids and CRM ids, and to the companies and partners that person acts for;
• it turns conversations and system events into facts, open items and promises, each with the turns that prove it;
• it compiles a short context for each agent, holding back what the customer's verification level does not allow, and records a receipt of every read.

Your agents keep their own models, prompts and vendors. Niadra is the memory layer they share, and it stays neutral: no vendor reads another vendor's memory, and no framework owns it.

Quickstart

from niadra import Niadra, phone

niadra = Niadra(channel="whatsapp")  # reads NIADRA_API_KEY
customer = phone("+5511912345678")

with niadra.conversation("thread-81", subject=customer) as conversation:
    conversation.customer(incoming_text)  # what the customer wrote
    context = conversation.context()  # what this agent needs to know now
    prompt = instructions + "\n\n" + context.system_block + "\n\n" + context.turn_block
    conversation.mark_injected(context)
    reply = your_model(prompt)
    conversation.agent(reply)  # what the agent answered

Three moments cover most agents: read the context before the model call, record the turns after it, and record an action() when the agent does something in a system (a refund, a new delivery date). wrap() does the reading and recording for you around an OpenAI-compatible client.

Requires Python 3.10+. Depends on httpx and pydantic only; each integration's framework comes with its extra, such as pip install 'niadra[livekit]'.

What your agent gets

Benchmark

Niadra and ten other memory systems for agents (thirteen configurations) were measured on the same dataset, in the same AWS region (us-east-2), with the same agent and the same judge for every system. The script, the dataset and every result file are in this repository, under benchmarks/, so anyone can run it again. The full page, with every metric and where Niadra does not lead, is at niadra.com/benchmark.

Combined run of 02/10/2026 (2026-09-30-5ea7a2), cross-channel accuracy on the same 165 valid cases, judged by the same model:

With a different user id on each channel (agents from different vendors), Mem0 open source drops to 28.5%, AgentCore Memory to 27.9%, and Niadra stays at 98.7%. Sensitive values reaching an unverified conversation: Niadra 0 of 32; Mem0 and AgentCore Memory 32 of 32.

Typed state (typed/2026-09-30-b59a5d): with the state and constraints blocks, 84.9% correct [78, 90] against 36.6% without them; declared constraints respected went from 0% to 94.4%, and what changed since the customer last saw it from 0% to 100%, with no extra tokens on a turn that asks for no block.

Where Niadra does not lead in that run: ingest acknowledgement against systems that write in the background, opening a single history item, history search p95 against LangMem by a few ms, and tokens per turn against readers that return answer text. The benchmark is produced by Niadra; that is why every input and output is public.

Integrations

Every integration wires the same five things into the framework's own extension points: the context before the model call (right after your instructions, the news of other channels at the end), the turns (the customer's and the agent's, with the provider's usage), the history tools bound to the customer (never a parameter the model sees), what the platform proves about who is there (verify() before the first context), and handoffs. With agent_memory=True the agent's own notes go before the customer's context and search_agent_memory (and remember) join the tools. None of them can fail your agent: Niadra slow or down means no context, never an error.

Each module's docstring is its guide, and examples/ has one script per integration. The versions tested are pinned in uv.lock; CI runs each integration against the framework's real types, on the lowest and highest Python it supports.

Voice

from niadra import AsyncNiadra
from niadra.integrations.livekit import NiadraAgent, conversation_for

niadra = AsyncNiadra(channel="voice")


async def entrypoint(ctx):
    caller = await ctx.wait_for_participant()
    conversation = conversation_for(niadra, caller, room=ctx.room)  # SIP number and call id
    await session.start(
        NiadraAgent(conversation, instructions=INSTRUCTIONS, attestation=level), room=ctx.room
    )

Pipecat takes NiadraMemoryProcessor(conversation) between aggregators.user() and the LLM, plus memory.observe(aggregators) for the turns and, right after the STT service, memory.prefetcher(). Both adapters start the caller's first read when the call starts (begin()), so it runs while the call is set up, and the first model call waits for it there (ready(), within 1.5 s). After that no turn waits on a round trip to the region for the pack: the pinned pack is the same bytes for the whole call, so it comes from memory at once and is revalidated by ETag in the background. While the caller is still speaking they prefetch() the partial transcript (LiveKit's user_input_transcribed, Pipecat's interim transcripts); once the words stop changing for 200 ms the SDK reads the turn with them, and the turn takes that read's slots and delta. A turn waits at most 200 ms for such a read still on its way; past that it gets the pack without slots and the next turn catches up. A prefetch never holds or fails a turn. For hosted platforms the SDK answers their webhooks: ElevenLabsWebhooks (the context as the niadra_context dynamic variable, server tools whose caller comes from ElevenLabs' system variables, and a signed post-call transcript recorded turn by turn) and VapiServer (assistant-request, tool-calls, end-of-call-report and transfers). Their call start webhooks wait for the first read within 1.5 s, while the phone rings. Every handler is a plain function of the body and the headers, for any web framework. A carrier's STIR/SHAKEN attestation proves V2 at level A and V1 at B or C; Twilio's StirVerstat is read as is.

Channels

from niadra.integrations.whatsapp import parse_webhook, sent

for message in parse_webhook(body, headers, APP_SECRET) or []:
    with niadra.conversation(f"wa-{message.wa_id}", subject=message.subject) as chat:
        message.record(chat)  # the wamid as idempotency key, verification_hint V1
        ...
        sent(chat, reply, graph_response)

Agent frameworks

from niadra.integrations.openai_agents import NiadraAgentsMemory

memory = NiadraAgentsMemory(conversation, agent_memory=True)
agent = Agent(name="Support", instructions=INSTRUCTIONS, tools=memory.tools)
await Runner.run(agent, text, hooks=memory.hooks, run_config=memory.run_config())

The other frameworks follow the same pattern in their own terms: middleware for LangGraph's create_agent, a callback handler and a runnable for LangChain, callbacks for Google ADK, a hook provider for Strands, a capability for Pydantic AI, a memory for LlamaIndex, kickoff callbacks for CrewAI, instructions and hooks for Agno, and a context provider for the Microsoft Agent Framework.

Model SDKs

from niadra.integrations.anthropic import wrap

claude = wrap(Anthropic())  # also niadra.integrations.bedrock.wrap and google_genai.wrap
with niadra.conversation(thread_id, subject=customer):
    claude.messages.create(model="claude-sonnet-4-5", max_tokens=512, system=INSTRUCTIONS, messages=history)

Each wrap() places the context where the provider expects instructions (Anthropic's system blocks, Bedrock's system, Gemini's system_instruction), keeps a prompt cache you set up (a cache_control or cachePoint you already use covers the pack too), and records the answer with the provider's usage. LiteLLM gets completion() and acompletion(); Azure OpenAI works with the OpenAI wrap() as it is.

Agent memory

The agent's own working notes: procedures, how the company's tools behave, pitfalls. Never about a customer: a note with personal data is refused.

notes = conversation.agent_memory()  # the block for the prompt: after your instructions, before the context
niadra.remember("pitfall", "Scheduling API", "Dates without a time zone are refused.", tags=["scheduling"])
niadra.search_agent_memory("reschedule a visit")
kit = conversation.tools(agent_memory=True, write_agent_memory=True)  # adds search_agent_memory and remember

The block is cached like the context and revalidated by ETag. With the agent memory off in the space, agent_memory() answers enabled=False and empty text. Writing needs a key with the agent_memory:write scope; a refused note tells the model exactly why, so it can rewrite it.

Questions people ask

How do I give my AI agent memory of past conversations on other channels? Record the turns with track() or conversation() in every agent, and read context() before each model call. Niadra ties the turns to the person, whichever id each channel uses.

How is this different from keeping chat history in my database or in a vector store? Stored history is raw text for one channel. Niadra keeps derived facts and open items with evidence, resolves identity across channels and systems, closes items when a system of record confirms an action, and filters what each agent may read by the verification level of the conversation.

What happens if Niadra is slow or down? The agent keeps answering without the memory. Every call has its own time budget for the whole call, retries included (300 ms for context; in a voice call the pack is served from memory and a turn waits at most 200 ms for its slots), and returns an empty value instead of raising, unless you ask for strict=True.

What about privacy and LGPD or GDPR? Items carry a verification level and a purpose, and the policy decides what each agent sees. Every read leaves a receipt, and a person can be erased or exported on request. The SDK never logs handles or message text.

Which models and frameworks does it work with? Any. The context is text you place in your prompt and the tools follow the common function-calling format. The integrations do the wiring for you (see Integrations): voice (LiveKit, Pipecat, ElevenLabs, Vapi, Retell, Twilio), WhatsApp, agent frameworks (OpenAI Agents, LangChain, LangGraph, Google ADK, Strands, Pydantic AI, LlamaIndex, CrewAI, Agno, Microsoft Agent Framework, Semantic Kernel, Haystack, CAMEL-AI, DSPy, AG2), model SDKs (OpenAI, Azure OpenAI, Anthropic, Bedrock, Google GenAI, LiteLLM), Langflow and Dify.

The client

Niadra is synchronous and thread-safe; AsyncNiadra has the same methods, awaited. One instance per process is enough. The key names its region and space, and the SDK derives the address from it: nia_sk_live_us-east-2_acme-prod_... talks to https://acme-prod.us-east-2.api.niadra.com. Pass base_url= (or set NIADRA_BASE_URL) to point it elsewhere, such as the local emulator.

Context

context() returns the context pack for a person, an organization or a business object: the few lines that matter now, compiled by Niadra from every channel and every agent.

context = niadra.context(
    phone("+5511912345678"),
    view="voice",  # chat, voice, brief, full, account, partner, task:<name>
    verification="V1",  # what you have proven about who is on the line
    conversation_id=call_id,
    target="openai/gpt-4.1",  # lets Niadra aim the pack at the model's prompt-cache floor
)
context.system_block  # the pinned pack: place it after your own instructions
context.turn_block  # live turns from other channels, this turn's slots and the delta: at the end
context.withheld  # items the policy held back at this verification level

With format="json", context.pack also carries the pack as data (context-pack.v1: typed sections with stable names, the preamble, the stamp and this turn's slots), for programs that build their own prompt.

turn= is the customer's last turn; a conversation sends it for you (below). It goes as query, and the answer adds context.slots: what that turn needs from memory that the pinned pack left out (the protocol number the customer asks for, the earlier conversations on the same topic with their dates, or a line saying memory has no record of it), in turn_block after the live turns and before the delta. The pack itself stays the pinned one. prefetch(subject, text=...) sends a partial transcript while the customer is still speaking; it runs in the background, one at a time per conversation (the newest text waits), and never raises.

Pass object="invoice:erp:0823" instead of a subject to center the pack on a business object, and about= to add what the organization the person acts for has that matters here.

With a conversation_id (or task_id) the server pins the pack to the same bytes on every turn. delta=True asks for what changed since this agent last read the subject; the server sends each change once, and conversation() keeps them for you (below).

The context cache

Inside a conversation or task, 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.

context.origin says which of these served the read, and context.age_ms how long ago Niadra sent or confirmed that pack.

Refreshes send the cached ETag, so an unchanged pack costs a not_modified answer instead of the full text, and only one background 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.

Configure it with CacheOptions(ttl, stale_while_revalidate, max_stale, max_entries), in seconds, or turn it off with CacheOptions(enabled=False).

Search, timeline and open

For what does not fit in the pack, the whole history is one call away.

found = niadra.search(customer, "technician visit", conversation_id=call_id)
found.recurrence  # occurrences=3, window_days=90: "this happened before"
page = niadra.timeline(customer, limit=20)  # most recent first; pass page.next_cursor to go on
item = niadra.open(found.items[0].id)  # what was asked, promised and by whom, the outcome

Filters take when, a time phrase in the customer's own words (last week, semana passada, en marzo); the answer's window says the period read and ignored lists what the server could not read. Every row's id is a bare UUID and kind says what it is; open() takes the id of an episode or object row, and feedback and closes take the id of a fact or an open item as listed. What an event states can expire (valid_until on the event): expired items leave the context and the search unless show_expired=True. An opened object lists its versions.

The handle, the search and the conversation id go in request bodies, never in a URL: a conversation id may be a phone number or an e-mail. open() sends POST /v1/history/open, and the tool kit adds the bound customer to it, so the server opens only that customer's items.

Track

track() records a message, a system event or an action. It never blocks: items go to a bounded local queue and a background thread (a task, with AsyncNiadra) sends them in batches of 15 or every second, whichever comes first, with three attempts and backoff. A message with a conversation_id is a turn the other agents read in live, so it leaves at once (QueueOptions.turn_interval, 0 by default), taking whatever else is waiting along. One batch is in flight per client, so turns queued while it is answered leave together in the next one: a burst of turns costs one request per round trip, not one per turn.

niadra.track(
    {
        "conversation_id": thread_id,
        "handles": [phone("+5511912345678")],
        "speaker": {"role": "customer"},
        "content": {"text": "I was charged twice"},
    }
)

Items take their idempotency_key from you (use the provider's message id when there is one) or from a UUIDv7 minted by the SDK, so a retried batch never duplicates anything. Call flush() to send what is queued now; close() and the exit hook flush for you.

Identity and verification

niadra.identify([phone("+5511912345678"), email("marina@example.com")])
niadra.verify("otp_whatsapp", "V2", handle=phone("+5511912345678"), conversation_id=thread_id)

identify() states that handles belong to the same person. verify() raises the verification level of one conversation after you proved it; Niadra never infers it. Both are sent at once rather than queued, so the next context() sees them, and verify() drops the cached pack of that conversation.

Actions

niadra.action(
    "credit",
    subject=customer,
    object="invoice:erp:0823",
    result="R$ 40",
    closes={"object": "invoice:erp:0823", "operation": "credit"},
)

An action is what an agent did in a system of record. closes names the open item or promise it fulfils. It shows in every other agent's pack as declared until the system of record confirms it.

Conversations and tasks

with niadra.conversation(thread_id, subject=customer, view="chat") as conversation:
    conversation.customer(incoming_text)
    context = conversation.context()  # pinned, cached, with every delta since the pin
    conversation.mark_injected(context)  # when the pack went into the prompt
    conversation.agent(reply_text)  # carries the context stamp
    conversation.verify("otp_whatsapp", "V2")
    conversation.handoff("human", reason="asked for a person")
# conversation.ended is emitted on exit, even if the block raised

The first context() of a conversation gets the pack the server pins for it. Later reads also ask for the delta, and the conversation keeps every delta it receives, in order, in context.delta, so turn_block carries all the changes since the pin, after the live turns (and this turn's slots). When the server pins a new pack, after verify() for instance, the kept deltas are dropped: the new pack already has them. query= picks this read's slots by other words than the turn; the pack is the pinned one all the same.

Every read sends the customer's last turn, the text of the last customer(), and the answer carries what it needs from memory as slots in turn_block. Pass turn= when the platform has the turn before customer() recorded it, or turn=None to read without one. In a voice call, conversation.prefetch(partial_transcript) sends the turn so far while the customer speaks, and conversation.begin() at call start (ringing, the inbound webhook) starts the first read so it runs while the call is set up; conversation.ready() waits for it there. Ending the conversation drops the pack and everything else the SDK kept in memory for it.

Call mark_injected() each time you put the pack in a prompt. The agent's turns and actions that follow carry it as context_stamp, with the pack's etag, which is how Niadra tells a context that arrived after the agent spoke from one the agent had and did not use. context_injected_at and first_agent_turn_at keep the first of each, for your own checks.

task() is the same for internal agents (billing, orders, tickets): it takes an object and a task view such as task:billing (the default is brief), centers the pack on the object, and emits task.ended.

Wrapping the OpenAI client

from openai import OpenAI
from niadra import wrap

openai = wrap(OpenAI())
with niadra.conversation(thread_id, subject=customer):
    openai.chat.completions.create(model="gpt-4.1", messages=messages)

Inside a conversation block, chat.completions.create and chat.completions.parse, sync or async, streaming or not, get the pack after your leading system messages and the turn block at the end. The injection is stamped, 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. Nothing the wrapper does can fail your call: a context it cannot fetch is left out, and a failure to record the answer is logged, without content.

The provider's prompt cache

The agent's turn also carries the usage the provider reported for the call: every input token (usage.prompt_tokens), the ones read from its prompt cache (usage.prompt_tokens_details.cached_tokens) and, through gateways that pass Anthropic's fields along, the ones written to it. A stream reports usage only when you ask for it with stream_options={"include_usage": True}; the wrapper never changes your request to get it. Niadra sums the usage per agent, vendor and model, and the Console shows the cache's hit rate and the estimated savings next to the rest of the space's usage.

Without wrap(), pass the provider's response with the turn. An OpenAI response and an Anthropic one are both understood (Anthropic's usage.input_tokens, cache_read_input_tokens and cache_creation_input_tokens):

message = anthropic.messages.create(model="claude-sonnet-4-5", system=[...], messages=messages)
conversation.agent(message.content[0].text, usage=message)

# or build it yourself
from niadra import ModelUsage

conversation.agent(
    reply, usage=ModelUsage(provider="openai", model="gpt-4.1", prompt_tokens=3000, cached_tokens=2048)
)

A response without usage is left out, and the turn is recorded either way.

Business objects

invoice = niadra.object_state("invoice:erp:0823")  # each field with its logical value and freshness
page = niadra.object_timeline("invoice:erp:0823", limit=20)

The state comes only from what the systems of record reported; an agent's action counts once a system confirms it. The timeline lists 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 with a slash cannot be addressed that way.

Feedback

niadra.feedback("correct_fact", customer, fact_id=fact.id, value="prefers e-mail")
niadra.feedback("resolve_open_item", customer, open_item_id=item.id)
niadra.feedback("conversation_outcome", customer, conversation_id=thread_id, value="resolved")

A correction of what Niadra derived. It is sent at once and recorded as an event, so it is audited like any other; retract_fact takes the fact out of every pack.

Media

upload = niadra.upload_media(recording, "audio/wav", subject=customer)
if upload:
    niadra.track(
        {
            "conversation_id": call_id,
            "handles": [customer],
            "speaker": {"role": "customer"},
            "content": {
                "type": "audio",
                "media_ref": upload.media_ref,
                "media_sha256": upload.media_sha256,
                "transcript": transcript,
            },
        }
    )

Media never travels inside an event. upload_media() reserves an upload, sends the bytes straight to storage over a signed URL (HTTPS only, with exactly the headers the signature covers and nothing else, so never your key), and returns the reference and digest for the event. 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.

Tools and MCP

tools() gives any model the history kit as function-calling tools, bound to one customer. The customer is not a parameter the model can see or change, which is what stops a prompt injection from switching customers.

kit = niadra.tools(customer, conversation_id=thread_id)
response = openai.chat.completions.create(model="gpt-4.1", messages=messages, tools=kit.definitions)
for call in response.choices[0].message.tool_calls or []:
    result = kit.call(call.function.name, call.function.arguments)  # always a string for the model

Inside a conversation or task, conversation.tools() gives the same kit bound to it, at the session's current verification level, including a level raised by verify() after the kit was made. kit.anthropic_definitions() returns the same tools in the Anthropic shape. For agents that speak MCP, mint a subject token from your backend and open the connection with it:

token = niadra.subject_token(customer, conversation_id=thread_id, verification="V1")
# connect to niadra.mcp_url with the source key as Bearer and token.headers

The agent core

What an agent does in a turn, recorded and checked in its own process, with Niadra never on the agent's path. Every feature below is off until the space turns it on; a space that did not ask sees no change.

from niadra import Niadra, phone

niadra = Niadra(channel="whatsapp")


def shown(p: dict) -> list[dict]:  # the objects a result showed, for the claim contract
    return [{"ref": f"product:store:{p['sku']}", "fields": {"price_sale": p["price_sale"]}}]


@Niadra.tool("check_price", provenance=shown)
def check_price(sku: str) -> dict:
    return catalog[sku]


with niadra.conversation("thread-82", subject=phone("+5511912345678"), agent_id="store") as conversation:
    conversation.customer("Quanto está o vestido PX?")
    with conversation.turn(build=Niadra.build(prompts={"store": "v16"}, model="gpt-4.1-mini")):
        context = conversation.context(include=["state", "constraints"])
        check_price("PX-4471")
        guarded = conversation.claims.guard_text(draft)  # the claim contract acts before the text goes
        conversation.agent(guarded.text)

• Turn records leave from a bounded queue in the background, in the content mode the space names: the values themselves, pointers to your own bucket (turns.store()), or only their digests. Closing a turn never waits for the network.
• Include blocks come in the same read as the pack. The state view's lines and the constraints go in the turn block after the slots, inside one <niadra> section; a read without include keeps its turn block byte for byte. With Niadra down, the last good blocks serve, marked degraded.
• Coordination decides by each purpose's direction when Niadra does not answer within 200 ms: a customer's message and service go, marketing, retention, collection and an effect with a key wait, and the local copy of the opt-out list always holds. Every check about an outbound contact keeps that copy, read again in the background once a minute.
• Tool bindings live in the space's tool-bindings document and come in the SDK profile, never in code: a tool measures the constraints block through the one served for its name, the counterfactual runs through it, and its capabilities.mask_output decides the masking when the code leaves mask_output unset.
• The niadra command runs replay (exit 1 on a regression), counterfactual, resolver-worker (the space's refresh requests, read with your resolvers inside your boundary; a watch fires only on a value the worker confirmed, and a resolver returns niadra.resolvers.NOT_FOUND when the source no longer has the object), types derive and contract test.
• niadra.api has one typed method per route of these features; unlike the rest of the SDK, it raises.

Failure behavior

Niadra must never take your agent down.

• No key: the client is a no-op and warns once.
• 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.
• Own time budgets, for the whole call: context() gives up after 200 ms for voice views and 300 ms otherwise (in a voice conversation the pack comes from memory and the 200 ms only bound the wait for a turn's slots; the first read of a call has 1.5 s, at call start); search, timeline, open and the object reads after 300 ms (voice) or 600 ms; the writes you wait for (identify, verify, feedback, subject_token) after 5 s; a media upload after 60 s. Retries and backoff happen inside the budget, and an answer that arrives late is dropped rather than waited for. An identify or verify that runs out of time stays queued for the background sender. track() never waits. Adjust them with Timeouts.
• Retries: 5xx, 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 are final.
• With Niadra down, a read serves the conversation's last good pack, degraded, with age_ms saying how old it is; the opt-out holds by the local copy of the suppression list; turn records, events and declarations wait in their queues and leave once Niadra answers again, each stored once. The pause between attempts doubles while Niadra stays down, up to 30 s. tests/test_chaos.py kills Niadra's process, silences its network, answers 503 and answers late in the middle of a conversation, and checks all of it.
• Logs carry method names, status codes, error codes and request ids, never handles or text.

Pass strict=True to raise instead, which is what you want in tests.

Testing without the cloud

niadra-mock 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 or inject failures (fail_next, revoke, cut, put_in_holdout). From a shell, niadra-mock --port 8765 serves it over HTTP.

A read's query never changes the pinned pack: it picks slots (the customer's lines outside the pack that share a word with it, and a no_record line for a number nothing holds). POST /v1/context/prefetch answers 202 and keeps each request in mock.cell.prefetches.

Documentation in Portuguese

The documentation is also available in Portuguese at docs.niadra.com, and the contact page in Portuguese at niadra.com/enterprise.

License

Apache 2.0. See LICENSE.