Python SDK

Use the sync and async Python clients to manage Orca and consume run streams.

Orca has three Python packages:

• orcapods is the generated Conductor client packaged from sdks/python/src/orcapods.
• agent-orc-vfs is the VirtualFS workspace client in virtualfs/sdk/py, imported as agent_orc_vfs.
• agent-orc-vfs-openai is the VirtualFS sandbox protocol adapter in virtualfs/sdk/adapters/openai-agents, imported as agent_orc_vfs_openai.

The orcapods package is not yet published to PyPI. Install it from the repository source; when your deployment publishes it to a package index, install from there instead.

pip install ./sdks/python

For local development against the VirtualFS package:

pip install ./virtualfs/sdk/py
pip install ./virtualfs/sdk/adapters/openai-agents

---

Client setup

import os

from orcapods import OrcapodsClient

client = OrcapodsClient(
    base_url=os.getenv("ORCA_BASE_URL", "https://api.orcapods.ai"),
    api_key=os.getenv("ORCA_API_KEY"),
    timeout=120.0,
)

Useful client options:

Option	Purpose
base_url	Override the conductor URL
environment	Use OrcapodsClientEnvironment.DEFAULT
api_key	Bearer token sent as Authorization: Bearer ao_...; the tenant is derived from the key
headers	Add custom deployment headers
timeout	Default timeout in seconds
max_retries	Default retry count
httpx_client	Use a preconfigured httpx.Client
logging	Configure SDK logging

Per-request options use request_options:

profiles = client.profiles.list(
    request_options={
        "timeout_in_seconds": 10,
        "max_retries": 1,
        "additional_headers": {"X-Request-ID": "onboarding-001"},
    }
)

---

Create a profile

client.profiles.create(
    name="researcher",
    runtime="general",
    model="anthropic:claude-sonnet-4-6",
    system_prompt="You are a research assistant. Be concise and cite sources.",
    tools=["@default", "web_search", "web_extract"],
)

Runtime/model patterns:

Runtime	Model format	Typical use
general	anthropic:..., openai:..., google:..., groq:...	Multi-provider agents through the Vercel AI SDK sidecar
claude	Claude model names	Claude-specific sidecar behavior
codex	Codex/OpenAI coding model names	Coding-focused sidecar behavior

---

Submit a run

The API can mint runId and sessionId. The current generated Python signature still requires id and session_id, so pass empty strings when you want Orca to assign them. If you supply either identifier, it must be 1-128 characters and use only letters, numbers, underscores, or hyphens.

created = client.runs.create(
    id="",
    profile="researcher",
    session_id="",
    title="SDK onboarding",
    prompt="Explain what Orca profiles and runs are in three bullets.",
)

print(created.run_id, created.session_id)

Reuse a session by passing the previous session_id:

followup = client.runs.create(
    id="",
    profile="researcher",
    session_id=created.session_id,
    title="Follow-up",
    prompt="Now turn that into onboarding copy for a new user.",
)

---

Stream run events

The Conductor stream is Server-Sent Events where each data: payload is a JSON Event object. The generated Python client includes client.runs.stream(id=...), but the current generated stream schema is str while Orca sends JSON objects. For production event handling, use the SDK for run creation and an explicit httpx SSE reader for the stream.

import json
import os

import httpx

created = client.runs.create(
    id="",
    profile="researcher",
    session_id="",
    title="Streaming example",
    prompt="Write a short onboarding checklist.",
)

base_url = os.getenv("ORCA_BASE_URL", "https://api.orcapods.ai")
api_key = os.getenv("ORCA_API_KEY", "")

