如何在 Jupyter 中识别验证码而不触发 asyncio.run 错误

要在 Jupyter Notebook 中识别验证码，先用 %pip 安装 capskip 包，让它装进内核自己的 Python。然后像在脚本里一样，在单元格中调用 CapSkip，或者直接在单元格的顶层 await 它的 AsyncCapSkip 客户端。千万不要做的，是把异步调用包进 asyncio.run()。内核本身已经在运行一个事件循环，所以 asyncio.run 会抛出 RuntimeError，Playwright 的同步 API 也会因为同样的原因拒绝启动。Jupyter Notebook 验证码识别还有两个特有的陷阱：停止按钮并不总能停下一次识别；在 Windows 上，旧版 ipykernel 会让 Playwright 无法启动浏览器。本指南涵盖安装、两种客户端、浏览器和远程内核，并已在 ipykernel 7.4 和 7.3 上测试。

你需要什么

• 在 Windows 机器上运行的 CapSkip。示例识别的是 reCAPTCHA v2，SDK 支持的其他所有类型在单元格里的用法都一样。
• Python 3.10 或更高版本，以及最新的 capskip 包（撰写本文时为 1.3.0），另外还需要 requests 和 Beautiful Soup 来处理页面。只有需要浏览器时才要装 Playwright。
• 任何形式的 Jupyter 都行：Notebook 7、JupyterLab，或者 VS Code 中的笔记本编辑器。三者都在 ipykernel 上运行你的单元格，下文提到的所有行为都源自它。
• 识别工具的地址。Local 模式只在 127.0.0.1 上响应，仅供本机使用；Server 模式监听你的网络地址或公网 IP，这样另一台机器上的内核就能通过 API 调用它。两者都位于 连接设置中，最后一节会说明什么时候需要 Server 模式。

第 1 步：用 %pip 安装到内核的 Python 中

在笔记本的单元格里运行安装命令，前面加上百分号：

# pip install capskip requests beautifulsoup4, run inside the notebook:
%pip install capskip requests beautifulsoup4

百分号很关键。IPython 的 %pip 魔法命令 会用内核自己的解释器（即 sys.executable）来运行 pip，所以包会装到你的单元格导入时查找的位置。换成感叹号，这一行就会交给 shell，由它去运行 PATH 上排在最前面的那个 pip。在装了不止一个 Python 的机器上，这是两个不同的安装环境。在一次测试中，内核里的 %pip 报告的是 capskip 1.3.0，而 shell 的 pip（来自全局的 Python 3.14）报告的却是旧的 1.0.1，没有较新的验证码类型。继续之前，先确认你用的是哪个 Python、装的是哪个版本：

import sys
import capskip

print(sys.executable)        # the Python this kernel runs on
print(capskip.__version__)   # 1.3.0 or later

如果某个包在你升级它时已经被导入过，就重启内核，让新版本加载进来。IPython 在执行 %pip 之后也会打印同样的提醒。

第 2 步：在单元格中用同步客户端识别

在笔记本里，同步客户端不需要任何特殊处理。用 requests.Session 获取页面，找到小组件，然后识别：

import requests
from bs4 import BeautifulSoup
from capskip import CapSkip

PAGE_URL = "https://example.com/login"

# pollingInterval caps the wait between polls. At 1 instead of the
# default 5, the stop button takes effect within about a second.
solver = CapSkip(host="127.0.0.1", port=8080, pollingInterval=1)

session = requests.Session()
page = session.get(PAGE_URL, timeout=30)
soup = BeautifulSoup(page.text, "html.parser")
widget = soup.select_one(".g-recaptcha[data-sitekey]")
form = widget.find_parent("form")

token = solver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL)["code"]
len(token)

然后在下一个单元格里提交，带上表单自身的字段，并使用同一个会话：

from urllib.parse import urljoin

def form_fields(form):
    """The fields a browser would send: hidden, text and checked boxes."""
    fields = {}
    for tag in form.select("input[name]"):
        kind = (tag.get("type") or "text").lower()
        if kind in ("submit", "button", "image", "file", "reset"):
            continue
        if kind in ("checkbox", "radio") and not tag.has_attr("checked"):
            continue
        fields[tag["name"]] = tag.get("value", "on" if kind in ("checkbox", "radio") else "")
    return fields

