Metadata-Version: 2.4
Name: opal-agent-sdk
Version: 0.5.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.10
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 over a Personal Access Token (PAT).

## 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 |
| `retry_max_attempts` | — | `3` | Max retry attempts for idempotent 5xx |
| `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.

## 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["content"], end="", flush=True)
```

## Multi-turn Chat

```python
async with client.agents.specialized.chat_stream(
    agent_id="your-agent-uuid",
) as chat:
    async for event in chat.send("Hello"):
        ...  # handle events

    async for event in chat.send("Tell me more"):
        ...  # chat.memory_id and chat.etag are managed automatically
```

## 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).

## 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.
