Metadata-Version: 2.4
Name: bystro
Version: 2.1.2
Classifier: License :: OSI Approved :: Mozilla Public License 2.0 (MPL 2.0)
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Requires-Dist: requests>=2.31.0,<3
Requires-Dist: python-socketio[client]>=5.11.0,<6
Requires-Dist: openpyxl==3.1.2
Requires-Dist: boto3==1.28.9
Requires-Dist: liftover==1.2.2
Requires-Dist: msgspec==0.18.6
Requires-Dist: numba==0.60.0
Requires-Dist: opensearch-py[async]==2.5.0
Requires-Dist: numpy==1.26.4
Requires-Dist: pandas==2.2.2
Requires-Dist: pyarrow==16.1.0
Requires-Dist: pystalk==0.7.0
Requires-Dist: ruamel-yaml==0.17.31
Requires-Dist: scikit-allel==1.3.8
Requires-Dist: scikit-learn==1.5.1
Requires-Dist: skops==0.7.post0
Requires-Dist: tqdm==4.66.3
Requires-Dist: cloudpickle==3.0.0
Requires-Dist: torch==2.2
Requires-Dist: psutil==5.9.6
Requires-Dist: matplotlib==3.7.1
Requires-Dist: ecos==2.0.13
Requires-Dist: osqp==0.6.5
Requires-Dist: cvxpy==1.5.2
Requires-Dist: pyro-ppl==1.9.1
Requires-Dist: somadata==1.0.0
Requires-Dist: statsmodels==0.14.2
Requires-Dist: nest-asyncio==1.6.0
Summary: Statistical genomics tools and the Bystro Think agent API
License: MPL-2.0
Requires-Python: >=3.11.0, <3.13.0
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Documentation, https://github.com/bystrogenomics/bystro/blob/main/python/THINK_API.md
Project-URL: Homepage, https://bystro.io
Project-URL: Repository, https://github.com/bystrogenomics/bystro

# Bystro Think Python API

The Think SDK submits durable agent workloads, streams visible output and
structured progress, handles human-input pauses, uploads large files in chunks,
reuses Bystro datasets and conversations as context, and downloads protected
results.

## Install

Bystro 2.1.2 supports CPython 3.11 and 3.12:

```sh
python --version
python -m pip install "bystro>=2.1.2,<2.2"
```

Production uses publicly trusted HTTPS certificates. Customers do not need a
custom CA bundle or TLS override; those are only for local development servers
using a private certificate authority.

## Authenticate once, then use the cached login

Use `getpass` for the one-time interactive login so secrets do not appear in
source code, notebook output, shell history, or environment listings:

```python
from getpass import getpass

from bystro.api import auth


email = input("Bystro email: ").strip()
site_access_code = getpass("Site access code (leave blank if not required): ")
auth.login(
    email,
    getpass("Bystro password: "),
    site_access_code=site_access_code or None,
)
```

The login JWT is stored in `~/.bystro/bystro_authentication_token.json`; the
directory is mode `0700` and the atomically replaced file is mode `0600`. The
site-access code is used only for that login session and is not cached.

Normal scripts then use the cached login without handling a password:

```python
from bystro.think import ThinkClient

client = ThinkClient.from_cached_login()
```

New accounts must accept the current legal assertions once. Use
`LegalConsent.accepted(name)` with `auth.signup(...)`, or complete signup in the
dashboard before running the login snippet above.

## Canonical interactive workflow

This is the recommended customer experience. It prints lifecycle changes,
backend-owned phases such as web search and source verification, visible answer
chunks, and an elapsed heartbeat if no server frame arrives for 30 seconds.
`interact()` prompts for any number of clarification or plan-review pauses.

```python
from bystro.think import NeedsInput, RunResult, ThinkClient, show_progress


with ThinkClient.from_cached_login(on_event=show_progress) as client:
    run = client.submit_with_progress(
        "Research the latest CAR-T therapies and cite primary sources."
    )
    outcome = run.interact(timeout=3600)

    if isinstance(outcome, NeedsInput):
        # interact() handles clarification and plan-review pauses itself.
        # A returned NeedsInput is a billing pause that must be resolved in
        # the dashboard, followed by run.refresh(). Unlimited accounts should
        # not enter this branch.
        print(outcome)
    else:
        assert isinstance(outcome, RunResult)
        print("\nFinal Markdown is also available as outcome.output")
```

