getDh123VideoStatus
Check the progress of 123 Digital Human video synthesis using task ID. This tool queries the status of video generation tasks on the Baidu Digital Human MCP Server.
Instructions
#工具说明：查询123数字人视频合成进度。
样例1：
用户输入：查一下taskid为xxx的123数字人视频好了没有 思考过程: 1.用户想要查询taskid为xxx的123数字人视频，需要使用“getDh123VideoStatus”工具。 2.工具需要task ID这些参数。 3.task ID的值为xxx
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| taskId | No | 视频任务ID |
Implementation Reference
- The main handler function implementing the tool logic: retrieves DH API client and calls get_video_status with the provided taskId, returning MCPVideoStatusResponse.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 'getDh123VideoStatus' tool with the FastMCP server, including name and detailed usage description.@mcp.tool( name="getDh123VideoStatus", description=( """ #工具说明：查询123数字人视频合成进度。 # 样例1： 用户输入：查一下taskid为xxx的123数字人视频好了没有 思考过程: 1.用户想要查询taskid为xxx的123数字人视频，需要使用“getDh123VideoStatus”工具。 2.工具需要task ID这些参数。 3.task ID的值为xxx """) )
- Pydantic model defining the structure of the tool's output response, used for validation and serialization.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
- Supporting method in DHApiClient that makes the actual API request to retrieve video status, handles response validation and error mapping to MCPVideoStatusResponse. Called by the handler with isAdvanced=False.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}" )