跳转至

RestClient

改编自 Tinker SDK（Apache-2.0）

参考文档改编自 Tinker SDK，已为 Mint Recursive 做了调整。详见归属说明与 Apache 2.0 许可。

class mint.RestClient ( holder )

用于执行 REST API 操作的客户端，如列出 checkpoint 和查询元数据。

RestClient 提供多个 REST 端点的访问能力，可以查询模型信息、checkpoint 及其他资源。通常通过 service_client.create_rest_client() 获取实例。

主要方法：

  • list_checkpoints() —— 列出可用的模型 checkpoint（训练和采样）
  • list_user_checkpoints() —— 列出当前用户所有训练任务下的全部 checkpoint
  • get_training_run() —— 获取模型信息与元数据，返回 ModelEntry
  • delete_checkpoint() —— 删除某个训练任务下已有的 checkpoint
  • get_checkpoint_archive_url() —— 获取下载 checkpoint 归档的签名 URL
  • publish_checkpoint_from_tinker_path() —— 发布 checkpoint，使其公开可访问
  • unpublish_checkpoint_from_tinker_path() —— 取消发布 checkpoint，恢复为私有
  • set_checkpoint_ttl_from_tinker_path() —— 设置或移除 checkpoint 的 TTL
  • assign_session_project() —— 将 session 归入某个项目
  • whoami() —— 获取调用方的用户 URN 和邮箱
  • export_session_trace() —— 将 session 的时间线导出为 Perfetto trace 并获取签名下载 URL
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}")

参数：

  • holder ( InternalClientHolder ) —— 管理 HTTP 连接和异步操作的内部客户端

get_training_run ( training_run_id , access_scope='owned' )

获取训练任务信息。

参数：

  • training_run_id ( types. ModelID ) —— 要查询的训练任务 ID
  • access_scope ( Literal['owned', 'accessible'] ，默认： 'owned' )

返回： 包含训练任务信息的 Future

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

异步变体： get_training_run_async()

get_training_run_by_tinker_path ( mindlab-toolkit_path , access_scope='owned' )

获取训练任务信息。

参数：

  • mindlab-toolkit_path ( str ) —— 指向 checkpoint 的 mindlab-toolkit 路径
  • access_scope ( Literal['owned', 'accessible'] ，默认： 'owned' )

返回： 包含训练任务信息的 Future

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

异步变体： get_training_run_by_tinker_path_async()

get_weights_info_by_tinker_path ( mindlab-toolkit_path )

从 mindlab-toolkit 路径获取 checkpoint 信息。

参数：

  • mindlab-toolkit_path ( str ) —— 指向 checkpoint 的 mindlab-toolkit 路径

返回： 包含 checkpoint 信息的 APIFuture。该 future 可被 await。

future = rest_client.get_weights_info_by_tinker_path("mindlab-toolkit://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 )

分页列出训练任务。

参数：

  • limit ( int ，默认： 20 ) —— 最多返回的训练任务数（默认 20）
  • offset ( int ，默认： 0 ) —— 分页偏移量（默认 0）
  • access_scope ( Literal['owned', 'accessible'] ，默认： 'owned' )
  • project_id ( str | None ，默认： None ) —— 如指定，只返回该项目下的训练任务

返回： 包含 TrainingRunsResponse 的 Future，其中带有训练任务列表和游标信息

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()

异步变体： list_training_runs_async()

list_checkpoints ( training_run_id )

列出可用的 checkpoint（训练和采样）。

参数：

  • training_run_id ( types. ModelID ) —— 要列出 checkpoint 的训练任务 ID

返回： 包含 CheckpointsListResponse 的 Future，其中列出可用 checkpoint

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}")

异步变体： list_checkpoints_async()

get_checkpoint_archive_url ( training_run_id , checkpoint_id )

获取下载 checkpoint 归档的签名 URL。

参数：

  • training_run_id ( types. ModelID ) —— 要下载权重的训练任务 ID
  • checkpoint_id ( str ) —— 要下载的 checkpoint ID

返回： 包含 CheckpointArchiveUrlResponse 的 Future，其中带有签名 URL 和过期时间

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

异步变体： get_checkpoint_archive_url_async()

delete_checkpoint ( training_run_id , checkpoint_id )

