z3t-ai-agent-sdk

Python SDK for building agents on the z3t.ai platform.

This is the Python counterpart to @z3t-ai/agent-sdk — same wire/HTTP contract.

Units differ from the TypeScript SDK. Durations here (timeout, reconnect_delay, max_reconnect_delay) are in seconds, not milliseconds, to match Python/asyncio convention (asyncio.sleep, asyncio.wait_for). The effective defaults are the same: 25s handler timeout, 1s initial reconnect backoff, 60s backoff ceiling.

---

Installation

pip install z3t-ai-agent-sdk

Install whichever LLM provider extras your agent uses:

pip install "z3t-ai-agent-sdk[openai,anthropic,google]"
# or just the ones you need, e.g.:
pip install "z3t-ai-agent-sdk[anthropic]"

Requires Python 3.10+.

---

Quick start

import asyncio
import os

from z3t_ai_agent import Agent, VersionSchema, s

agent = Agent(api_key=os.environ["Z3T_AGENT_KEY"])

contract_schema_v1 = VersionSchema(
    input=s.object({
        "document": s.file_uri(title="Contract PDF", accept=["application/pdf"]),
        "language": s.enum(["en", "fr", "de"], title="Language"),
        "notes": s.string(display="textarea", title="Notes").optional(),
    }),
    output=s.object({
        "summary": s.markdown(title="Summary"),
        "confidence": s.percent(title="Confidence"),
        "report": s.file_output(title="Full PDF report"),
    }),
)


@agent.handle(version=1, schema=contract_schema_v1)
async def handle_v1(input: dict, ctx) -> dict:
    await ctx.progress("downloading", "Downloading contract...", 0.1)
    file = await ctx.files.download(input["document"])

    await ctx.progress("analysing", "Analysing with AI...", 0.4)
    response = await ctx.llm.anthropic.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": f"Summarise this contract in {input['language']}"}],
    )

    await ctx.progress("uploading", "Generating report...", 0.8)
    report_uri = await ctx.files.upload(pdf_bytes, "report.pdf", "application/pdf")

    return {
        "summary": response.content[0].text,
        "confidence": 0.92,
        "report": report_uri,
    }


asyncio.run(agent.start())

See examples/ for complete, runnable agents — a minimal default handler, a versioned schema agent, a credentials-vault integration agent, a taxonomy-driven mapping agent, and an agent-to-agent chaining example.

---

From your machine to the marketplace

Writing the handler is only half the job — here's how you connect it to the platform, test it privately, and publish it.

1. Create the agent in the dashboard. In the z3t.ai app, go to Developer → Agents → New agent and fill in its name, description, category, and pricing. A new agent starts private — only your organization (and any orgs you explicitly grant access) can see or call it.

2. Create an agent key. Open the agent and, under API Keys, create a key. The raw key is shown once — copy it and set it as the environment variable your process reads:

export Z3T_AGENT_KEY="z3t_agentkey_..."

Each key authenticates exactly one agent. Rotate or revoke keys from the same panel.

3. Run and test privately. Start your process (asyncio.run(agent.start())). It connects to the relay and — if you declared schemas with s.* — registers each version as draft (see Versioning lifecycle). While the agent is private you can run test calls against it yourself from the dashboard (the Test action on a schema version, which works on drafts too) without it being visible to anyone else. Iterate freely: draft schemas are mutable, so restart as often as you like.

4. Publish. When you're happy:

• set status="active" on the VersionSchema and restart — this freezes and publishes the contract; and
• switch the agent's visibility to Public in the dashboard to list it in the marketplace. (Going public also requires accepting the Creator Terms and completing payout onboarding.)

These two axes are independent: schema status (draft → active) controls whether the contract is published; agent visibility (private → public) controls whether the agent is listed. A public agent still needs at least one active version before consumers can call it.

---

Registering handlers

Agent.handle() is a decorator factory:

@agent.handle()                          # default handler — all schema versions
@agent.handle(version=1)                  # version-specific, no schema
@agent.handle(version=1, schema=schema)    # version-specific, typed schema

A version-specific handler takes priority over the default handler for calls targeting that version. If neither matches, the relay receives an error and your handler is never invoked.

---

Schema builder (s)

Every field is required by default. Call .optional() to make it optional. Options that were object literals in the TypeScript builder (s.string({ title, ... })) are keyword arguments here (s.string(title=...)).

Input fields

