NoLang MCP Server

Instructions

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")