analyze_douyin_video
Analyzes Douyin videos from share links or text, extracting structured content like summaries, outlines, and step-by-step tutorials.
Instructions
分析抖音视频内容。可以接收抖音口令或直接的抖音链接。自动提取链接、获取视频并使用 AI 进行详细分析，返回视频描述、大纲、教程步骤等结构化内容。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| text | Yes | 抖音口令或链接。例如：'0.05 09/23 III:/ e@B.tE 今年农户都不易啊... https://v.douyin.com/DGHl69ciWp4/ 复制此链接...' 或直接 'https://v.douyin.com/DGHl69ciWp4/' |
Implementation Reference
- src/douyin_mcp_server.py:55-83 (registration)The tool 'analyze_douyin_video' is registered via the @server.list_tools() decorator, defining its name, description, and inputSchema (requires a 'text' string containing a Douyin link/phrase).
@server.list_tools() async def handle_list_tools() -> list[Tool]: """ 列出可用的工具 """ return [ Tool( name="analyze_douyin_video", description=( "分析抖音视频内容。可以接收抖音口令或直接的抖音链接。" "自动提取链接、获取视频并使用 AI 进行详细分析，" "返回视频描述、大纲、教程步骤等结构化内容。" ), inputSchema={ "type": "object", "properties": { "text": { "type": "string", "description": ( "抖音口令或链接。例如：'0.05 09/23 III:/ e@B.tE 今年农户都不易啊... " "https://v.douyin.com/DGHl69ciWp4/ 复制此链接...' 或直接 " "'https://v.douyin.com/DGHl69ciWp4/'" ) } }, "required": ["text"] } ) ] - src/douyin_mcp_server.py:86-148 (handler)The main handler for the 'analyze_douyin_video' tool. It extracts a Douyin link from text (via link_extractor), gets the real video URL (via DouyinAPI), then analyzes the video content using AI (via DoubaoAPI).
@server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]: """ 处理工具调用 """ if name != "analyze_douyin_video": raise ValueError(f"未知工具: {name}") text = arguments.get("text", "") if not text: return [TextContent( type="text", text="错误：未提供文本内容" )] try: # 步骤 1: 提取抖音链接 link = extract_douyin_link(text) if not link: return [TextContent( type="text", text=f"错误：无法从文本中提取抖音链接。\n\n输入的文本：{text}" )] result_parts = [f"✓ 成功提取链接: {link}\n"] # 步骤 2: 获取视频真实 URL try: video_url = await douyin_client.get_video_url(link) result_parts.append(f"✓ 成功获取视频 URL\n") except DouyinAPIError as e: return [TextContent( type="text", text=f"错误：获取视频 URL 失败\n\n{str(e)}" )] # 步骤 3: 使用豆包分析视频 try: client = get_doubao_client() analysis = await client.analyze_video(video_url) result_parts.append(f"✓ 视频分析完成\n\n{'='*50}\n\n{analysis}") return [TextContent( type="text", text="\n".join(result_parts) )] except DoubaoAPIError as e: return [TextContent( type="text", text=f"错误：视频分析失败\n\n{str(e)}" )] except ValueError as e: return [TextContent( type="text", text=f"配置错误：{str(e)}" )] except Exception as e: return [TextContent( type="text", text=f"未知错误：{str(e)}" )] - src/link_extractor.py:7-35 (helper)Helper function extract_douyin_link() that uses regex to extract a Douyin video URL from text (supports v.douyin.com and www.douyin.com).
def extract_douyin_link(text: str) -> Optional[str]: """ 从文本中提取抖音视频链接 支持的格式： - https://v.douyin.com/xxx/ - http://v.douyin.com/xxx/ - https://www.douyin.com/xxx Args: text: 包含抖音链接的文本（可能是口令） Returns: 提取到的链接，如果未找到返回 None Example: >>> text = "0.05 09/23 III:/ e@B.tE 今年农户都不易啊 https://v.douyin.com/DGHl69ciWp4/ 复制此链接..." >>> extract_douyin_link(text) 'https://v.douyin.com/DGHl69ciWp4/' """ # 正则表达式匹配抖音链接 # 支持 v.douyin.com 和 www.douyin.com pattern = r'https?://(?:v\.|www\.)?douyin\.com/[A-Za-z0-9]+/?' match = re.search(pattern, text) if match: return match.group(0) return None - src/douyin_api.py:67-123 (helper)Helper method get_video_url() in DouyinAPIClient that resolves a Douyin share link to a real video URL by querying the video data API and selecting the lowest-bitrate play address.
async def get_video_url(self, url: str) -> Optional[str]: """ 从加密链接获取真实视频 URL 根据返回的 JSON 结构，从 data.video.bit_rate 数组中选择 bit_rate 最小的对象，然后获取其 play_addr.url_list[0] Args: url: 抖音视频链接（加密或真实链接） Returns: 真实视频 URL，如果获取失败返回 None Raises: DouyinAPIError: API 调用失败时抛出 """ response = await self.get_video_data(url) # 检查响应码 if response.get("code") != 200: raise DouyinAPIError( f"抖音 API 返回错误码: {response.get('code')}" ) # 获取 data 字段 data = response.get("data") if not data: raise DouyinAPIError("API 响应中缺少 data 字段") # 获取 video 字段 video = data.get("video") if not video: raise DouyinAPIError("API 响应中缺少 video 字段") # 获取 bit_rate 数组 bit_rate_list = video.get("bit_rate") if not bit_rate_list or not isinstance(bit_rate_list, list) or len(bit_rate_list) == 0: raise DouyinAPIError("API 响应中 bit_rate 数组为空或不存在") # 找到 bit_rate 最小的对象 min_bitrate_item = min(bit_rate_list, key=lambda x: x.get("bit_rate", float('inf'))) # 获取 play_addr.url_list[0] play_addr = min_bitrate_item.get("play_addr") if not play_addr: raise DouyinAPIError("最小码率视频缺少 play_addr 字段") url_list = play_addr.get("url_list") if not url_list or not isinstance(url_list, list) or len(url_list) == 0: raise DouyinAPIError("play_addr.url_list 为空") video_url = url_list[0] if not video_url: raise DouyinAPIError("无法获取视频 URL") return video_url - src/doubao_api.py:34-112 (helper)Helper method analyze_video() in DoubaoAPIClient that sends the video URL to the Doubao AI model for analysis, using a structured prompt from prompts.py, and returns the AI-generated analysis text.
async def analyze_video(self, video_url: str) -> str: """ 分析视频内容 Args: video_url: 视频的真实 URL Returns: AI 生成的结构化视频分析结果 Raises: DoubaoAPIError: API 调用失败时抛出 """ try: # 构建请求数据 payload = { "model": self.model, "messages": [ { "role": "user", "content": [ { "type": "video_url", "video_url": { "url": video_url } }, { "type": "text", "text": get_video_analysis_prompt() } ] } ] } headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post( self.api_endpoint, json=payload, headers=headers ) response.raise_for_status() data = response.json() # 提取 AI 生成的内容 if "choices" in data and len(data["choices"]) > 0: choice = data["choices"][0] if "message" in choice and "content" in choice["message"]: return choice["message"]["content"] raise DoubaoAPIError( f"无法从豆包 API 响应中提取内容。响应数据: {data}" ) except httpx.HTTPStatusError as e: error_detail = "" try: error_data = e.response.json() error_detail = f": {error_data}" except: error_detail = f": {e.response.text}" raise DoubaoAPIError( f"豆包 API 返回错误状态码 {e.response.status_code}{error_detail}" ) from e except httpx.RequestError as e: raise DoubaoAPIError( f"豆包 API 请求失败: {str(e)}" ) from e except Exception as e: raise DoubaoAPIError( f"调用豆包 API 时出错: {str(e)}" ) from e