OpenRath v2.0.0 Agent Server and Remote Client

The Agent Server turns a registered durable workflow into tenant-scoped HTTP and SSE resources. RemoteClient and AsyncRemoteClient provide the matching Python client surface.

Workflow + immutable revision
             |
             v
        AgentServer
             |
     +-------+--------+
     |                |
HTTP resource API     SSE Run events
     |
RemoteClient / AsyncRemoteClient

The /v1 prefix is the version of the Agent Server HTTP contract. The product release documented on this page is Stable OpenRath v2.0.0.

Install the server profile

Install the exact Stable release with the Agent Server dependencies:

python -m pip install \
  "openrath[server] @ https://github.com/Rath-Team/OpenRath/releases/download/v2.0.0/openrath-2.0.0-py3-none-any.whl"

The embedded example below uses SQLite. For a PostgreSQL deployment, install openrath[server,postgres] and follow the operations guide.

Create an embedded Agent Server

Save this application as server_app.py:

from pathlib import Path
from uuid import UUID

from rath.definition import EffectClass, step
from rath.flow import Workflow
from rath.runtime import LocalRuntime, SQLiteRunStore
from rath.security import (
    Principal,
    PrincipalKind,
    SecurityContext,
    StructuredAuditSink,
)
from rath.server import AgentServer, StaticTokenAuth
from rath.session import Session


class EchoWorkflow(Workflow):
    @step(entry=True, effects=EffectClass.READ_ONLY)
    def echo(self, state, context):
        return {**state, "completed": True}

    # The v1 callable façade remains available in v2.0.0.
    def forward(self, session: Session) -> Session:
        return session


store = SQLiteRunStore(Path("agent-server.db"))
runtime = LocalRuntime(store)
security = SecurityContext(
    principal=Principal(id="local-client", kind=PrincipalKind.SERVICE),
    tenant_id="local",
    grants=frozenset(
        {
            "assistant.read",
            "session.create",
            "session.read",
            "run.create",
            "run.read",
            "run.cancel",
            "run.resume",
            "interrupt.read",
            "interrupt.decide",
            "feedback.create",
            "metrics.read",
        }
    ),
)
server = AgentServer(
    store,
    runtime,
    auth=StaticTokenAuth({"replace-this-local-token": security}),
    audit_sink=StructuredAuditSink(),
    embedded_worker=True,
)
server.register_assistant(
    "echo",
    EchoWorkflow(),
    revision_id=UUID("00000000-0000-4000-8000-000000000001"),
)
app = server.app

StaticTokenAuth is the reference bearer-token provider for self-hosted deployments and tests. An integrated deployment can implement the AuthProvider protocol and produce its own SecurityContext.

Start the application:

openrath-server --app server_app:app --host 127.0.0.1 --port 8000

The server starts one embedded durable worker because the application sets embedded_worker=True.

Submit and inspect a Run

In another terminal, save this as client.py:

from time import sleep

from rath.client import RemoteClient


client = RemoteClient(
    "http://127.0.0.1:8000",
    token="replace-this-local-token",
)
try:
    session = client.create_session()
    run = client.create_run(
        assistant_id="echo",
        session_id=session["id"],
        state={"message": "hello from the Agent Server"},
        idempotency_key="agent-server-quickstart-1",
    )

    for _ in range(50):
        run = client.get_run(run["id"])
        if run["status"] not in {"queued", "running"}:
            break
        sleep(0.1)
    else:
        raise TimeoutError("the Run did not finish")

    assert run["status"] == "succeeded"
    print(run["state"])
    for event in client.events(run["id"]):
        print(event["sequence"], event["type"])
finally:
    client.close()

Run it:

python client.py

AsyncRemoteClient exposes the same resource operations as native async methods. Its events(...) method asynchronously iterates the durable event replay endpoint.

Follow Run events over SSE

The SSE endpoint accepts an event cursor through after or Last-Event-ID. With follow=true, it waits for new events and exits after a terminal Run has no more events to emit:

curl -N \
  -H "Authorization: Bearer replace-this-local-token" \
  "http://127.0.0.1:8000/v1/runs/RUN_ID/stream?after=0&follow=true"

Each event contains a durable sequence number, type, timestamp, Run ID, and JSON data. A client can reconnect with the last processed sequence without replaying earlier events.

Service endpoints

Endpoint	Purpose
GET /health/live	Process liveness.
GET /health/ready	Durable-store readiness.
GET /info	Server capabilities and bounds.
GET /openapi.json	Generated Agent Server contract.
GET /metrics	Prometheus metrics; requires metrics.read.
/v1/assistants	Deployment templates and tenant aliases.
/v1/sessions	Durable Session resources.
/v1/runs	Create, list, inspect, cancel, and resume Runs.
/v1/runs/{run_id}/events	Ordered event replay.
/v1/runs/{run_id}/stream	Cursor-resumable SSE events.
/v1/interrupts	Pending approvals and decisions.
/v1/feedback	Run feedback records.
/v1/store	Governed Memory operations when configured.

The full request and response schemas are in the OpenAPI document.

Server command-line tools

Command	Contract
openrath-server --app module:app	Runs the ASGI application. OpenRath uses one process worker and scales through service replicas.
openrath-worker --app module:server	Runs a separate durable worker for an AgentServer object.
openrath-migrate	Applies the PostgreSQL schema with a DDL-capable identity.
openrath-migrate --check	Verifies the current PostgreSQL schema without applying changes.

For a separate worker deployment, construct the server with embedded_worker=False, export the AgentServer object as server, and run:

openrath-server --app server_app:app
openrath-worker --app server_app:server --worker-id worker-1

Continue with Operations for PostgreSQL migration, immutable images, backup and restore, rollout, and incident workflows.