Files
Agent-Reach/agent_reach/channels/douyin.py
T
Shawn 3a3d38acce fix: YouTube JS runtime check, Douyin health check, cli bare except, config permission race (#104)
- YouTube: warn when only Node.js is installed but yt-dlp config file
  is missing (previously returned "ok" incorrectly)
- Douyin: use `mcporter list` instead of calling with a hardcoded
  invalid share URL that always fails
- cli: replace bare `except:` with `except Exception:` in
  `_detect_environment` to avoid catching KeyboardInterrupt/SystemExit
- cli: fix unclosed file handle for cloud VM detection
- config: use `os.open()` with 0o600 mode to eliminate permission
  race window when saving credentials
2026-03-08 21:24:05 +08:00

57 lines
2.3 KiB
Python

# -*- coding: utf-8 -*-
"""Douyin (抖音) — check if mcporter + douyin-mcp-server is available."""
import shutil
import subprocess
from .base import Channel
class DouyinChannel(Channel):
name = "douyin"
description = "抖音短视频"
backends = ["douyin-mcp-server"]
tier = 2
def can_handle(self, url: str) -> bool:
from urllib.parse import urlparse
d = urlparse(url).netloc.lower()
return "douyin.com" in d or "iesdouyin.com" in d
def check(self, config=None):
mcporter = shutil.which("mcporter")
if not mcporter:
return "off", (
"需要 mcporter + douyin-mcp-server。安装步骤:\n"
" 1. npm install -g mcporter\n"
" 2. pip install douyin-mcp-server\n"
" 3. 启动服务(见下方说明)\n"
" 4. mcporter config add douyin http://localhost:18070/mcp\n"
" 详见 https://github.com/yzfly/douyin-mcp-server"
)
try:
r = subprocess.run(
[mcporter, "config", "list"], capture_output=True,
encoding="utf-8", errors="replace", timeout=5
)
if "douyin" not in r.stdout:
return "off", (
"mcporter 已装但抖音 MCP 未配置。运行:\n"
" pip install douyin-mcp-server\n"
" # 启动服务后:\n"
" mcporter config add douyin http://localhost:18070/mcp"
)
except Exception:
return "off", "mcporter 连接异常"
# Verify MCP connectivity by listing available tools instead of
# calling with a hardcoded (invalid) share link that always fails.
try:
r = subprocess.run(
[mcporter, "list", "douyin"],
capture_output=True, encoding="utf-8", errors="replace", timeout=15
)
if r.returncode == 0 and r.stdout.strip():
return "ok", "完整可用(视频解析、下载链接获取)"
return "warn", "MCP 已连接但工具列表为空,检查 douyin-mcp-server 服务是否在运行"
except Exception:
return "warn", "MCP 连接异常,检查 douyin-mcp-server 服务是否在运行"