data = form_fields(form)
data["username"] = "YOUR_USERNAME"
data["g-recaptcha-response"] = token
res = session.post(urljoin(PAGE_URL, form.get("action") or PAGE_URL), data=data,
                   headers={"Referer": PAGE_URL}, timeout=30)
res.status_code

会话会把页面的 cookie 带到提交请求中，正是这些 cookie 把隐藏的 CSRF 字段和你这次访问绑定在一起。Referer 头用来代替浏览器会发送的那个：requests 不发送 Referer，而有些框架（Django 就是其中之一）会拒绝既没有 Referer 也没有 Origin 的 HTTPS 表单提交。一个 reCAPTCHA token 只能提交一次，有效期大约两分钟，所以识别单元格一运行完，就紧接着运行提交单元格，不要过一会儿再回来运行。

笔记本和脚本的不同之处，就在停止按钮上。中断会在你的单元格里引发 KeyboardInterrupt，但在 Windows 内核上，它要等当前调用返回后才会生效。一次识别的大部分时间里，SDK 都在两次轮询之间休眠，所以单元格要等这段休眠结束才会停下：使用默认的 pollingInterval 时最多要等五秒，按上面的设置则大约一秒。traceback 看起来有点怪：先是一个 NetworkException，然后是在处理它的过程中引发的 KeyboardInterrupt。那个 NetworkException 只是 SDK 表示尚未就绪的信号。

麻烦的情况是：CapSkip 接受了连接，却始终不应答。同步客户端没有给它的 HTTP 调用设置超时，所以在 Windows 上，单元格会卡在一次任何中断都够不着的 socket 读取上，而 recaptchaTimeout 也帮不上忙，因为 SDK 只在两次轮询之间检查它。重启内核可以让单元格解脱，但会丢掉所有变量。关闭 CapSkip 同样能让它解脱，而且变量都还在，因为连接一断，读取就结束了，你之前发出的中断会立刻生效。另外，中断只会停下你的单元格，永远不会停下 CapSkip：SDK 没有取消任务的调用，所以被放弃的识别会继续进行，而它的 token 永远不会被读取。

第 3 步：在单元格中使用异步：在顶层 await，绝不用 asyncio.run

在脚本里运行异步代码的常规写法，放到笔记本里就会失败：

import asyncio
from capskip import AsyncCapSkip

asolver = AsyncCapSkip(host="127.0.0.1", port=8080, pollingInterval=1)

asyncio.run(asolver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL))
# RuntimeError: asyncio.run() cannot be called from a running event loop

ipykernel 在它早已启动的事件循环里运行每一个单元格，而 asyncio.run 偏要启动一个新的。更老的写法 asyncio.get_event_loop().run_until_complete() 同样会失败，报错为 RuntimeError: This event loop is already running。这两种调用都到不了 CapSkip，所以不会提交任何任务。你传入的协程从未运行，因此之后的某个单元格可能会打印一条 RuntimeWarning，提示它从未被 await；这条警告无害。解决办法比问题本身还简单。IPython 允许你在单元格的顶层、在任何函数之外直接 await：

# No asyncio.run: the kernel's loop runs this await.
from capskip import AsyncCapSkip

asolver = AsyncCapSkip(host="127.0.0.1", port=8080, pollingInterval=1)
result = await asolver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL)
token = result["code"]

这让 AsyncCapSkip 成为笔记本里更好的客户端，除了写法风格之外还有两个原因。中断会立刻取消 await，所以单元格马上结束，而不必等到下一次轮询之后。另外，它的 HTTP 调用走的是 httpx，而 httpx 默认会在连接沉默五秒后放弃。这样一来，一个不再应答的 CapSkip 会让调用以 SDK 错误结束，而不是一直挂着：如果卡住的是轮询，会在 recaptchaTimeout 到期后不久抛出 TimeoutException；如果卡住的是提交，五秒后就会抛出 NetworkException。在此期间，停止按钮始终有效。要缩短这段等待，就在构造函数上设置 recaptchaTimeout；在 Python SDK 中，recaptcha() 不接受 timeout 参数，传入的话会抛出 TypeError。

要同时进行多个识别，一个 gather 就够了。CapSkip 默认用 10 个线程识别 reCAPTCHA（即设置中的 Max. Threads），排在它们后面的任务最多等待 250 秒，之后就会被 CapSkip 判为失败，所以要给这一批设个上限：

import asyncio

# Max. Threads for reCAPTCHA is 10 by default; more than that queues.
limit = asyncio.Semaphore(10)

