Workflow と連携する

Python Workers プラットフォームは FFI ↗ を使い、Cloudflare リソースへのバインディングにアクセスします。詳細は bindings のドキュメントを参照してください。

設定の観点では、Python Workflows を有効にするには、Wrangler 設定ファイルに python_workflows compatibility flag を追加します。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "workflows-starter",
	"main": "src/index.py",
	// Set this to today's date
	"compatibility_date": "2026-09-20",
	"compatibility_flags": ["python_workflows", "python_workers"],
	"workflows": [
		{
			// name of your workflow
			"name": "workflows-starter",
			// binding name env.MY_WORKFLOW
			"binding": "MY_WORKFLOW",
			// this is class that extends the Workflow class in src/index.py
			"class_name": "MyWorkflow",
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "workflows-starter"
main = "src/index.py"
# Set this to today's date
compatibility_date = "2026-09-20"
compatibility_flags = [ "python_workflows", "python_workers" ]

[[workflows]]
name = "workflows-starter"
binding = "MY_WORKFLOW"
class_name = "MyWorkflow"

Workflow でペイロードを使う方法は次のとおりです。

from workers import WorkflowEntrypoint

class DemoWorkflowClass(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do('step-name')
        async def first_step():
            payload = event["payload"]
            return payload

Workflow

Workflow バインディングは Workflow クラスへのアクセスを提供します。そのすべてのメソッドをバインディングから使えます。

`create`

指定した Workflow の新しいインスタンスを作成（起動）します。

create(options=None)* options - Workflow インスタンスに渡す、省略可能な オプションの辞書です。WorkflowInstanceCreateOptions 型と同じキーを含める必要があります。

from workers import WorkerEntrypoint, Response


class Default(WorkerEntrypoint):
    async def fetch(self, request):
        event = {"foo": "bar"}
        await self.env.MY_WORKFLOW.create(params=event)
        return Response.json({"status": "success"})

create メソッドは WorkflowInstance オブジェクトを返します。これを使って Workflow インスタンスの状態を照会できます。これは JavaScript オブジェクトであり、Python オブジェクトではありません。

`create_batch`

新しい Workflow インスタンスのバッチを作成（起動）します。一度に最大 100 インスタンスです。インスタンス作成の上限内で複数インスタンスを一度に作る必要がある場合に便利です。

create_batch(batch)* batch - インスタンス作成時に渡す WorkflowInstanceCreateOptions のリストです。ユーザー指定の ID とペイロードパラメーターを含みます。

batch リストの各要素には、id と params の両方のプロパティを含めます。

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
			# Create a new batch of 3 Workflow instances, each with its own ID and pass params to the Workflow instances
        instances = [
            {"id": "id-abc123", "params": {"hello": "world-0"}},
            {"id": "id-def456", "params": {"hello": "world-1"}},
            {"id": "id-ghi789", "params": {"hello": "world-2"}},
        ]
        await self.env.MY_WORKFLOW.create_batch(instances)
        return Response.json({"status": "success"})

`get`

ID で Workflow インスタンスを取得します。

get(id)* id - 取得する Workflow インスタンスの ID です。

WorkflowInstance オブジェクトを返します。これを使って Workflow インスタンスの状態を照会できます。

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        instance = await self.env.MY_WORKFLOW.get("abc-123")

        # FFI methods available for WorkflowInstance
        await instance.status()
        await instance.pause()
        await instance.resume()
        await instance.restart()
        await instance.terminate()
        return Response.json({"status": "success"})

`send_event`

Workflow インスタンスにイベントを送ります。

send_event(type, payload)* type - Workflow インスタンスへ送るイベントのタイプです。 * payload - Workflow インスタンスへ送るペイロードです。

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        await self.env.MY_WORKFLOW.send_event(type="my-event-type", payload={"foo": "bar"})
        return Response.json({"status": "success"})

REST API（HTTP）

Workflows REST API ドキュメントを参照してください。

コマンドライン（CLI）

コマンドラインから Workflows を管理・起動する方法は、CLI クイックスタートを参照してください。