dyadic

The official Python SDK for the Dyadic API: use Iris from Python with streaming, function calling, images and typed errors. Requests use your Dyadic credits.

pip install dyadic

from dyadic import Dyadic

client = Dyadic()  # reads DYADIC_API_KEY
print(client.ask("Hello!", model="iris"))

Keep your key on the server

A Dyadic API key is a long-lived secret that spends your credits. Use it only in server-side code; never put it in a mobile app, a desktop app you ship, or anything that runs in someone else's browser. Front-end code should call your own server, which calls Dyadic.

request() only ever sends the key to the API you configured: a relative path joins base_url, and an absolute URL on any other host, port or scheme (or with credentials in it) raises DyadicError instead of being sent. Redirects are not followed. Plain http:// is accepted only for a local or private-network API (localhost, a private IP, or a single-label name such as dyadic-api).

Retries

GET, HEAD, DELETE and OPTIONS are retried on busy or failing servers. Anything else (a completion, an upload, a run) may already have run, and been billed, when it fails, so it is retried only if it provably never started: the connection was never made, or Dyadic refused it with rate_limit_exceeded or concurrency_limit. Every attempt of one call carries the same Idempotency-Key; pass idempotency_key= to choose it, or idempotent=True to request() for a call that is safe to send twice.

Runs: your own agent, Dyadic's tools, your functions

An application brings its own agent (name and instructions), lets it use Dyadic's built-in tools, and gives it functions it runs itself. runs.run calls your functions whenever the agent asks and returns the finished run.

def lookup_order(order_id):
    """Look up an order by id."""
    return {"order_id": order_id, "status": "shipped"}

