Cloudflare Python Workers 教程：使用 Pyodide 在边缘运行 Python 代码 | SkillsMD

Cloudflare Python Workers 教程：使用 Pyodide 在边缘运行 Python 代码

cloudflare-python-workers by jezweb/claude-skills

316 周安装量

670 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/jezweb/claude-skills --skill cloudflare-python-workers

Python Web框架开发云服务

🇨🇳中文介绍

Cloudflare Python Workers

状态：测试版（需要 python_workers 兼容性标志） 运行时：Pyodide（Python 3.12+ 编译为 WebAssembly） 包版本：workers-py@1.7.0, workers-runtime-sdk@0.3.1, wrangler@4.58.0 最后验证：2026-01-21

快速开始（5分钟）

1. 先决条件

确保已安装：

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

2. 初始化项目

# 创建项目目录
mkdir my-python-worker && cd my-python-worker

# 初始化 Python 项目
uv init

# 安装 pywrangler
uv tool install workers-py

# 初始化 Worker 配置
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!")

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

相关 Skills

FlyClaw：零登录航班聚合查询工具，Python实现多源航班信息与价格搜索

4,000,000 周安装

find-skills 技能搜索工具 - Vercel Labs 开源智能体技能包管理器

776,000 周安装

Vercel React 最佳实践指南 | 58条Next.js性能优化规则与代码重构

261,300 周安装

Vercel Web界面规范检查工具 - 自动检测代码是否符合Web设计指南

210,800 周安装

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

uv run pywrangler dev
# 访问 http://localhost:8787

uv run pywrangler deploy

# 仅限于内置包
# 只能使用 httpx、aiohttp、beautifulsoup4 等
# 错误："You cannot yet deploy Python Workers that depend on
# packages defined in requirements.txt [code: 10021]"

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



uv tool install workers-py
uv run pywrangler deploy  # 现在可以工作了！

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # 通过 self.env 访问绑定
        value = await self.env.MY_KV.get("key")

        # 解析请求
        url = request.url
        method = request.method

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

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

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

        # R2 对象存储
        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")

from workers import WorkerEntrypoint, Response
import json

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

        # 查询参数
        url = URL(request.url)
        name = url.searchParams.get("name", "World")

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

from workers import handler

@handler
async def on_scheduled(event, env, ctx):
    # 按 cron 计划运行
    print(f"Cron triggered at {event.scheduledTime}")

    # 执行工作...
    await env.MY_KV.put("last_run", str(event.scheduledTime))

{
  "triggers": {
    "crons": ["*/5 * * * *"]  // 每 5 分钟
  }
}

await step.do("my step", async () => {
  // 内联回调
  return result;
});

@step.do("my step")
async def my_step():
    # 带装饰器的命名函数
    return result

result = await my_step()

import asyncio

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

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

# 并发执行（类似 Promise.all）
results = await asyncio.gather(step_a(), step_b())
# results = ["A", "B"]

from workers import WorkflowEntrypoint, WorkerEntrypoint, Response

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, 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()

        # 步骤 2：休眠
        await step.sleep("wait", "10 seconds")

        # 步骤 3：处理
        @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):
        # 创建工作流实例
        instance = await self.env.MY_WORKFLOW.create()
        return Response(f"Workflow started: {instance.id}")

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 等待 step_a 和 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"
    }
  ]
}

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

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

# ✅ 可用 - httpx（异步）
import httpx

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

# ✅ 可用 - aiohttp
import aiohttp

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

# ❌ 不可用 - requests（同步）
import requests  # 会失败！

from js import fetch, console, Response as JSResponse

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

        # 控制台日志
        console.log("Fetched data:", data)

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

from js import Object
from pyodide.ffi import to_js

# ❌ 错误 - ImportError！
from pyodide.ffi import to_py
python_data = to_py(js_data)

# ✅ 正确 - to_py() 是一个方法
async def fetch(self, request):
    data = await request.json()  # 返回 JS 对象
    python_data = data.to_py()   # 转换为 Python 字典

# 将 Python 字典转换为 JavaScript 对象
python_dict = {"name": "test", "count": 42}
js_object = to_js(python_dict, dict_converter=Object.fromEntries)

