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 an organization API key (zct_...) in ZooWork under Settings → API Keys and keep it on your server. It authenticates as the organization, not as one end user.

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"}},
            }
        ],
    }
)

Direct DingTalk binding uses dingtalk-connector with clientId/clientSecret and currently requires dm_policy: "open"; it does not have a guided setup route. Feishu document permission administration is enabled with permission_admin_enabled: True. Read the returned capabilities.feishu_documents.sync and .provider states, including missing_scopes and approval_state, instead of treating the write receipt as readiness. These contracts are source-reviewed, not live-verified here.

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.

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()]
    enqueue(event.type, event.id, run_id(event))  # do the work outside the request
    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.