query_video_status
Check the status and retrieve results of video generation tasks in the edu_video_gen system. Use this tool to monitor progress and access completed videos and images with properly formatted links.
Instructions
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- scenext_mcp/server.py:141-194 (handler)The primary handler function for the 'query_video_status' MCP tool. It extracts the API key from HTTP headers, queries the Scenext API for the video task status, and returns the result or error.
@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:41-62 (helper)Helper utility function used by the query_video_status tool (and gen_video) to retrieve the API key from HTTP request headers or fallback to environment variable.
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", "") - tests/test_basic.py:31-41 (registration)Test code verifying that the 'query_video_status' tool is properly registered in the MCP server alongside 'gen_video'.
# 检查工具是否正确注册 expected_tools = ['gen_video', 'query_video_status'] # 获取注册的工具（这需要访问内部 API） tools = mcp._tool_manager.list_tools() tool_names = [tool.name for tool in tools] for expected_tool in expected_tools: assert expected_tool in tool_names, f"工具 {expected_tool} 未注册" print(f"工具注册测试通过 - 注册工具: {tool_names}")