Python SDK

Installation

pip install hubrix

Or from the downloaded zip:

pip install hubrix-python-2026.05.04.zip

Authentication

from hubrix import HubrixClient

# With API key (recommended for server-side)
client = HubrixClient(api_key="hbx_your_api_key")

# With email/password (obtains token automatically)
from hubrix.auth import login
tokens = await login(email="you@example.com", password="YOUR_PASSWORD")
client = HubrixClient(api_key=tokens["access_token"])

Quickstart

import asyncio
from hubrix import HubrixClient

async def main():
    async with HubrixClient(api_key="YOUR_API_KEY") as c:
        # 1. Verify connection
        profile = await c.me()
        print(f"Connected as: {profile['email']}")

        # 2. Start a research report
        job = await c.research.create(
            "Latest developments in AI reasoning models",
            depth="standard",
        )
        print(f"Research job: {job['id']}")

        # 3. Poll until complete (~30-60s)
        report = await c.research.poll(job["id"], timeout=120.0)
        print(f"Report: {len(report.get('content', ''))} chars")

        # 4. Save to markdown
        md = await c.research.export_markdown(job["id"])
        with open("report.md", "w") as f:
            f.write(md)
        print("Saved to report.md")

        # 5. List available workflows
        wf = await c.workflows.list(limit=10)
        print(f"Found {wf['total']} workflows")

asyncio.run(main())

Recipes

Run a workflow and poll for results

async with HubrixClient(api_key="YOUR_API_KEY") as c:
    # Create a run with inputs
    run = await c.workflows.create_run(
        workflow_id="your-workflow-uuid",
        inputs={"query": "Summarise Q1 2026 sales", "format": "markdown"},
    )
    print(f"Run queued: {run['id']}")

    # Poll until done (default 300s timeout)
    result = await c.workflows.poll_run(
        workflow_id="your-workflow-uuid",
        run_id=run["id"],
        interval=3.0,
        timeout=120.0,
    )
    print(f"Status: {result['status']}")
    print(f"Output: {result.get('outputs', {})}")

Deep research report

async with HubrixClient(api_key="YOUR_API_KEY") as c:
    # Standard depth: ~30-60s, 40 credits
    job = await c.research.create(
        query="Competitive landscape: AI workspace tools 2026",
        depth="standard",
    )

    # Deep depth: ~2-3min, 120 credits
    # job = await c.research.create(query="...", depth="deep")

    report = await c.research.poll(job["id"], timeout=180.0)
    print(f"Sources: {len(report.get('sources', []))}")

    # Export as markdown
    md = await c.research.export_markdown(job["id"])
    print(md[:1000])

Error handling

from hubrix import HubrixClient
from hubrix.errors import (
    InsufficientCreditsError,
    RateLimitError,
    AuthenticationError,
)

async with HubrixClient(api_key="YOUR_API_KEY") as c:
    try:
        report = await c.research.create("AI trends")
    except InsufficientCreditsError as e:
        print(f"Not enough credits: {e.detail}")
        # Top up at app.hubrix.ai/settings/billing
    except RateLimitError as e:
        print(f"Rate limited. Retry after: {e.detail}")
    except AuthenticationError:
        print("Token expired — refresh your key")

Reference

Full API reference at api.hubrix.ai/reference.

Source: github.com/hubrix-ai/sdk-python