# 在 Response 中使用
return Response(to_js({"status": "ok"}))

# ❌ 旧版（已弃用）
@handler
async def on_fetch(request):
    return Response("Hello")

# ✅ 新版（当前）
class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello")

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

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

{
  "compatibility_flags": ["python_workers"]
}

# ❌ 错误 - 在步骤外进行 fetch
response = await fetch("https://api.example.com")
@step.do("use data")
async def use_data():
    return await response.json()  # 重试时 response 可能已过时

# ✅ 正确 - 在步骤内进行 fetch
@step.do("fetch and use")
async def fetch_and_use():
    response = await fetch("https://api.example.com")
    return await response.json()

@step.do("process")
async def process():
    # 将 datetime 转换为字符串
    return {"timestamp": datetime.now().isoformat()}

# ❌ 无效 - 分开的终端
# 终端 1: npx wrangler dev (JS worker)
# 终端 2: npx wrangler dev (Python worker)
# 结果：网络连接丢失错误

# ✅ 有效 - 单个 wrangler 实例
npx wrangler dev -c ts/wrangler.jsonc -c py/wrangler.jsonc

# ❌ 失败 - 在包含 data: URL 的 notebook HTML 上触发 HTMLRewriter
response = await fetch("https://origin.example.com/notebook.html")
return response  # 如果 HTML 包含大型 data: URL 则崩溃

# ✅ 有效 - 直接流式传输或使用 text/plain
response = await fetch("https://origin.example.com/notebook.html")
headers = {"Content-Type": "text/plain"}  # 绕过解析器
return Response(await response.text(), headers=headers)

import random

# ❌ 部署失败 - 模块级别的 PRNG 调用
random.seed(42)

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

# ✅ 有效 - 在处理程序内部调用 PRNG
class Default(WorkerEntrypoint):
    async def fetch(self, request):
        random.seed(42)  # 在处理程序内部初始化
        return Response(str(random.randint(1, 100)))

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):
        # 通过 FastAPI 路由
        return await app(request)

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

🇺🇸English

Cloudflare Python Workers

Status : Beta (requires python_workers compatibility flag) Runtime : Pyodide (Python 3.12+ compiled to WebAssembly) Package Versions : workers-py@1.7.0, workers-runtime-sdk@0.3.1, wrangler@4.58.0 Last Verified : 2026-01-21

Quick Start (5 Minutes)

1. Prerequisites

Ensure you have installed:

uv - Python package manager
Node.js - Required for Wrangler

2. Initialize Project

# 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. Create Entry Point

Create src/entry.py:

from workers import WorkerEntrypoint, Response

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

4. Configure wrangler.jsonc

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

5. Run Locally

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

6. Deploy

uv run pywrangler deploy

Migration from Pre-December 2025 Workers

If you created a Python Worker before December 2025, you were limited to built-in packages. With pywrangler (Dec 2025), you can now deploy with external packages.

Old Approach (no longer needed):

# 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]"

New Approach (pywrangler):

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



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

Historical Timeline :

April 2024 - Dec 2025 : Package deployment completely blocked
Dec 8, 2025 : Pywrangler released, enabling package deployment
Jan 2026 : Open beta with full package support

See : Package deployment issue history

Core Concepts

WorkerEntrypoint Class Pattern

As of August 2025, Python Workers use a class-based pattern (not global handlers):

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}")

Accessing Bindings

All Cloudflare bindings are accessed via 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")

Supported Bindings :

D1 (SQL database)
KV (key-value storage)
R2 (object storage)
Workers AI
Vectorize
Durable Objects
Queues
Analytics Engine

See Cloudflare Bindings Documentation for details.

Request/Response Handling

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}!")

Scheduled Handlers (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))

Configure in wrangler.jsonc:

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

Python Workflows

Python Workflows enable durable, multi-step automation with automatic retries and state persistence.

Why Decorator Pattern?

Python Workflows use the @step.do() decorator pattern because Python does not easily support anonymous callbacks (unlike JavaScript/TypeScript which allows inline arrow functions). This is a fundamental language difference, not a limitation of Cloudflare's implementation.

