rugbe-client (Python)

Typed HTTP + SSE client for rugbe-agent-runtime, sync and async. Backend subset of @bhanux/rugbe-client: workflows, runs (streaming), sessions, files, skills, models, me, node types, health.

from rugbe_client import RuntimeClient

with RuntimeClient(
    "https://runtime.example.com",
    get_auth_headers=lambda: {"Authorization": f"Bearer {key}"},
    on_auth_error=lambda: refresh_key(),  # optional: called once on 401
) as rt:
    session = rt.sessions.create(subject="user-42")
    for event in rt.runs.stream(session.workflow_id, input={"message": "hi"}, view="chat"):
        if event.type == "text":
            print(event.content, end="")

AsyncRuntimeClient has the same surface with async/await (async for on runs.stream); its credential hooks may be coroutines.

• Errors: every non-2xx raises RuntimeApiError (status, kind, issues, detail). to_draft_conflict(err) reads a lost-update 409 on workflows.save_draft.
• Only 401 triggers on_auth_error, once; a refreshed credential that is refused again surfaces. 403 is never retried.
• Unknown stream event types arrive as UnknownEvent instead of raising.

Development

uv sync
uv run pytest
uv run ruff check . --fix && uv run ruff format src tests

tests/test_conformance.py compares model fields to packages/rugbe-client/openapi.json (refresh with bun run generate-types there).