`submit_with_progress()` installs a progress renderer automatically. Supplying
`show_progress` on the client also includes connection and reconnect events.
Do not pass the same callback again to `run.wait(on_event=...)`.

Think uses authenticated Socket.IO transport. It normally upgrades to WebSocket
and retains HTTP polling as a compatibility fallback. Output frames are
cumulative short snapshots, not necessarily one event per tokenizer token; the
SDK turns them into exact append/replace/retract updates and never exposes
internal reasoning text.

To require native WebSocket and fail rather than fall back to polling:

```python
with ThinkClient.from_cached_login(
    on_event=show_progress,
    transports=("websocket",),
) as client:
    result = client.submit_with_progress("Draw a duck.").interact(timeout=3600)
```

## Non-interactive input callbacks

Applications can answer pauses without calling `input()`. Manual `wait()` and
`respond()` remain available when the application needs complete control.

```python
from bystro.think import NeedsInput, ThinkClient


def answer_clarification(request: NeedsInput) -> str:
    print("Clarification:", request.prompt)
    return "Cover all disease areas and the last 24 months."


def review_plan(request: NeedsInput) -> str:
    print("Proposed plan:", request.prompt)
    return "accept"


with ThinkClient.from_cached_login() as client:
    run = client.submit_with_progress("Research recent CAR-T therapies.")
    result = run.interact(
        timeout=3600,
        on_clarification=answer_clarification,
        on_plan_review=review_plan,
    )
```

For manual control:

```python
from bystro.think import InputKind, NeedsInput


outcome = run.wait(timeout=3600)
if isinstance(outcome, NeedsInput):
    if outcome.kind is InputKind.PLAN_REVIEW:
        run.respond("accept")
    elif outcome.kind is InputKind.CLARIFICATION:
        run.respond("Use case_control as the phenotype column")
```

The first live pause can precede its durable checkpoint commit. `respond()`
waits for the checkpoint replay before uploading attachments or dispatching the
answer, and ignores stale replayed checkpoints after reconnect.

## Choose a mode

The default mode is `base`. Pass `RunOptions` per submitted conversation:

```python
from bystro.think import RunOptions


run = client.submit_with_progress(
    "Research the latest CAR-T therapies and cite primary sources.",
    options=RunOptions(mode="plus2"),
)
```

| Value | Dashboard name | Intended use |
| --- | --- | --- |
| `base` | Base | Faster, token-efficient work with lighter research. |
| `plus` | Plus v1 | Verified analysis with deep research. |
| `plus2` | Plus v2 | Stronger experimental research workflow. |
| `phd` | PhD | Deepest reasoning for demanding analyses. |

Other controls are typed fields on `RunOptions`: `advanced_planning`,
`auto_compact`, `fast`, `verify`, `verify_sources`, and
`zero_data_retention`. Availability and billing follow the authenticated
account and deployment configuration.

## Submit files with the question

A path passed in `files` is uploaded to the authenticated user's personal
artifacts and attached to the same message. Large inputs use resumable bounded
10 MiB chunks, SHA-256 checksums, idempotent retries, and asynchronous
finalization polling.

```python
from bystro.think import ThinkClient, UploadProgress


def upload_progress(progress: UploadProgress) -> None:
    print(
        f"[upload:{progress.phase.value}] {progress.fraction:.0%}",
        flush=True,
    )


with ThinkClient.from_cached_login() as client:
    run = client.submit_with_progress(
        "Analyze the cohort using the attached phenotype table.",
        files=["cohort.vcf.gz", "phenotypes.tsv"],
        on_upload_progress=upload_progress,
    )
    result = run.interact(timeout=3600)
```

A single path can be passed directly:

```python
run = client.submit_with_progress(
    "Summarize this study protocol.",
    files="protocol.pdf",
)
```

Create a reusable artifact before submission with `upload_artifact()` (or its
short alias `upload()`):

```python
artifact = client.upload_artifact(
    "cohort.vcf.gz",
    artifact_path="study/cohort.vcf.gz",
    on_progress=upload_progress,
)
run = client.submit_with_progress("Run QC on this cohort.", files=[artifact])
```

