zoowork

Official Python SDK for the ZooWork Managed Agents API. Developer Preview.

The client is asynchronous, typed, and built on httpx. Request and response fields retain the public API's wire spelling, while methods use Python snake_case.

pip install zoowork

Quickstart

Create a Project API key (zwp_live_...) in ZooWork Platform and keep it on your server. It scopes Agent and Session access to that Project. Platform runtime creation requires initialized Organization billing and bound owner credentials; rebind the key after signing in when the API requests it.

import asyncio
import os

from zoowork import assistant_text, create_zoowork_client, is_run_finished


async def main() -> None:
    async with create_zoowork_client(os.environ["ZOOWORK_API_KEY"]) as client:
        models = await client.list_models()
        primary = next(
            model["model"]
            for model in models
            if model.get("selectable", True)
            and model["model"] == "litellm/gpt-5.6-terra"
        )
        agent = await client.create_agent(
            {"name": "research-agent", "model": {"primary": primary}}
        )
        agent_id = agent["agent_id"]

        await client.start_agent(agent_id)
        await client.wait_until_running(agent_id)
        session = await client.create_session(
            agent_id,
            {"initial_events": [{"type": "user.message", "content": "What can you do?"}]},
        )

        async for event in client.stream_events(agent_id, session["session_id"]):
            print(assistant_text(event), end="", flush=True)
            if is_run_finished(event):
                break


asyncio.run(main())

Set ZOOWORK_API_KEY and call create_zoowork_client() with no argument if you prefer. The client uses the production API by default; ZOOWORK_BASE_URL or base_url= selects another deployment.

list_models() can include rows whose retirement has started. Check model.get("selectable", True) before using a row in a new Agent or config. A non-selectable choice returns 409 model_not_selectable; expired_fallback_to contains the reviewed replacement when present.

Agent resource mappings also accept userTimezone, a named IANA timezone for prompt and message time context, and include_global_skills: False to disable automatic global Skills while preserving explicit installs. An explicit "skills": [] also opts out. Schedule timezones are configured separately.

Pagination

list_agents() returns one AgentPage. Iterate the page to continue lazily through every remaining page while retaining the original filters:

page = await client.list_agents(labels={"project": "research"})
print(page.data, page.total, page.next_page)

async for agent in page:
    print(agent["agent_id"])

Use await page.get_next_page() for manual navigation or client.iter_agents() when you do not need the first page's metadata.

Durable events

The event stream is session-scoped and does not close when one turn ends. Break on is_run_finished(event). Save event.cursor after consuming an event and pass it back as cursor= when reconnecting.

async for event in client.stream_events(agent_id, session_id, cursor=last_cursor):
    if event.cursor is not None:
        last_cursor = event.cursor
    if is_run_finished(event):
        break

list_events() reads one durable page. list_all_events() follows cursor pagination, with a safe fallback for older deployments.

Agent tools and channels

Agent resources keep the API's original field names. MCP runtime context is opt-in and is not authentication. permission sets the server default; tools overrides exact native tool names. Tool-policy selectors accept an exact name, global *, or one trailing prefix*.

agent = await client.create_agent(
    {
        "name": "support-agent",
        "mcp": [
            {
                "name": "operations",
                "url": "https://mcp.example.com",
                "context": {"meta": True},
                "permission": "always_ask",
                "tools": {"lookup_customer": {"permission": "always_allow"}},
            }
        ],
    }
)

Project keys use the managed Environment. Root Environment and Skill administration and Channel binding return 404 service_api.not_found. Inspect attached Skills with list_agent_skills; select catalog Skills by name or skill_id at create time.

Filtered sessions and application-executed tools

list_sessions() keeps the legacy numeric page. list_session_page() selects the filtered cursor lane and starts with sls1:0. Its cursor is opaque and valid only with the same channel, surface, runtime-mode and archive filters; use each row's list_cursor or the page's next_cursor to continue.

Pass include_deleted=True to include deletion tombstones for reconciliation. Returned rows then carry deleted, and the page has includes_deleted=True. This flag is part of the cursor scope, so do not reuse a cursor created without it.

