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

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.

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:

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

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:

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 and examples/b8_oauth_run_agent.py, and the spec, docs/tech-spec/agent-framework/agent-sdk/auth-code-grant.md.

Token exchange (product backends)

An Optimizely product backend (CMS, CMP, Experimentation, Configured Commerce, Campaign, Analytics) already holds the signed-in user's Opti ID access token. TokenExchangeAuth trades it for an Agent SDK token for that user and the one Opal instance the product instances belong to:

from opal_agent_sdk import OpalClient, TokenExchangeAuth

auth = (
    TokenExchangeAuth.builder()
    .client_name("Optimizely CMS")  # required: your product, as authz-server's audit log shows it
    .optiid_token(user_access_token)  # the user's Opti ID access token (or a provider, below)
    .cms_instance_id(cms_instance_id)  # one or more instance ids
    .build()
)
identity = await auth.exchange()  # identity.instance_id is the Opal instance
async with OpalClient(auth=auth) as client:  # the instance comes from the token
    result = await client.agents.specialized.run("agent-id", parameters={"query": "Hi"})
await auth.aclose()

• The instance methods are cms_instance_id, cmp_instance_id, exp_instance_id, configured_commerce_instance_id, campaign_instance_id, analytics_instance_id and opal_instance_id. Each product instance is connected to exactly one Opal instance, so name product instances from one Opal instance. opal_instance_id names (or narrows to) that Opal instance, which the user must hold a role on; it cannot choose between Opal instances.
• client_name is required: it names the registered client, which is how authz-server's audit log attributes each exchange to a product (for example "Optimizely CMS"). Each name gets its own client, so two products on one host never share one.
• build() raises ValueError without a client name, without an Opti ID token or provider (or with both), without any instance id, or when an id is blank. base_url, issuer, client_store, verify_ssl and http_client are optional and mean what they do for AuthorizationCodeAuth.
• The first exchange registers a confidential client and keeps it in the client store (client|<issuer>|token-exchange|<client_name>; by default the shared credentials file). The client holds a secret, so treat the store as secret storage. On a server, pass a client_store (a MemoryTokenStore, or your own backed by a secret manager); a default store that cannot be written (read-only filesystem, no writable home directory) is tolerated with a warning, and the client is kept in memory and reused for the life of the process. The Agent SDK token stays in memory. It has no refresh token: the SDK exchanges the Opti ID token again shortly before it expires, and after a 401. When that early exchange fails (a 5xx, a network error, a refusal), requests keep using the held token while it is valid, and the SDK tries again after 5 seconds; only once the token has expired does a failed exchange raise.
• Using the auth before exchange() (its bearer_token, or OpalClient(auth=auth) without a config) raises OpalLoginRequiredError.
• A refusal raises OpalOAuthError with code="token_error" and the server's oauth_error: invalid_grant (the Opti ID token is invalid or expired), access_denied (the user has no Opal instance for those ids, or an id was named under the wrong product), or invalid_request (the ids belong to different Opal instances: name product instances from one Opal instance). A 5xx raises OpalServerError.

The server never issues a token that outlives the Opti ID token it was exchanged with. For an auth that lives as long as the user's session, pass optiid_token_provider instead of optiid_token: a function, sync or async, returning the user's current Opti ID token (your backend refreshes it). The SDK calls it whenever it needs to exchange again, never on every request:

auth = (
    TokenExchangeAuth.builder()
    .client_name("Optimizely CMS")
    .optiid_token_provider(lambda: session.optiid_access_token)  # or an async def
    .cms_instance_id(cms_instance_id)
    .build()
)

With a static optiid_token, the auth works until that Opti ID token expires (typically about an hour), then raises OpalOAuthError with code="token_error" and oauth_error="invalid_grant": build a new auth with a fresh token. A refused Opti ID token is remembered, so while the provider still returns it, the next exchange raises the same error at once, with no request. Keep one auth, and one OpalClient, per user and instance set rather than exchanging on every request.

See examples/b9_token_exchange.py.

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

    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.

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

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