Artifact paths are relative, have at most 64 components, and must end with the
local file's exact name. Invalid paths fail locally before upload.

## Compose genetic, conversation, and artifact context

The context helpers accept either a string or an immutable
`MessageWithContext`, so they compose without constructing XML manually. The
SDK serializes an escaped XML preview with `message.to_xml()`, while the live
request keeps ownership-bearing references in structured metadata.

```python
from bystro.think import (
    add_artifact_context,
    add_genetic_context,
    add_previous_conversation_context,
)


message = "Re-evaluate the strongest phenotype associations."
message = add_genetic_context(
    "annotation-job-id",
    message,
    name="Case cohort",
    assembly="hg38",
)
message = add_previous_conversation_context(
    "prior-thread-id",
    message,
    name="Earlier analysis",
)
message = add_artifact_context(artifact, message)

run = client.submit_with_progress(message)
```

Reusable higher-order transforms are available:

```python
from bystro.think import (
    artifact_context,
    compose_context,
    genetic_context,
    previous_conversation_context,
)


study_context = compose_context(
    genetic_context("annotation-job-id", assembly="hg38"),
    previous_conversation_context("prior-thread-id"),
    artifact_context("existing-artifact-id"),
)

run = client.submit_with_progress(
    study_context("Compare the strongest signals.")
)
```

Think resolves every dataset, conversation, and artifact under the
authenticated user's ownership. User-authored XML is never an authorization
boundary.

## Structured progress and custom presentation

`ThinkEvent.progress` contains the server's current phase snapshot. Typical
phase kinds are `search`, `verify`, `compute`, `query`, and `think`.
`ThinkEvent.stream_update` contains safe visible-output deltas.

```python
from bystro.think import EventKind, ThinkEvent


def on_event(event: ThinkEvent) -> None:
    if event.progress is not None:
        phase = event.progress.active_phase
        if phase is not None:
            print(phase.kind, phase.label, phase.completed, phase.total)
        return

    update = event.stream_update
    if event.kind is EventKind.STREAM and update is not None:
        if update.operation == "append":
            print(update.delta, end="", flush=True)
        elif update.operation == "replace":
            print("\n[corrected output]\n", update.delta)
        elif update.operation == "retract":
            print(f"\n[removed message {update.message_id}]")
```

`ProgressRenderer(heartbeat_interval=30)` provides the canonical terminal
presentation. Generic `Thinking...` and `Processing...` states print at most
once per turn; meaningful phase/count changes print immediately; and the renderer emits
`Still working... (… elapsed)` during complete transport silence. Heartbeat
workers stop on input, completion, failure, cancellation, or local detach.

## Results, conversations, and downloads

Generated files are available directly on a successful `RunResult`. The
listing is authenticated and loaded once, on first access, so access it while
the client context is open:

```python
from pathlib import Path

from bystro.think import RunResult, ThinkClient


with ThinkClient.from_cached_login() as client:
    run = client.submit_with_progress("Draw and save a cartoon duck.")
    result = run.interact(timeout=3600)
    if not isinstance(result, RunResult):
        raise RuntimeError("The run paused for billing")

    for output_file in result.files:  # result.artifacts is the same tuple
        print(output_file.path, output_file.size)

    if result.files:
        first = run.download_file(
            result.files[0],
            Path("downloads") / result.files[0].path,
        )
        print("Downloaded:", first)

    archive = run.download_all(Path("downloads") / f"{run.id}.tar")
    print("Archive:", archive)
```

`download_file()` and `download_all()` stream to a temporary file and publish
the destination only after the authenticated download completes. Existing
targets are never replaced unless `overwrite=True` is explicit.

List and resume past conversations:

```python
conversations = client.list_conversations(search="CAR-T", limit=20)
for conversation in conversations:
    print(conversation.id, conversation.name, conversation.created_at)

previous = client.resume(conversations[0].id)
print(previous.messages)
print(previous.output_files())
```

Omit `limit` to traverse all cursor pages. `run.messages` excludes internal
reasoning and progress-card messages; `run.history` is bounded SDK event
history. A resumed run restores its submitted mode and other `RunOptions`, so
`run.follow_up(...)` continues with the original settings.

