Skip to content
View as Markdown

RestClient

Generated from tinker 0.30.4 at commit 1e5777e. Source links point at that snapshot.

class tinker.RestClient(holder)

Client for REST API operations like listing checkpoints and metadata.

The RestClient provides access to various REST endpoints for querying model information, checkpoints, and other resources. You typically get one by calling service_client.create_rest_client().

Key methods:

  • list_checkpoints() - list available model checkpoints (both training and sampler)
  • list_user_checkpoints() - list all checkpoints across all user's training runs
  • get_training_run() - get model information and metadata as ModelEntry
  • delete_checkpoint() - delete an existing checkpoint for a training run
  • get_checkpoint_archive_url() - get signed URL to download checkpoint archive
  • get_external_weights_urls() - get per-file signed URLs for an external weights checkpoint
  • publish_checkpoint_from_tinker_path() - publish a checkpoint to make it public
  • unpublish_checkpoint_from_tinker_path() - unpublish a checkpoint to make it private
  • set_checkpoint_ttl_from_tinker_path() - set or remove TTL on a checkpoint
  • assign_session_project() - move a session into a project
  • whoami() - get the calling principal's user URN and email
  • export_session_trace() - export a session's timeline as a Perfetto trace and get a signed download URL

Parameters:

Example:

rest_client = service_client.create_rest_client()
training_run = rest_client.get_training_run("run-id").result()
print(f"Training Run: {training_run.training_run_id}, LoRA: {training_run.is_lora}")
checkpoints = rest_client.list_checkpoints("run-id").result()
print(f"Found {len(checkpoints.checkpoints)} checkpoints")
for checkpoint in checkpoints.checkpoints:
    print(f"  {checkpoint.checkpoint_type}: {checkpoint.checkpoint_id}")

get_training_run(training_run_id, access_scope='owned')

Get training run info.

Parameters:

Returns:

  • A Future containing the training run information

Example:

future = rest_client.get_training_run("run-id")
response = future.result()
print(f"Training Run ID: {response.training_run_id}, Base: {response.base_model}")

Async variant: get_training_run_async()

get_training_run_by_tinker_path(tinker_path, access_scope='owned')

Get training run info.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns:

  • A Future containing the training run information

Example:

future = rest_client.get_training_run_by_tinker_path("tinker://run-id/weights/checkpoint-001")
response = future.result()
print(f"Training Run ID: {response.training_run_id}, Base: {response.base_model}")

Async variant: get_training_run_by_tinker_path_async()

get_weights_info_by_tinker_path(tinker_path)

Get checkpoint information from a tinker path.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint

Returns:

  • An APIFuture containing the checkpoint information. The future is awaitable.

Example:

future = rest_client.get_weights_info_by_tinker_path("tinker://run-id/weights/checkpoint-001")
response = future.result()  # or await future
print(f"Base Model: {response.base_model}, LoRA Rank: {response.lora_rank}")

list_training_runs(limit=20, offset=0, access_scope='owned', project_id=None)

List training runs with pagination support.

Parameters:

  • limit (int, default: 20) – Maximum number of training runs to return (default 20)
  • offset (int, default: 0) – Offset for pagination (default 0)
  • access_scope (Literal['owned', 'accessible'], default: 'owned')
  • project_id (str | None, default: None) – If provided, only return training runs in this project

Returns:

Example:

future = rest_client.list_training_runs(limit=50)
response = future.result()
print(f"Found {len(response.training_runs)} training runs")
print(f"Total: {response.cursor.total_count}")
# Get next page
next_page = rest_client.list_training_runs(limit=50, offset=50)
# Only runs in a given project
project_runs = rest_client.list_training_runs(project_id="my-project-id").result()

Async variant: list_training_runs_async()

list_checkpoints(training_run_id)

List available checkpoints (both training and sampler).

Parameters:

Returns:

Example:

future = rest_client.list_checkpoints("run-id")
response = future.result()
for checkpoint in response.checkpoints:
    if checkpoint.checkpoint_type == "training":
        print(f"Training checkpoint: {checkpoint.checkpoint_id}")
    elif checkpoint.checkpoint_type == "sampler":
        print(f"Sampler checkpoint: {checkpoint.checkpoint_id}")