Method	Widget rendered	Key options
s.string()	Text input	display="textarea"|"markdown"|"code"|"hidden", min_length, max_length, pattern
s.email()	Email input	—
s.url()	URL input	—
s.date()	Date picker	min, max
s.datetime()	Date + time picker	min, max
s.number()	Number input	display="slider", min, max, multiple_of
s.integer()	Integer input	display="slider", min, max
s.boolean()	Checkbox	display="toggle"
s.enum(["a", "b"])	Dropdown	display="radio"
s.array(s.string())	Tag/chip input	min_items, max_items
s.array(s.enum([...]))	Multi-select checkboxes	min_items, max_items
s.array(s.object({...}))	Repeatable form group	min_items, max_items
s.object({...})	Nested section	—
s.file_uri()	File upload picker	accept (MIME list), max_size_mb
s.array(s.file_uri())	Multi-file upload	min_items, max_items
s.taxonomy_ref()	Taxonomy dropdown	taxonomy_slug
s.integration_ref() ⚠️	Integration dropdown — coming soon, vault not yet available	provider

Output fields

Method	Frontend rendering
s.string()	Plain text
s.markdown()	Rendered Markdown
s.html()	Sanitized HTML
s.url()	Clickable link
s.code(language=...)	Syntax-highlighted code block
s.json()	Syntax-highlighted JSON block
s.image()	Inline image
s.percent()	Percentage bar (value 0–1)
s.file_output()	Download button
s.array(s.file_output())	Download link list (auto-detected)
s.array(s.object({...}), layout="table")	Data table with column headers
s.array(s.image(), layout="gallery")	Equal-sized image tile grid
s.array(s.object({...}), layout="grid")	Multi-column card grid
s.array(s.object({...}))	Vertical card list (default)
s.object({...})	Key-value detail card
s.object({...}, columns=2)	Fields arranged in a 2-column grid
s.pdf_reference()	Clickable chip → PDF preview modal — construct values with pdf_reference(file, page=..., hint=...)
s.typed_value()	Frontend picks renderer from format — construct values with typed_value.markdown(str), typed_value.number(str), etc.

Array layouts

Pass layout to s.array() to control how the output is rendered. The default (no layout) stacks items vertically as cards.

# Table — columns from object properties; add sortable/searchable for interactivity
results=s.array(s.object({
    "name":   s.string(title="Name"),
    "score":  s.number(title="Score"),
    "status": s.enum(["pass", "fail"], title="Status", color_map={"pass": "green", "fail": "red"}),
}), layout="table", sortable=True, searchable=True, title="Results")

# Gallery — equal-sized image tiles; use with s.image()
images=s.array(s.image(), layout="gallery", title="Generated images")

# Grid — compact multi-column cards; good for product/people lists
products=s.array(s.object({
    "name":  s.string(title="Product"),
    "price": s.number(title="Price"),
}), layout="grid", title="Products")

# Vertical card list (default) — each item fully expanded, stacked
items=s.array(s.object({...}))

# File download list — automatic when items are s.file_output(); no layout= needed
reports=s.array(s.file_output(), title="Reports")

---

Versioning lifecycle

Every declared schema carries a status. It's synced to the platform on agent.start():

Status	Visible to consumers?	Mutable?	Use when
"draft" (default)	No	Yes — resync freely on every restart	Building and testing a version
"active"	Yes — published	No — frozen once active	Ready to publish the contract

VersionSchema(
    input=...,
    output=...,
    status="draft",              # default — mutable, invisible to consumers
    # status="active",           # publishes and freezes the schema
    deprecates=[1],              # optional — earlier versions this one replaces
    deprecation_notice="v1 is replaced by v2. Add `language` and send `documents` as a list.",
    interactive=True,            # optional — this version may pause runs to ask (frozen once active)
)

On agent.start() the SDK:

• creates new versions (as draft unless you pass status="active"),
• deprecates every version listed in deprecates — those versions keep working, but their consumers are notified and shown your deprecation_notice,
• rejects the restart if you change the schema of an already-active version (active schemas are immutable). Draft versions are yours to change.

Any version that syncs back as draft is logged on startup, so you always know which versions aren't publicly visible yet.

---

Asking the consumer a question

Consumers rarely fill in a form perfectly. An interactive version can read what it received, notice what is missing or contradictory, and ask — before it does the expensive work:

@agent.handle(version=3, schema=VersionSchema(
    input=s.object({"invoices": s.array(s.file_uri())}),
    output=s.object({"notice": s.file_output()}),
    interactive=True,
))
async def handle(input, ctx):
    facts = await ctx.step("extract", lambda: extract_facts(input, ctx))   # paid for once

    if facts["missing_contract"]:
        r = await ctx.ask(
            "contract",
            message=f"Invoice 3 refers to contract **{facts['contract_no']}**, which wasn't uploaded. Can you upload it?",
            schema=s.object({"contract": s.file_uri(title="Contract").optional()}),
        )
        if r.answered and r.answers.get("contract"):
            facts["contract"] = r.answers["contract"]
        # "declined" | "expired" | "unavailable" → carry on with a stated assumption

    return await ctx.step("draft", lambda: draft_notice(facts, ctx))

