pycronet 是面向 Windows x64 的 Python HTTP/WebSocket 客户端。Python 通过 ctypes 调用小型 C++ Shim,Shim 再调用 Chromium Native Cronet;Python 不直接管理 Cronet 回调指针和原生对象生命周期。
当前包版本为 0.3.0,底层目标是 Chromium 145.0.7632.162。Chromium 145 之后官方不再提供 Native Cronet 接口;升级内核时必须同步维护补丁、ABI 和构建产物。
- Requests 风格的 Session、AsyncSession 和模块级请求函数。
- 完整响应与流式响应,支持取消、超时、重定向和 Cookie。
- HTTP、HTTPS、HTTP/HTTPS 代理、SOCKS5、SOCKS5H。
- 每个 Session 独立的 TLS profile、证书校验策略和 CookieStore。
- Cronet WebSocket:WSS、子协议、Origin、自定义请求头、文本和二进制消息。
- 文件上传和下载;不需要 CFFI、Rust、Cargo、PyO3、nanobind 或额外网络库。
运行时需要 Windows x64、Python 3.9 至 3.14 和 Microsoft Visual C++ 运行库。安装 wheel 后,包内应包含:
pycronet/
_native_bin/
pycronet_shim.dll
cronet.145.0.7632.162.dll
加载器只接受带完整版本号的 Cronet DLL。如果报 DLL 缺失,请确认这两个文件都在包内的 _native_bin 目录,且名称未被修改。
py -3.12 -m pip install -e .构建 Shim 需要 Visual Studio 2022 C++、Windows SDK、CMake,以及已经构建好的 Cronet DLL 和导入库。Chromium 源码可以放在项目目录旁边或其他位置,通过环境变量传入:
$env:CHROMIUM_ROOT = "../chromium"
$env:CRONET_ROOT = "../chromium/src/out/Cronet145"
cmake -S native_shim -B build/native-shim -G "Visual Studio 17 2022" -A x64 -DCHROMIUM_ROOT=$env:CHROMIUM_ROOT -DCRONET_ROOT=$env:CRONET_ROOT
cmake --build build/native-shim --config ReleaseCRONET_ROOT 必须包含 cronet.145.0.7632.162.dll 和 cronet.145.0.7632.162.dll.lib。构建会把 Shim 和 Cronet DLL 复制到 pycronet/_native_bin/;.lib 只在编译时使用,.pdb 只用于调试,不应放入 wheel。也可以运行 py -3.12 setup.py build_py,它读取相同的环境变量。
重复请求应复用 Session,以复用 Engine、连接池、Cookie 和 TLS 配置:
import pycronet
with pycronet.Session(timeout=15.0, chrometls="chrome_150") as session:
response = session.get("https://example.com/")
response.raise_for_status()
print(response.status_code, response.text)timeout 单位是秒,可在单次请求覆盖。支持 get、post、put、delete、patch、head、options,参数包括 params、headers、cookies、data 和 json:
with pycronet.Session(headers={"User-Agent": "my-client/1.0"}) as session:
response = session.post(
"https://example.com/api",
params={"page": 1},
json={"name": "test"},
headers={"X-Request-ID": "demo"},
)模块级 pycronet.get、pycronet.post 等函数适合一次性同步请求;每次调用都会创建并关闭临时 Session:
response = pycronet.get("https://example.com/api", params={"page": 1}, timeout=10.0)
print(response.status_code, response.json())异步的一次性请求使用 pycronet.async_get、pycronet.async_post 等函数,必须在异步函数中等待:
async def fetch_once():
response = await pycronet.async_get("https://example.com/api", timeout=10.0)
return response.json()
print(asyncio.run(fetch_once()))需要连续请求、连接复用、共享 Cookie 或共享 TLS 配置时,应创建显式 Session。异步代码也可以使用 AsyncSession:
import asyncio
import pycronet
async def main():
async with pycronet.AsyncSession(chrometls="chrome_150") as session:
response = await session.get("https://example.com/")
print(response.status_code, response.text)
asyncio.run(main())Response 提供 status_code、headers、url、content、text、json() 和 cookies。raise_for_status() 在 4xx/5xx 时抛出 HTTPStatusError;网络、超时和 Cronet 错误抛出 RequestError。
大响应使用 stream(),通过有界事件队列逐块消费:
with pycronet.Session() as session:
with session.stream("GET", "https://example.com/large.bin") as response:
response.raise_for_status()
with open("large.bin", "wb") as output:
for chunk in response.iter_content(64 * 1024):
output.write(chunk)异步流式读取使用 aiter_content() 或 aiter_lines()。流关闭后会释放底层请求,建议始终使用 with/async with。
Session 默认维护 Python CookieJar。需要让 Chromium 在自动重定向等原生流程中携带 Cookie 时,开启内存 CookieStore:
with pycronet.Session(enable_cookie_store=True) as session:
session.get("https://example.com/login")
response = session.get("https://example.com/account")proxies 支持 http://、https://、socks5:// 和 socks5h://:
with pycronet.Session(proxies="http://127.0.0.1:8080") as session:
response = session.get("https://example.com/")
with pycronet.Session(proxies="socks5://user:password@127.0.0.1:1080") as session:
response = session.get("https://example.com/")用户名和密码只用于代理认证,不会写入日志、错误文本或公开代理字符串;每个凭据最多 255 个 UTF-8 字节。非法 URI、未知 scheme 或超长凭据会在创建 Session 时报告错误。不同 Session 不共享 Cookie、代理或 TLS 配置。
内置 profile 位于 pycronet/tls_profiles.json,常用名称有 chrome_144、chrome_150 和测试 profile:
response = pycronet.get("https://example.com/", chrometls="chrome_150", verify=False)verify=False 仅关闭该 Engine 的证书校验,适合本地自签名测试,不建议用于生产。cipher、曲线和签名算法名称会在 Python 层转换为 Chromium experimental options 使用的十六进制 ID;0x 开头的值会规范化,未知名称会跳过,全部无效时回退 Chromium 默认值。TLS_GREASE 不转换为固定数字,而由 TLS 栈按连接生成。
可运行时注册 profile:
pycronet.add_tls_profile("local_test", {
"cipher_suites": ["TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "0xC014"],
"tls_curves": ["X25519", "P-256"],
"tls_extensions": [],
"signature_algorithms": ["rsa_pss_rsae_sha256"],
})
with pycronet.Session(chrometls="local_test") as session:
response = session.get("https://example.com/")profile 不能突破 Chromium 145/BoringSSL 的实际能力。TLS 1.3 cipher 顺序、部分扩展顺序和签名算法可用性可能与其他版本不同。
通过 Session 创建 WebSocket,推荐使用 async with:
import pycronet
async with pycronet.AsyncSession(chrometls="chrome_150") as session:
async with session.websocket(
"wss://echo.websocket.events",
subprotocols=["chat"],
origin="https://example.com",
headers={"X-Client": "pycronet"},
) as ws:
await ws.send_text("hello")
print(await ws.recv())
await ws.send_binary(b"data")subprotocols 是子协议候选列表,会放入 Sec-WebSocket-Protocol 请求头。服务端从中选择一个��在握手响应中确认;没有支持的协议时可能拒绝连接。它不是 Content-Type,也不会改变单条消息的文本或二进制类型。
send_text() 发送 UTF-8 文本,send_binary() 发送 bytes,recv() 返回 str 或 bytes。正常关闭会抛出 ConnectionClosed;网络错误抛出 WebSocketError。
需要回调风格时使用 WebSocketApp。run_forever() 默认阻塞当前线程,传入 blocking=False 可在后台线程运行。回调参数依次是 WebSocketApp、消息、关闭码和原因,或异常���象:
def on_open(ws):
ws.send("hello")
def on_message(ws, message, is_text):
print("received:", message, "text=" + str(is_text))
def on_close(ws, code, reason, was_clean):
print("closed:", code, reason, "clean=" + str(was_clean))
def on_error(ws, error, net_error):
print("error:", error, "net_error=" + str(net_error))
with pycronet.Session() as session:
app = pycronet.WebSocketApp(
session, "wss://echo.websocket.events",
on_open=on_open, on_message=on_message,
on_close=on_close, on_error=on_error,
subprotocols=["chat"],
)
app.run_forever()with pycronet.Session() as session:
session.upload_file("https://example.com/upload", "image.png",
field_name="file", additional_fields={"tag": "demo"})
result = session.download_file("https://example.com/archive.zip",
"downloads/archive.zip")
print(result["size"])一次性同步函数 pycronet.upload_file 和 pycronet.download_file 会自动创建临时 Session:
uploaded = pycronet.upload_file(
"https://example.com/upload", "image.png",
field_name="file", additional_fields={"tag": "demo"}, timeout=30.0,
)
downloaded = pycronet.download_file(
"https://example.com/archive.zip", "downloads/archive.zip", timeout=60.0,
)
print(uploaded.status_code, downloaded["file_path"], downloaded["size"])异步函数 pycronet.async_upload_file 和 pycronet.async_download_file 需要使用 await:
async def transfer():
uploaded = await pycronet.async_upload_file(
"https://example.com/upload", "image.png", field_name="file"
)
downloaded = await pycronet.async_download_file(
"https://example.com/archive.zip", "downloads/archive.zip"
)
return uploaded.status_code, downloaded["file_path"]py -3.12 -m pip install -e .[test]
py -3.12 -m pytest -q
py -3.12 -m pip install tox
toxtox 配置覆盖 Python 3.9 至 3.14,未安装的解释器会跳过。基准脚本在独立子进程中比较 pycronet、requests、httpx、aiohttp 和 curl_cffi:
py -3.12 benchmarks/run_all.py
py -3.12 benchmarks/run_all.py --quick结果写入 benchmarks/results/,按相同模式、payload、并发条件分组,再按 RPS 降序和 p50 延迟升序排列。该基准主要衡量本地 HTTP 桥接和连接复用开销,不能替代公网 HTTPS、代理、上传或 WebSocket 专项测试。
pycronet/
__init__.py 公共导出
api.py 模块级请求函数
_session.py Session 和 AsyncSession
_transport.py 同步/异步传输层
response.py Response 和 StreamResponse
cookies.py CookieJar
profiles.py TLS profile 注册与序列化
tls_ids.py TLS 名称到十六进制 ID 的映射
websocket.py WebSocket 和 WebSocketApp
_native/ ctypes 绑定和事件分发
_native_bin/ 运行时 DLL
native_shim/ C++17 Shim 工程
patch_chromium_145_0_7632_162/
Chromium 145 补丁和构建说明
tests/ Python 测试
benchmarks/ 性能脚本和结果
tls_fingerprint_server/ 本地 TLS 指纹测试服务
带下划线的模块是内部实现细节,不建议业务代码直接导入。
- 仅支持 Windows x64,不支持 ARM64、Linux 或 macOS。
- 不支持 free-threaded Python。
- Shim 和 Cronet DLL 必须使用同一版本和架构。
- TLS profile 不能突破 Chromium 145/BoringSSL 的能力。
- verify=False 会降低证书安全性,只应在受控测试环境使用。
- 升级 Chromium 后必须重新验证 ABI、DLL 和 profile。
本项目参考了 cronet-cloak、cyCronet 和 python-cronet。感谢这些项目及其贡献者提供的实践经验。