tilebox-workflow-automations

Tilebox Automation を使用する

tilebox automation と、tilebox job submit --automation による単発のオートメーションタスクの送信に、このスキルを使用します。エージェントのワークフローでは --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 オートメーションタスクを単発で送信する

Python の CronTask クラスを、cron が実行されたかのように一度だけ実行する場合に使用します。

今すぐトリガーする：

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 ルールが適用されます。

• フィールドが 1 つの場合：値をそのまま送信します。
• フィールドが複数の場合：フィールド名をキーにした JSON オブジェクトを送信します。
• リストフィールドが 1 つの場合：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 で args のバイト列を表示します。

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 を使用します。