Metadata-Version: 2.5
Name: qoni
Version: 0.1.3
Summary: Unified Python SDK for Qoni Agent delegation, GUMem memory, and WebAgent automation.
Project-URL: Homepage, https://github.com/QONIAI/qoni-sdk-python
Project-URL: Issues, https://github.com/QONIAI/qoni-sdk-python/issues
License: MIT License
        
        Copyright (c) 2026 Eazo
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: agent,delegation,gumem,qoni,sdk,webagent
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.9
Requires-Dist: httpx>=0.24
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == 'dev'
Description-Content-Type: text/markdown

# qoni

Unified Python SDK for Qoni Agent delegation, GenAuth user management, GUMem memory, WebAgent automation, web search, and monitoring.

`qoni` gives a trusted server one compact way to use Qoni AK/SK. For runtime product calls, your service verifies the GenAuth user, requests short-lived delegation with `delegate_token`, and then uses the returned token in GUMem, Do Anything, Web Search, Deep Research, and Track calls. For GenAuth user management, the SDK exchanges AK/SK for a standard GenAuth management token internally and calls GenAuth v3 users APIs with that token.

AK/SK credentials must stay on a trusted server. Do not ship them to browsers, mobile apps, public CLI config, or untrusted Agent runtimes.

> This is the Python counterpart of [`@qoniai/qoni`](https://github.com/QONIAI/qoni-sdk-node) (the Node.js SDK). The API surface mirrors it, adapted to Python conventions: snake_case naming, keyword arguments, exceptions, generators, and dataclasses.

## Why Qoni

Agents that perform real work need more than a backend API key. They need a user boundary, explicit scopes, expiry, observable execution, and audit metadata that explains what happened later.

Qoni keeps that model small:

- One SDK entry: `Qoni(access_key=..., secret_key=...)`.
- One discovery path: `host` can override the Qoni Console/SDK gateway for private or local deployments, and the SDK reads downstream runtime URLs from `/api/v3/eak/runtime-config`.
- One delegation entry: call `delegate_token`; silent mode returns `data["token"]`, while interactive mode returns an authorization URL and later completes on your server.
- One management path: call `genauth.users.*`, and the SDK exchanges AK/SK for a GenAuth management token before calling GenAuth v3 users APIs.
- Capability-first namespaces: `genauth`, `gumem`, `do_anything`, `web_search`, `deep_research`, and `track`.
- Readable scope strings for least-privilege authorization.
- Typed exceptions with `request_id`, `trace_id`, `audit_id`, and `retryable`.

## Installation

```bash
pip install qoni
```

Requirements:

- Python 3.9 or later.
- Qoni `access_key` and `secret_key` created in Qoni Console.
- For silent runtime product calls, a real GenAuth user id from the userpool bound to the Qoni credential. For smoke tests, call `qoni.resolve_any_bound_user()` to grab the first bound user; in application code, resolve it with `current_user` or your existing server-side user session.
- For `genauth.users.*` management calls, no user id is required. The SDK uses AK/SK to request a GenAuth management token from Qoni.
- Optional `host` for private or local Qoni deployments. Leave it unset for hosted Qoni.

## Quick Start

```python
import os

from qoni import Qoni

qoni = Qoni(
    access_key=os.environ["QONI_ACCESS_KEY"],
    secret_key=os.environ["QONI_SECRET_KEY"],
)
```

### GenAuth User Management

`genauth.users.*` is a management-plane capability. It does not need a user id or `delegate_token`; the SDK exchanges AK/SK for a GenAuth management token internally.

```python
users = qoni.genauth.users.list(page=1, limit=20)
print("GenAuth users:", users.data)

created = qoni.genauth.users.create(
    username="sdk-demo",
    password=os.environ["GENAUTH_DEMO_USER_PASSWORD"],
)

profile = qoni.genauth.users.get(user_id=created.data["userId"])

qoni.genauth.users.update(
    user_id=profile.data["userId"],
    nickname="SDK demo user",
)

# Optional smoke-test cleanup:
# qoni.genauth.users.delete_batch(user_ids=[profile.data["userId"]])
```

### Runtime Product Delegation

GUMem, Do Anything, Web Search, Deep Research, and Track act for an end user. Silent delegation calls need a real GenAuth user id, then a `delegate_token` result:

```python
user_id = os.environ["QONI_USER_ID"]

# `products` is per-product authorization sugar; `agent` defaults to "sdk".
delegation = qoni.delegate_token(user_id=user_id, products=["do_anything"])
token = delegation.data["token"]
```

Interactive delegation returns an authorization URL instead; complete it on your server after the user authorizes:

```python
started = qoni.delegate_token(
    mode="interactive",
    redirect_uri="https://example.com/callback",
    state="opaque-state",
    products=["do_anything"],
)
print("send the user to:", started.data["authorizationUrl"])

# later, in the redirect handler:
completed = qoni.complete_delegate_token(
    grant_id=started.data["grantId"],
    code=code_from_callback,
    state="opaque-state",
)
token = completed.data["token"]
```

### Do Anything

```python
from qoni import CaptureOptions, QoniEventTypes

run = qoni.do_anything.run(
    token=token,  # passed once — the handle holds it; handle methods never take a token
    prompt="Open https://en.wikipedia.org/wiki/Singapore and summarize the country's key facts.",
    capture=CaptureOptions(screenshots=True),
)

# Stream semantic events (a generator; ends at the terminal event):
for event in run.events():
    if event.type == QoniEventTypes.PROGRESS:
        print("progress:", event.data)
    elif event.type == QoniEventTypes.MESSAGE:
        print(f"[{event.data['role']}] {event.data['text']}")
    elif event.type == QoniEventTypes.SCREENSHOT:
        open(f"step-{event.data.get('step', 0)}.png", "wb").write(event.image.data)
    elif event.type == QoniEventTypes.DONE:
        print("done:", event.data["terminal_reason"])

# Or drive the run to a settled result in one call:
result = run.wait(timeout=600)
print(result.status, result.output)

# Reuse the same browser session for a follow-up run:
follow_up = qoni.do_anything.run(
    token=token,
    prompt="Now open the History section and summarize it.",
    session=run.session_ref,
)
```

`wait(timeout=...)` interrupts HTTP/1.x response reads even when SSE is silent,
sends only comments, or never finishes a frame. Requests and reconnect backoff
share the remaining budget. It does not cancel the server-side run or interrupt
synchronous user callbacks. The response-scoped deadline timer is stopped and
joined before its connection is released; shared client defaults are unchanged.
Multiplexed HTTP/2 and opaque custom streaming transports are rejected for timed
waits because they cannot be safely interrupted through the synchronous API.

A terminal event is only a signal to fetch authoritative run detail. `wait()`
propagates detail errors unchanged and requires an explicit terminal detail
status; it never substitutes the lean terminal event payload for a result.

Malformed, empty, or non-image screenshot data URIs become `progress` events
with the original wire frame on `event.raw`; they do not interrupt `wait()` or
invoke `on_screenshot`. Valid percent-encoded image bytes are preserved.

Reconnect to a run later from anywhere:

```python
run = qoni.do_anything.attach(run_id, token=token)
print(run.status().status)
run.cancel("no longer needed")  # idempotent — terminal runs return their state
```

### Human-in-the-loop interactions

When a run needs the user (site login, clarification, confirmation, take-control, wait), it emits an `interaction` event carrying a typed `Interaction`. Act on it via the handle's declared-action methods:

```python
def on_interaction(handle, event):
    if handle.type == "clarification":
        print("agent asks:", handle.interaction.payload["question"])
        handle.answer("Use the first option.")
    elif handle.type == "site_login":
        for site in handle.interaction.payload["sites"]:
            print("sign in at:", site["login_url"])
        handle.confirm_signed_in()

result = run.wait(on_interaction=on_interaction)
```

Calling a method the backend did not declare raises `QoniValidationError` — check `handle.can(kind)` first if unsure.

### Web Search

```python
search = qoni.web_search.run(
    token=qoni.delegate_token(user_id=user_id, products=["web_search"]).data["token"],
    prompt=["latest Qoni SDK release", "Qoni delegation model"],
    max_results_per_query=5,
)
result = search.wait(timeout=300)
print(result.output)
```

### Deep Research

```python
research = qoni.deep_research.run(
    token=qoni.delegate_token(user_id=user_id, products=["deep_research"]).data["token"],
    prompt="The state of server-side agent authorization in 2026",
    depth="standard",
)
result = research.wait(timeout=3600)
for artifact in result.artifacts:
    open(artifact.name or f"{artifact.id}.md", "wb").write(artifact.content())
```

### Track (monitors)

```python
monitor = qoni.track.create(
    token=qoni.delegate_token(user_id=user_id, products=["track"]).data["token"],
    prompt="Watch the pricing page of example.com and alert me on changes.",
)

monitor.run_now()
for event in monitor.events():  # resident stream — break when done observing
    if event.type == QoniEventTypes.TRIGGERED:
        print("change detected:", event.data)
        break

monitor.pause()
monitor.refine(schedule={"kind": "interval", "interval_seconds": 3600})
monitor.resume()
print(monitor.runs(limit=10))
monitor.delete()
```

### GUMem memory

```python
gumem_token = qoni.delegate_token(
    user_id=user_id,
    scopes=["gumem.memory:read", "gumem.memory:write"],
).data["token"]

qoni.gumem.create_session(token=gumem_token, session_id="demo", title="Demo")
qoni.gumem.add_messages(
    token=gumem_token,
    session_id="demo",
    messages=[{"role": "user", "content": "I prefer aisle seats."}],
)
context = qoni.gumem.recall(token=gumem_token, session_id="demo", query="seating preference")
print(context.data)
```

## Error handling

Every failure raises a typed exception from one hierarchy:

```python
from qoni import (
    QoniError,               # base — code / status / request_id / trace_id / audit_id / retryable / body
    QoniAuthError,           # 401
    QoniPermissionDeniedError,  # 403 / missing scopes (message lists known scopes)
    QoniValidationError,     # 400 / 422 / local pre-validation
    QoniRateLimitError,      # 429 (retryable)
    QoniUpstreamError,       # 5xx / network (retryable)
    QoniTimeoutError,        # request or wait() timeout (retryable)
    QoniTokenExpiredError,
    QoniDelegationRequiredError,
)

try:
    qoni.do_anything.run(token=token, prompt="...")
except QoniPermissionDeniedError as err:
    print(err.code, err.status, err.request_id)
except QoniError as err:
    if err.retryable:
        ...  # safe to retry with backoff
```

Local pre-validation fails fast with actionable messages: unknown products, malformed scopes, missing user ids, and unsupported run options are raised before any request is made.

## Scopes

```python
from qoni import QoniScopes, QONI_PRODUCT_SCOPES, QONI_SCOPE_BUNDLES

QoniScopes.DO_ANYTHING_MANAGE      # "webagent.do_anything:manage"
QONI_PRODUCT_SCOPES["do_anything"] # ("webagent.do_anything:read", "webagent.do_anything:manage")
QONI_SCOPE_BUNDLES["GUMEM_READONLY"]
```

`delegate_token(products=[...])` accepts `"do_anything"`, `"web_search"`, `"deep_research"`, `"track"` and expands each to its read + manage pair.

## Event model

`run.events()` and `run.wait(on_event=...)` deliver semantic `RunEvent` objects. Match on `event.type` against `QoniEventTypes` constants:

| type | `event.data` |
| --- | --- |
| `PROGRESS` | human-readable line (str) |
| `MESSAGE` | `{"text": str, "role": str}` |
| `INTERACTION` | typed `Interaction` (act via `run.interaction_handle(...)`) |
| `SCREENSHOT` | `{"page_url", "step"}` — decoded image on `event.image` |
| `DONE` | `{"output", "succeeded", "terminal_reason"}` (terminal) |
| `RESULTS_READY` | result count (int, web search) |
| `PHASE` / `SECTION_READY` | phase name / section title (deep research) |
| `MONITOR_CREATED` / `TRIGGERED` / `CHECK_COMPLETED` | monitor id / change summary / bool (track) |

All internal wire churn folds into `PROGRESS`; the original wire frame stays on `event.raw`. Dropped SSE connections reconnect automatically with `Last-Event-ID` catch-up (`sse_max_retries`, default 5).

## Notes

- The public package and API use the Qoni brand. The deployed server protocol has not migrated yet, so internal signed routes remain under `/api/v3/eak/*`, the delegation token claim remains `eak_delegation_token`, and existing `eak.*` backend error codes are surfaced unchanged. Applications should use the public Qoni API and must not construct these internal wire values themselves.
- The client is synchronous and thread-safe for independent calls; construct it once per process. It is also a context manager (`with Qoni(...) as qoni:`).
- The client honors proxy environment variables (`HTTP_PROXY` / `HTTPS_PROXY` / `ALL_PROXY` / `NO_PROXY`) by default. Behind a local SOCKS proxy, either add your gateway to `NO_PROXY`, install `httpx[socks]`, or pass `Qoni(..., trust_env=False)` to bypass system proxies entirely.

## License

MIT

## Do Anything file artifacts

After `wait()`, `result.artifacts` contains this run's persisted files, including when using `attach()` after reconnecting. Screenshots and recordings remain on their dedicated APIs. Temporary S3 URLs expire 900 seconds after signing; do not put them in public logs. Refreshing a link requires a valid delegation token with `webagent.do_anything:read`. Stored files remain subject to project retention.

```python
result = run.wait()
for artifact in result.artifacts:
    url = artifact.download_url
    url = artifact.refresh_download_url()  # mint a new link after expiry
    data = artifact.content()  # bytes, authenticated download

# Fetch existing run files without replaying events:
artifacts = client.do_anything.artifacts(token=token, run_id=run.id)
```

Wire methods: `client.do_anything.api.list_artifacts(...)` and `artifact_download_url(...)`.
