Metadata-Version: 2.4
Name: opal-agent-sdk
Version: 0.8.0
Summary: Async Python SDK for invoking Opal agents over PAT-authenticated REST and socket.io.
Author-email: Optimizely <opal-team@optimizely.com>
License: MIT
Project-URL: Homepage, https://github.com/optimizely/opal-app
Project-URL: Documentation, https://github.com/optimizely/opal-app/blob/main/docs/tech-spec/agent-framework/agent-sdk/python-sdk.md
Project-URL: Source, https://github.com/optimizely/opal-app/tree/main/sdks/agent-sdk/python
Keywords: opal,agent,sdk,ai,llm,async
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Framework :: AsyncIO
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27
Requires-Dist: python-socketio[asyncio_client]>=5.14.0
Requires-Dist: pydantic>=2.6
Requires-Dist: pyyaml>=6.0
Provides-Extra: cli
Requires-Dist: typer>=0.12; extra == "cli"
Provides-Extra: dev
Requires-Dist: pytest>=8.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23.0; extra == "dev"
Requires-Dist: pytest-httpx>=0.30.0; extra == "dev"
Requires-Dist: pytest-cov>=5.0.0; extra == "dev"
Requires-Dist: ruff>=0.6.0; extra == "dev"
Requires-Dist: mypy>=1.10.0; extra == "dev"
Requires-Dist: types-PyYAML>=6.0; extra == "dev"
Requires-Dist: typer>=0.12; extra == "dev"
Requires-Dist: click>=8.0.0; extra == "dev"
Dynamic: license-file

# opal-agent-sdk (Python)

Async Python SDK for invoking Opal agents programmatically, with a Personal Access Token (PAT)
or by signing in with your browser.

## Install

```bash
pip install opal-agent-sdk            # core
pip install "opal-agent-sdk[cli]"     # adds the `opal` CLI
```

Python 3.10+. Async-only.

## Quickstart

```python
import asyncio, os
from opal_agent_sdk import OpalClient, PATAuth


async def main() -> None:
    async with OpalClient(auth=PATAuth(os.environ["OPAL_PAT"])) as client:
        result = await client.agents.specialized.run(
            agent_id="your-agent-uuid",
            parameters={"query": "Where is order #1234?"},
        )
        print(result.output_text)


asyncio.run(main())
```

See [`examples/`](./examples) for more (streaming, multi-turn chat, workflow agents, canvas access).

## Configuration

| Parameter | Env var | Default | Description |
|-----------|---------|---------|-------------|
| `base_url` | `OPAL_BASE_URL` | `https://opal.optimizely.com` | API Gateway URL |
| `instance_id` | `OPAL_INSTANCE_ID` | (auto-discovered from PAT) | Opal instance ID |
| `timeout_s` | — | `30.0` | Per-request timeout in seconds. A run's event stream waits up to `max(timeout_s, 330)` seconds between bytes |
| `retry_max_attempts` | — | `3` | Max retry attempts for idempotent 5xx |
| `ws_connect_timeout_s` | — | `10.0` | How long one socket.io handshake may take |
| `verify_ssl` | — | `True` | SSL certificate verification. Set to `False` for localdev only — **do not disable in production** |

With PAT auth, the SDK extracts `instance_id`, `customer_id`, and `product_instances` from the JWT claims. Env vars are only used with `OpalConfig.from_env()` for non-PAT flows.

You can also use `OpalConfig.from_pat()` for explicit PAT-based config construction, or `OpalConfig.from_env()` to load from environment variables.

## Sign in with your browser

A person at a machine with a browser can skip the PAT. `AuthorizationCodeAuth`
signs them in with Opti ID (OAuth 2.1 authorization code grant with PKCE),
keeps the session in a private file, and refreshes it for up to seven days:

```python
from opal_agent_sdk import AuthorizationCodeAuth, OpalClient

async with AuthorizationCodeAuth.from_env() as auth:  # reads OPAL_BASE_URL
    await auth.login()  # opens the browser only when needed
    async with OpalClient(auth=auth) as client:  # instance and customer come from the token
        result = await client.agents.specialized.run("agent-id", parameters={"query": "Hi"})
```

- `login()` opens the browser only when there is no usable session;
  `login(force=True)` always does (for example, to pick another instance).
- `resume()` never opens a browser: it loads and refreshes the stored session,
  or raises `OpalLoginRequiredError`. Use it in scripts.
