tilebox-workflow-automations

使用 Tilebox 自动化

在使用 tilebox automation 和通过 tilebox job submit --automation 提交一次性自动化任务时，请使用此技能。在 Agent 工作流中使用 --json。

刷新文档和 CLI 元数据

在依赖记忆中的信息之前，先检查已安装版本的确切标志和架构：

tilebox agent-context automation --output-schema
tilebox agent-context job submit --output-schema

相关文档概念：

• 满足外部触发条件时，自动化会自动提交工作流作业。
• 支持的触发类型包括 cron 计划和存储事件；数据集事件触发器已列入路线图。
• 自动化任务具有常规的任务标识符、版本和输入参数，以及用于描述事件的 trigger 属性。
• Python 可以注册自动化；目前尚不支持 Go 注册。CLI 目前可以检查自动化/存储位置，并提交由自动化触发的一次性作业。
• Tilebox Console 可以创建、编辑、删除和检查自动化。

CLI 支持的功能

只读自动化发现：

tilebox automation list --json
tilebox automation get <automation-id> --json
tilebox automation storage-locations --json

提交一次性触发的自动化作业：

tilebox job submit --name <job-name> --task <task-name> --automation cron --json
tilebox job submit --name <job-name> --task <task-name> --automation storage --trigger <storage-location-uuid>:/path --json

需要确切的输出架构或标志元数据时，请使用 tilebox agent-context automation --output-schema 或 tilebox agent-context job submit --output-schema。agent-context 始终返回 JSON；不要为其添加 --json。

检查现有自动化

需要自动化 ID、触发器定义或任务原型时，请先列出自动化：

tilebox automation list --json

检查某个自动化的完整详情：

tilebox automation get <automation-id> --json

重要字段：

• id、name、disabled
• prototype.identifier.name 和 prototype.identifier.version
• prototype.cluster_slug、prototype.max_retries、prototype.input
• cron_triggers[].schedule
• storage_event_triggers[].storage_location 和 glob_pattern

提交一次性 Cron 自动化任务

如果要像 cron 触发一样运行 Python CronTask 类任务一次，请使用此方法。

立即触发：

tilebox job submit --name cron-once --task ProcessCron --automation cron --json

在指定时间触发：

tilebox job submit \
  --name cron-at-time \
  --task ProcessCron \
  --automation cron \
  --trigger 2026-05-21T12:00:00Z \
  --json

--trigger 必须是 RFC3339/RFC3339Nano 时间戳。如果省略，CLI 将使用当前 UTC 时间。

提交一次性存储自动化任务

如果要像存储位置中创建了对象一样运行 Python StorageEventTask 类任务一次，请使用此方法。

tilebox job submit \
  --name storage-once \
  --task ProcessStorage \
  --automation storage \
  --trigger <storage-location-uuid>:/path/in/location \
  --json

规则：

• 必须指定 --trigger。
• 格式为 <storage-location-uuid>:/path/in/location。
• 路径必须以 / 开头。
• 事件类型始终为 created。
• 提交前，CLI 会从 Tilebox API 获取存储位置。

存储位置

要查找有效的存储位置 UUID：

tilebox automation storage-locations --json

自动化任务输入

--input 和 --input-file 仍用于描述任务的常规参数载荷。提交前，CLI 会将该载荷与自动化触发事件封装在一起。

对于 Python 自动化任务类，参数序列化遵循常规的 serialize_task 规则：

• 单个字段：直接提交该值。
• 多个字段：提交以字段名为键的 JSON 对象。
• 单个列表字段：提交 JSON 数组。

示例：

# Single string field
tilebox job submit --name cron-scene --task ProcessCronScene --automation cron --input S2A_001 --json

# Single integer field; submits number 5, not string "5"
tilebox job submit --name cron-count --task ProcessCronCount --automation cron --input 5 --json

# Multiple fields
tilebox job submit --name cron-structured --task ProcessCron --automation cron --input '{"scene_id":"S2A_001","other_arg":3}' --json

# One list field
tilebox job submit --name storage-scenes --task ProcessStorageScenes --automation storage --trigger <storage-location-uuid>:/batch.json --input '["S2A_001","S2B_002"]' --json

如果不确定，请让 Python 显示参数字节：

from tilebox.workflows.automations import CronTask, StorageEventTask

class ExampleCronTask(CronTask):
    name: str
    value: int

assert ExampleCronTask("test", 42)._serialize_args() == b'{"name": "test", "value": 42}'

与持久化自动化的关系

tilebox job submit --automation ... 会提交一个带有合成触发事件的一次性作业。它不会创建或更新持久化自动化。

对于持久化自动化定义，请使用 SDK/API。目前的 CLI 可以列出和检查自动化及存储位置，但不能创建、更新、启用/禁用或删除自动化。

使用 Python 注册持久化自动化

使用 Python SDK 注册持久化自动化。定义自动化任务类（CronTask 或 StorageEventTask），然后调用 client.automations() 上对应的 create_*_automation 方法。

Cron 自动化示例：

from tilebox.workflows import Client, ExecutionContext
from tilebox.workflows.automations import CronTask


class SendHeartbeat(CronTask):
    message: str

    def execute(self, context: ExecutionContext) -> None:
        context.logger.info(
            "Cron task triggered",
            message=self.message,
            trigger_time=self.trigger.time,
        )


client = Client()
automations = client.automations()

cron_automation = automations.create_cron_automation(
    "send-heartbeat",
    SendHeartbeat(message="hello"),
    cron_schedules=[
        "15 * * * *",   # every hour at minute 15
        "45 18 * * *",  # every day at 18:45
    ],
)
print(cron_automation)

存储事件自动化示例：

from tilebox.workflows import Client, ExecutionContext
from tilebox.workflows.automations import StorageEventTask, StorageEventType


class LogObjectCreation(StorageEventTask):
    head_bytes: int

    def execute(self, context: ExecutionContext) -> None:
        if self.trigger.type == StorageEventType.CREATED:
            path = self.trigger.location
            data = self.trigger.storage.read(path)
            context.logger.info(
                "Object created",
                path=path,
                size_bytes=len(data),
                preview=data[: self.head_bytes].hex(),
            )


client = Client()
automations = client.automations()
storage_locations = automations.storage_locations()

# Choose the registered storage location you want to watch.
gcs_bucket = storage_locations[0]

storage_automation = automations.create_storage_event_automation(
    "log-text-object-creations",
    LogObjectCreation(head_bytes=20),
    triggers=[
        (gcs_bucket, "**.txt"),  # match .txt files anywhere in the storage location
    ],
)
print(storage_automation)

注册后，通过 CLI 验证：

tilebox automation list --json
tilebox automation get <automation-id> --json

请确保有符合条件的任务运行器正在运行，且其任务标识符/版本和目标集群与自动化匹配；否则，触发的作业会一直排队，直到有运行器可用。

提交后调试

提交命令会返回一个任务 ID。然后使用任务工具：

tilebox job get <job-id> --json
tilebox job logs <job-id> --sort desc --limit 100 --json
tilebox job spans <job-id> --json

只有在已知有兼容的运行器可用且预计会执行任务时，才在提交命令中使用 --wait。