async def solve(sitekey, url):
    async with limit:
        return (await asolver.recaptcha(sitekey=sitekey, url=url))["code"]

# jobs: (sitekey, url) pairs; here the Step 2 page three times
jobs = [(widget["data-sitekey"], PAGE_URL)] * 3
tokens = await asyncio.gather(*(solve(k, u) for k, u in jobs), return_exceptions=True)

gather 会在最后一次性交回所有 token，所以要及时提交，或者改用 asyncio.as_completed，每拿到一个就提交一个。设置 return_exceptions=True 后，失败的识别会以它的异常出现在列表中，而不是从单元格里向外抛出，所以只提交那些是字符串的项。同一批里的识别共用一台 CapSkip 机器，所以每一个都会比单独识别时慢一些。客户端还会为每次轮询新建一个 HTTP 客户端，在 Windows 上每次大约占用内核进程 0.1 秒的 CPU 时间，所以一大批识别会让一个 CPU 核心一直忙着。更大批量的处理模式详见 在 Python 中并行识别验证码的指南.

针对 asyncio.run 报错的一些老答案会推荐 nest_asyncio。别用它。这个项目已于 2024 年 3 月归档。在 Python 3.14 上应用它会把内核搞坏：从下一个单元格起，每次 httpx 调用都会失败，抛出一个关于 NoneType 对象弱引用的 TypeError，AsyncCapSkip 也不例外，直到你重启为止。顶层 await 根本不需要任何补丁。

第 4 步：在笔记本中使用浏览器

Selenium 没有自己的事件循环，所以 webdriver 会话在单元格里的表现和在脚本里完全一样：用第 2 步的同步客户端识别，再把 token 写进页面。Playwright 的同步 API 会检查是否有正在运行的事件循环，并以如下报错拒绝运行：Error: It looks like you are using Playwright Sync API inside the asyncio loop. Please use the Async API instead. 所以在笔记本里，要通过 Playwright 的异步 API 来驱动它，并在顶层使用 await：

# pip install playwright into the kernel, then download a browser once:
%pip install playwright
import sys
!"{sys.executable}" -m playwright install chromium

from playwright.async_api import async_playwright

# start() instead of "async with", so the browser outlives this cell.
pw = await async_playwright().start()
browser = await pw.chromium.launch()
page = await browser.new_page()
await page.goto(PAGE_URL)

sitekey = await page.locator(".g-recaptcha").get_attribute("data-sitekey")
token = (await asolver.recaptcha(sitekey=sitekey, url=page.url))["code"]

# The widget's hidden textarea is the field the form submits.
await page.locator("#g-recaptcha-response").evaluate("(el, t) => { el.value = t; }", token)
await page.locator("form button[type=submit]").click()

把 pw 和 browser 保存在变量里，你就能在后面的单元格里继续摆弄这个页面，而这正是笔记本最大的意义所在。用完之后，用 await browser.close() 和 await pw.stop() 把两者都关掉。调用内核自己的 sys.executable，可以让下载的浏览器与你的单元格所导入的 Playwright 相匹配，道理和 %pip 一样。引号能让带空格的路径保持完整，而这行感叹号命令会在单元格里显示下载进度。关于如何驱动浏览器本身，更多内容见 Playwright 验证码识别指南.

在 Windows 上，调试 Playwright 之前先检查你的 ipykernel 版本。在 7.3 及更早的版本中，ipykernel 在 Windows 上用 selector 事件循环运行单元格，而这种循环无法启动 asyncio 子进程，所以 chromium.launch() 会直接抛出一个不带任何说明的 NotImplementedError。2026 年 9 月底发布的 7.4.0 版把 Windows 上的单元格改到了 Proactor 循环上，同样的单元格就能正常运行了。打印 ipykernel.__version__ 就能看到你用的是哪个版本。要升级的话，在笔记本里运行 %pip install -U ipykernel，然后重启内核。如果无法升级，换用工作线程也没用，因为 selector 策略作用于整个内核进程。这时就改用 Selenium 驱动浏览器，它启动 chromedriver 时不依赖 asyncio。

内核不在你的电脑上时：Server 模式

