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 Agents and Skills

Agents

# List + iterate
page = await client.agents.list(type="specialized")
for agent in page:
    print(agent.agent_id, agent.name)
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")

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/"))

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

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.