query_video_status
Check the status of a video generation task and retrieve results, including formatted video and image links, using the specified task ID.
Instructions
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | Yes |
Implementation Reference
- scenext_mcp/server.py:141-194 (handler)The handler function decorated with @mcp.tool(), implementing the logic to query video generation status using the Scenext API. Handles authentication from headers, API calls, and error responses.@mcp.tool() async def query_video_status(task_id: str) -> Dict[str, Any]: """ 查询视频生成状态 Args: task_id: 视频生成任务ID Returns: 包含任务状态信息与任务结果的字典，任务结果中的视频链接请用"[链接文本](视频URL)"形式展示给用户，图片链接请用""形式展示给用户 """ # 从请求头获取API KEY api_key = get_api_key_from_headers() if not api_key or api_key == "YOUR_API_KEY": return {"error": "请在请求头中提供有效的API Key，支持Authorization: Bearer <token>或X-API-Key: <token>"} if not task_id.strip(): return {"error": "任务ID不能为空"} url = f"{API_BASE_URL}/get_status/{task_id}" headers = { "Authorization": f"Bearer {api_key}" } try: async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as response: if response.status == 200: result = await response.json() logger.info(f"状态查询成功: {result}") if result.get("status") == "success": return result.get("data", {}) else: return { "error": "API返回错误状态", "details": result } else: error_text = await response.text() logger.error(f"状态查询失败: {response.status} - {error_text}") return { "error": f"状态查询失败: {response.status}", "details": error_text } except aiohttp.ClientError as e: logger.error(f"网络请求错误: {e}") return {"error": f"网络请求错误: {str(e)}"} except Exception as e: logger.error(f"未知错误: {e}") return {"error": f"未知错误: {str(e)}"}
- scenext_mcp/server.py:141-141 (registration)The @mcp.tool() decorator registers the query_video_status function as an MCP tool.@mcp.tool()
- scenext_mcp/server.py:41-62 (helper)Helper function to extract API key from HTTP headers or fallback to environment variable, used by the tool for authentication.def get_api_key_from_headers() -> str: """从HTTP请求头中获取API KEY""" try: headers = get_http_headers(include_all=True) # 尝试从Authorization header获取Bearer token auth_header = headers.get("authorization", "") if auth_header.startswith("Bearer "): return auth_header[7:] # 移除 "Bearer " 前缀 # 尝试从自定义header获取 api_key = headers.get("x-api-key", "") if api_key: return api_key # 如果都没有，返回环境变量中的默认值 return os.getenv("SCENEXT_API_KEY", "") except RuntimeError: # 如果不在HTTP请求上下文中，使用环境变量 return os.getenv("SCENEXT_API_KEY", "")
- scenext_mcp/server.py:143-151 (schema)Docstring providing the input schema (task_id: str) and output description for the tool.""" 查询视频生成状态 Args: task_id: 视频生成任务ID Returns: 包含任务状态信息与任务结果的字典，任务结果中的视频链接请用"[链接文本](视频URL)"形式展示给用户，图片链接请用""形式展示给用户 """