JavaScript Pattern (doesn't translate):

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

Python Pattern (required):

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

result = await my_step()

Source : Python Workflows Blog

Concurrency with asyncio.gather

Pyodide captures JavaScript promises (thenables) and proxies them as Python awaitables. This enables Promise.all-equivalent behavior using standard Python async patterns:

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"]

Why This Works : JavaScript promises from workflow steps are proxied as Python awaitables, allowing standard asyncio concurrency primitives.

Source : Python Workflows Blog

Basic Workflow

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 Dependencies

Define step dependencies for parallel execution:

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()

Workflow Configuration

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

Package Management

pyproject.toml Configuration

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

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

Supported Packages

Python Workers support:

Pure Python packages from PyPI
Pyodide packages (pre-built for WebAssembly)

See Pyodide packages list.

HTTP Clients

Only async HTTP libraries work:

# ✅ 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!

Requesting New Packages

Request support for new packages at: https://github.com/cloudflare/workerd/discussions/categories/python-packages

FFI (Foreign Function Interface)

Access JavaScript APIs from Python via Pyodide's FFI:

JavaScript Globals

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!")

Type Conversions

Important : to_py() is a METHOD on JavaScript objects, not a standalone function. Only to_js() is a function.

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"}))

Source : GitHub Issue #3322 (Pyodide maintainer clarification)

Known Issues Prevention

This skill prevents 11 documented issues :

Issue #1: Legacy Handler Pattern

Error : TypeError: on_fetch is not defined

Why : Handler pattern changed in August 2025.

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

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

Issue #2: Sync HTTP Libraries

Error : RuntimeError: cannot use blocking call in async context

Why : Python Workers run async-only. Sync libraries block the event loop.

# ❌ 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")

Issue #3: Native/Compiled Packages

Error : ModuleNotFoundError: No module named 'numpy' (or similar)

Why : Only pure Python packages work. Native C extensions are not supported.

Solution : Use Pyodide-compatible alternatives or check Pyodide packages.

Issue #4: Missing Compatibility Flags

Error : Error: Python Workers require the python_workers compatibility flag

Fix : Add to wrangler.jsonc:

{
  "compatibility_flags": ["python_workers"]
}

For Workflows, also add "python_workflows".

Issue #5: I/O Outside Workflow Steps

Error : Workflow state not persisted correctly

Why : All I/O must happen inside @step.do for durability.

# ❌ 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()

Issue #6: Type Serialization Errors

Error : TypeError: Object of type X is not JSON serializable

Why : Workflow step return values must be JSON-serializable.

Fix : Convert complex objects before returning:

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

Issue #7: Cold Start Performance

Note : Python Workers have higher cold starts than JavaScript. With Wasm memory snapshots (Dec 2025), heavy packages like FastAPI and Pydantic now load in ~1 second (down from ~10 seconds previously), but this is still ~2x slower than JavaScript Workers (~50ms).

Performance Numbers (as of Dec 2025):

Before snapshots : ~10 seconds for FastAPI/Pydantic
After snapshots : ~1 second (10x improvement)
JavaScript equivalent : ~50ms

Mitigation :

Minimize top-level imports
Use lazy loading for heavy packages
Consider JavaScript Workers for latency-critical paths
Wasm snapshots automatically improve cold starts (no config needed)

Source : Python Workers Redux Blog | InfoQ Coverage

Issue #8: Package Installation Failures

Error : Failed to install package X

Causes :

Package has native dependencies
Package not in Pyodide ecosystem
Network issues during bundling

Fix : Check package compatibility, use alternatives, or request support.

Issue #9: Dev Registry Breaks JS-to-Python RPC

Error : Network connection lost when calling Python Worker from JavaScript Worker Source : GitHub Issue #11438

Why It Happens : Dev registry doesn't properly route RPC calls between separately-run Workers in different terminals.

Prevention :

# ❌ 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

Run both workers in a single wrangler instance to enable proper RPC communication.

Issue #10: HTMLRewriter Memory Limit with Data URLs

Error : TypeError: Parser error: The memory limit has been exceeded Source : GitHub Issue #10814

Why It Happens : Large inline data: URLs (>10MB) in HTML trigger parser memory limits. This is NOT about response size—10MB plain text works fine, but 10MB HTML with embedded data URLs fails. Common with Python Jupyter Notebooks that use inline images for plots.

Prevention :

# ❌ 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)

Workarounds :

Avoid HTMLRewriter on notebook content (stream directly)
Pre-process notebooks to extract data URLs to external files
Use text/plain content-type to bypass parser

Issue #11: PRNG Cannot Be Seeded During Initialization

Error : Deployment fails with user error Source : Python Workers Redux Blog

Why It Happens : Wasm snapshots don't support PRNG initialization before request handlers. If you call pseudorandom number generator APIs (like random.seed()) during module initialization, deployment FAILS.

Prevention :

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)))