with httpx.stream(
    "GET",
    f"{base_url}/api/runs/{created.run_id}/stream",
    headers={"Accept": "text/event-stream", "Authorization": f"Bearer {api_key}"},
    timeout=300.0,
) as response:
    response.raise_for_status()
    for line in response.iter_lines():
        if not line.startswith("data:"):
            continue
        event = json.loads(line.removeprefix("data:").strip())
        if event["type"] in {"assistant", "result"}:
            print(event.get("message", ""), end="")
        elif event["type"] == "tool_call":
            print(f"\nTool: {event.get('toolName')}")
        elif event["type"] == "error":
            raise RuntimeError(event.get("message", "run failed"))

Use GET /api/runs/{runId}/events when you want the persisted newline-delimited event log instead of a live stream.

---

Async client

The async client mirrors the sync client. Use it in web services and async worker processes.

import asyncio
import os

from orcapods import AsyncOrcapodsClient

client = AsyncOrcapodsClient(
    base_url=os.getenv("ORCA_BASE_URL", "https://api.orcapods.ai"),
    api_key=os.getenv("ORCA_API_KEY"),
)


async def main() -> None:
    await client.misc.health()
    created = await client.runs.create(
        id="",
        profile="researcher",
        session_id="",
        title="Async run",
        prompt="Summarize Orca in one sentence.",
    )
    print(created.run_id)


asyncio.run(main())

---

Raw responses

Use .with_raw_response when you need headers, status codes, or the underlying httpx response metadata.

response = client.profiles.with_raw_response.list()

print(response.status_code)
print(response.headers)
print(response.data)

---

Error handling

All non-2xx responses raise ApiError or a generated subclass such as BadRequestError, ConflictError, or NotFoundError.

from orcapods.core.api_error import ApiError
from orcapods.errors import ConflictError

try:
    client.profiles.create(name="researcher", runtime="general")
except ConflictError:
    print("Profile already exists")
except ApiError as exc:
    print(exc.status_code)
    print(exc.body)

---

Common calls

client.misc.health()
client.misc.seed_orca()
client.misc.list_capability_bundles()

client.profiles.list()
client.sessions.list()
client.runs.list()
client.runs.retrieve(id="run-...")
client.runs.cancel(id="run-...")

client.skills.list()
client.mcp.list()
client.memory.get_bank()
client.storage.info()
client.topology.retrieve()
client.stats.summary()

See the SDK method map for the full generated surface.

---

VirtualFS workspace client

agent_orc_vfs.Workspace is the standalone VirtualFS HTTP client (sync + async), and agent_orc_vfs_openai.SandboxClient is the matching virtual-sandbox adapter for agent runtimes. Both packages live under virtualfs/sdk/ and target the /vfs/... HTTP API directly rather than the conductor.

See VirtualFS SDKs for installation, mount factories, typed file ops, OpenAI function tools, error mapping, and adapter usage.

Strict mount validation

Every mount the client declares is checked against the server’s live mount table. Call validate() (async) or validate_sync() (sync) before doing real work — typically right after constructing Workspace — so a misconfigured server fails fast instead of silently routing reads to the wrong store.

from agent_orc_vfs import Workspace, r2, MountMismatch, MountUnavailable

ws = Workspace(
    {
        "/": r2(bucket="agent-orc-prod", endpoint="https://<acct>.r2.cloudflarestorage.com"),
    },
    base_url=os.getenv("VFS_BASE_URL", "https://api.orcapods.ai"),
    token=os.environ["VFS_TOKEN"],
)

try:
    await ws.validate()
except MountMismatch as exc:
    # declared path missing on server, or kind/bucket/endpoint mismatch
    raise
except MountUnavailable as exc:
    # server has the mount but backend failed to initialize; exc carries reason
    raise

Validation raises:

• MountMismatch — a declared path is not registered on the server, or the backend kind / bucket / endpoint does not match what the server reports.
• MountUnavailable — the server has the mount but its backend failed to initialize at boot; the server-reported reason is included.

The r2() factory signature:

r2(
    bucket: str,
    *,
    endpoint: str = "",   # S3-compatible endpoint URL; empty = default AWS S3
    prefix: str = "",     # optional key prefix inside the bucket
) -> R2Mount