cloudflare-python-workers

Cloudflare Python Workers

状态：Beta（需要启用 python_workers 兼容性标志） 运行时：Pyodide（编译为 WebAssembly 的 Python 3.12+） 包版本：[email protected]、[email protected]、[email protected] 上次验证时间：2026-01-21

快速开始（5 分钟）

1. 前置条件

确保已安装：

• uv - Python 包管理器
• Node.js - Wrangler 所需

2. 初始化项目

# Create project directory
mkdir my-python-worker && cd my-python-worker

# Initialize Python project
uv init

# Install pywrangler
uv tool install workers-py

# Initialize Worker configuration
uv run pywrangler init

3. 创建入口点

创建 src/entry.py：

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from Python Worker!")

4. 配置 wrangler.jsonc

{
  "name": "my-python-worker",
  "main": "src/entry.py",
  "compatibility_date": "2025-12-01",
  "compatibility_flags": ["python_workers"]
}

5. 本地运行

uv run pywrangler dev
# Visit http://localhost:8787

6. 部署

uv run pywrangler deploy

---

从 2025 年 12 月之前的 Workers 迁移

如果你在 2025 年 12 月之前创建了 Python Worker，就只能使用内置包。借助 pywrangler（2025 年 12 月），现在可以部署使用外部包的项目。

旧方法（已不再需要）：

# Limited to built-in packages only
# Could only use httpx, aiohttp, beautifulsoup4, etc.
# Error: "You cannot yet deploy Python Workers that depend on
# packages defined in requirements.txt [code: 10021]"

新方法（pywrangler）：

# pyproject.toml
[project]
dependencies = ["fastapi", "any-pyodide-compatible-package"]

uv tool install workers-py
uv run pywrangler deploy  # Now works!

历史时间线：

• 2024 年 4 月 - 2025 年 12 月：完全无法部署包
• 2025 年 12 月 8 日：Pywrangler 发布，支持部署包
• 2026 年 1 月：全面支持包的公开 Beta 版

参阅：包部署问题历史

---

核心概念

WorkerEntrypoint 类模式

自 2025 年 8 月起，Python Workers 使用基于类的模式（而非全局处理程序）：

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Access bindings via self.env
        value = await self.env.MY_KV.get("key")

        # Parse request
        url = request.url
        method = request.method

        return Response(f"Method: {method}, URL: {url}")

访问绑定

所有 Cloudflare 绑定均通过 self.env 访问：

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # D1 Database
        result = await self.env.DB.prepare("SELECT * FROM users").all()

        # KV Storage
        value = await self.env.MY_KV.get("key")
        await self.env.MY_KV.put("key", "value")

        # R2 Object Storage
        obj = await self.env.MY_BUCKET.get("file.txt")

        # Workers AI
        response = await self.env.AI.run("@cf/meta/llama-2-7b-chat-int8", {
            "prompt": "Hello!"
        })

        return Response("OK")

支持的绑定：

• D1（SQL 数据库）
• KV（键值存储）
• R2（对象存储）
• Workers AI
• Vectorize
• Durable Objects
• Queues
• Analytics Engine

详情请参阅 Cloudflare 绑定文档。

请求/响应处理

from workers import WorkerEntrypoint, Response
import json

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Parse JSON body
        if request.method == "POST":
            body = await request.json()
            return Response(
                json.dumps({"received": body}),
                headers={"Content-Type": "application/json"}
            )

        # Query parameters
        url = URL(request.url)
        name = url.searchParams.get("name", "World")

        return Response(f"Hello, {name}!")

定时处理程序（Cron）

from workers import handler

@handler
async def on_scheduled(event, env, ctx):
    # Run on cron schedule
    print(f"Cron triggered at {event.scheduledTime}")

    # Do work...
    await env.MY_KV.put("last_run", str(event.scheduledTime))

在 wrangler.jsonc 中配置：

{
  "triggers": {
    "crons": ["*/5 * * * *"]  // Every 5 minutes
  }
}

---

Python 工作流

Python Workflows 支持构建持久化的多步骤自动化流程，并提供自动重试和状态持久化功能。

为什么采用装饰器模式？

