Instructions

Implementation Reference

• src/mcp_server_baidu_digitalhuman/dhserver.py:388-400 (registration)

  Registration of the 'getText2AudioStatus' tool using @mcp.tool decorator, including name, description with usage examples.

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

• src/mcp_server_baidu_digitalhuman/dhserver.py:401-416 (handler)

  Handler function that executes the tool: obtains DHApiClient and calls its get_text2audio_status method with the taskId, returns MCPText2AudioStatusResponse.

  async def getText2AudioStatus(
          taskId: Annotated[str, Field(description="语音合成任务ID", default=None)]
  ) -> MCPText2AudioStatusResponse:
      """
      Retrieve the status of generated audio via the DH API.
  
      Args:
          taskId: 任务ID
      """
      try:
          client = await getDhClient()
          ret = await client.get_text2audio_status(taskId)
          return ret
      except Exception as e:
          return MCPText2AudioStatusResponse(error=str(e))

• src/mcp_server_baidu_digitalhuman/dhapi.py:334-344 (schema)

  Pydantic model defining the output schema for the getText2AudioStatus tool response.

  class MCPText2AudioStatusResponse(BaseDHResponse):
      """ MCP 文本转语音状态响应 """
      taskId: Optional[str] = None
      status: Optional[str] = None
      failedCode: Optional[int] = None
      failedMessage: Optional[str] = None
      audioUrl: Optional[str] = None
      duration: Optional[int] = None
      textTimestamp: List[TextTimeStampInfo]
      createTime: Optional[str] = None
      updateTime: Optional[str] = None

• src/mcp_server_baidu_digitalhuman/dhapi.py:707-741 (helper)

  Supporting method in DHApiClient that performs the actual API call to query text-to-audio status and transforms the response into MCP format.

  async def get_text2audio_status(self, taskId: str) -> MCPText2AudioStatusResponse:
      """Get the status of a generated video from the API."""
  
      async def api_call():
          endpoint = f"api/digitalhuman/open/v1/tts/text2audio/task?taskId={taskId}"
          return await self._make_request(endpoint)
  
      try:
          result = await api_call()
  
          validated_response = Text2AudioStatusResponse.model_validate(result)
          data = validated_response.result
  
          return MCPText2AudioStatusResponse(
              taskId=data.taskId,
              status=data.status,
              audioUrl=data.audioUrl,
              duration=data.duration,
              textTimestamp=data.textTimestamp,
              createTime=data.createTime,
              updateTime=data.updateTime,
              failedCode=validated_response.code,
              failedMessage="" if validated_response.message is None else validated_response.message.global_field,
          )
      except httpx.RequestError as exc:
          return MCPText2AudioStatusResponse(error=f"HTTP Request failed: {exc}")
      except httpx.HTTPStatusError as exc:
          return MCPText2AudioStatusResponse(
              error=f"HTTP Error: {exc.response.status_code} - {exc.response.text}"
          )
      except ValidationError as ve:
          return MCPText2AudioStatusResponse(error=f"ValidationError")
      except Exception as e:
          return MCPText2AudioStatusResponse(error=f"An unexpected error occurred: {e}")