> ## Documentation Index
> Fetch the complete documentation index at: https://docs.orcapods.ai/llms.txt
> Use this file to discover all available pages before exploring further.

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

<Note>
  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.
</Note>

```bash theme={null}
pip install ./sdks/python
```

For local development against the VirtualFS package:

```bash theme={null}
pip install ./virtualfs/sdk/py
pip install ./virtualfs/sdk/adapters/openai-agents
```

***

## Client setup

```python theme={null}
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`:

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

***

## Create a profile

```python theme={null}
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.

```python theme={null}
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`:

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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`.

```python theme={null}
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

```python theme={null}
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](/sdk/reference) 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](/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.

```python theme={null}
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:

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