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.

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.