opal-agent-sdk (Python)

Async Python SDK for invoking Opal agents programmatically over a Personal Access Token (PAT).

Install

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

Python 3.10+. Async-only.

Quickstart

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/ for more (streaming, multi-turn chat, workflow agents, canvas access).

Configuration

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:

# 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

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

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

# 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

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

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.

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.

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

# 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

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/ — see its README for setup instructions.