Async variant: list_checkpoints_async()

get_checkpoint_archive_url(training_run_id, checkpoint_id)

Get signed URL to download checkpoint archive.

Parameters:

Returns:

Example:

future = rest_client.get_checkpoint_archive_url("run-id", "checkpoint-123")
response = future.result()
print(f"Download URL: {response.url}")
print(f"Expires at: {response.expires_at}")
# Use the URL to download the archive with your preferred HTTP client

Async variant: get_checkpoint_archive_url_async()

get_external_weights_urls(tinker_path)

Get signed download URLs, one per file, for an external weights checkpoint.

Parameters:

  • tinker_path (str) – The checkpoint's tinker path, as returned by save_weights_external (tinker://<training_run_id>/external_weights/<name>)

Returns:

  • A Future containing the ExternalWeightsUrlsResponse: urls maps each file path (relative to the checkpoint root) to a signed URL, valid until expires

Async variant: get_external_weights_urls_async()

delete_checkpoint(training_run_id, checkpoint_id)

Delete a checkpoint for a training run.

Parameters:

Returns: ConcurrentFuture[None]

Async variant: delete_checkpoint_async()

delete_checkpoint_from_tinker_path(tinker_path)

Delete a checkpoint referenced by a tinker path.

Parameters:

Returns: ConcurrentFuture[None]

Async variant: delete_checkpoint_from_tinker_path_async()

get_audit_log(event_type='all', day=None)

Get an audit log of events for the caller's organization.

Requires the tinker-admin RBAC role (VIEW_AUDIT_LOG capability).

Parameters:

  • event_type (Literal['all', 'checkpoints', 'projects', 'teams', 'organizations'], default: 'all') – Which resource's events to include: "checkpoints", "projects", "teams", "organizations", or "all". Defaults to "all".
  • day (date | None, default: None) – The date to query (default: today). The window covers midnight to midnight UTC.

Returns:

Example:

from datetime import date

future = rest_client.get_audit_log()
response = future.result()
print(f"Found {len(response.entries)} audit entries")
for entry in response.entries:
    print(f"  {entry.timestamp}: {entry.event} {entry.event_details}")

# Query a specific day
future = rest_client.get_audit_log(day=date(2025, 1, 15))

Async variant: get_audit_log_async()

get_checkpoint_archive_url_from_tinker_path(tinker_path)

Get signed URL to download checkpoint archive.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint

Returns:

Async variant: get_checkpoint_archive_url_from_tinker_path_async()

publish_checkpoint_from_tinker_path(tinker_path)

Publish a checkpoint referenced by a tinker path to make it publicly accessible.

Only the exact owner of the training run can publish checkpoints. Published checkpoints can be unpublished using the unpublish_checkpoint_from_tinker_path method.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint (e.g., "tinker://run-id/weights/0001")

Returns:

  • A Future that completes when the checkpoint is published

Raises:

HTTPException: 400 if checkpoint identifier is invalid HTTPException: 404 if checkpoint not found or user doesn't own the training run HTTPException: 409 if checkpoint is already public HTTPException: 500 if there's an error publishing the checkpoint

Example:

future = rest_client.publish_checkpoint_from_tinker_path("tinker://run-id/weights/0001")
future.result()  # Wait for completion
print("Checkpoint published successfully")

Async variant: publish_checkpoint_from_tinker_path_async()

unpublish_checkpoint_from_tinker_path(tinker_path)

Unpublish a checkpoint referenced by a tinker path to make it private again.

Only the exact owner of the training run can unpublish checkpoints. This reverses the effect of publishing a checkpoint.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint (e.g., "tinker://run-id/weights/0001")

Returns:

  • A Future that completes when the checkpoint is unpublished

Raises:

HTTPException: 400 if checkpoint identifier is invalid HTTPException: 404 if checkpoint not found or user doesn't own the training run HTTPException: 409 if checkpoint is already private HTTPException: 500 if there's an error unpublishing the checkpoint

Example:

future = rest_client.unpublish_checkpoint_from_tinker_path("tinker://run-id/weights/0001")
future.result()  # Wait for completion
print("Checkpoint unpublished successfully")