lookup_order.schema = {"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]}

run = client.runs.run(
    agent={"id": "helpdesk", "name": "Helpdesk", "instructions": "You answer customers of a small shop."},
    messages=[{"role": "user", "content": "Where is order 1042? Also, what's the weather in Lisbon today?"}],
    builtin_tools=["data.web_search", "data.web_fetch"],
    functions={"lookup_order": lookup_order},
    end_user="customer-17",           # your own id for who asked, kept with the run's usage
    conversation_id="ticket-88",
)
print(run.output_text, run.credits)

runs.create(..., stream=True) streams events (run.started, status, tool.started, tool.completed, plan.updated, sources, image.generated, delta, requires_action, run.completed).

Plans: propose, approve, carry out

With the key's Plans switch on, a run can plan before it acts. plan_mode=True investigates read-only and proposes a plan of Steps; nothing that changes anything runs. Your user reviews it (edit it with plans.update), you approve the revision they saw, and a run with plan_id carries it out step by step. The service checks the approval: a plan that isn't approved at its current revision doesn't run (plan_not_runnable). A run that ends with steps left pauses the plan; another run with the same plan_id resumes it.

proposal = client.runs.create(agent="iris", messages=[{"role": "user", "content": "Move our blog to a static site."}],
                              plan_mode=True, conversation_id="ticket-88")
plan = client.plans.retrieve(proposal.plans[0].id)
plan = client.plans.approve(plan.id, revision=plan.revision)
client.runs.create(agent="iris", messages=[{"role": "user", "content": "Go ahead."}], plan_id=plan.id,
                   conversation_id="ticket-88")

for event in client.plans.stream(plan.id):   # live; resumes after the last event and ends once the plan is finished
    print(event.event, event.data.get("step"), event.data.get("status"))

client.plans also has list(conversation=, status=, limit=, after=), create(title=, steps=, mode="plan" | "live"), update(id, revision=, ...), cancel(id), move_step(id, key, status=, note=, verified_by=) and events(id, after=). A key sees only its own plans.

Decisions: typed answers, fast and cheap

Ask one or more questions about some text (or text and images) and get typed answers back: a probability (predicate), one of your values (choice) or a rubric level (score). A fraction of a second and a fraction of a cent; billed at cost.

answer = client.decisions.create(
    input="I can't log in after resetting my password.",
    questions=[
        {"type": "predicate", "name": "urgent", "instructions": "The customer is blocked from using the product."},
        {"type": "choice", "name": "team", "instructions": "Which team should handle this.",
         "choices": [{"value": "account", "description": "Sign-in and account access"},
                     {"value": "billing", "description": "Payments and invoices"},
                     {"value": "other", "description": "Anything else"}]},
    ],
)
answers = {a["name"]: a for a in answer["answers"]}  # urgent: probability; team: choice

Connected apps: saving into a person's Workspace

With the person's approval, an application can save into their Dyadic Workspace, naming them by the Discord account linked to their Dyadic account.

from dyadic import DyadicError

status = client.connections.discord(discord_id)
if not status.authorized:
    print("Allow this app:", status.authorize_url)   # show the person this link

try:
    saved = client.workspace.saves.create(discord_id=discord_id, kind="message", title="From #general",
                                          text="Worth keeping", url="https://discord.com/channels/1/2/3")
    print(saved.url)
except DyadicError as error:
    if error.code == "authorization_required":
        print("Allow this app:", error.body["authorize_url"])

kind is message, conversation (with messages, saved as a chat the person can continue unless as_chat=False), image or file (with files: base64, up to 10 of 8 MB).

Conversations, stored completions, containers, vector stores, skills

The rest of the OpenAI-compatible API, named as OpenAI's SDK names it (the same calls on AsyncDyadic, awaited):

conversation = client.conversations.create(metadata={"ticket": "88"})
client.responses.create(model="iris", input="Hello", conversation=conversation.id)
items = client.conversations.items.list(conversation.id, order="asc")

client.responses.input_tokens.count(model="iris", input="How long is this?")   # free: an estimate
compacted = client.responses.compact(model="iris", previous_response_id="resp_...")  # send compacted.output next

stored = client.chat.completions.list(metadata={"team": "support"})    # completions made with store=True
client.chat.completions.messages.list(stored.data[0].id)

container = client.containers.create(name="scratch")                   # the key's code interpreter switch
client.containers.files.create(container.id, file="data.csv")

store = client.vector_stores.create(name="docs", file_ids=[file.id])    # file search switch; no credits
hits = client.vector_stores.search(store.id, query="refund policy")

skill = client.skills.create(files="./csv-skill")                       # a folder (or one zip, or a list of files)

Also conversations.retrieve/update/delete and items.create/retrieve/delete, chat.completions.retrieve/update/delete, containers.files.list/retrieve/delete/content.retrieve, vector_stores.files.* and vector_stores.file_batches.*, and skills.versions.*.

The Anthropic API and OpenRouter's, on the same key

client.anthropic speaks the Anthropic API at /api/anthropic/v1 (Claude's request and answer shapes, behaving as Anthropic's do) and client.openrouter OpenRouter's API through the AI gateway at /api/openrouter/v1, both with this client's key and beside its base URL:

message = client.anthropic.messages.create(model="iris", max_tokens=1024, system="Be brief.",
                                           messages=[{"role": "user", "content": "Hello"}])
print(message.text, message.stop_reason, message.usage.output_tokens)   # Anthropic's answer; .text joins its text blocks

stream = client.anthropic.messages.create(model="iris", max_tokens=1024, stream=True,
                                          messages=[{"role": "user", "content": "Tell me a story"}])
for event in stream:                                  # Anthropic's events: message_start, content_block_delta, ...
    if event.type == "content_block_delta" and event.delta.type == "text_delta":
        print(event.delta.text, end="")

batch = client.anthropic.messages.batches.create(requests=[{"custom_id": "a", "params": {
    "model": "iris", "max_tokens": 256, "messages": [{"role": "user", "content": "Hi"}]}}])
results = client.anthropic.messages.batches.results(batch.id)   # once processing_status is "ended"

# OpenRouter's models through the AI gateway (Dyadic turns it on for an app's key on request):
reply = client.openrouter.chat.completions.create(model="anthropic/claude-haiku-5.5",
                                                  messages=[{"role": "user", "content": "Hello"}])
print(client.openrouter.key().data.limit_remaining)   # this key's limits, answered by Dyadic

Also anthropic.messages.count_tokens (free), anthropic.messages.batches.list/retrieve/cancel/delete, anthropic.models.list/retrieve, anthropic.files.create/list/retrieve/content/delete, and openrouter.completions, openrouter.embeddings, openrouter.models.list/endpoints, openrouter.generation and openrouter.credits. The gateway is a platform decision: Dyadic turns it on for an app's key on request, the key's owner can turn it off, and a key without it gets 403 gateway_not_allowed. Anthropic's own SDKs and Claude Code work too, with the base URL https://dyadic.app/api/anthropic.

Agent Black Box: a flight recorder for your own agent

Record what your agent does into Dyadic's tamper-evident Black Box: every event is chained by hash and signed, so you (or anyone you give the evidence to) can check later that the record wasn't changed. Switch Black Box: send telemetry (and read, to read runs back) on for the key first; both are off by default.

from dyadic.blackbox import Recorder

recorder = Recorder()   # or client.blackbox.recorder()
with recorder.run("Book a room", agent="planner") as run:
    run.declare(allowed_tools=["search_rooms", "book_room"], allowed_hosts=["rooms.example.com"])
    with run.model_call("gpt-4.1") as call:
        call.responded(usage={"input_tokens": 812, "output_tokens": 40}, tool_calls=[{"id": "call_1", "name": "book_room"}])
    with run.tool_call("book_room", effect=True, hosts=["rooms.example.com"], call_id="call_1") as tool:
        tool.result(book_room("Orion"))
    run.verified("booking", "confirmation ORN-1042 read back from the booking API")

• Recording never blocks your agent: a bounded queue, a background thread, batches with an Idempotency-Key. Every event carries your process's own sequence number, so lost events show up as gaps.
• Only metadata leaves your process unless you pass capture_content=True (then secrets are redacted first, and the account's own settings decide what's kept).
• Subagents in other services: send run.traceparent() and start theirs with recorder.run(..., traceparent=header).
• Already on OpenTelemetry? BlackBoxSpanExporter() sends your spans (GenAI semantic conventions or OpenInference). Like the Recorder, it keeps content attributes (gen_ai.input.messages, tool arguments and results, input.value, ...) in your process unless capture_content=True, and passes every value through redact_secrets first (credential-named attributes are withheld; redact=None turns this off). MCP: instrument_mcp(session, run, server="notes").
• The Recorder sends from a background thread, so it takes the synchronous Dyadic client (AsyncDyadic().blackbox.recorder() raises TypeError); its runs and spans work under asyncio all the same.
• Read back and check: client.blackbox.runs.list(), .graph(id), .verify(id), .export(id, format="zip") (the bundle includes a dependency-free verify.py), and python -m dyadic.blackbox verify bundle.json --key KEY_ID=PUBLIC_KEY. An anchored bundle's RFC 3161 timestamp is checked too (verify_bundle(..., tsa_roots=[der])).
• client.blackbox.runs.simulate(id, mode="simulate" | "counterfactual" | "reconstruct"): nothing ever runs in a simulation; it is not historical evidence.

Reading, triaging and managing

With the key's read switch: client.blackbox.overview(), .settings(), runs.list/retrieve/events(types=[...])/graph/verify/checkpoints/otlp/export, diff(a, b), findings.list(status=, severity=, rule=) (calling findings(...) still works), rules.list() and rules.preview(rules) (what detection would raise with other settings; nothing is saved). Write: bookmarks.create(run_id, seq=, note=), findings.update(id, status="dismissed"). Content: runs.content(id). Manage: runs.erase(id), runs.delete(id) and the sinks below. A run of the key's that was deleted raises dyadic.RunDeletedError (HTTP 410). OTLP protobuf goes as it is: client.blackbox.send_otlp(raw_bytes).

Live

with client.blackbox.runs.stream(run_id) as events:
    for event in events:
        if event.event == "event":
            print(event.id, event.data.type)          # id: the event's sequence number
        elif event.event == "end":
            print(event.data.status, event.data.outcome)  # the stream stops after end (or deleted)

for event in client.blackbox.watch():                  # this key's runs as they start, change and grow
    if event.event == "run":
        print(event.data.id, event.data.status)

Both reconnect by themselves when the service asks or the connection drops (a run's stream resumes after its last event); on AsyncDyadic they are async iterators (async for). Break out of the loop or close() to end them.

Webhooks and collectors

client.blackbox.sinks.create(kind="webhook", url=..., events=["finding.created", "run.failed"], min_severity="high") sends signed JSON as findings are raised and runs end (keep its secret: it is shown once; sinks.rotate(id) makes a new one); kind="otlp" exports each finished run to an OpenTelemetry collector (protocol="http/protobuf", headers=). Check every webhook before trusting it, with the body exactly as it arrived:

from dyadic.blackbox import verify_webhook, WebhookVerificationError

try:
    message = verify_webhook(request.body, request.headers, secret)   # whsec_...
except WebhookVerificationError:
    ...  # answer 400

(Standard Webhooks: HMAC-SHA256, several signatures during a rotation, a timestamp within 5 minutes by default; skip repeats by the webhook-id header.)

Tests

pytest runs the SDK's own tests against a mocked transport: nothing reaches a server and nothing is spent. The live checks in live_tests/ call a real API and run only when DYADIC_LIVE_API_KEY is set (DYADIC_LIVE_BASE_URL for another deployment): free listings and token counts, and two messages of at most 16 tokens. Use a short-lived key with a small credit cap.

See https://dyadic.app/developers/ for the full guide.