Trajectory Class
The core runtime class for creating, managing, and exporting durable agent trajectories.
Overview
The Trajectory class is the primary entry point for the Trajectory-IR SDK. It wraps an agent's execution context inside a DBOS durable workflow, providing crash-safe state management and cryptographic sealing of every decision.
Constructor
class Trajectory:
def __init__(
self,
tenant_id: str,
trajectory_id: Optional[str] = None,
backend: str = "dbos"
) -> None:Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
tenant_id | str | required | Unique identifier for the tenant/user owning this trajectory. |
trajectory_id | str | None | None | Optional UUID. Auto-generated if not provided. |
backend | str | "dbos" | The durable execution backend to use. Currently only "dbos" is supported. |
Class Methods
Trajectory.start()
Creates and initializes a new Trajectory instance with an active DBOS execution context.
@classmethod
def start(
cls,
tenant_id: str,
trajectory_id: Optional[str] = None,
) -> "Trajectory":Returns: A fully initialized Trajectory object.
Example:
from trajectory_ir.runtime import Trajectory
traj = Trajectory.start(tenant_id="demo-user")
print(traj.trajectory_id) # e.g., "a1b2c3d4-..."Trajectory.tool()
A decorator that registers a function as a durable tool with a specific EffectClass.
@staticmethod
def tool(
effect_class: EffectClass = EffectClass.PURE,
idempotency_key: Optional[str] = None,
) -> Callable:Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
effect_class | EffectClass | PURE | The safety classification for this tool. |
idempotency_key | str | None | None | Optional key for idempotent write operations. Auto-generated if not provided. |
Example:
from trajectory_ir.runtime import Trajectory
from trajectory_ir.effects import EffectClass
@Trajectory.tool(effect_class=EffectClass.NON_IDEMPOTENT_WRITE)
def send_email(to: str, body: str) -> str:
# This is wrapped in a DBOS durable step
return f"Email sent to {to}"Instance Methods
traj.append_observation()
Appends a tool execution result as an Observation node to the trajectory log.
def append_observation(
self,
result: Any,
metadata: Optional[dict] = None,
) -> str:Returns: The SHA256 hash (Node Identity) of the appended observation.
traj.export()
Exports the trajectory as a portable .tir package.
def export(
self,
mode: Literal["full", "thin", "redacted"] = "full",
output_path: Optional[str] = None,
) -> str:Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
mode | str | "full" | Export mode. "full" includes all data, "thin" omits binary artifacts, "redacted" strips sensitive fields. |
output_path | str | None | None | File path. Defaults to ~/.trajectory-ir/exports/. |
Returns: The file path of the exported .tir package.
