Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/deployment.py:249-306 (handler)

  The main handler function for the MCP tool 'set_deployment_schedule'. It is decorated with @mcp.tool for registration and implements the logic to set a Prefect deployment's schedule using either cron or interval-based scheduling via the Prefect client.

  @mcp.tool
  async def set_deployment_schedule(
      deployment_id: str,
      cron: Optional[str] = None,
      interval_seconds: Optional[int] = None,
      anchor_date: Optional[str] = None,
      timezone: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Set a deployment's schedule.
      
      Args:
          deployment_id: The deployment UUID
          cron: Cron expression for the schedule
          interval_seconds: Alternative to cron - interval in seconds
          anchor_date: Required for interval schedules - the anchor date
          timezone: Timezone for the schedule
          
      Returns:
          Updated schedule details
      """
      async with get_client() as client:
          # Check schedule type
          if cron is not None and interval_seconds is not None:
              return [types.TextContent(
                  type="text",
                  text="Cannot specify both cron and interval_seconds. Choose one schedule type."
              )]
          
          if cron is not None:
              # Set cron schedule
              schedule = await client.set_deployment_schedule(
                  deployment_id=UUID(deployment_id),
                  schedule={"cron": cron, "timezone": timezone}
              )
          elif interval_seconds is not None:
              # Set interval schedule
              if not anchor_date:
                  return [types.TextContent(
                      type="text",
                      text="anchor_date is required for interval schedules"
                  )]
              
              schedule = await client.set_deployment_schedule(
                  deployment_id=UUID(deployment_id),
                  schedule={
                      "interval": interval_seconds,
                      "anchor_date": anchor_date,
                      "timezone": timezone
                  }
              )
          else:
              return [types.TextContent(
                  type="text",
                  text="Must specify either cron or interval_seconds to set a schedule"
              )]
          
          return [types.TextContent(type="text", text=str(schedule.model_dump()))]