Instructions

Implementation Reference

• src/mcp_server_baidu_digitalhuman/dhserver.py:332-350 (handler)

  The handler function that implements the core logic of the 'getDhVideoStatus' tool by querying the DH API client for video status.

  async def getDhVideoStatus(
          taskId: Annotated[str, Field(description="基础数字人视频任务ID", default=None)],
  ) -> MCPVideoStatusResponse:
      """
      Retrieve the status of a video generated via the DH API.
  
      Args:
          taskId: 任务ID
  
      Returns:
          任务状态
      """
      try:
          client = await getDhClient()
          ret = await client.get_video_status(taskId)
          return ret
      except Exception as e:
          return MCPVideoStatusResponse(error=str(e))

• src/mcp_server_baidu_digitalhuman/dhserver.py:318-331 (registration)

  The FastMCP @mcp.tool decorator registering the 'getDhVideoStatus' tool, including its name and description.

  @mcp.tool(
      name="getDhVideoStatus",
      description=(
      """
  #工具说明：查询基础数字人视频合成进度。
  # 样例1：
  用户输入：查一下taskid为xxx的数字人视频好了没有
  思考过程:
  1.用户想要查询taskid为xxx的数字人视频好了没有，需要根据context来做判断，最近调用了“generateDhVideo”工具，\
  需要使用“getDhVideoStatus”工具，如果没有查询到，则需要使用“getDh123VideoStatus”工具继续查询。
  2.工具需要task ID这些参数。
  3.task ID的值为xxx
      """)
  )

• src/mcp_server_baidu_digitalhuman/dhapi.py:303-313 (schema)

  Pydantic BaseModel defining the output schema (response structure) for the 'getDhVideoStatus' tool.

  class MCPVideoStatusResponse(BaseDHResponse):
      """ MCP 视频状态响应 """
      taskId: Optional[str] = None
      status: Optional[str] = None
      duration: Optional[float] = None
      videoUrl: Optional[str] = None
      requestId: Optional[str] = None
      failedCode: Optional[int] = None
      createTime: Optional[str] = None
      failedMessage: Optional[Dict[str, Any]] = None

• src/mcp_server_baidu_digitalhuman/dhapi.py:640-686 (helper)

  The DHApiClient method that performs the actual API call to retrieve video status, used by the tool handler.

  async def get_video_status(self, taskId: str, isAdvanced: bool = False) -> MCPVideoStatusResponse:
      """Get the status of a generated video from the API."""
  
      async def api_call():
          endpoint = f"api/digitalhuman/open/v1/video/advanced/task?taskId={taskId}" \
              if isAdvanced else f"api/digitalhuman/open/v1/video/task?taskId={taskId}"
          return await self._make_request(endpoint)
  
      try:
          result = await api_call()
  
          validated_response = VideoStatusResponse.model_validate(result)
  
          data = validated_response.result
  
          error_details = None
          if validated_response.code != 0 :
              error_details = {
                  "code": validated_response.code,
                  "message": validated_response.message.global_field,
              }
  
          return MCPVideoStatusResponse(
              taskId=data.taskId,
              status=data.status,
              duration=data.duration,
              videoUrl=data.videoUrl,
              createTime=data.createTime,
              failedMessage=error_details,
          )
      except httpx.RequestError as exc:
          return MCPVideoStatusResponse(
              error=f"HTTP Request failed: {exc}"
          )
      except httpx.HTTPStatusError as exc:
          return MCPVideoStatusResponse(
              error=f"HTTP Error: {exc.response.status_code} - {exc.response.text}"
          )
      except ValidationError as ve:
          return MCPVideoStatusResponse(
              error=f"ValidationError: {ve}"
          )
      except Exception as e:
          return MCPVideoStatusResponse(
              error=f"An unexpected error occurred: {e}"
          )