- `logout()` revokes the session on the server and deletes it locally.
- An auth stays with the identity (issuer, user, instance) it started with. If
  another sign-in stores a session for someone or somewhere else under the same
  profile (`login(force=True)` picking another instance, say), this auth raises
  `OpalSessionIdentityError` on its next token call rather than switching under
  a live client. Create a new `AuthorizationCodeAuth` and `OpalClient` for that
  session. **Use one `profile` per instance or user** to hold several at once:
  `AuthorizationCodeAuth.from_env(profile="commerce")`.
- Web apps use `authorization_url()` and `exchange_code()` with a pre-registered
  `client_id`; see `register_public_client()` in `opal_agent_sdk.oauth`, and
  [Web apps](#web-apps) below.
- The session file is `credentials.json` in `OPAL_CONFIG_DIR`, else your per-user
  config directory (`~/.config/opal-agent-sdk` on Linux). It is shared with the
  other Opal SDKs on the machine. Pass `session_store=` (and `client_store=`) to
  keep it elsewhere; see below.

### Web apps

One web server signs in many people, so each person needs their own session.
Give every user their own `profile` (or their own `session_store`):

```python
auth = AuthorizationCodeAuth(
    base_url="https://opal.optimizely.com",
    client_id=CLIENT_ID,  # from register_public_client()
    profile=f"user-{user.id}",  # one session per user
)
request = await auth.authorization_url("https://app.example.com/opal/callback")
# keep request.to_dict() in the user's server-side session, redirect to request.url;
# then, on the callback:
await auth.exchange_code(AuthorizationRequest.from_dict(saved), callback_url)
# later requests for that user: the same profile, then resume()
```

With the default store and the default profile, every user's sign-in would go
to the one shared session, the last one would win, and a later `resume()` for
one user could act as another. So `authorization_url()` and `exchange_code()`
raise `ValueError` in that configuration. A web app that genuinely has a single
user can pass `allow_shared_store=True`. `login()` and `resume()` (the
browser-on-this-machine flow) are not affected.

### Custom credential store

A sign-in keeps two things, and each has its own store:

- the **registered client** (`client|<issuer>|loopback`), which is no secret:
  a `ClientRegistrationStore`;
- the **session** (`session|<issuer>|<profile>`), which holds the tokens: a
  `SessionStore`, optionally with a lock (`LockingSessionStore`).

`FileTokenStore` (the default) and `MemoryTokenStore` are both kinds, and so is
anything else with the same three methods. Which store gets what:

- **Neither given:** one `FileTokenStore` holds both, in one file.
- **Only `session_store=`:** it holds the registration too.
- **Only `client_store=`:** the session goes to the default file.
- **Both:** each goes only to its own store.

This keeps the tokens in the OS keychain and the registration in the shared
file:

```python
from contextlib import asynccontextmanager
import asyncio
import keyring  # any backend works; keyring is only an example
import keyring.errors

from opal_agent_sdk import AuthorizationCodeAuth
from opal_agent_sdk.oauth import FileTokenStore


class KeyringSessionStore:
    """A SessionStore: three async methods over opaque strings."""

    async def load(self, key: str) -> str | None:
        return await asyncio.to_thread(keyring.get_password, "opal-agent-sdk", key)

    async def save(self, key: str, value: str) -> None:
        await asyncio.to_thread(keyring.set_password, "opal-agent-sdk", key, value)

    async def delete(self, key: str) -> None:
        try:
            await asyncio.to_thread(keyring.delete_password, "opal-agent-sdk", key)
        except keyring.errors.PasswordDeleteError:
            pass  # already gone

    # Optional: makes it a LockingSessionStore (see below).
    @asynccontextmanager
    async def lock(self, key: str):
        yield


auth = AuthorizationCodeAuth.from_env(
    session_store=KeyringSessionStore(),
    client_store=FileTokenStore(),
)
```

The contract, for both kinds:

- **Values are opaque strings.** The SDK owns their format (JSON). A store only
  keeps them. A session's value holds tokens, so a session store must protect it
  at rest.
- **`load()` returns `None` for a missing key**, and `delete()` of a missing key
  is not an error.
- **The methods are async.** Do blocking I/O in a thread (`asyncio.to_thread`)
  so the event loop keeps running.
- **Locking is optional, only on the session store, and serialises refreshes
  across processes.** Every refresh spends the refresh token and the server
  revokes the old one, so two processes must not refresh the same session at
  once. A session store with an `async` context manager `lock(key)` gets every
  read-refresh-write, and every write (the registration's too), run inside it.
  The SDK never nests it and never holds it while a browser sign-in waits. It
  must exclude every process that shares the store; `FileTokenStore` uses a
  lock file. Without a lock, the SDK still runs one refresh at a time per
  process, and before it gives up on a session it re-reads the store in case
  another process rotated it first. A client store is never asked for a lock.

**Use a PAT instead** for anything that runs without a person: servers, CI,
scheduled jobs, daemons. A browser sign-in needs someone at a browser, and its
refresh token dies after seven days however often it is used.

See [`examples/b7_oauth_login.py`](./examples/b7_oauth_login.py) and
[`examples/b8_oauth_run_agent.py`](./examples/b8_oauth_run_agent.py), and the spec,
`docs/tech-spec/agent-framework/agent-sdk/auth-code-grant.md`.

## PAT Scoping

PATs are scoped to a specific Opal instance at creation time. Optionally, they can also
be scoped to specific product connections (CMP, EXP, CMS, etc.).

The SDK reads the instance scope directly from the PAT's JWT claims — no separate
`OPAL_INSTANCE_ID` configuration is needed:

```python
# Instance auto-discovered from PAT — no config required
async with OpalClient(auth=PATAuth(os.environ["OPAL_PAT"])) as client:
    ...

# Explicit config (overrides PAT claims)
from opal_agent_sdk import OpalConfig

async with OpalClient(
    auth=PATAuth(os.environ["OPAL_PAT"]),
    config=OpalConfig.from_pat(
        os.environ["OPAL_PAT"], base_url="https://opal-localdev.optimizely.com"
    ),
) as client:
    ...
```

Create PATs in **Settings > Developer > Personal Access Tokens** in the Opal UI.

## Streaming

```python
async with OpalClient(auth=PATAuth(pat)) as client:
    async for event in client.agents.specialized.stream(
        agent_id="your-agent-uuid",
        parameters={"query": "Summarize this article"},
    ):
        if event.event_type == "response_chunk":
            print(event.payload.get("content", ""), end="", flush=True)
```

Event payloads are passed through exactly as the server sends them: text is
under `content`, a `tool_call` carries `name` and `input_params`, and its
`execution_id` is the tool call's id, which the matching `tool_result` shares.

When an execution fails, `run()`, `chat().start()` and `chat().send()` raise
`OpalServerError`, with the server's `error_type` as `code` and the event
payload as `payload`.
A run that fails after it has started ends with an `execution_complete` whose
`status` is not `COMPLETED`, so streaming callers check `status` there:

```python
    if event.event_type == "execution_complete" and event.payload["status"] != "COMPLETED":
        print(f"failed: {event.payload.get('error')}")
```

## Multi-turn Chat

A chat needs an agent with `conversation_mode` on. Its first turn carries the
agent's declared inputs, as a `run()` does; every `send()` after that is a user
message. How the chat starts depends on the agent:

- **With a prompt template** (a "seeded" agent), `start()` runs turn 1 from the
  parameters and returns the agent's reply. Call it before `send()`.
- **With an empty template** (an "open chat"), turn 1 only opens the
  conversation, with no model call. `start()` returns a turn with empty text,
  and a first `send()` without `start()` opens it for you.

```python
chat = client.agents.specialized.chat("your-agent-uuid", parameters={"search_query": "lighthouses"})
first = await chat.start()  # a seeded agent's reply; "" for an open chat
print(first.output_text)
turn = await chat.send("Summarize that in one sentence.")
print(turn.output_text)  # chat.memory_id and chat.etag are managed automatically

# Streaming: start() still returns turn 1 aggregated; send() yields events.
async with client.agents.specialized.chat_stream("open-chat-agent-uuid") as chat:
    async for event in chat.send("Hello"):  # opens the conversation first
        ...
```

A single-turn agent has no chat: `start()` and `send()` raise
`OpalUnsupportedError`; use `run()`. `parameters` apply to a new chat only; to
continue a saved one, pass `memory_id` (and `etag`) and call `send()`.

A `start()` that fails after the server created the conversation leaves the
chat holding its `memory_id`, so `start()` again raises `OpalUnsupportedError`
("chat already started"). To try again, call `reset()`, then `start()`.

## Catalog: list / export / import / retire Agents and Skills

### Agents

```python
# List + iterate. Each row carries its creator: `created_by` is the raw
# Turnstile id, `creator` the resolved name + email (one POST /users/lookup
# per page). Pass resolve_users=False to skip that call.
page = await client.agents.list(type="specialized")
for agent in page:
    print(
        agent.agent_id,
        agent.name,
        agent.creator.display_name if agent.creator else agent.created_by,
    )
async for agent in client.agents.iter(type="all"):
    print(agent.agent_id)

# Export — single agent (returns dict, optionally writes file)
doc = await client.agents.export("agent-id", type="specialized")
await client.agents.export("agent-id", type="specialized", output_file=Path("/tmp/a.json"))

# Export — recursive workflow tree (returns AgentExportTree)
tree = await client.agents.export("wf-id", type="workflow", recursive=True)
await client.agents.export(
    "wf-id",
    type="workflow",
    recursive=True,
    output_file=Path("/tmp/agents/"),
)

# Import — single agent, or recursive tree (topological sort + cycle detection)
result = await client.agents.import_(doc, type="specialized")
results = await client.agents.import_tree(Path("/tmp/agents/"))

# Shares
grants = await client.agents.list_shares("agent-id", type="specialized")

# Rename — a partial update, so this touches nothing but the name
await client.agents.update("agent-id", type="specialized", name="ARCHIVE-Weekly Digest")

# Retire, reversibly. There is no delete: deactivation is what a retention
# policy needs, and the irreversible step stays a human action in the UI.
await client.agents.deactivate("agent-id", type="specialized")
await client.agents.activate("agent-id", type="specialized")
```

### Skills

```python
# List
page = await client.skills.list(scope="org")
for skill in page:
    print(skill.skill_guid, skill.title)

# Export — JSON envelope, SKILL.md, or whole plugin bundle
doc = await client.skills.export("skill-guid", scope="org")
await client.skills.export(
    "skill-guid",
    scope="org",
    format="skill_md",
    output_file=Path("/tmp/skills/"),  # directory; SKILL.md is a directory artifact
)
result = await client.skills.export_as_plugin(
    ["g1", "g2"],
    scope="org",
    plugin_name="my-plugin",
    plugin_description="...",
    output_dir=Path("/tmp/plugin/"),
)

# Import — JSON, SKILL.md, or whole plugin bundle
r = await client.skills.import_(doc, scope="org")
rs = await client.skills.import_plugin(Path("/tmp/plugin/"))

# Retitle and retire. A skill's title is its name; the body and trigger keep
# round-tripping through export / import. No delete here either.
await client.skills.update("skill-guid", scope="org", title="ARCHIVE-Brand Voice")
await client.skills.deactivate("skill-guid", scope="org")
await client.skills.activate("skill-guid", scope="org")
```

Putting those together — the retention sweep this surface was built for.
**Collect the whole catalog first, then mutate**; do not mutate inside the
`iter()` loop:

```python
cutoff = datetime.now(timezone.utc) - timedelta(days=60)

# Step 1 — read. Snapshot the catalog before changing anything.
stale = [
    agent
    async for agent in client.agents.iter(type="all")
    if agent.created_at and agent.created_at < cutoff and is_stale(agent)
]

# Step 2 — write, against the snapshot.
for agent in stale:
    owner = agent.creator.email if agent.creator else agent.created_by
    await client.agents.update(agent.agent_id, type=agent.type, name=f"ARCHIVE-{agent.name}")
    await client.agents.deactivate(agent.agent_id, type=agent.type)
    notify(owner, agent)
```

The two-step shape is not stylistic. `iter()` pages with an offset, and both
mutations move the ground under it: `deactivate` drops the row out of the
default (active-only) query, and `update` bumps `updated_at`, which the listing
sorts on descending. Mutating mid-iteration shifts later rows into windows that
have already been read, so the sweep silently skips agents — the same class of
quiet truncation as reading the wrong pagination field.

For the wire shapes and round-trip semantics, see
[`docs/tech-spec/agent-framework/agent-sdk/skills-agents-export-import.md`](../../../docs/tech-spec/agent-framework/agent-sdk/skills-agents-export-import.md).

## Agent Directory — browse and install published agents

`client.agents` is your instance's own catalog. `client.directory` is the
**Agent Directory**: the published, cross-region catalog you browse and install
*from*. Seven verbs, no `publish` — authoring happens in the source region and
the UI.

```python
# Discover. Tag names feed straight back into the filter.
tags = await client.directory.tags(limit=10)
page = await client.directory.browse(tags=["reporting"], search="digest", page_size=20)

for agent in page:
    # browse populates install state for YOUR instance, so one listing answers
    # both "what exists" and "what do I already have".
    print(agent.agent_id, agent.name, agent.is_installed, agent.installed_version)

# Full detail, including the dependencies an install would pull in.
detail = await client.directory.get("weekly-digest")
print(detail.release_notes, [d.agent_id for d in detail.dependencies])

# Add `include_agent_data=True` for the complete export JSON (the same envelope
# `agents.export()` returns). Off by default — it is the whole agent.
raw = (await client.directory.get("weekly-digest", include_agent_data=True)).agent_data

# Install, with dependencies, in topological order.
result = await client.directory.install("weekly-digest")
print(result.status, result.dependencies_installed)

# What this instance has, and what has moved on since.
for row in await client.directory.installations():
    print(row.agent_id, row.directory_version, row.is_active)

for update in await client.directory.check_updates():
    print(f"{update.agent_name}: v{update.current_version} -> v{update.latest_version}")
    await client.directory.install(update.agent_id)  # applying one is just an install
```

**`install` does not overwrite by default.** It matches the server (a missing
`force_update` reads as false) and the UI, which defaults false *and* confirms
before overwriting an installed agent. Installing over an existing installation
fails with HTTP 409 unless you opt in.

Pass `force_update=True` for idempotent provisioning. Two things follow: local
edits to an installed agent are overwritten with no undo, and since no pinned
`version` means *latest*, a downgrade only happens if you deliberately pin an
older one.

**Pass `user_id` when you have it.** Without it the server records
`installed_by` as null and creates the imported agent as the literal user
`"system"`, so an audit of who installed what is empty for SDK-driven
installs.

**The target instance is never a parameter.** It comes from your config, which
is what the server accepts anyway — it 403s unless the id matches the caller's
own instance.

**`featured()` returns a different type on purpose.** The server builds featured
rows without ever looking up install state, so `is_installed` and friends would
arrive as schema defaults — `False` even for an agent that *is* installed.
`DirectoryFeaturedAgent` has no such attributes, which turns a silently wrong
answer into an `AttributeError`. Use `browse(featured=True)` when you need
featured agents *with* trustworthy install state.

**`installations()` includes orphans.** A row whose local agent has since been
deleted reports `is_active=False` rather than vanishing — filter on it when you
want live agents only.

## Execution logs

Fetch an agent's past executions (the history the Opal UI shows), paginated:

Mirrors the Opal UI's Logs view: **`list`/`iter` return lightweight summaries**, and
**`get` returns the full detail** for one execution (so you don't pull large
inputs/outputs for every row).

```python
# Lightweight rows (OpalExecutionSummary) — newest first
page = await client.executions.list(agent_id="your-agent-uuid", type="specialized")
for row in page:
    print(row.execution_id, row.status, row.created_at, row.duration_seconds)
if page.has_more:
    nxt = await client.executions.list(agent_id="your-agent-uuid", type="specialized", page=2)

# Auto-paginate the full history (still lightweight summaries)
async for row in client.executions.iter(agent_id="your-agent-uuid", type="workflow"):
    print(row.execution_id, row.status)

# Full detail for one execution (OpalExecutionLog) — inputs, output, log, steps
detail = await client.executions.get(execution_id="exec-...", type="specialized")
print(detail.parameters, detail.result, detail.log)
```

`OpalExecutionSummary` (list rows) is bounded — the Logs-table metadata plus identifiers:
`execution_id`, `created_at`, `status`, `duration_seconds`, `agent_version`, `total_credits`,
`evaluation_score`, `user_id` (triggering user), `token_usage`, `llm_provider`,
`conversation_mode`. `OpalExecutionLog` (`get`) is a superset that adds the potentially-large
fields: `parameters` (input variables), `result` (output), `log`, and (workflow)
`step_executions`. The agent's display *name* is on neither — resolve it from `agent_id` via
`client.agents.list(...)`.

`type` is `"specialized"` or `"workflow"`. Filters: `status`, `start_date`, `end_date`,
`user_id`; `agent_version` (specialized only) and `search` (workflow agent-name search only).
Live event subscription for an in-flight execution is separate — see
`client.executions.subscribe(...)`.

## Development

```bash
make install        # pip install -e ".[dev,cli]"
make test           # pytest
make lint           # ruff check
make format         # ruff format
make typecheck      # mypy --strict
make check          # all of the above
```

## Demo App

A FastAPI + HTML demo app is included in [`demo/`](./demo/) — see its [README](./demo/README.md) for setup instructions.
