Typed Python client for the Orloj v1 REST API: synchronous and asynchronous usage (httpx), Pydantic v2 models, SSE watch streams, cursor-based pagination, tasks.run(wait=True) with typed output, and a small graph-as-code authoring layer.
- Python 3.10+
uv add orloj-sdkThen import the orloj_sdk package (from orloj_sdk import OrlojClient). pip install orloj-sdk works too.
From a git checkout:
uv syncThe client reads defaults from the environment (same order as orlojctl):
| Setting | Constructor | Environment (first wins) |
|---|---|---|
| API token | api_token= |
ORLOJCTL_API_TOKEN, ORLOJ_API_TOKEN |
| Base URL | base_url= |
ORLOJCTL_SERVER, ORLOJ_SERVER |
If unset, the base URL defaults to http://127.0.0.1:8080 and the namespace defaults to default. HTTP max_retries defaults to 3 for production-friendly transient failure handling.
from orloj_sdk import OrlojClient
client = OrlojClient(api_token="your-token")
who = client.auth.whoami()
print(who.authenticated, who.username)
for agent in client.agents.list_all():
print(agent.metadata.name)from pydantic import BaseModel
from orloj_sdk import OrlojClient
class Answer(BaseModel):
summary: str
client = OrlojClient(api_token="your-token")
answer = client.tasks.run(
name="job-1",
system="demo-pipeline",
input={"query": "hello"},
wait=True,
output_type=Answer,
output_key="result",
)
print(answer.summary)from orloj_sdk import OrlojClient
from orloj_sdk.authoring import agent, edge, node, system
@agent("researcher", model_ref="default-model")
def researcher() -> None:
"""Research the query and list key facts."""
pipeline = (
system("demo-pipeline")
.add_agent(researcher)
.set_graph({"researcher": node()})
)
pipeline.apply(OrlojClient(api_token="your-token"))from orloj_sdk import AsyncOrlojClient
async with AsyncOrlojClient(api_token="your-token") as client:
page = await client.agents.list(limit=20)Orloj durability (leases, retries, idempotency, dead-letter) lives in the server. See docs/durability.md for guarantees, terminal phases, production server baseline, and SDK wait defaults. Temporal interop is intentionally deferred — Orloj remains the control-plane identity.
The SDK vendors Orloj’s OpenAPI under openapi/ (see openapi/PIN.json). CI fails if:
- hand-written models lag schema properties (
scripts/check_schema_parity.py), or - generated models under
src/orloj_sdk/models/_generated/are stale (scripts/generate_models.py).
Workflow: sync from orloj → regenerate → fix parity → commit. Details in CONTRIBUTING.md.
Resource accessors on OrlojClient / AsyncOrlojClient:
| Attribute | Path prefix |
|---|---|
agents |
/v1/agents |
agent_systems |
/v1/agent-systems |
model_endpoints |
/v1/model-endpoints |
tools |
/v1/tools |
secrets |
/v1/secrets |
sealed_secrets |
/v1/sealed-secrets (+ get_public_key) |
memories |
/v1/memories |
context_adapters |
/v1/context-adapters |
eval_datasets |
/v1/eval-datasets |
eval_runs |
/v1/eval-runs (start/cancel/export/finalize/annotate/compare) |
agent_policies |
/v1/agent-policies |
agent_roles |
/v1/agent-roles |
tool_permissions |
/v1/tool-permissions |
tool_approvals |
/v1/tool-approvals |
task_approvals |
/v1/task-approvals |
tasks |
/v1/tasks (run/wait/output, logs/messages/metrics/cancel/retry/watch) |
task_schedules |
/v1/task-schedules |
task_webhooks |
/v1/task-webhooks |
workers |
/v1/workers |
mcp_servers |
/v1/mcp-servers |
auth |
auth/tokens/health/capabilities (cli_token, …) |
events |
/v1/events/watch |
a2a |
A2A registry, agent cards, JSON-RPC invoke |
Runnable scripts (require a running Orloj API):
| Script | Description |
|---|---|
examples/quickstart.py |
whoami and list agents |
examples/watch_tasks.py |
SSE watch on a task until completion |
examples/multi_agent_system.py |
Create an AgentSystem with a graph |
examples/authoring_system.py |
Graph-as-code authoring + apply |
examples/async_example.py |
AsyncOrlojClient with asyncio.gather |
export ORLOJ_API_TOKEN=...
uv run python examples/quickstart.pySee CONTRIBUTING.md for setup, checks, and pull request expectations.
uv sync
uv run ruff check src/orloj_sdk tests examples
uv run ruff format src/orloj_sdk tests examples
uv run mypy src/orloj_sdk
uv run pytest --cov=orloj_sdkApache-2.0