Only call PRNG functions inside request handlers, not at module level.

Best Practices

Always Do

Use WorkerEntrypoint class pattern
Use async HTTP clients (httpx, aiohttp)
Put all I/O inside workflow steps
Add python_workers compatibility flag
Use self.env for all bindings
Return JSON-serializable data from workflow steps

Never Do

Use sync HTTP libraries (requests)
Use native/compiled packages
Perform I/O outside workflow steps
Use legacy @handler decorator for fetch
Expect JavaScript-level cold start times

Framework Note: FastAPI

FastAPI can work with Python Workers but with limitations:

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)

Limitations :

Async-only (no sync endpoints)
No WSGI middleware
Beta stability

See Cloudflare FastAPI example for details.

Official Documentation

Dependencies

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

Note : Always pin versions for reproducible builds. Check PyPI workers-py for latest releases.

Production Validation

Cloudflare changelog: Dec 8, 2025 (Pywrangler + cold start improvements)
workers-py 1.7.0: Latest stable (Jan 2026)
Python Workflows beta: Aug 22, 2025
Handler pattern change: Aug 14, 2025

Compatibility Date Guidance :

Use 2025-12-01 for new projects (latest features including pywrangler improvements)
Use 2025-08-01 only if you need to match older production Workers

Weekly Installs

316

Repository

jezweb/claude-skills

GitHub Stars

652

First Seen

Jan 20, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykWarn

Installed on

claude-code264

gemini-cli211

opencode206

cursor200

antigravity194

codex183

Cloudflare Python Workers 教程：使用 Pyodide 在边缘运行 Python 代码

🇨🇳中文介绍

Cloudflare Python Workers

快速开始（5分钟）

1. 先决条件

2. 初始化项目

3. 创建入口点

相关 Skills

4. 配置 wrangler.jsonc

5. 本地运行

6. 部署

从 2025 年 12 月之前的 Workers 迁移

核心概念

WorkerEntrypoint 类模式

访问绑定

请求/响应处理

计划任务处理程序（Cron）

Python Workflows

为什么使用装饰器模式？

使用 asyncio.gather 实现并发

基本工作流

DAG 依赖关系

工作流配置

包管理

pyproject.toml 配置

支持的包

HTTP 客户端

请求新包

FFI（外部函数接口）

JavaScript 全局对象

类型转换

已知问题预防

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

问题 #2：同步 HTTP 库

问题 #3：原生/编译包

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

问题 #5：在工作流步骤之外进行 I/O 操作

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

问题 #7：冷启动性能

问题 #8：包安装失败

问题 #9：开发注册表破坏 JS 到 Python 的 RPC

问题 #10：HTMLRewriter 对数据 URL 的内存限制

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

最佳实践

始终要做

永远不要做

框架说明：FastAPI

官方文档

依赖项

生产验证

🇺🇸English

Cloudflare Python Workers

Quick Start (5 Minutes)

1. Prerequisites

2. Initialize Project

3. Create Entry Point

4. Configure wrangler.jsonc

5. Run Locally

6. Deploy

Migration from Pre-December 2025 Workers

Core Concepts

WorkerEntrypoint Class Pattern

Accessing Bindings

Request/Response Handling

Scheduled Handlers (Cron)

Python Workflows

Why Decorator Pattern?

Concurrency with asyncio.gather

Basic Workflow

DAG Dependencies

Workflow Configuration

Package Management

pyproject.toml Configuration

Supported Packages

HTTP Clients

Requesting New Packages

FFI (Foreign Function Interface)

JavaScript Globals

Type Conversions

Known Issues Prevention