| name | webai2api |
|---|---|
| description | Reverse-engineer any web AI chat interface into an OpenAI-compatible API. Upload a .har file (browser DevTools export), and automatically detect the chat endpoint, extract auth headers/cookies, analyze SSE streaming, identify PoW challenges, and generate a proxy server. Use when you need to use a web-only AI model through any OpenAI SDK (Claude Code, Cursor, Continue, etc.). |
User uploads a .har file. You parse it, extract every detail of the chat API, then generate an OpenAI-compatible proxy. Supports both HTTP (SSE/JSON) and WebSocket transports — auto-detected from the HAR file.
- Open target AI chat page, F12 → Network, check "Preserve log"
- Send a chat message, wait for full reply
- Right-click any request → "Save all as HAR with content"
- Upload the
.harfile
Optionally: specify model name (default gpt-4o).
from har_parser import parse_har
analysis = parse_har("uploaded.har")It scores every HAR entry by these rules (higher = more likely):
| Criterion | Points |
|---|---|
| Method is POST | +0 (required, else -100) |
URL path contains /api/chat, /v1/chat, /conversation, /completion etc. |
+50 |
Request body has "messages" array |
+40 |
"messages" has ≥1 item with non-empty "content" |
+20 |
Body has "prompt" field |
+30 |
Body has "stream" field |
+10 |
Body has "model" field |
+10 |
Body has "temperature", "max_tokens", "top_p" |
+5 each |
Response Content-Type is text/event-stream |
+30 |
Response Content-Type is application/json |
+15 |
| Response body is JSON with a long string field (>50 chars) | +5 |
URL path contains .js, .css, .png, analytics, favicon |
-50 each |
The entry with the highest score >= 0 is the chat API.
If no HTTP POST entry scored >= 0, or if a response.status == 101 (Switching Protocols) entry is found, parse_har tries WebSocket detection:
- Find entries with
status == 101that have a_webSocketMessagesarray - Require at least 1 "send" + 1 "receive" message in the array
- Score the messages: JSON body contains
message/content/promptkeywords, long string values,streamflag - If score >= 3, mark
analysis.has_websocket = Trueand populateanalysis.ws.*
WS scoring criteria:
| Criterion | Points |
|---|---|
Entry has _webSocketMessages with ≥1 send + ≥1 receive |
required |
| Send frame JSON has chat-like keys (message, content, prompt, text, query) | +2 each key |
Send frame JSON has stream field |
+1 |
| Send frame JSON has a string value longer than 10 characters | +2 |
| Receive frame JSON has a chat-like key with a long string value | +2 per frame |
| Minimum score to qualify | 3 |
The WS entry also provides:
- Send template: JSON structure of the first send message (used as payload blueprint)
- Init messages: any send messages that lack chat content keys (likely auth/init frames to replay on connect)
- Streaming detection: if ≥3 receive frames for a single send →
receive_is_streaming = True
analysis.base_url # "https://chat.example.com"
analysis.chat_endpoint # "/api/v0/chat/completion"
analysis.headers # dict of ALL request headers from the matched entry
analysis.cookies # raw Cookie header string
analysis.auth_header # Authorization header or None
analysis.auth_type # "none" | "pow" | "oauth"
analysis.request_body_template # parsed JSON body (a dict)
analysis.content_field_path # dot-path to content in non-streaming JSON response
analysis.is_streaming # True if response Content-Type was text/event-stream
analysis.sse_event_type # SSE event name from "event: ..." lines
analysis.sse_data_field # key name inside SSE data JSON that holds text
analysis.sse_format # "plain_token" | "path_op_value" | "nested" | "raw_text"
analysis.has_pow # True if any entry hit a path with "challenge"/"pow"
analysis.pow_endpoint # "/api/v0/chat/create_pow_challenge"
analysis.all_endpoints # list of all unique URL paths in HAR
analysis.chat_entry_index # index of the matched entry
analysis.supported_params # ["temperature", "top_p", "max_tokens", ...]
# inferred from HAR bodies (zero-cost probe)
analysis.has_websocket # True / False — WebSocket detected
analysis.ws.ws_url # "wss://chat.example.com/ws" (WS URL)
analysis.ws.send_template # JSON dict of the first send frame
analysis.ws.input_field # which key in send_template holds user msg
analysis.ws.receive_field # which key in receive frames holds AI text
analysis.ws.receive_is_streaming # True if multiple receives per send
analysis.ws.type_field # if frames use a type/event discriminator
analysis.ws.extra_send_fields # non-input keys to carry in every send
analysis.ws.init_messages # auth/init frames to replay after connectOpen analysis.request_body_template and inspect it. Below are the exhaustive set of patterns, ordered by detection priority. Pick the first match.
# HAR body: {"messages": [{"role":"user","content":"Hello"}], "stream":true, "model":"deepseek"}
# Strategy: pass messages through directly, keep all other keys
payload = {
"messages": messages, # OpenAI messages → pass straight through
"stream": stream,
}
# Also carry over any extra HAR keys not related to messages content:
for k, v in analysis.request_body_template.items():
if k not in ("messages", "stream"):
payload[k] = v # e.g. model, temperature, session_id# HAR body: {"prompt":"Hello", "stream":true}
# Strategy: extract last user message as plain string
last = messages[-1]["content"] if messages else ""
if isinstance(last, list):
# ContentPart array: extract text parts joined
last = " ".join(p.get("text","") for p in last if p.get("type")=="text")
payload = {"prompt": last, "stream": stream}
for k, v in analysis.request_body_template.items():
if k not in ("prompt", "stream"):
payload[k] = v# HAR body: {"query":"Hello", "history":[{"role":"user","content":"Hi"}], "stream":true}
# Strategy: query = last message, convert OpenAI messages to history format
last = messages[-1]["content"] if messages else ""
payload = {"query": last, "stream": stream}
if "history" in analysis.request_body_template:
payload["history"] = messages[:-1] # all except last
for k, v in analysis.request_body_template.items():
if k not in ("query", "history", "stream"):
payload[k] = vpayload = {"inputs": last, "stream": stream}
for k, v in analysis.request_body_template.items():
if k not in ("inputs", "stream"):
payload[k] = v# Find the field with the longest string value in the template
# That's the "user input" field. Also track if there's a history array.
input_field = None
history_field = None
max_len = 0
for k, v in analysis.request_body_template.items():
if isinstance(v, str) and len(v) > max_len:
max_len = len(v)
input_field = k
if isinstance(v, list) and v and isinstance(v[0], dict) and "role" in v[0]:
history_field = k
# Reconstruct
payload = dict(analysis.request_body_template)
last = messages[-1]["content"] if messages else ""
payload[input_field] = last
if history_field:
payload[history_field] = messages[:-1]
payload["stream"] = streamThe target API uses WebSocket, not HTTP. Do NOT modify adapter.py. Instead, generate ws_adapter.py from templates/ws_adapter.py.
Send frame format — derived from analysis.ws.send_template:
# HAR show first send frame was: {"type": "chat", "content": "Hello", "stream": true}
# analysis.ws.input_field = "content"
# analysis.ws.extra_send_fields = {"type": "chat"}
payload = dict(analysis.ws.extra_send_fields) # {"type": "chat"}
last = messages[-1]["content"] if messages else ""
if isinstance(last, list):
last = " ".join(p.get("text","") for p in last if p.get("type")=="text")
payload[analysis.ws.input_field] = last # {"type": "chat", "content": "Hello"}
if stream:
payload["stream"] = TrueIf input_field is "messages", pass the full messages array instead:
# HAR: {"messages": [...], "model": "gpt-4o"}
payload = dict(analysis.ws.extra_send_fields)
payload["messages"] = messages
if stream:
payload["stream"] = TrueReceive extraction — based on analysis.ws.receive_field. Same dot-path logic as Step 2.2.
Init messages — if analysis.ws.init_messages is non-empty, send them in order right after WebSocket connect before the first user message:
await ws.send(json.dumps(init_msg))
# ... for each init_msg in analysis.ws.init_messagesType field filtering — if analysis.ws.type_field is set, skip receive frames whose type is error, ping, pong, ack, status, typing, etc.
analysis.supported_params contains every OpenAI-compatible parameter that the target API was observed using in the HAR. Use this list to dynamically decide which kwargs to pass through:
# In __init__, store the supported list:
self.supported_params = analysis.supported_params
# e.g. ["temperature", "top_p", "max_tokens", "stop", "presence_penalty"]
# In convert_request, after building the payload:
for param in self.supported_params:
# User's request may have passed this param via kwargs
val = kwargs.get(param)
if val is not None:
payload[param] = valHow it works at runtime:
| User sends via OpenAI SDK | analysis.supported_params contains |
adapter puts in payload |
|---|---|---|
temperature=0.7 |
"temperature" |
temperature: 0.7 |
temperature=0.7 |
not in list | omitted — target doesn't support it |
max_tokens=4096, temperature=0.5 |
["temperature"] |
only temperature: 0.5 passed, max_tokens dropped |
stop=["\n\n"], frequency_penalty=0.3 |
["stop"] |
only stop: ["\n\n"] passed |
How har_parser.py infers the list:
1. Collect all HAR POST entries whose URL path matches the chat endpoint.
2. For each entry, extract JSON body keys.
3. Match keys against a known alias table:
"temperature" → "temperature"
"max_tokens" → "max_tokens"
"max_length" → "max_tokens" (alias)
"max_new_tokens" → "max_tokens" (alias)
"top_p" → "top_p"
"top_k" → "top_k"
"presence_penalty" → "presence_penalty"
"frequency_penalty" → "frequency_penalty"
"repetition_penalty" → "repetition_penalty"
"stop" → "stop"
"stop_sequences" → "stop" (alias)
"n" → "n"
"seed" → "seed"
"user" → "user"
4. If a key appears in any request with a non-null, non-zero value,
its canonical name is added to supported_params.
This is zero-cost: it reads what the browser already sent, without sending any extra probe requests.
Verification (optional, added to Step 6):
# If temperature is supported, send two requests with extreme values
# and confirm the responses are meaningfully different
if "temperature" in adapter.supported_params:
cold = await _send_with_temp(adapter, 0.0)
hot = await _send_with_temp(adapter, 2.0)
if len(cold) == len(hot) and cold == hot:
print("[WARN] temperature set but responses identical — target may ignore it")if tools and self.dsml_enabled and self.dsml_ready:
if tool_choice != "none":
dsml_prompt = build_dsml_tool_prompt(tools, tool_choice)
messages = self._inject_dsml_prompt(messages, dsml_prompt)Place this before the payload construction so messages is already modified.
If analysis.has_websocket is True, generate ws_adapter.py from templates/ws_adapter.py. Do NOT modify adapter.py.
Modification table — map each analysis.ws.* field to the corresponding WebSocketChatAdapter attribute:
ws_adapter.py attribute/method |
Set from analysis.ws.* |
|---|---|
self.ws_url |
analysis.ws.ws_url |
self.send_template |
analysis.ws.send_template |
self.input_field |
analysis.ws.input_field |
self.receive_field |
analysis.ws.receive_field |
self.receive_is_streaming |
analysis.ws.receive_is_streaming |
self.type_field |
analysis.ws.type_field |
self.extra_send_fields |
analysis.ws.extra_send_fields |
self.init_messages |
analysis.ws.init_messages |
_connect() |
Override only if target requires custom subprotocols |
_extract_content_from_frame() |
Uses receive_field + type_field — no change needed |
_is_done_frame() |
Override only if done signal differs from defaults |
The template already uses self.input_field and self.extra_send_fields in convert_request() — just fill them in __init__ and it works.
Key behavioral differences from HTTP adapter to note:
- Connection lifecycle: Each request opens a fresh WS connection (connect → send → receive → close). If the HAR shows multiple send/receive pairs on one WS (persistent session), change to reuse the connection.
- Init messages: If
init_messagesis non-empty,_connect()automatically sends them after connecting. No additional code needed. - Frame done detection:
_is_done_frame()checks multiple patterns (finish_reason, done flag, type field,[DONE]). If the target uses a unique signal, override this method. - Non-streaming mode: When
receive_is_streamingis False, the adapter reads exactly one receive frame and returns it. When True, it reads frames until_is_done_frame()returns True. - Binary frames: WebSocketChatAdapter only handles text frames (opcode 1). Binary frames (opcode 2) are not supported — report this limitation.
analysis.is_streaming is False. The HAR response is JSON.
Extract analysis.content_field_path which is a dot-path like:
"answer"→data["answer"]"choices[0].message.content"→data["choices"][0]["message"]["content"]"data.text"→data["data"]["text"]
Modify _extract_content_from_json:
def _extract_content_from_json(self, data: dict) -> Optional[str]:
"""Navigate the content_field_path to extract the AI response text."""
# Build the field path from analysis.content_field_path
path = analysis.content_field_path
if path:
import functools
try:
parts = path.replace("[", ".").replace("]", "").split(".")
val = functools.reduce(lambda d, k: d[int(k) if k.isdigit() else k] if isinstance(d, (dict, list)) else None, parts, data)
if isinstance(val, str) and val.strip():
return val
except (KeyError, IndexError, TypeError):
pass
# Fallback: scan common top-level keys
for key in ["answer", "text", "content", "reply", "response", "output", "completion", "result"]:
val = data.get(key)
if isinstance(val, str) and val.strip():
return val
# Deep fallback: recursively find longest string in full JSON
def deepest(obj):
if isinstance(obj, dict):
for v in obj.values():
r = deepest(v)
if r: return r
elif isinstance(obj, list):
for v in obj:
r = deepest(v)
if r: return r
elif isinstance(obj, str) and len(obj) > 20:
return obj
return None
return deepest(data)analysis.is_streaming is True. The HAR response has Content-Type: text/event-stream.
Modify _extract_content_from_data according to the table below. Only keep the matching branch + one generic fallback. Delete all other branches.
sse_format |
sse_data_field |
Code to write in _extract_content_from_data |
|---|---|---|
"plain_token" |
"v" |
return data.get("v") if isinstance(data.get("v"), str) else None |
"plain_token" |
"content" |
return data.get("content") if isinstance(data.get("content"), str) else None |
"plain_token" |
"text" |
return data.get("text") if isinstance(data.get("text"), str) else None |
"plain_token" |
"delta" |
d = data.get("delta"); return d.get("content") if isinstance(d, dict) else (d if isinstance(d, str) else None) |
"plain_token" |
"token" |
return data.get("token") if isinstance(data.get("token"), str) else None |
"plain_token" |
"response" |
return data.get("response") if isinstance(data.get("response"), str) else None |
"path_op_value" |
"v" |
DeepSeek: return data.get("v") if data.get("o") == "APPEND" else None |
"nested" |
"v.content" |
v = data.get("v", {}); return v.get("content") or v.get("response", {}).get("content") |
"nested" |
"delta.content" |
d = data.get("delta", {}); return d.get("content") |
"nested" |
"choices[0].delta.content" |
c = data.get("choices", []); return c[0].get("delta", {}).get("content") if c else None |
"raw_text" |
— | SSE data line is raw text, not JSON: return data if isinstance(data, str) else None |
Fallback that always stays at the end:
# Final fallback: any string field in the data dict
if isinstance(data, dict):
for key in ("content", "text", "answer", "v", "response", "token", "delta"):
val = data.get(key)
if isinstance(val, str) and val:
return val
if isinstance(val, dict):
sub = val.get("content") or val.get("text")
if sub:
return sub
return NoneReplace the hardcoded field lookup:
# BEFORE (in template):
content = response.get("answer") or response.get("text") or json.dumps(response)
# AFTER:
content = self._extract_content_from_json(response)
if not content:
content = json.dumps(response, ensure_ascii=False)Before configuring auth, check references/bypass-guides/ for detailed techniques on handling CAPTCHA, WAF, and PoW challenges. The guides contain ready-to-use code snippets and detection methods.
# auth_type is "none"
# Nothing extra needed, Cookie in headers is sufficientif analysis.auth_header:
self.headers["Authorization"] = analysis.auth_headerself.auth_type = "pow"
self._challenge_endpoint = analysis.pow_endpoint # "/api/v0/chat/create_pow_challenge"
# In _ensure_auth_headers: fetch challenge, solve, return header
def _ensure_auth_headers(self) -> dict:
if self.auth_type != "pow":
return {}
challenge = self._fetch_challenge()
answer = self._solve_challenge(challenge)
return {"X-DS-PoW-Response": self._encode_pow_answer(challenge, answer)}
def _fetch_challenge(self) -> dict:
resp = httpx.post(
f"{self.base_url}{self._challenge_endpoint}",
json={"target_path": "/api/v0/chat/completion"},
headers=self.headers,
timeout=15,
)
return resp.json()
def _solve_challenge(self, challenge_data: dict) -> int:
# Extract from nested DeepSeek response format
biz = challenge_data.get("data", {}).get("biz_data", {}).get("challenge", challenge_data)
# Use wasm solver if wasm binary available, else implement hash loop
# Typical: hash = sha256(salt + expire_at + nonce), find nonce where hash[:difficulty_hex] == target
import hashlib
salt = biz["salt"]
expire_at = biz["expire_at"]
difficulty = biz["difficulty"]
target = biz.get("target_path", "")
nonce = 0
while True:
h = hashlib.sha256(f"{salt}{expire_at}{nonce}{target}".encode()).hexdigest()
if h.startswith("0" * difficulty):
return nonce
nonce += 1self.auth_type = "token_refresh"
# In _ensure_auth_headers: if 401, call refresh endpoint, update Cookie in self.headersThe __init__ must produce headers identical to what the browser sent:
def __init__(self, cookies: str, base_url: str, dsml_enabled: bool = True):
self.headers = {}
# 1. Take ALL headers from HAR except content-length
for key, val in analysis.headers.items():
if key.lower() != "content-length":
self.headers[key] = val
# 2. Ensure User-Agent exists (HAR always has it, but be safe)
if "User-Agent" not in self.headers:
self.headers["User-Agent"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
# 3. Ensure Content-Type is set
if "Content-Type" not in self.headers:
self.headers["Content-Type"] = "application/json"
# 4. Override with explicit auth if detected
if analysis.auth_header:
self.headers["Authorization"] = analysis.auth_header
self.base_url = base_url.rstrip("/")
self.chat_endpoint = analysis.chat_endpoint
self.auth_type = analysis.auth_type
self.dsml_enabled = dsml_enabled
self.dsml_ready = FalseSome APIs require state across requests (session ID, chat session ID).
Check analysis.request_body_template or analysis.ws.send_template for keys like:
chat_session_id,session_id,conversation_id,conversationId
def __init__(self, ...):
...
# Session state from HAR
self.chat_session_id = analysis.request_body_template.get("chat_session_id")
self.conversation_id = analysis.request_body_template.get("conversation_id")
# If the API returns a new session ID per response, update it:
self._auto_update_session = "chat_session_id" in analysis.request_body_template
def convert_request(self, ...):
...
payload["chat_session_id"] = self.chat_session_id
# After receiving response, extract new session ID:
# response.get("chat_session_id") → update self.chat_session_idIf the HAR shows multiple send/receive pairs on the same WebSocket (check if _webSocketMessages has alternating send/receive across the whole entry), the target expects a persistent connection. Modify WebSocketChatAdapter to keep the WS open:
def __init__(self, ...):
...
self._ws = None # persistent connection
self._session_id = analysis.ws.extra_send_fields.get("chat_session_id")
self._auto_update = "chat_session_id" in analysis.ws.extra_send_fields
async def _get_connection(self):
"""Reuse or create a persistent WebSocket."""
if self._ws is None or self._ws.closed:
self._ws = await self._connect()
return self._wsThen replace _connect() calls in send_request/stream_request with _get_connection(). This avoids reconnecting and re-sending init messages for every request.
Copy templates/config_tool.py → output config_tool.py. This is a standalone Tkinter GUI that lets the user re-configure .env from any HAR file without re-running the AI.
Map every key in the info dict (from parse_har_file()) to a Chinese label:
FIELD_LABELS = {
"base_url": "目标地址",
"chat_endpoint": "聊天端点",
"cookies": "Cookie",
"auth_header": "Authorization",
"auth_type": "认证类型",
"is_streaming": "流式支持",
"has_websocket": "WebSocket",
"has_pow": "PoW 挑战",
"content_field_path": "内容字段路径",
"supported_params": "支持的参数",
# HAR特有的字段，从 analysis 中取:
# "chat_session_id": "会话 ID",
}Only include keys that exist in analysis. If the target has unique fields (DeepSeek's chat_session_id, Gemini's bl/f.sid/at/sn), add them here.
Map (env_key, info_key, comment) for every field that should appear in .env:
ENV_MAPPING = [
("HAR_PATH", "har_path", "HAR 文件路径（用于重新解析）"),
("TARGET_URL", "base_url", "目标网站地址"),
("CHAT_ENDPOINT", "chat_endpoint", "聊天 API 端点路径"),
("COOKIES", "cookies", "登录 Cookie"),
("AUTH_HEADER", "auth_header", "Authorization 令牌"),
("AUTH_TYPE", "auth_type", "认证类型"),
("STREAMING", "is_streaming", "是否支持流式输出"),
("WEBSOCKET", "has_websocket", "是否使用 WebSocket"),
# 目标特有:
# ("CHAT_SESSION_ID", "chat_session_id", "会话 ID"),
]Control which fields appear in the TreeView and in what order:
DISPLAY_FIELDS = [
("目标地址", "base_url"),
("聊天端点", "chat_endpoint"),
("认证类型", "auth_type"),
("Cookie", "cookies"),
("Authorization", "auth_header"),
# ...
]The default implementation calls har_parser.parse_har() and extracts common fields. If the target has unique parameters not covered by har_parser.py, add extraction logic in parse_har_file() or use the _extract_raw_har() helper which can regex-search the raw HAR text for specific keys.
Example for DeepSeek (adds chat_session_id):
def parse_har_file(har_path: str) -> dict:
info = _default_parse(har_path) # calls har_parser.parse_har
# Extract target-specific fields
chat_session_id = _extract_raw_har(har_path, "chat_session_id")
if chat_session_id:
info["chat_session_id"] = chat_session_id
FIELD_LABELS["chat_session_id"] = "会话 ID"
ENV_MAPPING.append(("CHAT_SESSION_ID", "chat_session_id", "DeepSeek 会话 ID"))
DISPLAY_FIELDS.append(("会话 ID", "chat_session_id"))
return infoThis is the key innovation. .env is split into two sections. AI decides which keys go where based on the nature of each field:
放入 MUTABLE_KEYS（异变） |
不放入（不易变） |
|---|---|
| Cookie、Authorization 等会过期的鉴权凭证 | PORT、HOST 等服务器监听配置 |
| Token、session_id 等每次登录变化的参数 | MODEL_NAME、API_KEY 等固定标识 |
| TARGET_URL、CHAT_ENDPOINT（站点结构可能变） | DSML_ENABLED 等功能开关 |
| STREAMING、WEBSOCKET 等传输方式标记 | 用户自定义的运行时参数 |
| 从 HAR 中解析出的任何易变字段 | 不会在 Cookie 过期时一起变化的值 |
规则：凡是会随登录过期/变化的 key → 放入 MUTABLE_KEYS；凡是服务器本身的配置 → 不放入。
运行时 merge_env_with_auth() 的行为：
- 读取现有
.env - 只覆写
MUTABLE_KEYS中的 key - 其他 key（PORT、MODEL_NAME 等）保留原值
- 如果
.env不存在，不易变部分用DEFAULT_IMMUTABLE填充
示例输出 .env：
# ============================================
# 不易变部分 — 服务器配置（配置工具不会修改）
# ============================================
MODEL_NAME=gpt-4o
HOST=0.0.0.0
PORT=8000
...
# ============================================
# 异变部分 — 账号鉴权凭证（配置工具只更新此段）
# ============================================
COOKIES=...
AUTH_HEADER=...
Add to the output deliverables. The user can double-click config_tool.py to:
- Select a HAR file from disk
- Click "解析" to extract all config fields
- View the generated .env in a syntax-highlighted preview
- Click "保存到 .env" to persist
- Click "启动代理服务器" to run
server.py
When Cookie expires, the user just re-opens config_tool.py, selects the same HAR (or a fresh one), clicks "解析" → "保存" → "启动", all without touching the AI.
Run this exact verification protocol, in order:
import httpx
async def verify_non_streaming():
payload = adapter.convert_request(
[{"role": "user", "content": "Hello"}],
stream=False,
)
async with httpx.AsyncClient(headers=adapter.headers, timeout=30) as client:
resp = await client.post(
f"{adapter.base_url}{adapter.chat_endpoint}",
json=payload,
)
assert resp.status_code == 200, f"Non-streaming: status {resp.status_code}"
data = resp.json()
content = adapter._extract_content_from_json(data)
assert content and len(content) > 0, f"Non-streaming: empty content from {list(data.keys())}"
print(f"[OK] Non-streaming: got {len(content)} chars")
return Trueasync def verify_streaming():
payload = adapter.convert_request(
[{"role": "user", "content": "Hello"}],
stream=True,
)
chunks = []
async with httpx.AsyncClient(headers=adapter.headers, timeout=30) as client:
async with client.stream("POST", f"{adapter.base_url}{adapter.chat_endpoint}", json=payload) as resp:
assert resp.status_code == 200, f"Streaming: status {resp.status_code}"
async for line in resp.aiter_lines():
line = line.strip()
if not line: continue
if line.startswith("data: "):
raw = line[6:]
if raw.strip() == "[DONE]": break
chunks.append(raw)
assert len(chunks) > 0, "Streaming: no data chunks received"
print(f"[OK] Streaming: got {len(chunks)} chunks")
return Trueimport websockets
async def verify_ws_non_streaming():
payload = adapter.convert_request(
[{"role": "user", "content": "Hello"}],
stream=False,
)
ws = await adapter._connect()
try:
await ws.send(json.dumps(payload, ensure_ascii=False))
frames = []
if adapter.receive_is_streaming:
async for raw in ws:
text = adapter._extract_content_from_frame(raw)
if text:
frames.append(text)
if adapter._is_done_frame(raw):
break
else:
raw = await ws.recv()
text = adapter._extract_content_from_frame(raw)
if text:
frames.append(text)
assert len(frames) > 0, "WS non-streaming: no content frames"
print(f"[OK] WS non-streaming: got {len(frames)} frames")
return True
finally:
await ws.close()
async def verify_ws_streaming():
if not adapter.receive_is_streaming:
print("[SKIP] WS streaming: target is non-streaming")
return True
payload = adapter.convert_request(
[{"role": "user", "content": "Hello"}],
stream=True,
)
ws = await adapter._connect()
try:
await ws.send(json.dumps(payload, ensure_ascii=False))
chunks = []
async for raw in ws:
text = adapter._extract_content_from_frame(raw)
if text:
chunks.append(text)
if adapter._is_done_frame(raw):
break
assert len(chunks) > 1, f"WS streaming: only {len(chunks)} chunks (need >1)"
print(f"[OK] WS streaming: got {len(chunks)} chunks")
return True
finally:
await ws.close()| Failure | Root cause | Fix |
|---|---|---|
| Status 401/403 | Cookie expired or auth missing | Tell user Cookie expired, ask for fresh HAR |
| Status 404 | Wrong endpoint | Check analysis.chat_endpoint; try common alternatives |
| Status 400 | Wrong payload shape | Compare generated payload against analysis.request_body_template byte-for-byte |
| Empty content in JSON | Wrong content field path | Re-scan JSON keys; pick the longest string value path |
| No SSE chunks | Wrong streaming mode or field | Read raw HAR SSE text; manually identify which key holds text |
| Connection refused | Target blocked by CORS/geo | Add Origin and Referer headers from HAR |
| WS connection refused | Wrong URL scheme/port | Check analysis.ws.ws_url; try both wss:// and ws:// |
| WS closed immediately | Auth/init messages missing | Check analysis.ws.init_messages and verify they replay correctly |
| WS receive empty | Wrong receive_field |
Re-scan HAR receive frame JSON keys; pick the longest string value |
| WS stream stuck | Wrong done signal | Read HAR receive frames manually; find what signals completion |
| WS binary frame | Target sends opcode 2 | Cannot support — tell user this AI uses binary WS frames |
Max 3 retry rounds. After each round, adjust and re-run verification. If still failing after 3, output what was tried and ask user for a fresh HAR file.
pip install fastapi uvicorn httpx python-dotenv websockets
python server.py &
sleep 2
# Non-streaming
curl -s http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o","messages":[{"role":"user","content":"Hello"}],"stream":false}' \
| python -c "import sys,json; d=json.load(sys.stdin); print('OK:', d['choices'][0]['message']['content'][:80])"
# Streaming
curl -s -N http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o","messages":[{"role":"user","content":"Hello"}],"stream":true}' \
| head -5adapter.py— modified HTTP adapter (if HTTP detected)ws_adapter.py— WebSocket adapter (ifhas_websocketis True)config_tool.py— GUI configurator (HAR → .env, with Cookie update support)server.py— from template (no changes needed).env.example:TARGET_URL=https://chat.example.com COOKIES=... MODEL_NAME=gpt-4o HOST=0.0.0.0 PORT=8000 API_KEY=sk-web2api-placeholder DSML_ENABLED=truerequirements.txt:fastapi,uvicorn,httpx,python-dotenv,pydantic,websockets- Start command and E2E verification output
- Integration guide:
OPENAI_API_BASE=http://localhost:8000/v1
| Field | Type | Source | Example |
|---|---|---|---|
base_url |
str |
Scheme + netloc of matched entry | https://chat.example.com |
chat_endpoint |
str |
URL path + query | /api/v0/chat/completion |
headers |
dict |
All request headers | {Cookie: ..., Origin: ...} |
cookies |
str |
Cookie header value | __session=abc; token=xyz |
auth_header |
Optional[str] |
Authorization header | Bearer eyJ... |
auth_type |
str |
"none" / "pow" / "oauth" |
— |
request_body_template |
dict |
Parsed POST body | {messages: [...], stream: true} |
content_field_path |
str |
Deepest non-streaming content field | choices[0].message.content |
is_streaming |
bool |
text/event-stream in Content-Type |
— |
sse_event_type |
str |
SSE event: lines |
"append" or "" |
sse_data_field |
str |
Key inside SSE data JSON | "v" / "content" / "delta" |
sse_format |
str |
Structure of SSE data payload | "plain_token" / "path_op_value" / "nested" / "raw_text" |
has_pow |
bool |
Challenge endpoint in HAR | — |
pow_endpoint |
str |
PoW challenge URL path | /api/v0/chat/create_pow_challenge |
all_endpoints |
list[str] |
All unique URL paths in HAR | — |
supported_params |
list[str] |
OpenAI params seen in HAR request bodies | ["temperature", "top_p"] |
chat_entry_index |
int |
Index in HAR entries list | 42 |
has_websocket |
bool |
WebSocket entry detected in HAR | True or False |
ws.ws_url |
str |
WebSocket URL (wss:// or ws://) | wss://chat.example.com/ws |
ws.send_template |
dict |
JSON of first WS send frame | {type: "chat", content: ""} |
ws.input_field |
str |
Key in send_template for user msg | "content" / "messages" |
ws.receive_field |
str |
Key in receive frames for AI text | "content" / "answer" |
ws.receive_is_streaming |
bool |
Multiple receives per single send | True / False |
ws.type_field |
str |
Frame type discriminator key | "type" / "event" / "" |
ws.extra_send_fields |
dict |
Non-input keys to carry in every send | {type: "chat", id: 1} |
ws.init_messages |
list[dict] |
Auth/init frames to replay on connect | [{type: "auth", ...}] |
har_parser.py classifies SSE data into these formats. Understand them to write correct extraction code:
Each data: line contains a JSON object. One key holds the incremental text directly.
// data: {"v": "Hello"}
// data: {"content": " world"}
// data: {"token": "!"}→ The value at sse_data_field is a plain string.
Each data: line has p (path), o (operation), v (value). Text arrives as APPEND operations.
// event: append
// data: {"p": "response/content", "o": "APPEND", "v": "Hello"}→ Only extract when o == "APPEND", take v.
The value at sse_data_field is a dict, and content is nested deeper.
// data: {"delta": {"content": "Hello"}}
// data: {"choices": [{"delta": {"content": " world"}}]}→ Navigate one level deeper: data[sse_field]["content"].
The data: line content is not JSON at all — it's a plain string.
// data: Hello
// data: world
→ The content IS the raw string itself.
| Type | Endpoint | Payload | Content Field | SSE |
|---|---|---|---|---|
| DeepSeek Chat | /api/v0/chat/completion |
{chat_session_id, prompt, stream} |
APPEND v |
path_op_value |
| ChatGPT Next Web | /api/chat |
{messages, model} |
choices[0].message.content |
nested / choices[0].delta.content |
| ChatGPT Official | /backend-api/conversation |
{action, messages} |
message.content.parts[0] |
nested |
| LobeChat | /api/chat |
{messages, model} |
choices[0].delta.content |
nested |
| Open WebUI | /chat/completions |
{messages, model} |
OpenAI standard | nested |
| Gradio Chat | /api/chat |
{data: [...]} |
data[0] |
raw_text or plain_token |
| Custom NextJS | /api/chat |
{prompt, history} |
text / answer |
varies |
| WebSocket Generic | wss://.../ws |
send: {content: "..."} / {messages: [...]} |
receive content |
receive_is_streaming auto-detected |
DSML injects XML-style instructions into the system prompt to support tool calling on models without native function calling.
The tool_dsml.py template now includes these capabilities:
| Function | Purpose |
|---|---|
strip_dsml_markup() |
Remove ` |
parse_dsml_invoke() |
Parse DSML → OpenAI tool_calls (handles both `< |
format_tool_calls_for_prompt() |
Convert OpenAI tool_calls back to DSML for history messages |
build_dsml_tool_prompt() |
Generate DSML instruction prompt from tool definitions |
strip_dsml_tags() |
Clean all DSML/XML markup from text |
_auto_type() |
Smart type inference (string "42" → int 42, "true" → True) |
_cdata() |
CDATA wrapping with proper ]]> escaping |
has_dsml_content() |
Quick check for DSML tags (supports both prefixed and bare formats) |
Multi-turn tool calls: When building prompts for multi-turn conversations, format_tool_calls_for_prompt() converts previous tool_calls back to DSML format so the model sees the complete history.
| Tag | Purpose |
|---|---|
<|DSML|tool_calls> |
Root container for one or more invocations |
<|DSML|invoke name="fn_name"> |
Single tool call start |
<|DSML|parameter name="key"><![CDATA[value]]></|DSML|parameter> |
Key-value parameter |
</|DSML|invoke> |
Tool call end |
</|DSML|tool_calls> |
Root end |
# User sends tools → adapter injects DSML prompt into system message
messages_with_dsml = adapter._inject_dsml_prompt(messages, tools, tool_choice)
# Target model responds with DSML tags embedded in text
# adapter._extract_content_from_data captures the text
# StreamSieve character-by-character state machine separates:
# - Normal text → OpenAI delta.content chunks
# - DSML blocks → parsed into OpenAI delta.tool_calls chunksThe tool_sieve.py template now features:
| Feature | Description |
|---|---|
| Multi-tag detection | Recognizes <|DSML|tool_calls>, <tool_calls>, <invoke, etc. |
_split_safe() |
Prevents mid-tag cut-off by checking if trailing chars match a known tag prefix |
| Dual-buffer design | _pending buffer for safe text, _capture_buf for tool call capture |
| Event model | SieveEvent with type='text' or type='tool_calls' for structured output |
| Flush fallback | Incomplete captures are flushed as plain text at stream end |
StreamSieve handles: split TCP chunks, embedded CDATA, mixed text+tools, abrupt stream end.
tool_choice mapping: "auto" → optional, "none" → skip, "required" → force.
When has_pow is True (DeepSeek pattern):
- PRE-request:
POST /api/v0/chat/create_pow_challenge→ returns{data: {biz_data: {challenge: {algorithm, challenge, salt, expire_at, difficulty, signature, target_path}}}} - Solve: find nonce such that
sha256(salt + expire_at + nonce + target_path)starts withdifficultyzero hex chars - Encode:
base64({algorithm, challenge, salt, answer, signature, target_path}) - Send as
X-DS-PoW-Responseheader with the main chat request
If WASM solver binary is available from the target site, use wasmtime instead of Python hashlib for performance.
- Text-only: no multimodal (images/files)
- Tool calling via DSML injection, not native function calling — depends on model's XML comprehension
seed/response_format/json_modenot supported- Parameters (
temperature,max_tokens, ...) automatically detected from HAR — only those the target API actually uses will be passed through - PoW solver is site-specific (DeepSeek pattern)
- Cookie-based auth expires; user must refresh HAR when needed
- WebSocket only supports text frames (opcode 1) — binary frames not supported
- WebSocket adapter requires
websocketslibrary (added to requirements) - WS init messages are replayed on every new connection — ensure they are idempotent