getDhVideoStatus
Check the progress of digital human video synthesis by providing the task ID to monitor completion status.
Instructions
#工具说明：查询基础数字人视频合成进度。
样例1：
用户输入：查一下taskid为xxx的数字人视频好了没有 思考过程: 1.用户想要查询taskid为xxx的数字人视频好了没有，需要根据context来做判断，最近调用了“generateDhVideo”工具，需要使用“getDhVideoStatus”工具，如果没有查询到，则需要使用“getDh123VideoStatus”工具继续查询。 2.工具需要task ID这些参数。 3.task ID的值为xxx
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| taskId | No | 基础数字人视频任务ID |
Implementation Reference
- 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 """) ) - 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 - 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}" )