opal-agent-sdk 0.2.1

Async Python SDK for invoking Opal agents over PAT-authenticated REST and socket.io.

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
import os
from opal_agent_sdk import OpalClient, OpalConfig, PATAuth

async def main() -> None:
    async with OpalClient(
        auth=PATAuth(os.environ["OPAL_PAT"]),
        config=OpalConfig(instance_id="your-instance-id"),
    ) 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

Parameter	Env var	Default	Description
base_url	OPAL_BASE_URL	https://api.opal.optimizely.com	API Gateway URL
instance_id	OPAL_INSTANCE_ID	(required)	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

You can also use OpalConfig.from_env() to load from environment variables.

Streaming

async with OpalClient(auth=PATAuth(pat), config=config) 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.