手搓阿里云盘 MCP：完整代码贴出来

上篇讲了为什么自己写以及三个实测踩过的坑。这篇把能跑的完整代码贴出来照着建三个文件、配一下 MCP 就能用。没看过上篇的建议先翻一下不然下面有些写法看着怪。比如刷新 token 那个grant_type字段是踩了坑才改成那样的。文件结构aliyundrive-mcp/ ├── server.py # MCP Server9 个工具 ├── auth.py # 扫码授权、换 token、验证 └── test_connection.py # 连接自检排错用token.json 不用你手建跑完 auth.py 会自动生成。server.py完整代码阿里云盘 MCP Server (社区托管路线 - AList OAuth) 授权流程: 浏览器打开 https://alistgo.com/tool/aliyundrive/request 扫码 - 拿到 refresh_token - python auth.py refresh_token - 换 access_token 写入 token.json 注意: refresh_token 约 30 天失效失效后需重新扫码授权。 import os import sys import json import time import threading from typing import Optional, List, Dict, Any import requests from fastmcp import FastMCP # ---------- 相对独立: 路径不写死跟随脚本目录 / 环境变量 ---------- BASE_DIR os.path.dirname(os.path.abspath(__file__)) TOKEN_FILE os.environ.get(ALIPAN_TOKEN_FILE) or os.path.join(BASE_DIR, token.json) TOKEN_FILE os.path.abspath(TOKEN_FILE) API_BASE https://openapi.alipan.com # 社区托管刷新端点 (官方 api.nn.ci 已被阻断); 官方个人开发者申请自 2025-07 暂停 OAUTH_TOKEN_URL os.environ.get(ALIPAN_OAUTH_TOKEN_URL) or https://api.alistgo.com/alist/ali_open/token mcp FastMCP(aliyundrive-mcp) _lock threading.Lock() _session requests.Session() # ---------- token 管理 ---------- def _load_token() - Dict[str, Any]: if not os.path.exists(TOKEN_FILE): raise RuntimeError( f未找到 token 文件: {TOKEN_FILE}\n 请先授权: 浏览器打开 https://alistgo.com/tool/aliyundrive/request 扫码拿 refresh_token, 然后运行: python auth.py refresh_token ) with open(TOKEN_FILE, r, encodingutf-8) as f: return json.load(f) def _save_token(data: Dict[str, Any]) - None: try: with open(TOKEN_FILE, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) except OSError as e: # 写权限不足时只警告不阻断 (内存态 token 仍可用到过期) print(f[warn] 无法写回 token 文件 {TOKEN_FILE}: {e}, filesys.stderr) def _refresh(refresh_token: str) - Dict[str, Any]: 用 refresh_token 换 access_token。 实测(2026-08-30): 社区代理端点 api.alistgo.com/alist/ali_open/token 只认 {grant_type: refresh_token, refresh_token: ...}; 传 {code: ...} 会被 判为 authorization_code 模式并报 Incorrect GrantType。 r _session.post( OAUTH_TOKEN_URL, json{grant_type: refresh_token, refresh_token: refresh_token}, timeout30, ) if r.status_code 400: raise RuntimeError(f刷新 token 失败 ({r.status_code}): {r.text[:300]}) body r.json() if not body.get(access_token): raise RuntimeError(f刷新 token 返回异常: {json.dumps(body, ensure_asciiFalse)[:300]}) data _load_token() if os.path.exists(TOKEN_FILE) else {} data[access_token] body[access_token] data[refresh_token] body.get(refresh_token) or refresh_token data[expires_in] body.get(expires_in, 7200) data[expires_at] time.time() int(data[expires_in]) - 300 # 提前 5 分钟刷新 _save_token(data) return data def get_token() - Dict[str, Any]: with _lock: data _load_token() if data.get(expires_at, 0) time.time(): data _refresh(data.get(refresh_token, )) return data # ---------- 通用请求 ---------- def api(path: str, payload: Optional[Dict[str, Any]] None, retry: bool True) - Dict[str, Any]: tok get_token() url f{API_BASE}/adrive/v1.0/{path.lstrip(/)} headers { Authorization: fBearer {tok[access_token]}, Content-Type: application/json, } r _session.post(url, jsonpayload or {}, headersheaders, timeout60) # 401/403 且未重试过 - 强制刷新后重试一次 if r.status_code in (401, 403) and retry: with _lock: _refresh(_load_token().get(refresh_token, )) return api(path, payload, retryFalse) if r.status_code 400: raise RuntimeError(f阿里网盘 API 错误 ({r.status_code}): {r.text[:500]}) if not r.content: return {} try: return r.json() except ValueError: return {raw: r.text[:500]} _drive_id_cache: Optional[str] None def drive_id(force: bool False) - str: 获取默认 drive_id 并缓存避免每个工具都重复请求用户信息。 global _drive_id_cache if _drive_id_cache is None or force: d api(user/getDriveInfo) _drive_id_cache d.get(default_drive_id) or d.get(backup_drive_id) if not _drive_id_cache: raise RuntimeError(未能获取 drive_id账号可能未开通开放平台权限) return _drive_id_cache def _fmt_item(it: Dict[str, Any]) - Dict[str, Any]: return { file_id: it.get(file_id), name: it.get(name), type: it.get(type), size: it.get(size), updated_at: it.get(updated_at), parent_file_id: it.get(parent_file_id), } # ---------- 工具 ---------- mcp.tool() def get_user_info() - Dict[str, Any]: 获取阿里云盘账号信息用户昵称、容量使用情况、默认 drive_id。 d api(user/getDriveInfo) # 容量不在 getDriveInfo 里需单独查 getSpaceInfo (实测 2026-08-30) try: sp api(user/getSpaceInfo).get(personal_space_info) or {} except Exception: sp {} used, total sp.get(used_size), sp.get(total_size) gb 1024 ** 3 return { user_name: d.get(nick_name) or d.get(user_name), user_id: d.get(user_id), default_drive_id: d.get(default_drive_id), backup_drive_id: d.get(backup_drive_id), used_size: used, total_size: total, used_gb: round(used / gb, 2) if used else None, total_gb: round(total / gb, 2) if total else None, used_percent: round(used / total * 100, 2) if (used and total) else None, } mcp.tool() def list_files(parent_file_id: str root, limit: int 100, order_by: str updated_at, order_direction: str DESC) - List[Dict[str, Any]]: 列出指定目录下的文件和文件夹。parent_file_id 默认为 root根目录。 payload { drive_id: drive_id(), parent_file_id: parent_file_id, limit: min(max(limit, 1), 200), order_by: order_by, order_direction: order_direction, fields: *, } d api(openFile/list, payload) return [_fmt_item(i) for i in d.get(items, [])] mcp.tool() def search_files(query: str, limit: int 50) - List[Dict[str, Any]]: 按文件名关键词搜索阿里云盘文件。 payload { drive_id: drive_id(), query: fname match {query}, limit: min(max(limit, 1), 100), order_by: updated_at DESC, } d api(openFile/search, payload) return [_fmt_item(i) for i in d.get(items, [])] mcp.tool() def get_file_info(file_id: str) - Dict[str, Any]: 获取单个文件或文件夹的详细信息。 d api(openFile/get, { drive_id: drive_id(), file_id: file_id, }) return d mcp.tool() def create_folder(name: str, parent_file_id: str root) - Dict[str, Any]: 在指定目录下新建文件夹返回新文件夹的 file_id。 d api(openFile/create, { drive_id: drive_id(), parent_file_id: parent_file_id, name: name, type: folder, check_name_mode: refuse, }) return {file_id: d.get(file_id), name: d.get(name), type: d.get(type)} mcp.tool() def rename_file(file_id: str, new_name: str) - Dict[str, Any]: 重命名文件或文件夹。 d api(openFile/update, { drive_id: drive_id(), file_id: file_id, name: new_name, check_name_mode: refuse, }) return {file_id: d.get(file_id), name: d.get(name), updated_at: d.get(updated_at)} mcp.tool() def move_files(file_ids: List[str], target_parent_id: str) - Dict[str, Any]: 批量移动文件/文件夹到目标目录移动不是复制。 drive_id drive_id() results, errors [], [] for fid in file_ids: try: d api(openFile/update, { drive_id: drive_id, file_id: fid, to_parent_file_id: target_parent_id, }) results.append({file_id: fid, ok: True, name: d.get(name)}) except Exception as e: errors.append({file_id: fid, ok: False, error: str(e)}) return {moved: results, failed: errors} mcp.tool() def delete_file(file_id: str) - Dict[str, Any]: 删除文件或文件夹移入回收站可恢复不是永久销毁。 api(openFile/recyclebin/trash, { drive_id: drive_id(), file_id: file_id, }) return {file_id: file_id, trashed: True} mcp.tool() def get_download_url(file_id: str, expire_sec: int 3600) - Dict[str, Any]: 获取文件的临时下载链接默认有效期 1 小时。文件夹没有下载链接。 info api(openFile/get, {drive_id: drive_id(), file_id: file_id}) if info.get(type) folder: return { file_id: file_id, url: None, error: 这是文件夹没有下载链接。请传入具体文件的 file_id。, } d api(openFile/getDownloadUrl, { drive_id: drive_id(), file_id: file_id, expire_sec: expire_sec, }) return { file_id: file_id, name: info.get(name), url: d.get(url), expiration: d.get(expiration), size: info.get(size) or d.get(size), } if __name__ __main__: mcp.run()上篇说的几个坑在代码里对应的位置刷新 token 的grant_type写法在_refresh()函数里。容量要单独查getSpaceInfo在get_user_info()里。文件夹没有下载链接的提示在get_download_url()末尾。token 这块get_token()会在 access_token 快过期时自动调_refresh()外部工具无感。刷新失败比如 refresh_token 作废会抛错这时候跑 auth.py 重新扫码就行。auth.py完整代码阿里云盘 MCP 授权工具 用法: python auth.py # 打印授权页链接与步骤说明 python auth.py refresh_token # 换取 access_token 写入 token.json并验证连通性 注意: refresh_token 约 30 天失效失效后需重新扫码授权。 import os import sys import json import time import requests BASE_DIR os.path.dirname(os.path.abspath(__file__)) TOKEN_FILE os.environ.get(ALIPAN_TOKEN_FILE) or os.path.join(BASE_DIR, token.json) TOKEN_FILE os.path.abspath(TOKEN_FILE) OAUTH_TOKEN_URL os.environ.get(ALIPAN_OAUTH_TOKEN_URL) or https://api.alistgo.com/alist/ali_open/token API_BASE https://openapi.alipan.com AUTH_PAGE https://alistgo.com/tool/aliyundrive/request def usage() - None: print( * 60) print(阿里云盘 MCP 授权) print( * 60) print(f\n1. 浏览器打开授权页:\n {AUTH_PAGE}) print(\n2. 页面上有两种方式任选其一:) print( - 桌面浏览器: 点「前往登录」- 阿里云盘扫码/登录) print( - 手机 APP: 点「扫描二维码」- 阿里云盘 APP 扫) print(\n3. 授权成功后页面会显示一串 refresh_token复制它) print(\n4. 回到这里执行:) print( python auth.py 粘贴的 refresh_token) print(f\nToken 将保存到: {TOKEN_FILE}) print(注意: refresh_token 约 30 天失效失效后需重新走一遍上面的流程。) print( * 60) def exchange(refresh_token: str) - dict: 用 refresh_token 换 access_token。 注意: 阿里云盘 refresh_token 一次性消耗试错须按概率排序成功即止。 社区代理端点(api.alistgo.com)对字段名敏感: 传 codeauthorization_code 模式 传 refresh_token续期模式。用错会报 Incorrect GrantType。 attempts [ {refresh_token: refresh_token}, {code: refresh_token}, {grant_type: refresh_token, refresh_token: refresh_token}, ] last_err for payload in attempts: field ,.join(payload.keys()) r requests.post(OAUTH_TOKEN_URL, jsonpayload, timeout30) if r.status_code 400: body r.json() if body.get(access_token): print(f (成功字段格式: {field})) return body last_err f返回异常[{field}]: {json.dumps(body, ensure_asciiFalse)[:300]} else: last_err fHTTP {r.status_code}[{field}]: {r.text[:300]} # 只在字段不被认识类错误时继续尝试避免误耗 token if GrantType not in last_err and Invalid request not in last_err: break raise RuntimeError(f换取 token 失败。最后一次: {last_err}) def _call(access_token: str, path: str) - dict: r requests.post( f{API_BASE}/adrive/v1.0/{path.lstrip(/)}, headers{Authorization: fBearer {access_token}, Content-Type: application/json}, json{}, timeout30, ) if r.status_code 400: raise RuntimeError(f连通性验证失败 ({path} HTTP {r.status_code}): {r.text[:400]}) return r.json() def verify(access_token: str): return _call(access_token, user/getDriveInfo), _call(access_token, user/getSpaceInfo) def main() - int: if len(sys.argv) 2 or not sys.argv[1].strip(): usage() return 0 refresh_token sys.argv[1].strip().strip().strip() print(正在换取 access_token ...) body exchange(refresh_token) data { access_token: body[access_token], refresh_token: body.get(refresh_token) or refresh_token, expires_in: body.get(expires_in, 7200), expires_at: time.time() int(body.get(expires_in, 7200)) - 300, } with open(TOKEN_FILE, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) print(f已保存: {TOKEN_FILE}) print(正在验证连通性 ...) info, space verify(data[access_token]) sp space.get(personal_space_info) or {} gb 1024 ** 3 used, total sp.get(used_size) or 0, sp.get(total_size) or 0 print(\n授权成功:) print(f 用户: {info.get(nick_name) or info.get(user_name)}) print(f drive_id: {info.get(default_drive_id)}) print(f 容量: {used / gb:.2f} GB / {total / gb:.2f} GB (f ({used / total * 100:.1f}%) if total else )) print(\n现在可以在 MCP 里重新连接 aliyundrive-mcp 了。) return 0 if __name__ __main__: try: sys.exit(main()) except Exception as e: print(f\n[失败] {e}, filesys.stderr) sys.exit(1)exchange()里那几个 attempts就是为应对 refresh_token 一次性消耗按可能性排序试第一个拿到的格式就记下来用不瞎试浪费。test_connection.py这个不是必须的但建议留着。MCP 面板报连不上时先跑它能直接分清是 token 过期还是代码挂了阿里云盘 MCP 连接自检 用法: python test_connection.py 作用: 以真实 MCP stdio 方式启动 server.py跑一遍只读工具 确认「token 有效 API 路径正确 工具可用」。 MCP 面板报 Connection closed 时先跑这个定位是 token 问题还是代码问题。 import json import os import subprocess import sys BASE_DIR os.path.dirname(os.path.abspath(__file__)) PY sys.executable SERVER os.path.join(BASE_DIR, server.py) TOKEN os.environ.get(ALIPAN_TOKEN_FILE) or os.path.join(BASE_DIR, token.json) if not os.path.exists(TOKEN): print(f[FAIL] 未找到 token 文件: {TOKEN}) print( 请先运行: python auth.py refresh_token) sys.exit(1) p subprocess.Popen([PY, -u, SERVER], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.DEVNULL, textTrue, encodingutf-8, bufsize1, env{ALIPAN_TOKEN_FILE: TOKEN, PATH: , SYSTEMROOT: C:\\Windows}) def rpc(payload): p.stdin.write(json.dumps(payload) \n) p.stdin.flush() return json.loads(p.stdout.readline()) def tool(name, argsNone): r rpc({jsonrpc: 2.0, id: 99, method: tools/call, params: {name: name, arguments: args or {}}}) if error in r: return {__error__: r[error].get(message, str(r[error]))} txt r[result][content][0][text] try: return json.loads(txt) except ValueError: return txt fails [] def check(label, res): ok not (isinstance(res, dict) and __error__ in res) print(f[{PASS if ok else FAIL}] {label}) if not ok: fails.append(label) print( , res.get(__error__)) return res try: r rpc({jsonrpc: 2.0, id: 1, method: initialize, params: {protocolVersion: 2024-11-05, capabilities: {}, clientInfo: {name: selfcheck, version: 1}}}) print(fserver - {r[result][serverInfo][name]} v{r[result][serverInfo][version]}) p.stdin.write(json.dumps({jsonrpc: 2.0, method: notifications/initialized}) \n) p.stdin.flush() r rpc({jsonrpc: 2.0, id: 2, method: tools/list}) print(ftools - {len(r[result][tools])} 个\n) info check(get_user_info (账号容量), tool(get_user_info)) if isinstance(info, dict) and user_name in info: print(f {info[user_name]} | {info.get(used_gb)} GB / {info.get(total_gb)} GB) files check(list_files (根目录), tool(list_files, {limit: 3})) if isinstance(files, list): for i in files[:3]: print(f [{i[type]}] {i[name]}) check(search_files (搜索), tool(search_files, {query: e, limit: 1})) finally: p.terminate() print(\n * 44) if fails: print(f自检失败: {, .join(fails)}) sys.exit(1) print(自检通过token 有效MCP 可正常连接。)接到 MCP把下面这段加进你的 mcp.json路径换成你自己的{ mcpServers: { aliyundrive-mcp: { command: 你的 python 解释器路径, args: [-u, aliyundrive-mcp/server.py], env: { ALIPAN_TOKEN_FILE: aliyundrive-mcp/token.json } } } }启动命令里那个-u别省Python 非 tty 下 stdout 块缓冲会卡死 MCP 的 stdio 握手。token 路径走环境变量不写死进脚本换机器也好使。从零跑起来拿到三个文件按这个顺序来装依赖pip install fastmcp requestsPython 3.10 以上fastmcp 3.x 需要。建目录放文件新建一个文件夹把 server.py / auth.py / test_connection.py 放进去。token.json 不用手建跑授权时会自动生成。拿 refresh_token浏览器打开 Get Aliyundrive Refresh Token | AList Docs 电脑点前往登录手机用阿里云盘 App 扫二维码。授权成功页面会给一串 refresh_token复制下来。换 token在文件夹里开终端python auth.py 粘贴的refresh_token成功会打印账号名和容量并生成 token.json。接进 MCP 客户端把上节那段配置加进你的 mcp.json路径换成自己的。注意那个-u别省Python 非 tty 下 stdout 块缓冲会卡死 stdio 握手。自检python test_connection.py全绿就说明 token 有效、9 个工具可用。之后在客户端里就能翻网盘了。refresh_token 约 30 天失效到期重跑第 4 步重新扫码。收尾三个文件建好跑python auth.py refresh_token拿到 tokenMCP 面板连上就能让 AI 助手翻你网盘了。依赖就两个fastmcp 和 requests。环境我用的是隔离 venv你本机有这俩包也能直接跑。下篇就到这。要是跑起来有报错把报错贴出来我看看是哪步没对上。顺带一句手搓工具只是顺手。我之前主业是药企实验室 IT 自动化物理化学本科加计算机化学硕士13 年一线。Office一键报表、仪器数据对接、LIMS 定制这类活也会。想把你实验室那些重复劳动直接干掉可以找我聊聊。