ctx.ask does not wait for the answer — that could take days. It ends the turn: your handler stops, the platform shows the consumer the question (and emails them), and when they answer, skip, or the deadline passes, the call is dispatched again — possibly to another replica of your agent. Your handler runs from the top: ctx.step returns what it returned last time instead of running again, and ctx.ask returns an AskResult:

result.action	When
"answered"	The consumer filled in the form — result.answers matches your schema
"declined"	The consumer skipped the question
"expired"	Nobody answered within the deadline (7 days) — the run is not refunded, so finish with your best effort
"unavailable"	This run can't take questions (an API integration that didn't opt in, another agent, or the call used its 3 questions). Nothing was paused.

Rules of thumb

• Put expensive or side-effecting work in ctx.step (the function may be sync or async). Code outside steps runs again on every resume — uploads belong inside a step.
• Step results must be JSON-serializable and come back in their JSON form (a tuple becomes a list) — on the first run too. The whole journal is capped at 2 MB.
• Keys are unique per call, across steps and questions.
• The pause unwinds your handler by raising SuspendSignal, a BaseException — except Exception: won't catch it. Don't catch it deliberately; the SDK suspends anyway and logs a warning.
• The handler timeout applies per turn.

Nothing about the call is cached on your machine: the platform hands the journal back with the resume, and deletes it when the call ends.

---

CallContext reference

Passed as the second argument to every handler.

Member	Signature	Notes
ctx.call_id / ctx.schema_version	str / int	—
ctx.progress(step, message, progress=None)	async	Fire-and-forget milestone — one activity-log row per call; progress is 0–1
ctx.subprogress(message, progress=None)	async	Detail within the current step — replaces the previous sub-line instead of adding a row
ctx.step(key, fn)	async -> Any	Runs fn once per call; on a resumed turn returns the recorded result (see above)
ctx.ask(key, *, message, schema)	async -> AskResult	Pauses the run to ask the consumer (interactive versions only)
ctx.turn / ctx.can_ask	int / bool	0 on the first dispatch, +1 per resume / whether ask can pause this run
ctx.files.download(uri)	async -> DownloadResult	DownloadResult(buffer: bytes, filename: str, mime_type: str)
ctx.files.upload(data, filename, mime_type)	async -> str	Returns the new z3t://files/{id} URI
ctx.taxonomies.entries(uri)	async -> list[TaxonomyEntry]	—
ctx.taxonomies.lookup(uri, key)	async -> TaxonomyEntry | None	None on not-found
ctx.integrations.credentials(uri) ⚠️	async -> dict[str, str]	Coming soon — vault not yet available. Shape depends on the integration's auth type
ctx.llm.openai / .anthropic / .google	client instances or None	None if the optional provider package isn't installed
ctx.agents.call(agent_id, plan_id, input, *, schema_version=None, consumer_org_id=None, timeout=None)	async -> Any	timeout in seconds; always suppresses progress events on the downstream call

ctx.llm.google is built on the modern google-genai package (genai.Client), not the legacy google-generativeai. Use ctx.llm.google.models.generate_content(...) or the async variant ctx.llm.google.aio.models.generate_content(...).

---

Configuration

Agent(
    api_key: str,                              # required — from the z3t dashboard
    base_url: str | None = None,                # default: "https://relay.z3t.ai/v1"
    relay_urls: list[str] | None = None,        # override — skips bootstrap; for local dev/tests
    timeout: float | None = None,               # seconds, default 25.0
    max_concurrent_calls: int | None = None,     # default 10
    reconnect_delay: float | None = None,        # seconds, default 1.0
    max_reconnect_delay: float | None = None,    # seconds, default 60.0
    heartbeat_interval: float | None = None,     # seconds, default 30.0 — keepalive ping period; 0 disables
    heartbeat_timeout: float | None = None,      # seconds, default 20.0 — reconnect if no pong within this
    logger: Logger | None = None,                # default: ConsoleLogger() — needs info/warn/error
)

---

Testing

pip install -e ".[dev]"
pytest                                    # unit + integration
pytest --cov=z3t_ai_agent --cov-fail-under=90  # with coverage

Integration tests spin up an in-process mock relay (tests/helpers/mock_relay.py, built on websockets.serve) — no external services required.

---

License

MIT © z3t.ai