Python Workers 平台利用 FFI ↗ 访问 Cloudflare 资源的绑定。有关更多信息，请参阅绑定文档。
从配置角度来看，启用 Python Workflows 需要在 Wrangler 配置文件中添加 python_workflows 兼容性标志。
{
"$schema": "./node_modules/wrangler/config-schema.json",
"name": "workflows-starter",
"main": "src/index.py",
// Set this to today's date
"compatibility_date": "2026-08-17",
"compatibility_flags": ["python_workflows", "python_workers"],
"workflows": [
{
// workflow 名称
"name": "workflows-starter",
// 绑定名称 env.MY_WORKFLOW
"binding": "MY_WORKFLOW",
// 这是在 src/index.py 中扩展 Workflow 类的 class
"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-08-17"
compatibility_flags = [ "python_workflows", "python_workers" ]
[[workflows]]
name = "workflows-starter"
binding = "MY_WORKFLOW"
class_name = "MyWorkflow"以下是在 workflow 中使用 payload 的方式：
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 payloadWorkflow 绑定提供对 Workflow 类的访问。其所有方法都可在绑定上使用。
创建（触发）给定 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 对象。
创建（触发）一批新的 workflow 实例，一次最多 100 个实例。如果你需要一次创建多个实例且在实例创建限制内，这很有用。
create_batch(batch)*batch- 创建实例时传递的WorkflowInstanceCreateOptions列表，包括用户提供的 ID 和 payload 参数。
batch 列表的每个元素都应包含 id 和 params 属性：
from workers import WorkerEntrypoint, Response
class Default(WorkerEntrypoint):
async def fetch(self, request):
# 创建 3 个 Workflow 实例的新批次，每个都有自己的 ID 并传递 params
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"})按 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")
# WorkflowInstance 可用的 FFI 方法
await instance.status()
await instance.pause()
await instance.resume()
await instance.restart()
await instance.terminate()
return Response.json({"status": "success"})向 workflow 实例发送事件。
send_event(type, payload)*type- 要发送给 workflow 实例的事件类型。 *payload- 要发送给 workflow 实例的 payload。
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"})请参阅 CLI 快速入门 了解如何通过命令行管理和触发 Workflows。