上面的所有内容都假设内核和 CapSkip 运行在同一台机器上，所以 127.0.0.1 能访问到它。这是最常见的 Jupyter Notebook 验证码识别环境，但不是唯一的一种。由服务器上的 JupyterHub 提供的笔记本、VS Code 中的远程内核，或者 Google Colab 这类托管笔记本，都会在别处运行你的单元格，而在那里，127.0.0.1 指的是那台机器本身。于是第一次调用就会失败，抛出 NetworkException，提示连接被拒绝。在连接设置里把 CapSkip 切换到 Server 模式，它就会监听你的网络地址或公网 IP，远程内核以这个地址作为 host，就能通过同一套 API 调用它。

如果链路要经过公网，请使用静态公网 IP。Server 模式启动时，CapSkip 会向所有地址开放它的端口，所以要想只让你预期的机器访问，就为其他地址添加一条 Windows Firewall 阻止规则。这种做法适用于内核地址固定的情况，比如在你自己的 JupyterHub 服务器上。托管笔记本的出口地址可能在不同会话之间发生变化，所以在那种情况下，要在 CapSkip 的设置里打开 API 密钥校验，并把密钥作为 apiKey 传入。硬件仍然是你自己的，用量也仍然不计量。客户端不会自己读取任何环境变量，所以要在单元格里用 os.getenv 读取 CAPSKIP_HOST 并传进去，就像完整示例那样。

完整可运行示例

三个单元格，全程异步：这是能经受住停止按钮考验的最短完整 Jupyter Notebook 验证码识别流程。

# Cell 1: pip install capskip requests beautifulsoup4, then restart the kernel.
%pip install capskip requests beautifulsoup4

# Cell 2: the client and a helper. Run once per kernel.
import os
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup
from capskip import AsyncCapSkip, CapSkipError

PAGE_URL = "https://example.com/login"
solver = AsyncCapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
    pollingInterval=1,
)

def form_fields(form):
    fields = {}
    for tag in form.select("input[name]"):
        kind = (tag.get("type") or "text").lower()
        if kind in ("submit", "button", "image", "file", "reset"):
            continue
        if kind in ("checkbox", "radio") and not tag.has_attr("checked"):
            continue
        fields[tag["name"]] = tag.get("value", "on" if kind in ("checkbox", "radio") else "")
    return fields

# Cell 3: fetch, solve, post. Run it again for a fresh token.
session = requests.Session()
page = session.get(PAGE_URL, timeout=30)
soup = BeautifulSoup(page.text, "html.parser")
widget = soup.select_one(".g-recaptcha[data-sitekey]")
if widget is None:
    raise RuntimeError("No reCAPTCHA widget in the HTML; the page may build it with JavaScript.")
form = widget.find_parent("form")

try:
    result = await solver.recaptcha(sitekey=widget["data-sitekey"], url=PAGE_URL)
except CapSkipError as err:
    raise RuntimeError(f"solve failed: {type(err).__name__}: {err}") from err

data = form_fields(form)
data["username"] = "YOUR_USERNAME"
data["g-recaptcha-response"] = result["code"]
res = session.post(urljoin(PAGE_URL, form.get("action") or PAGE_URL), data=data,
                   headers={"Referer": PAGE_URL}, timeout=30)
print(res.status_code)

需要反复运行的是单元格 3，每运行一次就是一次新的识别、一个新的 token，因为一个 reCAPTCHA token 只能用一次。如果页面用 JavaScript 生成小组件，获取到的 HTML 里就没有 sitekey；这时改用第 4 步中的 Playwright 单元格，或者从渲染小组件的脚本里读取 sitekey。

常见错误及其含义

常见问题

简短版结论

Jupyter Notebook 验证码识别归结起来就是五个习惯。用 %pip 安装，绝不要通过 shell 安装。同步的 CapSkip 客户端在单元格里的用法和在脚本里一样；把 pollingInterval 设为 1，让中断能很快生效。使用异步时，在单元格的顶层 await AsyncCapSkip，绝不要把它包进 asyncio.run，也不要求助于 nest_asyncio。需要浏览器时，直接使用 Selenium，或者使用 Playwright 的异步 API，并且在 Windows 上运行 ipykernel 7.4 或更高版本。如果内核在另一台机器上，就把 CapSkip 切换到 Server 模式，并传入它的地址。

• Python 包能识别的所有验证码类型： Python 验证码识别页面.
• 这个小组件本身如何运作、识别工具覆盖哪些内容： reCAPTCHA v2 识别页面.

在笔记本里，爬虫是靠一次次重跑逐步搭建起来的。把 验证码绕过 工具放在你自己的机器上运行，每一次重跑就都是免费的。