trelent-agents 0.2.10

Python SDK for the Agent Orchestration API.

trelent-agents

Python SDK for the Trelent Agent Orchestration API.

Installation

pip install trelent-agents

Quick Start

Without authentication

from trelent_agents import Client

client = Client(api_url="http://localhost:8000")

sandboxes = client.sandboxes.list()
print(sandboxes)

run = client.runs.create(
    sandbox="example-agent:latest",
    prompt="Create hello.txt with a greeting",
)
print(run.id, run.status, run.harness.kind)

With authentication

When the API has authentication enabled, provide your OAuth2 client credentials:

from trelent_agents import Client

client = Client(
    api_url="https://agents.trelent.com",
    client_id="your-client-id",
    client_secret="your-client-secret",
)

sandboxes = client.sandboxes.list()
run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Do something useful",
)

The SDK handles token acquisition automatically — it exchanges your client credentials for a JWT via the API's /token endpoint before making authenticated requests.

If you need to call the token proxy directly, use client.tokens.create(...).

The client also supports context manager usage:

with Client(api_url="https://agents.trelent.com", client_id="...", client_secret="...") as client:
    runs = client.runs.list()

Client Options

client = Client(
    api_url="https://agents.trelent.com",  # API base URL (default: https://agents.trelent.com)
    client_id="...",                        # OAuth2 client ID (optional)
    client_secret="...",                    # OAuth2 client secret (optional)
    scope="...",                            # Override default OAuth2 scope (optional)
)

Both client_id and client_secret must be provided together, or both omitted.

Resources

client.sandboxes

sandboxes = client.sandboxes.list()
# => list[RegistrySandbox] — [RegistrySandbox(name="my-sandbox", tags=["latest", "v1"])]

When auth is enabled, sandboxes are automatically filtered to your namespace. Sandbox names are returned without the namespace prefix.

client.tokens

token = client.tokens.create(
    client_id="your-client-id",
    client_secret="your-client-secret",
    scope="AgentOrchestrator:runs:list",
)

client.runs

from trelent_agents import ClaudeCodeHarnessSpec, LocalImporter, ObjectStorageDumpExporter, RunSecretBinding, SharedDrive

# Create a run
run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Build a web server",
    harness=ClaudeCodeHarnessSpec(),  # optional, default: ClaudeCodeHarnessSpec()
    workdir="/workspace",          # optional, default: image WORKDIR or /workspace
    timeout_seconds=3600,          # optional, default: 3600
    imports=[LocalImporter(path="./data")],  # optional
    exports=[ObjectStorageDumpExporter()],   # optional
    shared_drives=[SharedDrive(name="my-drive")],  # optional
    secrets=[                                # optional
        RunSecretBinding(name="openai", env_var="OPENAI_API_KEY"),
    ],
)

# List runs (optionally filter by sandbox)
runs = client.runs.list()
filtered = client.runs.list(sandbox="my-sandbox:latest")

# Get a specific run
run = client.runs.get("run-id")

# Check status
status = client.runs.get_status("run-id")

# Cancel a run
cancel = client.runs.cancel("run-id")

# Get checkpoints
chain = client.runs.get_checkpoint_chain("run-id")
checkpoint = client.runs.get_checkpoint("run-id", "checkpoint-id")

Run operations

# Fork (resume from checkpoint)
forked = run.fork(
    "Continue from where you left off",
    workdir="/workspace/project",
    timeout_seconds=1800,
)

# Refresh run state
run.refresh()
print(run.status)  # updated status

Structured outputs

Pass output_schema as a Pydantic class to get a typed result back. The SDK converts your class to JSON Schema before sending, and re-validates the harness's structured output through it on the way back via run.output_parsed:

from pydantic import BaseModel
from trelent_agents import CodexHarnessSpec

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Alice and Bob are going to a science fair on Friday. Extract the event.",
    harness=CodexHarnessSpec(),
    output_schema=CalendarEvent,
)

# poll until done, then:
run.refresh()
event: CalendarEvent | None = run.output_parsed
if event:
    print(event.name, event.date, event.participants)

Fetching a typed run later (or from another process). When a different process or a later poll fetches the run, pass the same class to get() and output_parsed hydrates the same way:

run = client.runs.get("run-id", output_schema=CalendarEvent)
event: CalendarEvent | None = run.output_parsed

You can also pass a raw JSON Schema dict if you don't want to use Pydantic:

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="...",
    output_schema={
        "type": "object",
        "properties": {"name": {"type": "string"}},
        "required": ["name"],
        "additionalProperties": False,
    },
)
# Read the raw dict — `output_parsed` is `None` when no class was provided.
run.refresh()
print(run.result.structured_output)

Codex (OpenAI) is stricter than Claude. Codex requires every property in required, additionalProperties: false on every object, and rejects patternProperties, format, optional-field keywords, etc. If your schema violates these the API returns a 422 at request time with a JSON-pointer path to the bad node. Claude accepts most schemas as-is.

Streaming Events

Subscribe to real-time events from a run using Server-Sent Events (SSE). Events are streamed as the agent executes, providing live visibility into reasoning, messages, and tool usage.

Basic streaming

from trelent_agents import Client, EventType

client = Client(api_url=api_url, client_id=client_id, client_secret=client_secret)

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Build a simple web server",
)

# Subscribe to the event stream
for event in run.stream():
    match event.type:
        case EventType.MESSAGE_DELTA:
            print(event.data.delta.text, end="", flush=True)
        case EventType.TOOL_CALL_STARTED:
            print(f"\nTool: {event.data.name}")
        case EventType.SESSION_COMPLETED:
            print(f"\nCompleted with exit code: {event.data.exit_code}")