删除某个训练任务下的 checkpoint。

参数：

  • training_run_id ( types. ModelID )
  • checkpoint_id ( str )

返回： ConcurrentFuture[None]

异步变体： delete_checkpoint_async()

delete_checkpoint_from_tinker_path ( mindlab-toolkit_path )

删除由 mindlab-toolkit 路径引用的 checkpoint。

参数：

  • mindlab-toolkit_path ( str )

返回： ConcurrentFuture[None]

异步变体： delete_checkpoint_from_tinker_path_async()

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

获取调用方所属组织的审计日志。

需要 tinker-admin RBAC 角色（VIEW_AUDIT_LOG 权限）。

参数：

  • event_type ( Literal['all', 'checkpoints'] ，默认： 'all' ) —— 包含的事件类型。"all" 与 "checkpoints" 目前等效，默认为 "all"。
  • day ( date | None ，默认： None ) —— 查询的日期（默认为当天）。时间窗口覆盖 UTC 当天 0 点到次日 0 点。

返回： 包含 AuditLogResponse 的 Future，其中带有审计日志条目

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.tinker_path})")
# Query a specific day
future = rest_client.get_audit_log(day=date(2025, 1, 15))

异步变体： get_audit_log_async()

get_checkpoint_archive_url_from_tinker_path ( mindlab-toolkit_path )

获取下载 checkpoint 归档的签名 URL。

参数：

  • mindlab-toolkit_path ( str ) —— 指向 checkpoint 的 mindlab-toolkit 路径

返回： 包含 CheckpointArchiveUrlResponse 的 Future，其中带有签名 URL 和过期时间

异步变体： get_checkpoint_archive_url_from_tinker_path_async()

publish_checkpoint_from_tinker_path ( mindlab-toolkit_path )

发布由 mindlab-toolkit 路径引用的 checkpoint，使其公开可访问。

只有训练任务的精确所有者才能发布 checkpoint。 已发布的 checkpoint 可用 unpublish_checkpoint_from_tinker_path 方法取消发布。

参数：

  • mindlab-toolkit_path ( str ) —— 指向 checkpoint 的 mindlab-toolkit 路径（如 "mindlab-toolkit://run-id/weights/0001"）

返回： checkpoint 发布完成后完成的 Future

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

异步变体： publish_checkpoint_from_tinker_path_async()

unpublish_checkpoint_from_tinker_path ( mindlab-toolkit_path )

取消发布由 mindlab-toolkit 路径引用的 checkpoint，将其恢复为私有。

只有训练任务的精确所有者才能取消发布 checkpoint。 此操作是发布 checkpoint 的逆操作。

参数：

  • mindlab-toolkit_path ( str ) —— 指向 checkpoint 的 mindlab-toolkit 路径（如 "mindlab-toolkit://run-id/weights/0001"）

返回： checkpoint 取消发布后完成的 Future

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

异步变体： unpublish_checkpoint_from_tinker_path_async()

set_checkpoint_ttl_from_tinker_path ( mindlab-toolkit_path , ttl_seconds )

设置或移除由 mindlab-toolkit 路径引用的 checkpoint 的 TTL。

如指定 ttl_seconds，checkpoint 将在当前起经过该秒数后过期。 如 ttl_seconds 为 None，已有的过期时间将被移除。

参数：

  • mindlab-toolkit_path ( str ) —— 指向 checkpoint 的 mindlab-toolkit 路径（如 "mindlab-toolkit://run-id/weights/0001")
  • ttl_seconds ( int | None ) —— 过期前的秒数，或 None 表示移除 TTL

返回： TTL 设置完成后完成的 Future

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

异步变体： set_checkpoint_ttl_from_tinker_path_async()

list_user_checkpoints ( limit=100 , offset=0 )

列出当前用户所有训练任务下的全部 checkpoint。

此方法获取已认证用户拥有的所有训练任务的 checkpoint，按时间倒序排列，支持分页，便于处理大量 checkpoint。

参数：

  • limit ( int ，默认： 100 ) —— 最多返回的 checkpoint 数（默认 100）
  • offset ( int ，默认： 0 ) —— 分页偏移量（默认 0）

返回： 包含 CheckpointsListResponse 的 Future，其中带有 checkpoint 列表和游标信息

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)

