coworkers_client

Standalone async Python client library for the Coworkers chatbot API.

The PyPI distribution name is coworkers_client. The import package is also coworkers_client.

Testing With Injected HTTP Seams

CoworkersChatbot can be constructed with either a caller-managed http_client or a low-level transport.

• http_client= accepts an httpx.AsyncClient and stays caller-owned. await chatbot.close() will not close it.
• transport= accepts an httpx async transport and is wrapped in an internal httpx.AsyncClient. await chatbot.close() will close that internal client.
• Passing both http_client and transport is rejected with ValueError.
• retry_policy= accepts a CoworkersRetryPolicy for deterministic tests or bounded retries. When omitted, the client keeps the default 3-attempt retry behavior for transient 503/504 responses and httpx.ReadTimeout / httpx.ConnectTimeout with 10-20 second jitter.

Example with httpx.MockTransport:

import httpx

from coworkers_client import CoworkersChatbot, CoworkersRetryPolicy


async def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(
        200,
        request=request,
        json={
            "discussionID": 42,
            "userInfo": {"userToken": "test-user"},
            "messages": [
                {
                    "timestamp": 123,
                    "message": {"context": {}, "text": "mocked reply"},
                }
            ],
        },
    )


transport = httpx.MockTransport(handler)
chatbot = CoworkersChatbot(
    url="https://chatbot.example",
    transport=transport,
    retry_policy=CoworkersRetryPolicy.fixed_delay(0.0),
)

response = await chatbot.send_message("test-user", "hello")
assert response.get_text() == "mocked reply"

await chatbot.close()

Disable retries entirely for one-shot failure assertions:

from coworkers_client import CoworkersChatbot, CoworkersRetryPolicy

chatbot = CoworkersChatbot(
    url="https://chatbot.example",
    retry_policy=CoworkersRetryPolicy.disabled(),
)

CoworkersDiscussion also exposes deterministic test seams:

• user_token_factory= lets tests supply predictable generated tokens without patching uuid.
• restart(user_token=...) lets tests force a specific token for restart scenarios.
• start_dialog(...) now returns the underlying CoworkersResponse and updates the wrapper state from that response.

send_start_discussion(...) also accepts caller-provided scalar context values. They are serialized with the same payload shape as send_message(...) and are available in the initial chatbot response when the service supports them.

Consumer fakes and adapters can use the public ChatbotTransport protocol instead of importing from a private module:

from collections.abc import Mapping

from coworkers_client import ChatbotTransport, CoworkersDiscussion, CoworkersResponse


class FakeTransport:
    def __init__(self, response: CoworkersResponse) -> None:
        self.response = response

    async def send_start_discussion(
        self,
        user_token: str,
        context: Mapping[str, object] | None = None,
    ) -> CoworkersResponse:
        return self.response

    async def send_message(
        self,
        user_token: str,
        text: str,
        context: Mapping[str, object] | None = None,
    ) -> CoworkersResponse:
        return self.response

    async def start_dialog(
        self,
        user_token: str,
        dialog_id: int,
        context: Mapping[str, object] | None = None,
    ) -> CoworkersResponse:
        return self.response

    async def send_goals(self, user_token: str, goals: list[str]) -> dict[str, object]:
        return {"goals": goals}


prepared_response = CoworkersResponse()
prepared_response.load(
    {
        "discussionID": 42,
        "userInfo": {"userToken": "test-user"},
        "messages": [],
    }
)

transport: ChatbotTransport = FakeTransport(prepared_response)
discussion = CoworkersDiscussion(transport, user_token="test-user")

The stable import paths are coworkers_client.transport.ChatbotTransport and coworkers_client.ChatbotTransport.

Optional Event Endpoint

CoworkersChatbot.send_event(...) sends a generic event to POST /chat-api/v1/add-event.

• It accepts event_type, severity, subject, message, and optional details.
• It always includes the current sender userToken, includes instanceId when the client was configured with one, and posts the event fields inside the API's nested event object.
• It returns None.
• Failures are warning-only for this method, because not every live chatbot exposes the optional event endpoint.
• Severity is normalized to lowercase in the outbound payload to match the live Mailbot contract.

CoworkersChatbot.get_discussion(..., events=True) includes typed DiscussionEvent entries on the returned Discussion, and CoworkersChatbot.get_discussion_events(user_token) returns that event list directly for consumer assertions.

Example:

events = await chatbot.get_discussion_events(response.user_token)
assert any(event.subject == "send-event-subject" for event in events)

Error Handling

Public async client methods raise library-defined exceptions instead of exposing raw transport or decode errors directly.

• CoworkersClientError: base class for all public client failures
• CoworkersRequestError: request failed before a valid API response could be processed
• CoworkersTimeoutError: request timed out
• CoworkersHttpError: API returned an HTTP error response
• CoworkersResponseError: response body could not be decoded or did not match the expected payload shape

Each custom exception preserves the original underlying error as its cause, so callers can still inspect exc.__cause__ for the originating httpx or decode exception when needed.

Development

• Install uv
• Sync dependencies: uv sync --group dev
• Run all tests: make pytest
• Run unit tests only: make pytest-unit
• Run live integration tests only: make pytest-integration
• Run coverage for unit + integration suites: make coverage
• Run Ruff: make ruff
• Run Pyright: make pyright
• Build the package: make build

Release

Maintainers publish this package through the GitLab tag pipeline.

• The release tag must be vX.Y.Z or X.Y.Z.
• The tag pipeline rewrites [project].version from CI_COMMIT_TAG, builds the package, validates the artifacts, and uploads them to PyPI.
• The publish job requires the PYPI_TOKEN CI variable. It is used as a PyPI API token with twine.

Local release validation:

uv sync --group dev
make ruff
make pyright
make coverage
make pytest-integration
make clean update-version build check-wheel check-dist RELEASE_TAG=v0.1.0

The CI publish job runs make publish after the release build artifacts pass validation.

Live Integration Configuration

Live chatbot tests use the committed Python config in tests/integration_config.py.

• Base URL: https://mailbot.bot.daktela.com
• instance_id: 1
• dialog_id: 1

The UI URL is https://mailbot.bot.daktela.com/#/, but the client must use the normalized API base origin without the #/ fragment for HTTP requests.