Python Workflows 使用 @step.do() 装饰器模式，因为 Python 不易支持匿名回调（而 JavaScript/TypeScript 支持内联箭头函数）。这是语言本身的根本差异，并非 Cloudflare 实现的限制。

JavaScript 模式（无法直接转换）：

await step.do("my step", async () => {
  // Inline callback
  return result;
});

Python 模式（必需）：

@step.do("my step")
async def my_step():
    # Named function with decorator
    return result

result = await my_step()

来源：Python Workflows 博客

使用 asyncio.gather 并发执行

Pyodide 会捕获 JavaScript Promise（thenable），并将其代理为 Python 可等待对象。这样便可使用标准 Python 异步模式实现与 Promise.all 等效的行为：

import asyncio

@step.do("step_a")
async def step_a():
    return "A"

@step.do("step_b")
async def step_b():
    return "B"

# Concurrent execution (like Promise.all)
results = await asyncio.gather(step_a(), step_b())
# results = ["A", "B"]

原理：工作流步骤返回的 JavaScript Promise 会被代理为 Python 可等待对象，因此可以使用标准 asyncio 并发原语。

来源：Python Workflows 博客

基本工作流

from workers import WorkflowEntrypoint, WorkerEntrypoint, Response

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        # Step 1
        @step.do("fetch data")
        async def fetch_data():
            response = await fetch("https://api.example.com/data")
            return await response.json()

        data = await fetch_data()

        # Step 2: Sleep
        await step.sleep("wait", "10 seconds")

        # Step 3: Process
        @step.do("process data")
        async def process_data():
            return {"processed": True, "count": len(data)}

        result = await process_data()
        return result


class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Create workflow instance
        instance = await self.env.MY_WORKFLOW.create()
        return Response(f"Workflow started: {instance.id}")

DAG 依赖关系

定义步骤依赖关系以实现并行执行：

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do("step_a")
        async def step_a():
            return "A done"

        @step.do("step_b")
        async def step_b():
            return "B done"

        # step_c waits for both step_a and step_b
        @step.do("step_c", depends=[step_a, step_b], concurrent=True)
        async def step_c(result_a, result_b):
            return f"C received: {result_a}, {result_b}"

        return await step_c()

工作流配置

{
  "compatibility_flags": ["python_workers", "python_workflows"],
  "compatibility_date": "2025-12-01",
  "workflows": [
    {
      "name": "my-workflow",
      "binding": "MY_WORKFLOW",
      "class_name": "MyWorkflow"
    }
  ]
}

---

包管理

pyproject.toml 配置

[project]
name = "my-python-worker"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "beautifulsoup4",
    "httpx"
]

[dependency-groups]
dev = [
    "workers-py",
    "workers-runtime-sdk"
]

支持的包

Python Workers 支持：

• 来自 PyPI 的纯 Python 包
• Pyodide 包（为 WebAssembly 预构建）

请参阅 Pyodide 包列表。

HTTP 客户端

仅异步 HTTP 库可用：

# ✅ WORKS - httpx (async)
import httpx

async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com")

# ✅ WORKS - aiohttp
import aiohttp

async with aiohttp.ClientSession() as session:
    async with session.get("https://api.example.com") as response:
        data = await response.json()

# ❌ DOES NOT WORK - requests (sync)
import requests  # Will fail!

申请支持新包

可在此处申请支持新包： https://github.com/cloudflare/workerd/discussions/categories/python-packages

---

FFI（外部函数接口）

通过 Pyodide 的 FFI 从 Python 访问 JavaScript API：

JavaScript 全局对象

from js import fetch, console, Response as JSResponse

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Use JavaScript fetch
        response = await fetch("https://api.example.com")
        data = await response.json()

        # Console logging
        console.log("Fetched data:", data)

        # Return JavaScript Response
        return JSResponse.new("Hello!")

类型转换

重要：to_py() 是 JavaScript 对象上的方法，而不是独立函数。只有 to_js() 是函数。

from js import Object
from pyodide.ffi import to_js

# ❌ WRONG - ImportError!
from pyodide.ffi import to_py
python_data = to_py(js_data)

