douyin-mcp-server：抖音无水印视频下载与文案提取 MCP 服务器

# 克隆项目
git clone https://github.com/yzfly/douyin-mcp-server.git
cd douyin-mcp-server

# 安装依赖
uv sync

# 启动服务
uv run python web/app.py

{
  "mcpServers": {
    "douyin-mcp": {
      "command": "uvx",
      "args": ["douyin-mcp-server"],
      "env": {
        "API_KEY": "sk-xxxxxxxxxxxxxxxx"
      }
    }
  }
}

# 查看帮助
uv run python douyin-video/scripts/douyin_downloader.py --help

# 获取视频信息（无需 API）
uv run python douyin-video/scripts/douyin_downloader.py -l "分享链接" -a info

# 下载无水印视频
uv run python douyin-video/scripts/douyin_downloader.py -l "分享链接" -a download -o ./videos

# 提取文案（需要 API_KEY）
export API_KEY="sk-xxx"
uv run python douyin-video/scripts/douyin_downloader.py -l "分享链接" -a extract -o ./output

#!/usr/bin/env python3
"""
抖音无水印视频下载并提取文本的 MCP 服务器
"""
import os
import re
import json
import requests
import tempfile
import asyncio
from pathlib import Path
from typing import Optional
import dashscope
from mcp.server.fastmcp import FastMCP, Context

# 创建 MCP 服务器实例
mcp = FastMCP("Douyin MCP Server")

class DouyinProcessor:
    """抖音视频处理器"""

    def __init__(self, api_key: str, model: Optional[str] = None):
        self.api_key = api_key
        self.model = model or "paraformer-v2"
        self.temp_dir = Path(tempfile.mkdtemp())
        dashscope.api_key = api_key

    def parse_share_url(self, share_text: str) -> dict:
        """从分享文本中提取无水印视频链接"""
        # 提取分享链接
        urls = re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', share_text)
        if not urls:
            raise ValueError("未找到有效的分享链接")

        share_url = urls[0]
        share_response = requests.get(share_url, headers=HEADERS)
        video_id = share_response.url.split("?")[0].strip("/").split("/")[-1]

        # 获取视频页面内容并解析
        response = requests.get(share_url, headers=HEADERS)
        pattern = re.compile(r"window\._ROUTER_DATA\s*=\s*(.*?)</script>", re.DOTALL)
        find_res = pattern.search(response.text)

        json_data = json.loads(find_res.group(1).strip())
        data = json_data["loaderData"]["video_(id)/page"]["videoInfoRes"]["item_list"][0]

        # 获取无水印视频链接
        video_url = data["video"]["play_addr"]["url_list"][0].replace("playwm", "play")

        return {
            "url": video_url,
            "title": data.get("desc", "").strip() or f"douyin_{video_id}",
            "video_id": video_id
        }

@mcp.tool()
def get_douyin_download_link(share_link: str) -> str:
    """获取抖音视频的无水印下载链接"""
    try:
        processor = DouyinProcessor("")
        video_info = processor.parse_share_url(share_link)
        return json.dumps({
            "status": "success",
            "video_id": video_info["video_id"],
            "title": video_info["title"],
            "download_url": video_info["url"]
        }, ensure_ascii=False, indent=2)
    except Exception as e:
        return json.dumps({
            "status": "error",
            "error": f"获取下载链接失败: {str(e)}"
        }, ensure_ascii=False, indent=2)

@mcp.tool()
async def extract_douyin_text(share_link: str, model: Optional[str] = None, ctx: Context = None) -> str:
    """从抖音分享链接提取视频中的文本内容"""
    try:
        api_key = os.getenv('API_KEY')
        if not api_key:
            raise ValueError("未设置环境变量 API_KEY")

        processor = DouyinProcessor(api_key, model)
        ctx.info("正在解析抖音分享链接...")
        video_info = processor.parse_share_url(share_link)

        ctx.info("正在从视频中提取文本...")
        text_content = processor.extract_text_from_video_url(video_info['url'])
        return text_content
    except Exception as e:
        raise Exception(f"提取抖音视频文本失败: {str(e)}")

def main():
    """启动MCP服务器"""
    mcp.run()

if __name__ == "__main__":
    main()