Async variant: unpublish_checkpoint_from_tinker_path_async()

set_checkpoint_ttl_from_tinker_path(tinker_path, ttl_seconds)

Set or remove the TTL on a checkpoint referenced by a tinker path.

If ttl_seconds is provided, the checkpoint will expire after that many seconds from now. It must be between 1 hour (3600) and 10 years. If ttl_seconds is None, any existing expiration will be removed.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint (e.g., "tinker://run-id/weights/0001")
  • ttl_seconds (int | None) – Seconds until expiration (1 hour to 10 years), or None to remove TTL

Returns:

  • A Future that completes when the TTL is set

Raises:

HTTPException: 400 if checkpoint identifier is invalid or ttl_seconds is out of range HTTPException: 404 if checkpoint not found or user doesn't own the training run HTTPException: 500 if there's an error setting the TTL

Example:

future = rest_client.set_checkpoint_ttl_from_tinker_path("tinker://run-id/weights/0001", 86400)
future.result()  # Wait for completion
print("Checkpoint TTL set successfully")

Async variant: set_checkpoint_ttl_from_tinker_path_async()

list_user_checkpoints(limit=100, offset=0)

List all checkpoints for the current user across all their training runs.

This method retrieves checkpoints from all training runs owned by the authenticated user, sorted by time (newest first). It supports pagination for efficiently handling large numbers of checkpoints.

Parameters:

  • limit (int, default: 100) – Maximum number of checkpoints to return (default 100)
  • offset (int, default: 0) – Offset for pagination (default 0)

Returns:

Example:

future = rest_client.list_user_checkpoints(limit=50)
response = future.result()
print(f"Found {len(response.checkpoints)} checkpoints")
print(f"Total: {response.cursor.total_count if response.cursor else 'Unknown'}")
for checkpoint in response.checkpoints:
    print(f"  {checkpoint.training_run_id}/{checkpoint.checkpoint_id}")
# Get next page if there are more checkpoints
if response.cursor and response.cursor.offset + response.cursor.limit < response.cursor.total_count:
    next_page = rest_client.list_user_checkpoints(limit=50, offset=50)

Async variant: list_user_checkpoints_async()

get_session(session_id, access_scope='owned')

Get session information including all training runs and samplers.

Parameters:

  • session_id (str) – The session ID to get information for
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns:

  • A Future containing the GetSessionResponse with training_run_ids, sampler_ids, and user_metadata

Example:

future = rest_client.get_session("session-id")
response = future.result()
print(f"Training runs: {len(response.training_run_ids)}")
print(f"Samplers: {len(response.sampler_ids)}")
print(f"User metadata: {response.user_metadata}")

Async variant: get_session_async()

list_sessions(limit=20, offset=0, access_scope='owned')

List sessions with pagination support.

Parameters:

  • limit (int, default: 20) – Maximum number of sessions to return (default 20)
  • offset (int, default: 0) – Offset for pagination (default 0)
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns:

Example:

future = rest_client.list_sessions(limit=50)
response = future.result()
print(f"Found {len(response.sessions)} sessions")
# Get next page
next_page = rest_client.list_sessions(limit=50, offset=50)

Async variant: list_sessions_async()

assign_session_project(session_id, project_id)

Move a session (and all of its training runs/samplers) into a project.

Use this to attach a previously-created session to a project, or to move a session between projects. Clearing the project is not supported — sessions cannot be moved out of a project once placed.

Parameters:

Returns:

  • A Future that completes when the session has been moved

Raises:

HTTPException: 400 if project_id is missing HTTPException: 403 if the caller lacks access to the destination project HTTPException: 404 if the session is not found or not accessible

Example:

future = rest_client.assign_session_project("session-id", "project-id")
future.result()

Async variant: assign_session_project_async()

export_session_trace(session_id)

Export a session's timeline as a Perfetto trace and get a signed download URL.

Kicks off (or reuses) an async export job on the server that builds a Perfetto trace (.pftrace) of the session's training and sampling requests, polls until the file is uploaded, and returns a signed download URL. The URL expires after about an hour; call this method again to get a fresh one (the trace is not rebuilt if it is already up to date).