# ✅ CORRECT - to_py() is a method
async def fetch(self, request):
    data = await request.json()  # Returns JS object
    python_data = data.to_py()   # Convert to Python dict

# Convert Python dict to JavaScript object
python_dict = {"name": "test", "count": 42}
js_object = to_js(python_dict, dict_converter=Object.fromEntries)

# Use in Response
return Response(to_js({"status": "ok"}))

来源：GitHub Issue #3322（Pyodide 维护者说明）

---

已知问题预防

此技能可预防 11 个有记录的问题：

问题 #1：旧版处理程序模式

错误：TypeError: on_fetch is not defined

原因：处理程序模式于 2025 年 8 月发生了变化。

# ❌ OLD (deprecated)
@handler
async def on_fetch(request):
    return Response("Hello")

# ✅ NEW (current)
class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello")

问题 #2：同步 HTTP 库

错误：RuntimeError: cannot use blocking call in async context

原因：Python Workers 仅支持异步操作。同步库会阻塞事件循环。

# ❌ FAILS
import requests
response = requests.get("https://api.example.com")

# ✅ WORKS
import httpx
async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com")

问题 #3：原生／编译包

错误：ModuleNotFoundError: No module named 'numpy'（或类似错误）

原因：仅纯 Python 包可用。不支持原生 C 扩展。

解决方案：使用兼容 Pyodide 的替代方案，或查看 Pyodide packages。

问题 #4：缺少兼容性标志

错误：Error: Python Workers require the python_workers compatibility flag

修复方法：将以下内容添加到 wrangler.jsonc：

{
  "compatibility_flags": ["python_workers"]
}

对于 Workflows，还需添加 "python_workflows"。

问题 #5：在 Workflow 步骤之外执行 I/O

错误：Workflow 状态未正确持久化

原因：所有 I/O 都必须在 @step.do 内执行，以确保持久性。

# ❌ BAD - fetch outside step
response = await fetch("https://api.example.com")
@step.do("use data")
async def use_data():
    return await response.json()  # response may be stale on retry

# ✅ GOOD - fetch inside step
@step.do("fetch and use")
async def fetch_and_use():
    response = await fetch("https://api.example.com")
    return await response.json()

问题 #6：类型序列化错误

错误：TypeError: Object of type X is not JSON serializable

原因：Workflow 步骤的返回值必须可序列化为 JSON。

修复方法：返回前转换复杂对象：

@step.do("process")
async def process():
    # Convert datetime to string
    return {"timestamp": datetime.now().isoformat()}

问题 #7：冷启动性能

注意：Python Workers 的冷启动时间比 JavaScript 长。借助 Wasm 内存快照（2025 年 12 月），FastAPI 和 Pydantic 等大型包现在约需 1 秒加载（此前约需 10 秒），但仍比 JavaScript Workers（约 50 毫秒）慢约 2 倍。

性能数据（截至 2025 年 12 月）：

• 快照前：FastAPI／Pydantic 约需 10 秒
• 快照后：约需 1 秒（性能提升 10 倍）
• JavaScript 对应情况：约 50 毫秒

缓解措施：

• 尽量减少顶层导入
• 对大型包使用延迟加载
• 对延迟敏感的路径考虑使用 JavaScript Workers
• Wasm 快照会自动改善冷启动（无需配置）

来源：Python Workers Redux Blog | InfoQ Coverage

问题 #8：包安装失败

错误：Failed to install package X

原因：

• 包包含原生依赖项
• 包不属于 Pyodide 生态
• 打包期间出现网络问题

修复方法：检查包的兼容性、使用替代方案，或申请支持。

问题 #9：开发注册表导致 JS-to-Python RPC 失效

错误：从 JavaScript Worker 调用 Python Worker 时出现 Network connection lost 来源：GitHub Issue #11438

发生原因：开发注册表无法正确路由在不同终端中分别运行的 Workers 之间的 RPC 调用。

预防措施：

# ❌ Doesn't work - separate terminals
# Terminal 1: npx wrangler dev (JS worker)
# Terminal 2: npx wrangler dev (Python worker)
# Result: Network connection lost error

