NoLang MCP Server

Instructions

Implementation Reference

• nolang_mcp/server.py:115-118 (registration)

  Tool registration using FastMCP @mcp.tool decorator with the exact name.
  @mcp.tool( name="wait_video_generation_and_get_download_url", description="Polls until video generation completes and returns the download URL.", )
• nolang_mcp/server.py:119-156 (handler)

  The core handler function that polls the NoLang API for video status, handles progress reporting, different statuses, and returns the download URL upon completion.
  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 model defining the input parameters for the tool: video_id, max_wait_time, check_interval.
  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 model defining the output schema: 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")
• nolang_mcp/models.py:55-62 (schema)

  Enum used in the handler and output schema for video status values.
  class VideoStatusEnum(str, Enum): """Video generation status.""" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" EXPIRED = "expired"