In Jupyter, `RunResult` and `NeedsInput` implement `_repr_markdown_()`, so
placing either object at the end of a cell renders its Markdown naturally.

## Cancellation, detach, and reconnect

Cancellation is distinct from closing a local client:

```python
from bystro.think import RunCancelledError


run.cancel(timeout=60)  # waits for durable server cleanup to be released
try:
    run.wait()
except RunCancelledError:
    print("Cancelled")
```

`cancel()` sends the active task ID when available, ignores delayed lifecycle
events from older tasks, and reissues an interrupted stop after reconnect until
the server emits its durable release event.

Use `run.detach()` (or close the `ThinkClient`) to disconnect locally while the
server keeps working. Reattach from another process later:

```python
run_id = run.id
run.detach()

with ThinkClient.from_cached_login() as client:
    resumed = client.resume(run_id)
    outcome = resumed.wait(timeout=3600)
```

A `ThinkClient` owns one foreground conversation at a time. Use separate
clients for concurrently controlled conversations.

## Async applications

The submission API is synchronous; the live event iterator and terminal wait
also have event-loop-friendly async forms:

```python
import asyncio

from bystro.think import ThinkClient


async def main() -> None:
    with ThinkClient.from_cached_login() as client:
        run = client.submit("Research recent CAR-T approvals.")
        async for event in run.aevents(timeout=3600):
            print(event.kind.value)
        result = await run.await_result(timeout=30)
        print(result)


asyncio.run(main())
```

`aevents()` never blocks the event loop while waiting for Socket.IO events;
durable refreshes run outside the loop.

## Cloudflare configuration

No Cloudflare change is needed when the installed SDK connects, uploads, and
downloads successfully. If browser-only challenges intercept Python traffic,
create a **zone-level custom rule** with action **Skip** and match only the API
hosts/routes used by the SDK. Select only:

- All Super Bot Fight Mode rules
- Browser Integrity Check
- Security Level

Keep **Log matching requests** enabled. Do not select all remaining custom
rules, rate limiting rules, or managed WAF rules unless a specific logged false
positive proves one of those components is responsible. Cloudflare documents
that Skip can target these products independently, leaving other security
layers active: [Skip action](https://developers.cloudflare.com/waf/custom-rules/skip/)
and [available skip options](https://developers.cloudflare.com/waf/custom-rules/skip/options/).

For `ai.bystro.cloud`, the complete SDK transport surface is:

```text
/auth/cookie
/set-session-cookie
/ws/socket.io
/project/threads
/user/files/*
/api/user-output/*
```

For `bystro.cloud`, one-time programmatic login uses:

```text
/api/site-gate/authenticate
/api/user/auth/local
```

If customers also need to discover existing genetic-analysis jobs with
`bystro.api.annotation.get_jobs`, include the narrow `/api/jobs*` path prefix
on `bystro.cloud`. This route is optional when the caller already knows the
genetic job ID. It remains protected by Bystro authentication and authorization;
the Cloudflare rule skips only the selected browser-oriented checks.

Keep the route expression narrow rather than bypassing by user agent or a
shared customer token. Application authentication and ownership checks still
run at the origin, while Cloudflare DDoS protection, managed WAF, and rate
limits remain available.

An easy verification is to install the wheel in a clean environment, unset
`REQUESTS_CA_BUNDLE` and `SSL_CERT_FILE`, require
`transports=("websocket",)`, submit a small run, list conversations, upload a
file larger than 10 MiB, and download one result plus the tar archive. A
Cloudflare HTML challenge or `cf-ray` 403 indicates the route rule still does
not match; a typed JSON/application error means the request reached Bystro.

## Errors

Transport, HTTP, authentication, billing, admission, cancellation, timeout, and
protocol failures have typed exceptions under `bystro.think`. In particular:

- `ThinkAuthenticationError`: cached dashboard login is missing or expired.
- `ThinkBillingRequiredError`: initial admission requires plan/credit action.
- `RunRejectedError`: submission was rejected before dispatch.
- `RunTimeoutError`: a local wait deadline elapsed; the durable run may continue.
- `RunCancelledError`: server-side cancellation completed.
- `RunProtocolError`: the server returned contradictory or incomplete state.

Closing a client never implies cancellation.

