Instructions

Implementation Reference

• src/mcp_server_baidu_digitalhuman/dhserver.py:223-237 (handler)

  The async handler function implementing the tool's core logic: gets the DH client and calls get_video_status(taskId).

  async def getDh123VideoStatus(
          taskId: Annotated[str, Field(description="视频任务ID", default=None)],
  ) -> MCPVideoStatusResponse:
      """
      Retrieve the status of a 123 video generated via the DH API.
  
      Args:
          taskId: 视频任务ID
      """
      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:210-222 (registration)

  Registers the tool with @mcp.tool, defining name, description, and input schema via the function signature.

  @mcp.tool(
      name="getDh123VideoStatus",
      description=(
      """
  #工具说明：查询123数字人视频合成进度。
  # 样例1：
  用户输入：查一下taskid为xxx的123数字人视频好了没有
  思考过程:
  1.用户想要查询taskid为xxx的123数字人视频，需要使用“getDh123VideoStatus”工具。
  2.工具需要task ID这些参数。
  3.task ID的值为xxx
      """)
  )

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

  Pydantic model for the tool's output response structure and validation.

  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)

  DHApiClient method that handles the API request to the Baidu DH service for video status, parses response using VideoStatusResponse, and maps to MCPVideoStatusResponse.

  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}"
          )