异步变体： list_user_checkpoints_async()

get_session ( session_id , access_scope='owned' )

获取 session 信息，包括其中所有训练任务和采样器。

参数：

  • session_id ( str ) —— 要查询的 session ID
  • access_scope ( Literal['owned', 'accessible'] ，默认： 'owned' )

返回： 包含 GetSessionResponse 的 Future，其中带有 training_run_ids 和 sampler_ids

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)}")

异步变体： get_session_async()

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

分页列出 session。

参数：

  • limit ( int ，默认： 20 ) —— 最多返回的 session 数（默认 20）
  • offset ( int ，默认： 0 ) —— 分页偏移量（默认 0）
  • access_scope ( Literal['owned', 'accessible'] ，默认： 'owned' )

返回： 包含 ListSessionsResponse 的 Future，其中带有 session ID 列表

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)

异步变体： list_sessions_async()

assign_session_project ( session_id , project_id )

将 session（及其所有训练任务和采样器）归入某个项目。

用于将先前创建的 session 关联到项目，或在项目之间移动 session。不支持清除项目——session 一旦归入项目就无法再移出。

参数：

  • session_id ( str ) —— 要移动的 session ID
  • project_id ( str ) —— 目标项目 ID

返回： session 移动完成后完成的 Future

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

异步变体： assign_session_project_async()

export_session_trace ( session_id )

将 session 的时间线导出为 Perfetto trace 并获取签名下载 URL。

在服务端启动（或复用）一个异步导出任务，构建 session 训练与采样请求的 Perfetto trace（.pftrace），轮询直到文件上传完成，再返回签名下载 URL。URL 约一小时后过期；重新调用此方法可获取新 URL（trace 如已是最新则不会重建）。

下载 trace 时，直接对该 URL 发起普通 HTTPS GET（无需 auth headers）：

import urllib.request
url = rest_client.export_session_trace("session-id").result()
urllib.request.urlretrieve(url, "session-id.pftrace")
或从命令行：curl -o session.pftrace '<url>'（URL 含查询参数，需加引号）。用 https://ui.perfetto.dev 打开文件查看时间线，或用 mindlab-toolkit session export-trace <session-id> 从 CLI 完成全部操作。

参数：

  • session_id ( str ) —— 要导出 trace 的 session ID

返回： 包含 .pftrace 文件签名下载 URL 的 Future

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

异步变体： export_session_trace_async()

whoami ()

获取调用方的身份信息。

返回当前所用凭证关联的用户 URN；当 principal 由用户背书时，同时返回用户邮箱。

身份信息从 SDK 持有的 JWT claims 中读取（后台缓存并自动刷新），因此此调用不会发起服务端请求。

返回： 包含 WhoamiResponse 的 APIFuture。该 future 可被 await。

# 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 )

获取采样器信息。

参数：

  • sampler_id ( str ) —— 要查询的采样器 ID（sampling_session_id）

返回： 包含 GetSamplerResponse 的 APIFuture，其中带有采样器详情

# 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}")

异步变体： get_sampler_async()

get_billing_usage ( starting_on , ending_before )

获取组织的详细计费用量。

返回按小时分桶的用量，形式为 BillingUsageEvent 信封列表：公共归属信息（桶、基座模型、用户、session、项目）在信封上，按用量类型区分的具体载荷在 event_info 中——这是一个以 .type 区分的联合类型（training / sampling_prefill / sampling_sample / checkpoint / storage），每种各带适用字段（token_count、gigabyte_hours、count、prefill 缓存标记）。session 的 user_metadata 在 response.sessions 中按 session 出现一次，以 session_id 为键。不含美元金额。数据比实时滞后最多几小时。需要组织内的计费查看权限。

参数：

  • starting_on ( datetime | str ) —— 包含的窗口起点（RFC 3339 字符串或 datetime），对齐到 UTC 小时边界
  • ending_before ( datetime | str ) —— 不包含的窗口终点，对齐到 UTC 小时边界；最多比 starting_on 晚 14 天；不得从未来开始

返回： 包含 BillingUsageResponse 的 Future

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)

异步变体： get_billing_usage_async()

被引用

  • ServiceClient.create_rest_client