从网页到 MCP：我把「139 云盘分享下载」做成了一个可复用的连接器

地址表/
├── 香洲、横琴/   （87 个 .xls）
└── 金湾、斗门/   （67 个 .xls）

curl -s -L "https://yun.139.com/shareweb/#/w/i/2xTrGM6yFun6u"

grep -oE 'static/js/[a-zA-Z0-9._-]+\.js' index.html | sort -u
# app.3c33c3cd.js  chunk-xxx.js  chunk-vendors.xxx.js ...

grep -ohE '"[a-zA-Z0-9/._-]*(share|Share|outLink|OutLink)[a-zA-Z0-9/._-]*"' *.js | sort -u

/yun-share/richlifeApp/devapp/IOutLink/getOutLinkInfoV6   ← 列目录
/yun-share/richlifeApp/devapp/IOutLink/dlFromOutLinkV3    ← 取下载地址

curl -X POST "https://share-kd-njs.yun.139.com/yun-share/richlifeApp/devapp/IOutLink/getOutLinkInfoV6" \
  -H "Content-Type: application/json" \
  -H "Origin: https://yun.139.com" \
  -d '{"getOutLinkInfoReq":{
        "linkID":"2xTrGM6yFun6u","passwd":"f6ro",
        "pCaID":"root","bNum":1,"eNum":50,"srtDr":1,"caSrt":"","coSrt":""
      }}'

for p in "" "root" "-1" "0"; do ... done   # 只有 "root" 有数据

def walk(link_id, passwd, cookies, pcaid="root", prefix="", page_size=200):
    """递归列出全部分享文件，返回 [(相对路径, 字节数, 云盘内部 path)]"""
    out = []
    b = 1
    while True:
        d = list_dir(link_id, passwd, cookies, pcaid, b, b + page_size - 1)
        if not d:
            break
        # caLst = 子目录，coLst = 文件
        for c in d.get("caLst") or []:
            out += walk(link_id, passwd, cookies, c["caID"],
                        prefix + "/" + c.get("caName", ""), page_size)
        for c in d.get("coLst") or []:
            out.append((prefix + "/" + c.get("coName", ""),
                        int(c.get("coSize") or 0), c.get("path")))
        got = len(d.get("coLst") or [])
        if got < page_size and not d.get("nextPageCursor"):
            break
        b += page_size
    return out

{ "resultCode": "200000401", "desc": "IP鉴权失败、用户账号鉴权失败" }

sign = MD5( MD5(base64( sorted( encodeURIComponent(JSON.stringify(body)) ) )) 
          + MD5( 时间串 + ":" + 16位随机串 ) ).toUpperCase()

mcloud-sign = "{时间串},{16位随机串},{sign}"

def build_headers(data, cookies):
    c = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    a = _rand16()                      # 16 位 [A-Za-z0-9] 随机串
    if data:
        s = urllib.parse.quote(json.dumps(data, separators=(",", ":"), ensure_ascii=False),
                               safe="~()*!.'-")
        s = "".join(sorted(s))         # 注意：对整个编码后的串做字符级排序
        r = _md5(base64.b64encode(s.encode()).decode())
    else:
        r = _md5(base64.b64encode(b"").decode())
    sign = _md5(r + _md5(c + ":" + a)).upper()
    return {
        "Content-Type": "application/json;charset=UTF-8",
        "mcloud-sign": f"{c},{a},{sign}",
        "mcloud-client": "10702",
        "mcloud-version": "7.13.3",
        "mcloud-channel": "1000101",
        # ... 其余业务头
    }

const cookies = await ctx.cookies();
const wanted = ['authorization', 'auth_token', 'token', 'skey', 'a_k', 'ud_id',
                'ORCHES-I-ACCOUNT-ENCRYPT'];

if cookies.get("skey"):
    h["mcloud-skey"] = cookies["skey"]
if cookies.get("authorization"):
    h["Authorization"] = cookies["authorization"]
    h["x-m4c-token"] = cookies.get("auth_token", "")
    h["x-m4c-account"] = cookies.get("ORCHES-I-ACCOUNT-ENCRYPT", "")

Error: Target page closed.

def get_dl_url(link_id, passwd, cookies, fpath, account=""):
    d = post("/richlifeApp/devapp/IOutLink/dlFromOutLinkV3", {
        "dlFromOutLinkReqV3": {
            "account": account,            # 手机号，从加密 cookie 里 base64 解出来
            "linkID": link_id,
            "coIDLst": {"item": [fpath]},
        }}, cookies)
    return d

# 读一行 → 解析 JSON-RPC → 分发 → 回写一行
for line in sys.stdin:
    req = json.loads(line)
    if req.get("method") == "tools/list":
        reply({"tools": TOOLS})
    elif req.get("method") == "tools/call":
        reply(dispatch(req["params"]["name"], req["params"].get("arguments") or {}))

download         → 立刻返回 job_id（子进程独立跑，DETACHED_PROCESS）
download_status  → 读 state/jobs/<job_id>.json，返回进度
download_cancel  → 写 state/cancel/<job_id> 标记，子进程自行退出

def spawn(args, log_name):
    flags = 0x00000008 | 0x00000200 if os.name == "nt" else 0
    return subprocess.Popen(args, cwd=BASE, env=env, stdin=subprocess.DEVNULL,
                            stdout=logf, stderr=logf, creationflags=flags)

def fetch(url, dest, timeout=300, tries=3):
    for i in range(tries):
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r, open(dest, "wb") as f:
                while chunk := r.read(1 << 16):
                    f.write(chunk)
            return os.path.getsize(dest)
        except Exception as e:
            if i < tries - 1:
                time.sleep(2)

passwd = a.get("passcode") or "f6ro"    # ← 危险

def resolve_share(a):
    url = (a.get("share_url") or "").strip()
    pw = (a.get("passcode") or "").strip()
    if not url:
        return DEFAULT_SHARE, (pw or DEFAULT_PASSWD)   # 完全没给链接才用示例
    return url, pw        # 给了链接就绝不用内置提取码兜底

if code in AUTH_FAIL_CODES or "鉴权失败" in desc:
    raise NotLoggedIn(f"[{code}] {desc}")
if code in PASSCODE_FAIL_CODES or "提取码" in desc:
    raise PasscodeError(f"[{code}] {desc}")
if code in NOT_FOUND_CODES or "外链不存在" in desc or "已取消" in desc:
    raise ShareNotFound(f"[{code}] {desc}")