Streaming via client

You can also stream using the run ID directly:

for event in client.runs.stream_events("run-abc123"):
    print(event.type, event.seq)

Event types

Event Type	Description
SESSION_STARTED	Agent session initialized
SESSION_COMPLETED	Session finished (includes exit_code and usage)
TURN_STARTED	New conversation turn began
TURN_COMPLETED	Turn finished
TURN_FAILED	Turn failed with error
MESSAGE_STARTED	Assistant message started
MESSAGE_DELTA	Incremental text content
MESSAGE_COMPLETED	Full message content available
REASONING_STARTED	Extended thinking started
REASONING_DELTA	Incremental reasoning text
REASONING_COMPLETED	Reasoning block finished
TOOL_CALL_STARTED	Tool invocation started (includes name and kind)
TOOL_CALL_INPUT_DELTA	Streaming tool input JSON
TOOL_CALL_INPUT_COMPLETE	Full tool input available
TOOL_CALL_OUTPUT_DELTA	Streaming tool output
TOOL_CALL_COMPLETED	Tool execution finished (includes exit_code)
STRUCTURED_OUTPUT	Validated structured output (only when run was created with output_schema; emitted just before SESSION_COMPLETED)
ERROR	Error occurred

Event structure

Every event has a common envelope:

class CommonEvent(BaseModel):
    seq: int           # Sequence number (monotonically increasing)
    timestamp: str     # ISO 8601 timestamp
    type: EventType    # Event type discriminator
    data: EventData    # Type-specific payload

client.health()

health = client.health()
# => HealthResponse(status="ok")

Connectors

Importing local files

from trelent_agents import LocalImporter

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Process the data",
    imports=[LocalImporter(path="./my-data-dir")],
)

The LocalImporter tarballs and base64-encodes the local path, sending it inline with the request. The API stages large payloads onto a shared volume server-side, so the run-creation request itself stays the only network hop you need.

Exporting to S3

from trelent_agents import ObjectStorageDumpExporter

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Generate a report",
    exports=[ObjectStorageDumpExporter(bucket="my-bucket", path="reports/")],
)

Shared drives

Shared drives mount a user-scoped shared directory into the sandbox at /mnt/shared/<name>. They are specified via a dedicated shared_drives field on the run, separate from import/export connectors.

from trelent_agents import SharedDrive

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Read and write files in /mnt/shared/my-drive",
    shared_drives=[SharedDrive(name="my-drive")],
)

To mount a subdirectory of the shared drive:

shared_drives=[SharedDrive(name="my-drive", subpath="project-a")]

To export a shared drive (or a subdirectory of it) to S3 after the run:

from trelent_agents import SharedDrive, SharedDriveExporter

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Generate reports in /mnt/shared/my-drive",
    shared_drives=[SharedDrive(name="my-drive")],
    exports=[SharedDriveExporter(name="my-drive", directory="reports")],
)

Secrets

Create a user-scoped secret once. The API stores the value in the deployment's configured secret backend and only stores metadata in the database.

client.secrets.create("openai", "sk-...")

secrets = client.secrets.list()
secret = client.secrets.get("openai")
client.secrets.update("openai", "sk-new")
client.secrets.delete("openai")

Use secrets in runs by logical name. A string shorthand injects the secret as an environment variable with the same name:

run = client.runs.create(
    sandbox="my-sandbox:latest",
    prompt="Use the injected secret",
    secrets=["openai"],
)

For a custom environment variable:

from trelent_agents import RunSecretBinding

secrets=[RunSecretBinding(name="openai", env_var="OPENAI_API_KEY")]

To mount a secret as a file instead of an env var:

secrets=[RunSecretBinding(name="openai", mount=True)]

The runner writes mounted secrets to /mnt/secrets/<secret_name> inside the sandbox. You can also set both env_var and mount=True to expose both forms.

Docker Registry Setup

When auth is enabled, push sandbox images to your user namespace on the registry:

# Login with your OAuth2 credentials
docker login registry.example.com -u <client_id> -p <client_secret>

# Push to your namespace
docker tag my-sandbox:latest registry.example.com/<client_id>/my-sandbox:latest
docker push registry.example.com/<client_id>/my-sandbox:latest

When creating runs via the SDK, just use the sandbox name without the namespace prefix:

run = client.runs.create(
    sandbox="my-sandbox:latest",  # not "<client_id>/my-sandbox:latest"
    prompt="...",
)

The API resolves the full registry path automatically based on your authenticated identity.

Types

The SDK exports the following types:

from trelent_agents import (
    Client,
    ClaudeCodeHarnessSpec,
    CodexHarnessSpec,
    CommonEvent,
    EventType,
    Run,
    RunStatus,
    RunResult,
    RunStatusResponse,
    RegistrySandbox,
    CheckpointResponse,
    CancelRunResponse,
    ChatHistoryEntry,
    FileOutput,
    OutputFile,
    HealthResponse,
    HarnessInfo,
    HarnessKind,
    HarnessSpec,
    TokenRequest,
    TokenResponse,
    ToolKind,
    WorkflowIds,
    LocalImporter,
    SharedDrive,
    SharedDriveExporter,
    ObjectStorageDumpExporter,
    RunSecretBinding,
    SecretResponse,
    APIError,
    NotFoundError,
    ValidationError,
)