getText2AudioStatus
Check the progress and status of text-to-speech synthesis tasks by providing the task ID to monitor completion and retrieve results.
Instructions
#工具说明：查询音频合成进度。
样例1：
用户输入：查一下taskid为xxx的语音合成好了没有。 思考过程: 1.用户想要查询taskid为xxx的音频好了没有，需要使用“getText2AudioStatus”工具查询。 2.工具需要task ID这些参数。 3.task ID的值为xxx
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| taskId | No | 语音合成任务ID |
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 """) ) - 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)) - 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 - 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}")