# ✅ Works - single wrangler instance
npx wrangler dev -c ts/wrangler.jsonc -c py/wrangler.jsonc

在单个 wrangler 实例中运行两个 Worker，以启用正常的 RPC 通信。

问题 #10：HTMLRewriter 使用数据 URL 时超出内存限制

错误：TypeError: Parser error: The memory limit has been exceeded 来源：GitHub Issue #10814

发生原因：HTML 中较大的内联 data: URL（>10MB）会触发解析器内存限制。这与响应大小无关——10MB 纯文本可以正常工作，但包含嵌入式数据 URL 的 10MB HTML 会失败。使用内联图片显示绘图的 Python Jupyter Notebook 中常见此问题。

预防措施：

# ❌ FAILS - HTMLRewriter triggered on notebook HTML with data: URLs
response = await fetch("https://origin.example.com/notebook.html")
return response  # Crashes if HTML contains large data: URLs

# ✅ WORKS - Stream directly or use text/plain
response = await fetch("https://origin.example.com/notebook.html")
headers = {"Content-Type": "text/plain"}  # Bypass parser
return Response(await response.text(), headers=headers)

解决方法：

• 避免对笔记本内容使用 HTMLRewriter（直接进行流式传输）
• 预处理笔记本，将数据 URL 提取到外部文件
• 使用 text/plain 内容类型绕过解析器

问题 #11：初始化期间无法为 PRNG 设定种子

错误：部署因用户错误而失败 来源：Python Workers Redux Blog

原因：Wasm 快照不支持在请求处理程序运行前初始化 PRNG。如果你在模块初始化期间调用伪随机数生成器 API（例如 random.seed()），部署会失败。

预防措施：

import random

# ❌ FAILS deployment - module-level PRNG call
random.seed(42)

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response(str(random.randint(1, 100)))

# ✅ WORKS - PRNG calls inside handlers
class Default(WorkerEntrypoint):
    async def fetch(self, request):
        random.seed(42)  # Initialize inside handler
        return Response(str(random.randint(1, 100)))

仅在请求处理程序内调用 PRNG 函数，不要在模块级别调用。

---

最佳实践

始终执行

• 使用 WorkerEntrypoint 类模式
• 使用异步 HTTP 客户端（httpx、aiohttp）
• 将所有 I/O 操作放入工作流步骤中
• 添加 python_workers 兼容性标志
• 所有绑定都使用 self.env
• 从工作流步骤返回可序列化为 JSON 的数据

绝不要执行

• 使用同步 HTTP 库（requests）
• 使用原生／编译型软件包
• 在工作流步骤之外执行 I/O 操作
• 将旧版 @handler 装饰器用于 fetch
• 期待达到 JavaScript 级别的冷启动时间

---

框架说明：FastAPI

FastAPI 可以与 Python Workers 配合使用，但存在一些限制：

from fastapi import FastAPI
from workers import WorkerEntrypoint

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello from FastAPI"}

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Route through FastAPI
        return await app(request)

限制：

• 仅支持异步（不支持同步端点）
• 不支持 WSGI 中间件
• 稳定性仍处于 Beta 阶段

详情请参阅 Cloudflare FastAPI 示例。

---

官方文档

• Python Workers 概览
• Python Workers 基础知识
• Python Workers 的工作原理
• Python 软件包
• FFI（外部函数接口）
• Python 工作流
• Pywrangler CLI
• Pyodide 软件包列表

---

依赖项

{
  "workers-py": "1.7.0",
  "workers-runtime-sdk": "0.3.1",
  "wrangler": "4.58.0"
}

注意：始终固定版本，以确保构建可复现。请查看 PyPI workers-py 获取最新版本。

---

生产环境验证

• Cloudflare 更新日志：2025 年 12 月 8 日（Pywrangler + 冷启动改进）
• workers-py 1.7.0：最新稳定版本（2026 年 1 月）
• Python Workflows Beta：2025 年 8 月 22 日
• 处理程序模式变更：2025 年 8 月 14 日

兼容性日期指南：

• 新项目使用 2025-12-01（包含 pywrangler 改进等最新功能）
• 仅当需要与较旧的生产环境 Workers 保持一致时，才使用 2025-08-01