Declare application-executed tools in resource.custom_tools. At most 32 declarations are accepted. A declaration has name, description, an object input_schema, and optional timeoutMs (default 600,000 ms; maximum 86,400,000 ms).

When custom_tool_use(event) returns a requested call, execute it in your application and return the result with resolve_custom_tool_call(). list_custom_tool_calls(status="pending") recovers pending work after a restart. You can instead post user.custom_tool_result to the owning session. While paused, run_status is awaiting_approval; check pending_custom_tool_calls to distinguish it from a normal approval.

from zoowork import custom_tool_use

call = custom_tool_use(event)
if call is not None and call.phase == "requested":
    await client.resolve_custom_tool_call(
        agent_id,
        call.call_id,
        content=[{"type": "json", "value": {"price": 42}}],
        resolved_by="pricing-service",
    )

Result content contains 1–16 text, JSON, or base64 image blocks. Use an idempotency key when posting the session event. A pending REST resolution returns signaled: true before the row becomes terminal; signaled: false means it was already completed, timed out, or cancelled. This lifecycle is source-reviewed and still needs deployment verification.

Receiving webhooks

verify_webhook_signature() and unwrap_webhook() verify a received webhook using the standard library only. Pass the request bytes exactly as received: a re-serialized JSON body no longer matches the signature. The signature is checked before the body is parsed, so unverified bytes never reach the JSON parser. Supply accept_once using your durable storage and queue; a worker performs business processing after acknowledgement.

from zoowork import WebhookSignatureError, run_id, unwrap_webhook


def application(environ, start_response):
    length = int(environ.get("CONTENT_LENGTH") or 0)
    raw_body = environ["wsgi.input"].read(length)
    headers = {
        key[5:].replace("_", "-"): value
        for key, value in environ.items()
        if key.startswith("HTTP_")
    }
    try:
        event = unwrap_webhook(headers=headers, raw_body=raw_body)
    except WebhookSignatureError as error:
        start_response("400 Bad Request", [("content-type", "text/plain")])
        return [error.code.encode()]
    if event.id != headers.get("WEBHOOK-ID"):
        start_response("400 Bad Request", [])
        return []
    try:
        accept_once(event.id, event)  # atomic durable insert + worker item; duplicates succeed
    except Exception:
        start_response("503 Service Unavailable", [])
        return []
    start_response("204 No Content", [])
    return []

headers accepts any case-insensitive container: a plain dict, a single-valued mapping of lists, or a headers object with get(). secret defaults to ZOOWORK_WEBHOOK_SECRET, which may hold every active secret of a rotation window separated by whitespace or commas; pass secret=["whsec_...", "whsec_..."] when configuration comes from elsewhere. The timestamp window is ±300 seconds (tolerance_seconds) and the body ceiling is 16 KiB (max_body_bytes).

unwrap_webhook() returns a frozen WebhookEvent with the envelope's own field names: object, id, type, schema_version, created_at and data. The frame is validated strictly — object must be "event", id, type and created_at must be strings, schema_version must be present and an integer value (a JSON number, 1 or 1.0, never 1.5; it is read back as an int), and data must be an object — because the server sends all of them on every envelope and the TypeScript SDK rejects the same ones. type keeps the server's string even when this release does not know it, because event types are added server-side; check is_known_webhook_event_type() against WEBHOOK_EVENT_TYPES and ignore what you do not handle. data is passed through unchanged, and session_id(), run_id(), agent_id() and schedule_id() read its common fields.

Failures raise WebhookSignatureError, a ZooworkError with status 400 and a stable code: invalid_secret, body_too_large, missing_header, invalid_header, timestamp_out_of_window, signature_mismatch, or invalid_payload. Match the code, never the message; messages never contain a secret, a signature, or body bytes. The code is the cross-language contract: the TypeScript SDK raises its own standalone error class with the identical codes, so a repeated header or a malformed envelope is rejected the same way on both sides.

The wire format is plain Standard Webhooks, so a receiver can equally verify with the official PyPI standardwebhooks package. This SDK is byte-compatible with it: tests/fixtures/webhook_vectors.json is the fixed vector file the server and the TypeScript SDK assert against, copied verbatim.

API surface

The runtime client follows the TypeScript SDK's public capabilities:

• agents, lifecycle, models and paginated listing;
• channels, direct DingTalk, and guided Feishu/WeCom/WeChat setup;
• skill upload, versioning and agent attachment;
• sessions, filtered cursor listing, application-executed custom tools, events and SSE streaming;
• approvals, artifacts and system prompts;
• schedules, wake and sandbox exec;
• Environments and immutable Environment versions.

Methods return dictionaries containing the API response unchanged unless the SDK must normalize pagination or events. Unknown fields are intentionally preserved.

Errors

Every non-successful response raises ZooworkError. Match error.type or error.status, never the human-readable message. Diagnostic fields include content_type, body_snippet, cf_ray, request_id, and retryable.

Development

python -m venv .venv
. .venv/bin/activate
python -m pip install -e '.[dev]'
python -m pytest
python -m ruff check .
python -m mypy src
python -m build
python -m twine check dist/*

Unit tests are offline. The live staging procedure is documented in e2e/README.md and is never run by CI.

Developer API helpers

Check the installed SDK for these methods; if a release lacks one, use the documented HTTP endpoint. These examples require a release containing the helpers.

usage = await client.get_usage(range="7d", view="both")
endpoint = await client.create_agent_webhook(
    agent_id, {"url": "https://receiver.example/webhook", "event_types": ["run.finished"]},
    idempotency_key="register-hook-v1",
)
# Save signing_secret securely when present; an idempotent replay may return None.
hooks = await client.list_agent_webhooks(agent_id)  # hooks["webhooks"]
output = await client.get_run_output(agent_id, session_id, run_id)
approvals = await client.list_approval_page(agent_id, session_id=session_id)

Direct workspace Files and the database viewer are not supported production workflows. Supply text in Session messages, ask the Agent to create and publish Artifacts, and download those through the Artifact API. The Agent can use agent_db and return query results in its reply. Method presence is not production availability. Usage stays within current key scope. Paging returns next_cursor and has_more unchanged. get_approval and get_custom_tool_call read terminal as well as pending actions. Existing array-returning list methods remain available.

Webhook management includes get/update/delete, rotate_agent_webhook_secret, test_agent_webhook, get_agent_webhook_event, delivery list/detail and single/batch redelivery. Create, rotation, test and redelivery require a stable idempotency_key; SDK mutations never retry automatically. A 202 receipt acknowledges queuing. Query deliveries for the outcome.

Session input accepts runtime_mode: "active" to pin the active configuration at creation; omission resolves active configuration on subsequent turns. idle_compaction preserves false, null and omission. MCP tool overrides support requireConfirmation. Production update_agent currently rejects expected_config_version with 400 invalid_declared_key. Omit it for ordinary last-write-wins updates; serialize competing writes in your application. A GET then PUT is not atomic. Ownership-only changes do not increment configuration version. The separate upgrade_system_prompt version precondition remains supported.

Current production behavior

• Manual Schedule execution requires enabled: true, which also enables automatic firings. A disabled Schedule can return triggered: true and still be skipped. The receipt is not a run result; run rows can lack status/session linkage.
• Approval waiting is agent.approval / requested; resolved ends the approval wait. agent.tool / blocked ends a call without execution, with no later end. Reasons include policy denial, approval denial/timeout/cancellation, or interruption; inspect the event payload's deniedReason.
• Save each processed stream cursor with its Session ID. Pass it when reading a subsequent turn in that Session. No cursor means replay from the beginning, including old run.finished. REST events and post-event receipts do not supply a cursor; the last REST page has a null continuation token. There is no current-tail helper. Without a saved cursor, replay and reconstruct state or deliberately start a new conversation; do not synthesize a cursor from seq.
• First Agent deletion succeeds with 204; repeated deletion returns 404 through the public API. For cleanup retries, interpret 404 as absence only for a known Agent with unchanged key scope. Other-tenant or inaccessible resources also return 404.
• Invalid Usage parameters can return either 400 usage.invalid_query or 422 with no business error type. Correct the parameters rather than retrying unchanged. Match status as well as type.

See the public guides for supported workflows. These notes do not change SDK transport behavior or make unavailable endpoints usable.