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()
        agent = await client.create_agent(
            {"name": "research-agent", "model": {"primary": models[0]["model"]}}
        )
        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.

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.

API surface

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

• agents, lifecycle, models and paginated listing;
• channels and guided Feishu/WeCom/WeChat setup;
• skill upload, versioning and agent attachment;
• sessions, 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.