wait_video_generation_and_get_download_url
Monitor video generation progress and retrieve the download URL when complete. Use this tool to track AI-generated videos from text input and access them for download.
Instructions
Polls until video generation completes and returns the download URL.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| args | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| status | Yes | Current status of the video | |
| video_id | Yes | Unique identifier for the video | |
| download_url | No | Signed URL for the finished MP4 file |
Implementation Reference
- nolang_mcp/server.py:115-157 (handler)The main handler function decorated with @mcp.tool. It polls the API for video status at intervals until completion, failure, or timeout, reporting progress and returning the download URL upon success.
@mcp.tool( name="wait_video_generation_and_get_download_url", description="Polls until video generation completes and returns the download URL.", ) async def wait_video_generation_and_get_download_url( args: VideoWaitArgs, ctx: Context, ) -> VideoStatusResult: start = time.time() while time.time() - start < args.max_wait_time: try: status_response = await nolang_api.get_video_status(args.video_id) except httpx.HTTPStatusError as e: raise RuntimeError(format_http_error(e)) from e current_status = VideoStatusEnum(status_response.status) if current_status == VideoStatusEnum.RUNNING: await ctx.report_progress( progress=time.time() - start, total=args.max_wait_time ) # notify progress to client await asyncio.sleep(args.check_interval) continue if current_status == VideoStatusEnum.COMPLETED: await ctx.report_progress( progress=args.max_wait_time, total=args.max_wait_time ) # notify progress to client (100%) return VideoStatusResult( video_id=args.video_id, status=current_status, download_url=status_response.download_url, ) if current_status == VideoStatusEnum.FAILED: raise RuntimeError(f"Video generation failed. Video ID: {args.video_id}") # Unknown status – wait and retry await asyncio.sleep(args.check_interval) raise TimeoutError("Video generation did not complete within the time limit.") - nolang_mcp/models.py:268-284 (schema)Pydantic schema for the tool's input arguments: video_id (required), max_wait_time (default 600s), check_interval (default 10s).
class VideoWaitArgs(BaseModel): """Arguments for waiting for video generation completion.""" model_config = ConfigDict(extra="forbid") video_id: UUID = Field( ..., description="Video ID of the generation job", ) max_wait_time: int = Field( default=600, description="Maximum seconds to wait for generation completion", ge=1, le=3600, ) check_interval: int = Field(default=10, description="Interval (seconds) to check status", ge=1, le=60) - nolang_mcp/models.py:311-317 (schema)Pydantic schema for the tool's output: video_id, status, and download_url.
class VideoStatusResult(BaseModel): model_config = ConfigDict(extra="allow") video_id: UUID = Field(..., description="Unique identifier for the video") status: VideoStatusEnum = Field(..., description="Current status of the video") download_url: Optional[str] = Field(None, description="Signed URL for the finished MP4 file")