edu_video_gen

Instructions

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)"形式展示给用户，图片链接请用"![链接文本](图片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}")