无需Node.js/.NET及非Python依赖，如何连接SignalR WebSocket？

咱们完全可以用纯Python工具链来实现SignalR WebSocket的通信，不用碰Node.js或者.NET。SignalR的WebSocket不是普通的WebSocket，得先走完协商流程才能建立有效连接，下面一步步来操作：

1. 先装纯Python依赖 #

只需要两个轻量的纯Python库，直接用pip装就行：

pip install requests websockets cloudscraper

2. 第一步：获取SignalR协商参数 #

SignalR要求先发送一个HTTP GET请求到/negotiate端点，拿到connectionId这类核心参数。拿你提到的Bittrex服务举例，代码大概是这样：

import cloudscraper

def get_negotiate_info(base_url):
    negotiate_url = f"{base_url}/negotiate"
    # 有些服务会校验User-Agent，模拟个浏览器的头就行
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
    }
    # 用cloudscraper绕过Cloudflare，无需Node.js
    scraper = cloudscraper.create_scraper()
    res = scraper.get(negotiate_url, headers=headers)
    res.raise_for_status()  # 要是请求失败直接抛出错误
    return res.json()

# 替换成你要连接的SignalR服务基础URL
negotiate_data = get_negotiate_info("https://socket.bittrex.com/signalr")
connection_id = negotiate_data["ConnectionId"]

3. 第二步：建立WebSocket连接 #

拿到connectionId后，就能构建WebSocket的连接URL了。SignalR的WebSocket URL格式一般是{base_url替换成wss}/connect?transport=webSockets&connectionId={你的connectionId}，还要带上clientProtocol参数（一般是1.5）。然后就可以连接并收发消息了：

import asyncio
import websockets

async def connect_signalr(base_url, connection_id, target_pair):
    # 把http换成wss，构建WebSocket地址
    ws_url = f"{base_url.replace('https', 'wss')}/connect?transport=webSockets&connectionId={connection_id}&clientProtocol=1.5"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
    }
    
    async with websockets.connect(ws_url, extra_headers=headers) as ws:
        print("WebSocket连接成功啦！")
        # 发送订阅消息，格式要符合SignalR的要求：H是Hub名，M是方法名，A是参数数组，I是消息ID
        subscribe_msg = f'{{"H":"c2","M":"SubscribeToExchangeDeltas","A":["{target_pair}"],"I":1}}'
        await ws.send(subscribe_msg)
        print(f"已订阅{target_pair}的行情")
        
        # 持续接收消息
        try:
            async for msg in ws:
                print("收到消息：", msg)
        except websockets.exceptions.ConnectionClosed:
            print("连接被断开了")

# 运行异步任务
asyncio.run(connect_signalr("https://socket.bittrex.com/signalr", connection_id, "BTC-USD"))

4. 解决你遇到的Cloudflare问题 #

你之前碰到的OSError: Missing Node.js runtime是因为用了依赖Node.js的cfscrape，换成纯Python的cloudscraper就能解决，它不需要Node.js就能绕过Cloudflare的验证，上面的代码已经集成了这个方案。

完整可运行代码 #

把上面的步骤整合起来，直接就能跑的代码：

import cloudscraper
import asyncio
import websockets

def get_negotiate_info(base_url):
    negotiate_url = f"{base_url}/negotiate"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
    }
    scraper = cloudscraper.create_scraper()
    res = scraper.get(negotiate_url, headers=headers)
    res.raise_for_status()
    return res.json()

async def connect_and_listen(base_url, connection_id, target_pair):
    ws_url = f"{base_url.replace('https', 'wss')}/connect?transport=webSockets&connectionId={connection_id}&clientProtocol=1.5"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
    }
    
    async with websockets.connect(ws_url, extra_headers=headers) as ws:
        print("已连接到SignalR WebSocket")
        subscribe_msg = f'{{"H":"c2","M":"SubscribeToExchangeDeltas","A":["{target_pair}"],"I":1}}'
        await ws.send(subscribe_msg)
        print(f"已订阅{target_pair}的实时行情")
        
        try:
            async for msg in ws:
                print("收到行情消息：", msg)
        except websockets.exceptions.ConnectionClosed:
            print("连接意外断开，请检查网络或服务状态")

if __name__ == "__main__":
    signalr_base_url = "https://socket.bittrex.com/signalr"
    negotiate_data = get_negotiate_info(signalr_base_url)
    connection_id = negotiate_data["ConnectionId"]
    asyncio.run(connect_and_listen(signalr_base_url, connection_id, "BTC-USD"))

内容的提问来源于stack exchange，提问作者Greg