To download the trace, issue a plain HTTPS GET to the URL (no auth headers needed): 

import urllib.request

url = rest_client.export_session_trace("session-id").result()
urllib.request.urlretrieve(url, "session-id.pftrace")
or from a shell: curl -o session.pftrace '<url>' (quote the URL, it contains query parameters). Open the file in https://ui.perfetto.dev to view the timeline, or use tinker session export-trace <session-id> to do all of this from the CLI.

Parameters:

  • session_id (str) – The session ID to export a trace for

Returns:

  • A Future containing the signed download URL for the .pftrace file

Raises:

RuntimeError: if the export job fails

Example:

url = rest_client.export_session_trace("session-id").result()
print(f"Download URL: {url}")

Async variant: export_session_trace_async()

whoami()

Get the calling principal's identity.

Returns the user URN associated with the credential in use, and the user's email when the principal is user-backed.

The identity is read from the claims of the JWT the SDK already holds from auth (cached and refreshed in the background), so this makes no server requests.

Returns:

Example:

# Sync usage
response = rest_client.whoami().result()
print(f"URN: {response.user_urn}, email: {response.email}")

# Async usage
response = await rest_client.whoami()

get_sampler(sampler_id)

Get sampler information.

Parameters:

  • sampler_id (str) – The sampler ID (sampling_session_id) to get information for

Returns:

Example:

# Sync usage
future = rest_client.get_sampler("session-id:sample:0")
response = future.result()
print(f"Base model: {response.base_model}")
print(f"Model path: {response.model_path}")

# Async usage
response = await rest_client.get_sampler("session-id:sample:0")
print(f"Base model: {response.base_model}")

Async variant: get_sampler_async()

get_billing_usage(starting_on, ending_before)

Get detailed billing usage for your organization.

Returns hourly-bucketed usage as a list of BillingUsageEvent envelopes: the shared attribution (bucket, base model, user, session, project) lives on the envelope, and the usage-kind-specific payload is event_info — a union discriminated on .type (training / sampling_prefill / sampling_sample / checkpoint / storage). Each variant carries exactly the fields that apply (token_count, gigabyte_hours, count, the prefill cached flag). Session user_metadata comes once per session in response.sessions, keyed by session_id. Token and storage events include estimated_cost_usd, an estimated gross usage cost before credits and commits, plus the applicable effective rate per million tokens or per GB-month. These are not invoice amounts due. Token cost is the rate per million tokens multiplied by token_count / 1_000_000; storage cost is the rate per GB-month multiplied by gigabyte_hours / 720. A completed UTC day is priced only after its full-day usage quantities reconcile with usage line items from the finalized invoice export, or the latest draft when no finalized invoice is available. The current incomplete UTC day instead uses the published Tinker rate-card snapshot, so its estimate is not invoice-reconciled and can change after the day completes. Costs are null when a completed day has not reconciled or the current rate card has no applicable rate. cost_data_through is a conservative exclusive UTC completeness watermark. It advances only across consecutive reconciled completed billable-usage days and stops before the first gap. It is not the latest timestamp with any cost: current-day estimates do not advance it, and a later reconciled day may have costs beyond it. Data lags real time by up to a few hours. Requires billing view access in your organization.

Parameters:

  • starting_on (datetime | str) – Inclusive window start (RFC 3339 string or datetime), aligned to a UTC hour boundary; must not be in the future
  • ending_before (datetime | str) – Exclusive window end, aligned to a UTC hour boundary; at most 14 days after starting_on

Returns:

Example:

future = rest_client.get_billing_usage(
    "2026-07-13T00:00:00Z", "2026-07-14T00:00:00Z"
)
for event in future.result().data:
    match event.event_info:
        case types.StorageBillingEvent() as info:
            print(event.bucket_start, "storage", info.gigabyte_hours, "GB-h")
        case types.CheckpointBillingEvent() as info:
            print(event.bucket_start, "checkpoints", info.count)
        case info:
            print(event.bucket_start, info.type, event.base_model, info.token_count)

Async variant: get_billing_usage_async()

Referenced by