websocket握手错误需分层捕获:dns失败→socket.gaierror,tcp连接问题→oserror,tls失败→ssl.sslerror,http非101→invalidstatuscode,协议校验失败→invalidhandshake,非法响应→invalidmessage,并结合上下文智能分类。

WebSocket 异步长连接的握手阶段出错,不是靠通用 try-catch 就能“精准捕获+智能分类”的问题。它本质是异步协程中的协议级失败,错误来源分散、语义混杂,直接套用同步式异常处理容易漏判、误判,甚至掩盖真实原因。
关键在于:握手错误只发生在 await websockets.connect(...) 这一调用中,且绝大多数错误属于网络层或协议层异常,需按来源分层识别,而非统一 except Exception: 捕获。
? 握手阶段典型错误来源与对应捕获方式
WebSocket 握手本质是一次 HTTP 升级请求(GET + Upgrade: websocket),所以所有错误都源于这个过程:
- DNS 解析失败 →
socket.gaierror - TCP 连接超时/拒绝 →
OSError(含ConnectionRefusedError,TimeoutError) - TLS 握手失败(如证书无效、不支持的加密套件)→
ssl.SSLError - HTTP 响应非 101 →
websockets.exceptions.InvalidStatusCode(如 400/401/403/404/502/503) - Sec-WebSocket-Accept 校验失败 →
websockets.exceptions.InvalidHandshake - 服务端返回非 WebSocket 协议响应(如 HTML 页面)→
websockets.exceptions.InvalidMessage
这些异常类型明确、可区分,应逐类捕获,而非兜底。
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成,提供 184 种工具和 200 多条命令,用于安全分析和代码质量...
import websockets
import ssl
try:
async with websockets.connect(
"wss://example.com/ws",
ssl=ssl.create_default_context(), # 显式传入,便于控制 TLS 行为
open_timeout=10, # 控制握手总耗时
) as ws:
# 连接成功
pass
except socket.gaierror as e:
# DNS 解析失败:域名不存在、网络不可达
log_error("DNS_RESOLVE_FAILED", str(e))
except OSError as e:
if isinstance(e, (ConnectionRefusedError, TimeoutError)):
log_error("CONNECTION_FAILED", str(e))
else:
log_error("NETWORK_IO_ERROR", str(e))
except ssl.SSLError as e:
# TLS 层失败:证书问题、协议版本不兼容、SNI 错误等
log_error("TLS_HANDSHAKE_FAILED", str(e))
except websockets.exceptions.InvalidStatusCode as e:
# HTTP 状态码非 101:认证失败(401)、权限不足(403)、路由错误(404)、网关问题(502/503)
log_error(f"HTTP_STATUS_{e.status_code}", f"status={e.status_code}, headers={e.headers}")
except websockets.exceptions.InvalidHandshake as e:
# 协议握手字段缺失或校验失败:Sec-WebSocket-Accept 不匹配、缺少 Upgrade 头等
log_error("PROTOCOL_HANDSHAKE_INVALID", str(e))
except websockets.exceptions.InvalidMessage as e:
# 服务端返回了非法响应体(如 HTML 错误页、JSON API 响应),无法解析为 WebSocket 协议帧
log_error("UNEXPECTED_RESPONSE_BODY", str(e))
except Exception as e:
# 兜底:仅用于未预见的库内部异常(如 websockets 版本 bug)
log_error("UNKNOWN_HANDSHAKE_ERROR", type(e).__name__ + ": " + str(e))
? 智能分类的核心逻辑:不只是看异常类型,还要看上下文
光捕获异常类型不够,需结合连接参数和环境做二次判断,才能实现“智能分类”:
- 若使用
wss://但捕获ssl.SSLError,且错误信息含"certificate verify failed"→ 可归为 CERT_VERIFY_FAILED,提示用户检查证书信任链或临时禁用验证(仅测试); - 若
InvalidStatusCode(401)且 URL 中含 token 参数或已传Authorizationheader → 归为 AUTH_TOKEN_EXPIRED,触发刷新流程; - 若
InvalidStatusCode(403)且服务端域名是live.douyin.com→ 很可能因签名失效(如抖音ac_signature过期),应走签名重生成逻辑; - 若多次连续出现
ConnectionRefusedError或TimeoutError,且目标地址是内网 IP → 判定为 BACKEND_SERVICE_DOWN,而非客户端网络问题。
这类判断需封装成函数,避免硬编码:
def classify_handshake_error(exc, url: str, headers: dict = None):
if isinstance(exc, ssl.SSLError) and "verify failed" in str(exc).lower():
return "CERT_VERIFY_FAILED"
if isinstance(exc, websockets.exceptions.InvalidStatusCode):
if exc.status_code == 401:
return "AUTH_TOKEN_EXPIRED"
if exc.status_code == 403 and "douyin" in url:
return "SIGNATURE_INVALID"
return f"HTTP_{exc.status_code}"
if isinstance(exc, OSError) and "timed out" in str(exc).lower():
return "CONNECTION_TIMEOUT"
return type(exc).__name__.replace("Exception", "")
⚠️ 特别注意:不要在 except 里直接重连
很多开发者习惯在 except 块里立刻 await websockets.connect(...) 重试,这会引发两个严重问题:
- 没有退避机制 → 短时间内密集重试,可能被服务端限流或封禁;
- 未清理前置资源 → 如 DNS 缓存、SSL 上下文、连接池句柄,造成内存或 fd 泄漏。
正确做法是:
✅ 记录错误分类;
✅ 触发带指数退避的异步重试任务(用 asyncio.create_task() 启动);
✅ 在重试前清除相关缓存(如 socket.getaddrinfo 结果可清,但通常由 OS 管理);
✅ 若是认证类错误(401/403),先刷新凭证再重试。
不复杂但容易忽略。










