diff --git a/cmd/config/subcommand/sandbox/config_flags.go b/cmd/config/subcommand/sandbox/config_flags.go
index e9fbd5a6..7b025f54 100755
--- a/cmd/config/subcommand/sandbox/config_flags.go
+++ b/cmd/config/subcommand/sandbox/config_flags.go
@@ -50,7 +50,7 @@ func (Config) mustMarshalJSON(v json.Marshaler) string {
 // flags is json-name.json-sub-name... etc.
 func (cfg Config) GetPFlagSet(prefix string) *pflag.FlagSet {
 	cmdFlags := pflag.NewFlagSet("Config", pflag.ExitOnError)
-	cmdFlags.StringVar(&DefaultConfig.Source, fmt.Sprintf("%v%v", prefix, "source"), DefaultConfig.Source, "Path of your source code")
+	cmdFlags.StringVar(&DefaultConfig.DeprecatedSource, fmt.Sprintf("%v%v", prefix, "source"), DefaultConfig.DeprecatedSource, "deprecated,  path of your source code,  please build images with local daemon")
 	cmdFlags.StringVar(&DefaultConfig.Version, fmt.Sprintf("%v%v", prefix, "version"), DefaultConfig.Version, "Version of flyte. Only supports flyte releases greater than v0.10.0")
 	cmdFlags.StringVar(&DefaultConfig.Image, fmt.Sprintf("%v%v", prefix, "image"), DefaultConfig.Image, "Optional. Provide a fully qualified path to a Flyte compliant docker image.")
 	cmdFlags.BoolVar(&DefaultConfig.Prerelease, fmt.Sprintf("%v%v", prefix, "pre"), DefaultConfig.Prerelease, "Optional. Pre release Version of flyte will be used for sandbox.")
diff --git a/cmd/config/subcommand/sandbox/config_flags_test.go b/cmd/config/subcommand/sandbox/config_flags_test.go
index 0d7021b7..79f6e88f 100755
--- a/cmd/config/subcommand/sandbox/config_flags_test.go
+++ b/cmd/config/subcommand/sandbox/config_flags_test.go
@@ -106,7 +106,7 @@ func TestConfig_SetFlags(t *testing.T) {
 
 			cmdFlags.Set("source", testValue)
 			if vString, err := cmdFlags.GetString("source"); err == nil {
-				testDecodeJson_Config(t, fmt.Sprintf("%v", vString), &actual.Source)
+				testDecodeJson_Config(t, fmt.Sprintf("%v", vString), &actual.DeprecatedSource)
 
 			} else {
 				assert.FailNow(t, err.Error())
diff --git a/cmd/config/subcommand/sandbox/sandbox_config.go b/cmd/config/subcommand/sandbox/sandbox_config.go
index 463f751e..1dfce643 100644
--- a/cmd/config/subcommand/sandbox/sandbox_config.go
+++ b/cmd/config/subcommand/sandbox/sandbox_config.go
@@ -4,7 +4,7 @@ import "github.com/flyteorg/flytectl/pkg/docker"
 
 // Config holds configuration flags for sandbox command.
 type Config struct {
-	Source string `json:"source" pflag:",Path of your source code"`
+	DeprecatedSource string `json:"source" pflag:",deprecated, path of your source code, please build images with local daemon"`
 
 	// Flytectl sandbox only supports Flyte version available in Github release https://github.com/flyteorg/flyte/tags.
 	// Flytectl sandbox will only work for v0.10.0+.
diff --git a/cmd/demo/demo.go b/cmd/demo/demo.go
index 35149b68..23052175 100644
--- a/cmd/demo/demo.go
+++ b/cmd/demo/demo.go
@@ -6,6 +6,11 @@ import (
 	"github.com/spf13/cobra"
 )
 
+const (
+	flyteNs     = "flyte"
+	K8sEndpoint = "https://127.0.0.1:6443"
+)
+
 // Long descriptions are whitespace sensitive when generating docs using sphinx.
 const (
 	demoShort = `Helps with demo interactions like start, teardown, status, and exec.`
@@ -47,6 +52,9 @@ func CreateDemoCommand() *cobra.Command {
 		"start": {CmdFunc: startDemoCluster, Aliases: []string{}, ProjectDomainNotRequired: true,
 			Short: startShort,
 			Long:  startLong, PFlagProvider: sandboxCmdConfig.DefaultConfig, DisableFlyteClient: true},
+		"reload": {CmdFunc: reloadDemoCluster, Aliases: []string{}, ProjectDomainNotRequired: true,
+			Short: reloadShort,
+			Long:  reloadLong, PFlagProvider: sandboxCmdConfig.DefaultConfig, DisableFlyteClient: true},
 		"teardown": {CmdFunc: teardownDemoCluster, Aliases: []string{}, ProjectDomainNotRequired: true,
 			Short: teardownShort,
 			Long:  teardownLong, DisableFlyteClient: true},
diff --git a/cmd/demo/demo_test.go b/cmd/demo/demo_test.go
index 0ce33286..fdc2c09b 100644
--- a/cmd/demo/demo_test.go
+++ b/cmd/demo/demo_test.go
@@ -13,7 +13,8 @@ func TestCreateDemoCommand(t *testing.T) {
 	assert.Equal(t, demoCommand.Use, "demo")
 	assert.Equal(t, demoCommand.Short, "Helps with demo interactions like start, teardown, status, and exec.")
 	fmt.Println(demoCommand.Commands())
-	assert.Equal(t, len(demoCommand.Commands()), 4)
+
+	assert.Equal(t, len(demoCommand.Commands()), 5)
 	cmdNouns := demoCommand.Commands()
 	// Sort by Use value.
 	sort.Slice(cmdNouns, func(i, j int) bool {
@@ -24,16 +25,19 @@ func TestCreateDemoCommand(t *testing.T) {
 	assert.Equal(t, cmdNouns[0].Short, execShort)
 	assert.Equal(t, cmdNouns[0].Long, execLong)
 
-	assert.Equal(t, cmdNouns[1].Use, "start")
-	assert.Equal(t, cmdNouns[1].Short, startShort)
-	assert.Equal(t, cmdNouns[1].Long, startLong)
+	assert.Equal(t, cmdNouns[1].Use, "reload")
+	assert.Equal(t, cmdNouns[1].Short, reloadShort)
+	assert.Equal(t, cmdNouns[1].Long, reloadLong)
 
-	assert.Equal(t, cmdNouns[2].Use, "status")
-	assert.Equal(t, cmdNouns[2].Short, statusShort)
-	assert.Equal(t, cmdNouns[2].Long, statusLong)
+	assert.Equal(t, cmdNouns[2].Use, "start")
+	assert.Equal(t, cmdNouns[2].Short, startShort)
+	assert.Equal(t, cmdNouns[2].Long, startLong)
 
-	assert.Equal(t, cmdNouns[3].Use, "teardown")
-	assert.Equal(t, cmdNouns[3].Short, teardownShort)
-	assert.Equal(t, cmdNouns[3].Long, teardownLong)
+	assert.Equal(t, cmdNouns[3].Use, "status")
+	assert.Equal(t, cmdNouns[3].Short, statusShort)
+	assert.Equal(t, cmdNouns[3].Long, statusLong)
 
+	assert.Equal(t, cmdNouns[4].Use, "teardown")
+	assert.Equal(t, cmdNouns[4].Short, teardownShort)
+	assert.Equal(t, cmdNouns[4].Long, teardownLong)
 }
diff --git a/cmd/demo/reload.go b/cmd/demo/reload.go
new file mode 100644
index 00000000..f7441cf7
--- /dev/null
+++ b/cmd/demo/reload.go
@@ -0,0 +1,58 @@
+package demo
+
+import (
+	"context"
+	"fmt"
+
+	cmdCore "github.com/flyteorg/flytectl/cmd/core"
+	"github.com/flyteorg/flytectl/pkg/docker"
+	"github.com/flyteorg/flytectl/pkg/k8s"
+	"github.com/flyteorg/flytestdlib/logger"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	labelSelector = "app=flyte"
+)
+const (
+	reloadShort = "Power cycle the Flyte executable pod, effectively picking up an updated config."
+	reloadLong  = `
+If you've changed the ~/.flyte/state/flyte.yaml file, run this command to restart the Flyte binary pod, effectively
+picking up the new settings:
+
+Usage
+::
+
+ flytectl demo reload
+
+`
+)
+
+// reloadDemoCluster will kill the flyte binary pod so the new one can pick up a new config file
+func reloadDemoCluster(ctx context.Context, args []string, cmdCtx cmdCore.CommandContext) error {
+	k8sClient, err := k8s.GetK8sClient(docker.Kubeconfig, K8sEndpoint)
+	if err != nil {
+		fmt.Println("Could not get K8s client")
+		return err
+	}
+	pi := k8sClient.CoreV1().Pods(flyteNs)
+	podList, err := pi.List(ctx, v1.ListOptions{LabelSelector: labelSelector})
+	if err != nil {
+		fmt.Println("could not list pods")
+		return err
+	}
+	if len(podList.Items) != 1 {
+		return fmt.Errorf("should only have one pod running, %d found, %v", len(podList.Items), podList.Items)
+	}
+	logger.Debugf(ctx, "Found %d pods\n", len(podList.Items))
+	var grace = int64(0)
+	err = pi.Delete(ctx, podList.Items[0].Name, v1.DeleteOptions{
+		GracePeriodSeconds: &grace,
+	})
+	if err != nil {
+		fmt.Printf("Could not delete Flyte pod, old configuration may still be in effect. Err: %s\n", err)
+		return err
+	}
+
+	return nil
+}
diff --git a/cmd/demo/reload_test.go b/cmd/demo/reload_test.go
new file mode 100644
index 00000000..35ceea04
--- /dev/null
+++ b/cmd/demo/reload_test.go
@@ -0,0 +1,51 @@
+package demo
+
+import (
+	"context"
+	"testing"
+
+	cmdCore "github.com/flyteorg/flytectl/cmd/core"
+	"github.com/flyteorg/flytectl/pkg/k8s"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	testclient "k8s.io/client-go/kubernetes/fake"
+)
+
+var fakePod = corev1.Pod{
+	Status: corev1.PodStatus{
+		Phase:      corev1.PodRunning,
+		Conditions: []corev1.PodCondition{},
+	},
+	ObjectMeta: metav1.ObjectMeta{
+		Name:   "dummyflytepod",
+		Labels: map[string]string{"app": "flyte"},
+	},
+}
+
+func TestDemoReload(t *testing.T) {
+	ctx := context.Background()
+	commandCtx := cmdCore.CommandContext{}
+
+	t.Run("No errors", func(t *testing.T) {
+		client := testclient.NewSimpleClientset()
+		_, err := client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
+		assert.NoError(t, err)
+		k8s.Client = client
+		err = reloadDemoCluster(ctx, []string{}, commandCtx)
+		assert.NoError(t, err)
+	})
+
+	t.Run("Multiple pods will error", func(t *testing.T) {
+		client := testclient.NewSimpleClientset()
+		_, err := client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
+		assert.NoError(t, err)
+		fakePod.SetName("othername")
+		_, err = client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
+		assert.NoError(t, err)
+		k8s.Client = client
+		err = reloadDemoCluster(ctx, []string{}, commandCtx)
+		assert.Errorf(t, err, "should only have one pod")
+	})
+}
diff --git a/cmd/demo/start.go b/cmd/demo/start.go
index a800ed17..c542d615 100644
--- a/cmd/demo/start.go
+++ b/cmd/demo/start.go
@@ -3,6 +3,8 @@ package demo
 import (
 	"context"
 
+	"github.com/flyteorg/flytectl/pkg/docker"
+
 	"github.com/flyteorg/flytectl/pkg/sandbox"
 
 	sandboxCmdConfig "github.com/flyteorg/flytectl/cmd/config/subcommand/sandbox"
@@ -75,18 +77,21 @@ eg : for passing multiple environment variables
  flytectl demo start --env USER=foo --env PASSWORD=bar
 
 
-For just printing the docker commands for bringingup the demo container
+For just printing the docker commands for bringing up the demo container
 ::
 
  flytectl demo start --dryRun
 
-
 Usage
 `
 )
 
 func startDemoCluster(ctx context.Context, args []string, cmdCtx cmdCore.CommandContext) error {
-	sandboxDefaultConfig := sandboxCmdConfig.DefaultConfig
-	return sandbox.StartDemoCluster(ctx, args, sandboxDefaultConfig)
+	cfg := sandboxCmdConfig.DefaultConfig
+	err := cfg.ImagePullPolicy.Set(docker.ImagePullPolicyIfNotPresent.String())
+	if err != nil {
+		return err
+	}
 
+	return sandbox.StartDemoCluster(ctx, args, cfg)
 }
diff --git a/cmd/register/files.go b/cmd/register/files.go
index f87f353e..e8667431 100644
--- a/cmd/register/files.go
+++ b/cmd/register/files.go
@@ -151,7 +151,7 @@ func Register(ctx context.Context, args []string, cfg *config.Config, cmdCtx cmd
 			return fmt.Errorf("failed to upload source code from [%v]. Error: %w", sourceCodePath, err)
 		}
 
-		logger.Infof(ctx, "Source code successfully uploaded to [%v]", uploadLocation)
+		logger.Infof(ctx, "DeprecatedSource code successfully uploaded to [%v]", uploadLocation)
 	}
 
 	var registerResults []Result
diff --git a/go.mod b/go.mod
index cc7402ff..b1f6e1bc 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,6 @@ module github.com/flyteorg/flytectl
 go 1.18
 
 require (
-	github.com/avast/retry-go v3.0.0+incompatible
 	github.com/awalterschulze/gographviz v2.0.3+incompatible
 	github.com/disiqueira/gotree v1.0.0
 	github.com/docker/docker v20.10.7+incompatible
@@ -44,6 +43,7 @@ require (
 )
 
 require (
+	github.com/avast/retry-go v3.0.0+incompatible
 	github.com/flyteorg/flytepropeller v1.1.1
 	golang.org/x/text v0.3.7
 )
diff --git a/pkg/docker/docker.go b/pkg/docker/docker.go
index 84f9fb33..46ea3ea1 100644
--- a/pkg/docker/docker.go
+++ b/pkg/docker/docker.go
@@ -25,6 +25,8 @@ type Docker interface {
 	ContainerExecAttach(ctx context.Context, execID string, config types.ExecStartCheck) (types.HijackedResponse, error)
 	ContainerExecInspect(ctx context.Context, execID string) (types.ContainerExecInspect, error)
 	ImageList(ctx context.Context, listOption types.ImageListOptions) ([]types.ImageSummary, error)
+	ContainerStatPath(ctx context.Context, containerID, path string) (types.ContainerPathStat, error)
+	CopyFromContainer(ctx context.Context, containerID, srcPath string) (io.ReadCloser, types.ContainerPathStat, error)
 }
 
 type FlyteDocker struct {
diff --git a/pkg/docker/docker_util.go b/pkg/docker/docker_util.go
index bca12b19..da7ce9bc 100644
--- a/pkg/docker/docker_util.go
+++ b/pkg/docker/docker_util.go
@@ -24,11 +24,14 @@ import (
 )
 
 var (
-	Kubeconfig              = f.FilePathJoin(f.UserHomeDir(), ".flyte", "k3s", "k3s.yaml")
+	FlyteStateDir           = f.FilePathJoin(f.UserHomeDir(), ".flyte", "state")
+	Kubeconfig              = f.FilePathJoin(FlyteStateDir, "kubeconfig")
+	SandboxKubeconfig       = f.FilePathJoin(f.UserHomeDir(), ".flyte", "k3s", "k3s.yaml")
 	SuccessMessage          = "Deploying Flyte..."
 	FlyteSandboxClusterName = "flyte-sandbox"
 	Environment             = []string{"SANDBOX=1", "KUBERNETES_API_PORT=30086", "FLYTE_HOST=localhost:30081", "FLYTE_AWS_ENDPOINT=http://localhost:30084"}
 	Source                  = "/root"
+	StateDirMountDest       = "/srv/flyte"
 	K3sDir                  = "/etc/rancher/"
 	Client                  Docker
 	Volumes                 = []mount.Mount{
@@ -126,20 +129,18 @@ func GetSandboxPorts() (map[nat.Port]struct{}, map[nat.Port][]nat.PortBinding, e
 // GetDemoPorts will return demo ports
 func GetDemoPorts() (map[nat.Port]struct{}, map[nat.Port][]nat.PortBinding, error) {
 	return nat.ParsePortSpecs([]string{
-		"0.0.0.0:30080:30080", // Flyteconsole Port
-		"0.0.0.0:30081:30081", // Flyteadmin Port
-		"0.0.0.0:30082:30082", // K8s Dashboard Port
-		"0.0.0.0:30084:30084", // Minio API Port
-		"0.0.0.0:30086:30086", // K8s cluster
-		"0.0.0.0:30088:30088", // Minio Console Port
-		"0.0.0.0:30089:30089", // Postgres Port
-		"0.0.0.0:30090:30090", // Webhook service
+		"0.0.0.0:6443:6443",   // K3s API Port
+		"0.0.0.0:30080:30080", // HTTP Port
+		"0.0.0.0:30000:30000", // Registry Port
+		"0.0.0.0:30001:30001", // Postgres Port
+		"0.0.0.0:30002:30002", // Minio API Port (use HTTP port for minio console)
 	})
 }
 
 // PullDockerImage will Pull docker image
 func PullDockerImage(ctx context.Context, cli Docker, image string, pullPolicy ImagePullPolicy,
 	imagePullOptions ImagePullOptions, dryRun bool) error {
+
 	if dryRun {
 		PrintPullImage(image, imagePullOptions)
 		return nil
@@ -222,6 +223,7 @@ func PrintCreateContainer(volumes []mount.Mount, portBindings map[nat.Port][]nat
 // StartContainer will create and start docker container
 func StartContainer(ctx context.Context, cli Docker, volumes []mount.Mount, exposedPorts map[nat.Port]struct{},
 	portBindings map[nat.Port][]nat.PortBinding, name, image string, additionalEnvVars []string, dryRun bool) (string, error) {
+
 	// Append the additional env variables to the default list of env
 	Environment = append(Environment, additionalEnvVars...)
 	if dryRun {
@@ -252,6 +254,45 @@ func StartContainer(ctx context.Context, cli Docker, volumes []mount.Mount, expo
 	return resp.ID, nil
 }
 
+// CopyContainerFile try to create the container, see if the source file is there, copy it to the destination
+func CopyContainerFile(ctx context.Context, cli Docker, source, destination, name, image string) error {
+	resp, err := cli.ContainerCreate(ctx, &container.Config{Image: image}, &container.HostConfig{}, nil, nil, name)
+	if err != nil {
+		return err
+	}
+	var removeErr error
+	defer func() {
+		removeErr = cli.ContainerRemove(context.Background(), resp.ID, types.ContainerRemoveOptions{
+			Force: true,
+		})
+	}()
+	_, err = cli.ContainerStatPath(ctx, resp.ID, source)
+	if err != nil {
+		return err
+	}
+	reader, _, err := cli.CopyFromContainer(ctx, resp.ID, source)
+	if err != nil {
+		return err
+	}
+	tarFile := destination + ".tar"
+	outFile, err := os.Create(tarFile)
+	if err != nil {
+		return err
+	}
+	defer outFile.Close()
+	defer reader.Close()
+	_, err = io.Copy(outFile, reader)
+	if err != nil {
+		return err
+	}
+	r, _ := os.Open(tarFile)
+	err = f.ExtractTar(r, destination)
+	if err != nil {
+		return err
+	}
+	return removeErr
+}
+
 // ReadLogs will return io scanner for reading the logs of a container
 func ReadLogs(ctx context.Context, cli Docker, id string) (*bufio.Scanner, error) {
 	reader, err := cli.ContainerLogs(ctx, id, types.ContainerLogsOptions{
diff --git a/pkg/docker/docker_util_test.go b/pkg/docker/docker_util_test.go
index de1d6b09..79e7dae7 100644
--- a/pkg/docker/docker_util_test.go
+++ b/pkg/docker/docker_util_test.go
@@ -1,17 +1,21 @@
 package docker
 
 import (
+	"archive/tar"
 	"bufio"
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	f "github.com/flyteorg/flytectl/pkg/filesystemutils"
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/flyteorg/flytectl/pkg/docker/mocks"
+
 	"github.com/stretchr/testify/mock"
 
 	"github.com/docker/docker/api/types"
@@ -370,3 +374,78 @@ func TestInspectExecResp(t *testing.T) {
 	})
 
 }
+
+func TestDemoPorts(t *testing.T) {
+	_, ports, _ := GetDemoPorts()
+	assert.Equal(t, 5, len(ports))
+}
+
+func TestCopyFile(t *testing.T) {
+	ctx := context.Background()
+	// Create a fake tar file in tmp.
+	fo, err := os.CreateTemp("", "sampledata")
+	assert.NoError(t, err)
+	tarWriter := tar.NewWriter(fo)
+	err = tarWriter.WriteHeader(&tar.Header{
+		Typeflag: tar.TypeReg,
+		Name:     "flyte.yaml",
+		Size:     4,
+		Mode:     0640,
+		ModTime:  time.Unix(1245206587, 0),
+	})
+	assert.NoError(t, err)
+	cnt, err := tarWriter.Write([]byte("a: b"))
+	assert.NoError(t, err)
+	assert.Equal(t, 4, cnt)
+	tarWriter.Close()
+	fo.Close()
+
+	image := "some:image"
+	containerName := "my-container"
+
+	t.Run("No errors", func(t *testing.T) {
+		// Create reader of the tar file
+		reader, err := os.Open(fo.Name())
+		assert.NoError(t, err)
+		// Create destination file name
+		destDir, err := os.MkdirTemp("", "dest")
+		assert.NoError(t, err)
+		destination := filepath.Join(destDir, "destfile")
+
+		// Mocks
+		mockDocker := &mocks.Docker{}
+		mockDocker.OnContainerCreate(
+			ctx, &container.Config{Image: image}, &container.HostConfig{}, nil, nil, containerName).Return(
+			container.ContainerCreateCreatedBody{ID: containerName}, nil)
+		mockDocker.OnContainerStatPath(ctx, containerName, "some source").Return(types.ContainerPathStat{}, nil)
+		mockDocker.OnCopyFromContainer(ctx, containerName, "some source").Return(reader, types.ContainerPathStat{}, nil)
+		mockDocker.OnContainerRemove(ctx, containerName, types.ContainerRemoveOptions{Force: true}).Return(nil)
+		assert.Nil(t, err)
+
+		// Run
+		err = CopyContainerFile(ctx, mockDocker, "some source", destination, containerName, image)
+		assert.NoError(t, err)
+
+		// Read the file and make sure it's correct
+		strBytes, err := os.ReadFile(destination)
+		assert.NoError(t, err)
+		assert.Equal(t, "a: b", string(strBytes))
+	})
+
+	t.Run("Erroring on stat", func(t *testing.T) {
+		myErr := fmt.Errorf("erroring on stat")
+
+		// Mocks
+		mockDocker := &mocks.Docker{}
+		mockDocker.OnContainerCreate(
+			ctx, &container.Config{Image: image}, &container.HostConfig{}, nil, nil, containerName).Return(
+			container.ContainerCreateCreatedBody{ID: containerName}, nil)
+		mockDocker.OnContainerStatPath(ctx, containerName, "some source").Return(types.ContainerPathStat{}, myErr)
+		mockDocker.OnContainerRemove(ctx, containerName, types.ContainerRemoveOptions{Force: true}).Return(nil)
+		assert.Nil(t, err)
+
+		// Run
+		err = CopyContainerFile(ctx, mockDocker, "some source", "", containerName, image)
+		assert.Equal(t, myErr, err)
+	})
+}
diff --git a/pkg/docker/mocks/docker.go b/pkg/docker/mocks/docker.go
index b1f234f9..a2ddc27c 100644
--- a/pkg/docker/mocks/docker.go
+++ b/pkg/docker/mocks/docker.go
@@ -325,6 +325,45 @@ func (_m *Docker) ContainerStart(ctx context.Context, containerID string, option
 	return r0
 }
 
+type Docker_ContainerStatPath struct {
+	*mock.Call
+}
+
+func (_m Docker_ContainerStatPath) Return(_a0 types.ContainerPathStat, _a1 error) *Docker_ContainerStatPath {
+	return &Docker_ContainerStatPath{Call: _m.Call.Return(_a0, _a1)}
+}
+
+func (_m *Docker) OnContainerStatPath(ctx context.Context, containerID string, path string) *Docker_ContainerStatPath {
+	c_call := _m.On("ContainerStatPath", ctx, containerID, path)
+	return &Docker_ContainerStatPath{Call: c_call}
+}
+
+func (_m *Docker) OnContainerStatPathMatch(matchers ...interface{}) *Docker_ContainerStatPath {
+	c_call := _m.On("ContainerStatPath", matchers...)
+	return &Docker_ContainerStatPath{Call: c_call}
+}
+
+// ContainerStatPath provides a mock function with given fields: ctx, containerID, path
+func (_m *Docker) ContainerStatPath(ctx context.Context, containerID string, path string) (types.ContainerPathStat, error) {
+	ret := _m.Called(ctx, containerID, path)
+
+	var r0 types.ContainerPathStat
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) types.ContainerPathStat); ok {
+		r0 = rf(ctx, containerID, path)
+	} else {
+		r0 = ret.Get(0).(types.ContainerPathStat)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
+		r1 = rf(ctx, containerID, path)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 type Docker_ContainerWait struct {
 	*mock.Call
 }
@@ -368,6 +407,54 @@ func (_m *Docker) ContainerWait(ctx context.Context, containerID string, conditi
 	return r0, r1
 }
 
+type Docker_CopyFromContainer struct {
+	*mock.Call
+}
+
+func (_m Docker_CopyFromContainer) Return(_a0 io.ReadCloser, _a1 types.ContainerPathStat, _a2 error) *Docker_CopyFromContainer {
+	return &Docker_CopyFromContainer{Call: _m.Call.Return(_a0, _a1, _a2)}
+}
+
+func (_m *Docker) OnCopyFromContainer(ctx context.Context, containerID string, srcPath string) *Docker_CopyFromContainer {
+	c_call := _m.On("CopyFromContainer", ctx, containerID, srcPath)
+	return &Docker_CopyFromContainer{Call: c_call}
+}
+
+func (_m *Docker) OnCopyFromContainerMatch(matchers ...interface{}) *Docker_CopyFromContainer {
+	c_call := _m.On("CopyFromContainer", matchers...)
+	return &Docker_CopyFromContainer{Call: c_call}
+}
+
+// CopyFromContainer provides a mock function with given fields: ctx, containerID, srcPath
+func (_m *Docker) CopyFromContainer(ctx context.Context, containerID string, srcPath string) (io.ReadCloser, types.ContainerPathStat, error) {
+	ret := _m.Called(ctx, containerID, srcPath)
+
+	var r0 io.ReadCloser
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) io.ReadCloser); ok {
+		r0 = rf(ctx, containerID, srcPath)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(io.ReadCloser)
+		}
+	}
+
+	var r1 types.ContainerPathStat
+	if rf, ok := ret.Get(1).(func(context.Context, string, string) types.ContainerPathStat); ok {
+		r1 = rf(ctx, containerID, srcPath)
+	} else {
+		r1 = ret.Get(1).(types.ContainerPathStat)
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(context.Context, string, string) error); ok {
+		r2 = rf(ctx, containerID, srcPath)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
 type Docker_ImageList struct {
 	*mock.Call
 }
diff --git a/pkg/filesystemutils/file_system_utils.go b/pkg/filesystemutils/file_system_utils.go
index 2f0b756c..77ac4d8e 100644
--- a/pkg/filesystemutils/file_system_utils.go
+++ b/pkg/filesystemutils/file_system_utils.go
@@ -1,6 +1,9 @@
 package filesystemutils
 
 import (
+	"archive/tar"
+	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 )
@@ -20,3 +23,48 @@ func UserHomeDir() string {
 func FilePathJoin(elems ...string) string {
 	return filePathJoinFunc(elems...)
 }
+
+func ExtractTar(ss io.Reader, destination string) error {
+	tarReader := tar.NewReader(ss)
+
+	for {
+		header, err := tarReader.Next()
+
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			return err
+		}
+
+		switch header.Typeflag {
+		case tar.TypeDir:
+			if err := os.Mkdir(header.Name, 0755); err != nil {
+				return err
+			}
+		case tar.TypeReg:
+			fmt.Printf("Creating Flyte configuration file at: %s\n", destination)
+			outFile, err := os.Create(destination)
+			if err != nil {
+				return err
+			}
+			for {
+				// Read one 1MB at a time.
+				if _, err := io.CopyN(outFile, tarReader, 1024*1024); err != nil {
+					if err == io.EOF {
+						break
+					}
+					return err
+				}
+			}
+			outFile.Close()
+
+		default:
+			return fmt.Errorf("ExtractTarGz: unknown type: %v in %s",
+				header.Typeflag,
+				header.Name)
+		}
+	}
+	return nil
+}
diff --git a/pkg/filesystemutils/file_system_utils_test.go b/pkg/filesystemutils/file_system_utils_test.go
new file mode 100644
index 00000000..2bca38d8
--- /dev/null
+++ b/pkg/filesystemutils/file_system_utils_test.go
@@ -0,0 +1,112 @@
+package filesystemutils
+
+import (
+	"archive/tar"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	homeDirVal = "/home/user"
+	homeDirErr error
+)
+
+func FakeUserHomeDir() (string, error) {
+	return homeDirVal, homeDirErr
+}
+
+func TestUserHomeDir(t *testing.T) {
+	t.Run("User home dir", func(t *testing.T) {
+		osUserHomDirFunc = FakeUserHomeDir
+		homeDir := UserHomeDir()
+		assert.Equal(t, homeDirVal, homeDir)
+	})
+	t.Run("User home dir fail", func(t *testing.T) {
+		homeDirErr = fmt.Errorf("failed to get users home directory")
+		homeDirVal = "."
+		osUserHomDirFunc = FakeUserHomeDir
+		homeDir := UserHomeDir()
+		assert.Equal(t, ".", homeDir)
+		// Reset
+		homeDirErr = nil
+		homeDirVal = "/home/user"
+	})
+}
+
+func TestFilePathJoin(t *testing.T) {
+	t.Run("File path join", func(t *testing.T) {
+		homeDir := FilePathJoin("/", "home", "user")
+		assert.Equal(t, "/home/user", homeDir)
+	})
+}
+
+func TestTaring(t *testing.T) {
+	// Create a fake tar file in tmp.
+	text := "a: b"
+	fo, err := os.CreateTemp("", "sampledata")
+	assert.NoError(t, err)
+	tarWriter := tar.NewWriter(fo)
+	err = tarWriter.WriteHeader(&tar.Header{
+		Typeflag: tar.TypeReg,
+		Name:     "flyte.yaml",
+		Size:     4,
+		Mode:     0640,
+		ModTime:  time.Unix(1245206587, 0),
+	})
+	assert.NoError(t, err)
+	cnt, err := tarWriter.Write([]byte(text))
+	assert.NoError(t, err)
+	assert.Equal(t, 4, cnt)
+	tarWriter.Close()
+	fo.Close()
+
+	t.Run("Basic testing", func(t *testing.T) {
+		destFile, err := os.CreateTemp("", "sampledata")
+		assert.NoError(t, err)
+		reader, err := os.Open(fo.Name())
+		assert.NoError(t, err)
+		err = ExtractTar(reader, destFile.Name())
+		assert.NoError(t, err)
+		fileBytes, err := os.ReadFile(destFile.Name())
+		assert.NoError(t, err)
+		readString := string(fileBytes)
+		assert.Equal(t, text, readString)
+
+		// Try to extract the file we just extracted again. It's not a tar file obviously so it should error
+		reader, err = os.Open(destFile.Name())
+		assert.NoError(t, err)
+		err = ExtractTar(reader, destFile.Name())
+		assert.Errorf(t, err, "unexpected EOF")
+	})
+}
+
+func TestTarBadHeader(t *testing.T) {
+	// Create a fake tar file in tmp.
+	fo, err := os.CreateTemp("", "sampledata")
+	assert.NoError(t, err)
+	tarWriter := tar.NewWriter(fo)
+	// Write a symlink, we should not know how to parse.
+	err = tarWriter.WriteHeader(&tar.Header{
+		Typeflag: tar.TypeLink,
+		Name:     "flyte.yaml",
+		Size:     4,
+		Mode:     0640,
+		ModTime:  time.Unix(1245206587, 0),
+	})
+	assert.NoError(t, err)
+	tarWriter.Close()
+	fo.Close()
+
+	t.Run("Basic testing", func(t *testing.T) {
+		destFile, err := os.CreateTemp("", "sampledata")
+		assert.NoError(t, err)
+		reader, err := os.Open(fo.Name())
+		assert.NoError(t, err)
+		err = ExtractTar(reader, destFile.Name())
+		assert.Errorf(t, err, "ExtractTarGz: unknown type")
+	})
+}
diff --git a/pkg/filesystemutils/flile_system_utils_test.go b/pkg/filesystemutils/flile_system_utils_test.go
deleted file mode 100644
index 9698d170..00000000
--- a/pkg/filesystemutils/flile_system_utils_test.go
+++ /dev/null
@@ -1,42 +0,0 @@
-package filesystemutils
-
-import (
-	"fmt"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-var (
-	homeDirVal = "/home/user"
-	homeDirErr error
-)
-
-func FakeUserHomeDir() (string, error) {
-	return homeDirVal, homeDirErr
-}
-
-func TestUserHomeDir(t *testing.T) {
-	t.Run("User home dir", func(t *testing.T) {
-		osUserHomDirFunc = FakeUserHomeDir
-		homeDir := UserHomeDir()
-		assert.Equal(t, homeDirVal, homeDir)
-	})
-	t.Run("User home dir fail", func(t *testing.T) {
-		homeDirErr = fmt.Errorf("failed to get users home directory")
-		homeDirVal = "."
-		osUserHomDirFunc = FakeUserHomeDir
-		homeDir := UserHomeDir()
-		assert.Equal(t, ".", homeDir)
-		// Reset
-		homeDirErr = nil
-		homeDirVal = "/home/user"
-	})
-}
-
-func TestFilePathJoin(t *testing.T) {
-	t.Run("File path join", func(t *testing.T) {
-		homeDir := FilePathJoin("/", "home", "user")
-		assert.Equal(t, "/home/user", homeDir)
-	})
-}
diff --git a/pkg/k8s/k8s.go b/pkg/k8s/k8s.go
index 1a34a93b..705c6887 100644
--- a/pkg/k8s/k8s.go
+++ b/pkg/k8s/k8s.go
@@ -63,7 +63,7 @@ func (k *ContextManager) CheckConfig() error {
 	return err
 }
 
-// CopyKubeContext copies context srcCtxName part of srcConfigAccess to targetCtxName part of targetConfigAccess.
+// CopyContext copies context srcCtxName part of srcConfigAccess to targetCtxName part of targetConfigAccess.
 func (k *ContextManager) CopyContext(srcConfigAccess clientcmd.ConfigAccess, srcCtxName, targetCtxName string) error {
 	err := k.CheckConfig()
 	if err != nil {
diff --git a/pkg/sandbox/start.go b/pkg/sandbox/start.go
index 3a53923c..f7ac5ed9 100644
--- a/pkg/sandbox/start.go
+++ b/pkg/sandbox/start.go
@@ -20,6 +20,7 @@ import (
 	"github.com/flyteorg/flytectl/pkg/github"
 	"github.com/flyteorg/flytectl/pkg/k8s"
 	"github.com/flyteorg/flytectl/pkg/util"
+	"github.com/flyteorg/flytestdlib/logger"
 	"github.com/kataras/tablewriter"
 	corev1api "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -33,9 +34,12 @@ const (
 	taintEffect          = "NoSchedule"
 	sandboxContextName   = "flyte-sandbox"
 	sandboxDockerContext = "default"
-	k8sEndpoint          = "https://127.0.0.1:30086"
+	K8sEndpoint          = "https://127.0.0.1:6443"
+	sandboxK8sEndpoint   = "https://127.0.0.1:30086"
 	sandboxImageName     = "cr.flyte.org/flyteorg/flyte-sandbox"
-	demoImageName        = "cr.flyte.org/flyteorg/flyte-sandbox-lite"
+	demoImageName        = "cr.flyte.org/flyteorg/flyte-sandbox-bundled"
+	DefaultFlyteConfig   = "/opt/flyte/defaults.flyte.yaml"
+	k3sKubeConfigEnvVar  = "K3S_KUBECONFIG_OUTPUT=/srv/flyte/kubeconfig"
 )
 
 func isNodeTainted(ctx context.Context, client corev1.CoreV1Interface) (bool, error) {
@@ -140,9 +144,9 @@ func MountVolume(file, destination string) (*mount.Mount, error) {
 	return nil, nil
 }
 
-func UpdateLocalKubeContext(k8sCtxMgr k8s.ContextOps, dockerCtx string, contextName string) error {
+func UpdateLocalKubeContext(k8sCtxMgr k8s.ContextOps, dockerCtx string, contextName string, kubeConfigPath string) error {
 	srcConfigAccess := &clientcmd.PathOptions{
-		GlobalFile:   docker.Kubeconfig,
+		GlobalFile:   kubeConfigPath,
 		LoadingRules: clientcmd.NewDefaultClientConfigLoadingRules(),
 	}
 	return k8sCtxMgr.CopyContext(srcConfigAccess, dockerCtx, contextName)
@@ -158,16 +162,13 @@ func startSandbox(ctx context.Context, cli docker.Docker, g github.GHRepoService
 				return nil, err
 			}
 			fmt.Printf("Existing details of your sandbox")
-			util.PrintSandboxMessage(consolePort, sandboxConfig.DryRun)
+			util.PrintSandboxMessage(consolePort, docker.Kubeconfig, sandboxConfig.DryRun)
 			return nil, nil
 		}
 	}
-	if err := util.SetupFlyteDir(); err != nil {
-		return nil, err
-	}
 
 	templateValues := configutil.ConfigTemplateSpec{
-		Host:     "localhost:30081",
+		Host:     "localhost:30080",
 		Insecure: true,
 		Console:  fmt.Sprintf("http://localhost:%d", consolePort),
 	}
@@ -176,11 +177,25 @@ func startSandbox(ctx context.Context, cli docker.Docker, g github.GHRepoService
 	}
 
 	volumes := docker.Volumes
-	if vol, err := MountVolume(sandboxConfig.Source, docker.Source); err != nil {
+	// Mount this even though it should no longer be necessary. This is for user code
+	if vol, err := MountVolume(sandboxConfig.DeprecatedSource, docker.Source); err != nil {
 		return nil, err
 	} else if vol != nil {
 		volumes = append(volumes, *vol)
 	}
+
+	// This is the state directory mount, flyte will write the kubeconfig here. May hold more in future releases
+	// To be interoperable with the old sandbox, only mount if the directory exists, should've created by StartCluster
+	if fileInfo, err := os.Stat(docker.FlyteStateDir); err == nil {
+		if fileInfo.IsDir() {
+			if vol, err := MountVolume(docker.FlyteStateDir, docker.StateDirMountDest); err != nil {
+				return nil, err
+			} else if vol != nil {
+				volumes = append(volumes, *vol)
+			}
+		}
+	}
+
 	sandboxImage := sandboxConfig.Image
 	if len(sandboxImage) == 0 {
 		image, version, err := github.GetFullyQualifiedImageName(defaultImagePrefix, sandboxConfig.Version, defaultImageName, sandboxConfig.Prerelease, g)
@@ -253,6 +268,101 @@ func StartCluster(ctx context.Context, args []string, sandboxConfig *sandboxCmdC
 	}
 
 	ghRepo := github.GetGHRepoService()
+	if err := util.CreatePathAndFile(docker.Kubeconfig); err != nil {
+		return err
+	}
+
+	reader, err := startSandbox(ctx, cli, ghRepo, os.Stdin, sandboxConfig, defaultImageName, defaultImagePrefix, exposedPorts, portBindings, consolePort)
+	if err != nil {
+		return err
+	}
+
+	if reader != nil {
+		var k8sClient k8s.K8s
+		err = retry.Do(
+			func() error {
+				// This should wait for the kubeconfig file being there.
+				k8sClient, err = k8s.GetK8sClient(docker.Kubeconfig, K8sEndpoint)
+				return err
+			},
+			retry.Attempts(10),
+		)
+		if err != nil {
+			return err
+		}
+
+		// Live-ness check
+		err = retry.Do(
+			func() error {
+				// Have to get a new client every time because you run into x509 errors if not
+				fmt.Println("Waiting for cluster to come up...")
+				k8sClient, err = k8s.GetK8sClient(docker.Kubeconfig, K8sEndpoint)
+				if err != nil {
+					logger.Debugf(ctx, "Error getting K8s client in liveness check %s", err)
+					return err
+				}
+				req := k8sClient.CoreV1().RESTClient().Get()
+				req = req.RequestURI("livez")
+				res := req.Do(ctx)
+				return res.Error()
+			},
+			retry.Attempts(15),
+		)
+		if err != nil {
+			return err
+		}
+
+		// Readiness check
+		err = retry.Do(
+			func() error {
+				// No need to refresh client here
+				req := k8sClient.CoreV1().RESTClient().Get()
+				req = req.RequestURI("readyz")
+				res := req.Do(ctx)
+				return res.Error()
+			},
+			retry.Attempts(10),
+		)
+		if err != nil {
+			return err
+		}
+
+		// This will copy the kubeconfig from where k3s writes it () to the main file.
+		// This code is located after the waits above since it appears that k3s goes through at least a couple versions
+		// of the config keys/certs. If this copy is done too early, the copied credentials won't work.
+		if err = UpdateLocalKubeContext(k8sCtxMgr, sandboxDockerContext, sandboxContextName, docker.Kubeconfig); err != nil {
+			return err
+		}
+
+		// Watch for Flyte Deployment
+		if err := WatchFlyteDeployment(ctx, k8sClient.CoreV1()); err != nil {
+			return err
+		}
+		if primePod {
+			primeFlytekitPod(ctx, k8sClient.CoreV1().Pods("default"))
+		}
+	}
+	return nil
+}
+
+// StartClusterForSandbox is the code for the original multi deploy version of sandbox, should be removed once we
+// document the new development experience for plugins.
+func StartClusterForSandbox(ctx context.Context, args []string, sandboxConfig *sandboxCmdConfig.Config, primePod bool, defaultImageName string, defaultImagePrefix string, exposedPorts map[nat.Port]struct{}, portBindings map[nat.Port][]nat.PortBinding, consolePort int) error {
+	k8sCtxMgr := k8s.NewK8sContextManager()
+	err := k8sCtxMgr.CheckConfig()
+	if err != nil {
+		return err
+	}
+	cli, err := docker.GetDockerClient()
+	if err != nil {
+		return err
+	}
+
+	ghRepo := github.GetGHRepoService()
+
+	if err := util.CreatePathAndFile(docker.SandboxKubeconfig); err != nil {
+		return err
+	}
 
 	reader, err := startSandbox(ctx, cli, ghRepo, os.Stdin, sandboxConfig, defaultImageName, defaultImagePrefix, exposedPorts, portBindings, consolePort)
 	if err != nil {
@@ -266,7 +376,7 @@ func StartCluster(ctx context.Context, args []string, sandboxConfig *sandboxCmdC
 		var k8sClient k8s.K8s
 		err = retry.Do(
 			func() error {
-				k8sClient, err = k8s.GetK8sClient(docker.Kubeconfig, k8sEndpoint)
+				k8sClient, err = k8s.GetK8sClient(docker.SandboxKubeconfig, sandboxK8sEndpoint)
 				return err
 			},
 			retry.Attempts(10),
@@ -274,10 +384,11 @@ func StartCluster(ctx context.Context, args []string, sandboxConfig *sandboxCmdC
 		if err != nil {
 			return err
 		}
-		if err = UpdateLocalKubeContext(k8sCtxMgr, sandboxDockerContext, sandboxContextName); err != nil {
+		if err = UpdateLocalKubeContext(k8sCtxMgr, sandboxDockerContext, sandboxContextName, docker.SandboxKubeconfig); err != nil {
 			return err
 		}
 
+		// TODO: This doesn't appear to correctly watch for the Flyte deployment but doesn't do so on master either.
 		if err := WatchFlyteDeployment(ctx, k8sClient.CoreV1()); err != nil {
 			return err
 		}
@@ -293,17 +404,16 @@ func StartDemoCluster(ctx context.Context, args []string, sandboxConfig *sandbox
 	primePod := true
 	sandboxImagePrefix := "sha"
 	exposedPorts, portBindings, err := docker.GetDemoPorts()
-	if sandboxConfig.Dev {
-		exposedPorts, portBindings, err = docker.GetDevPorts()
-	}
 	if err != nil {
 		return err
 	}
+	// K3s will automatically write the file specified by this var, which is mounted from user's local state dir.
+	sandboxConfig.Env = append(sandboxConfig.Env, k3sKubeConfigEnvVar)
 	err = StartCluster(ctx, args, sandboxConfig, primePod, demoImageName, sandboxImagePrefix, exposedPorts, portBindings, util.DemoConsolePort)
 	if err != nil {
 		return err
 	}
-	util.PrintSandboxMessage(util.DemoConsolePort, sandboxConfig.DryRun)
+	util.PrintDemoMessage(util.DemoConsolePort, docker.Kubeconfig, sandboxConfig.DryRun)
 	return nil
 }
 
@@ -314,10 +424,10 @@ func StartSandboxCluster(ctx context.Context, args []string, sandboxConfig *sand
 	if err != nil {
 		return err
 	}
-	err = StartCluster(ctx, args, sandboxConfig, primePod, sandboxImageName, demoImagePrefix, exposedPorts, portBindings, util.SandBoxConsolePort)
+	err = StartClusterForSandbox(ctx, args, sandboxConfig, primePod, sandboxImageName, demoImagePrefix, exposedPorts, portBindings, util.SandBoxConsolePort)
 	if err != nil {
 		return err
 	}
-	util.PrintSandboxMessage(util.SandBoxConsolePort, sandboxConfig.DryRun)
+	util.PrintSandboxMessage(util.SandBoxConsolePort, docker.SandboxKubeconfig, sandboxConfig.DryRun)
 	return nil
 }
diff --git a/pkg/sandbox/start_test.go b/pkg/sandbox/start_test.go
index c3b22c6b..7bc204ac 100644
--- a/pkg/sandbox/start_test.go
+++ b/pkg/sandbox/start_test.go
@@ -103,7 +103,7 @@ func TestStartFunc(t *testing.T) {
 		Platform:     "",
 	}
 	assert.Nil(t, util.SetupFlyteDir())
-	assert.Nil(t, os.MkdirAll(f.FilePathJoin(f.UserHomeDir(), ".flyte", "k3s"), os.ModePerm))
+	assert.Nil(t, os.MkdirAll(f.FilePathJoin(f.UserHomeDir(), ".flyte", "state"), os.ModePerm))
 	assert.Nil(t, ioutil.WriteFile(docker.Kubeconfig, []byte(content), os.ModePerm))
 
 	fakePod.SetName("flyte")
@@ -146,7 +146,7 @@ func TestStartFunc(t *testing.T) {
 		assert.Nil(t, reader)
 	})
 	t.Run("Successfully run demo cluster with source code", func(t *testing.T) {
-		sandboxCmdConfig.DefaultConfig.Source = f.UserHomeDir()
+		sandboxCmdConfig.DefaultConfig.DeprecatedSource = f.UserHomeDir()
 		sandboxCmdConfig.DefaultConfig.Version = ""
 		sandboxSetup()
 		mockDocker.OnContainerList(ctx, types.ContainerListOptions{All: true}).Return([]types.Container{}, nil)
@@ -162,7 +162,7 @@ func TestStartFunc(t *testing.T) {
 		assert.Nil(t, err)
 	})
 	t.Run("Successfully run demo cluster with abs path of source code", func(t *testing.T) {
-		sandboxCmdConfig.DefaultConfig.Source = "../"
+		sandboxCmdConfig.DefaultConfig.DeprecatedSource = "../"
 		sandboxCmdConfig.DefaultConfig.Version = ""
 		sandboxSetup()
 		mockDocker.OnContainerList(ctx, types.ContainerListOptions{All: true}).Return([]types.Container{}, nil)
@@ -279,7 +279,7 @@ func TestStartFunc(t *testing.T) {
 	})
 	t.Run("Successfully run demo cluster command", func(t *testing.T) {
 		//	mockOutStream := new(io.Writer)
-		//cmdCtx := cmdCore.NewCommandContext(admin.InitializeMockClientset(), *mockOutStream)
+		// cmdCtx := cmdCore.NewCommandContext(admin.InitializeMockClientset(), *mockOutStream)
 		client := testclient.NewSimpleClientset()
 		k8s.Client = client
 		_, err := client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
@@ -306,7 +306,7 @@ func TestStartFunc(t *testing.T) {
 		}).Return(reader, nil)
 		mockK8sContextMgr := &k8sMocks.ContextOps{}
 		docker.Client = mockDocker
-		sandboxCmdConfig.DefaultConfig.Source = ""
+		sandboxCmdConfig.DefaultConfig.DeprecatedSource = ""
 		sandboxCmdConfig.DefaultConfig.Version = ""
 		k8s.ContextMgr = mockK8sContextMgr
 		ghutil.Client = githubMock
@@ -314,12 +314,10 @@ func TestStartFunc(t *testing.T) {
 		mockK8sContextMgr.OnCopyContextMatch(mock.Anything, mock.Anything, mock.Anything).Return(nil)
 		err = StartSandboxCluster(context.Background(), []string{}, config)
 		assert.Nil(t, err)
-		err = StartDemoCluster(context.Background(), []string{}, config)
-		assert.Nil(t, err)
 	})
 	t.Run("Error in running demo cluster command", func(t *testing.T) {
-		//mockOutStream := new(io.Writer)
-		//cmdCtx := cmdCore.NewCommandContext(admin.InitializeMockClientset(), *mockOutStream)
+		// mockOutStream := new(io.Writer)
+		// cmdCtx := cmdCore.NewCommandContext(admin.InitializeMockClientset(), *mockOutStream)
 		sandboxSetup()
 		docker.Client = mockDocker
 		mockDocker.OnContainerListMatch(mock.Anything, mock.Anything).Return([]types.Container{}, fmt.Errorf("failed to list containers"))
@@ -362,7 +360,6 @@ func TestMonitorFlyteDeployment(t *testing.T) {
 
 		err = WatchFlyteDeployment(ctx, client.CoreV1())
 		assert.NotNil(t, err)
-
 	})
 
 	t.Run("Monitor k8s deployment success", func(t *testing.T) {
@@ -385,13 +382,10 @@ func TestMonitorFlyteDeployment(t *testing.T) {
 
 		err = WatchFlyteDeployment(ctx, client.CoreV1())
 		assert.Nil(t, err)
-
 	})
-
 }
 
 func TestGetFlyteDeploymentCount(t *testing.T) {
-
 	ctx := context.Background()
 	client := testclient.NewSimpleClientset()
 	c, err := getFlyteDeployment(ctx, client.CoreV1())
diff --git a/pkg/util/util.go b/pkg/util/util.go
index afdb3596..ece88227 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -6,6 +6,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"os"
+	"path/filepath"
 	"strings"
 
 	"github.com/flyteorg/flytectl/pkg/configutil"
@@ -34,16 +35,36 @@ func WriteIntoFile(data []byte, file string) error {
 	return nil
 }
 
+func CreatePathAndFile(pathToConfig string) error {
+	p, err := filepath.Abs(pathToConfig)
+	if err != nil {
+		return err
+	}
+	if err := os.MkdirAll(filepath.Dir(p), os.ModePerm); err != nil {
+		return err
+	}
+
+	// Created a empty file with right permission
+	if _, err := os.Stat(p); err != nil {
+		if os.IsNotExist(err) {
+			if err := os.WriteFile(p, []byte(""), os.ModePerm); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // SetupFlyteDir will create .flyte dir if not exist
 func SetupFlyteDir() error {
-	if err := os.MkdirAll(f.FilePathJoin(f.UserHomeDir(), ".flyte", "k3s"), os.ModePerm); err != nil {
+	if err := os.MkdirAll(f.FilePathJoin(f.UserHomeDir(), ".flyte", "state"), os.ModePerm); err != nil {
 		return err
 	}
 
 	// Created a empty file with right permission
 	if _, err := os.Stat(docker.Kubeconfig); err != nil {
 		if os.IsNotExist(err) {
-			if err := ioutil.WriteFile(docker.Kubeconfig, []byte(""), os.ModePerm); err != nil {
+			if err := os.WriteFile(docker.Kubeconfig, []byte(""), os.ModePerm); err != nil {
 				return err
 			}
 		}
@@ -52,12 +73,36 @@ func SetupFlyteDir() error {
 	return nil
 }
 
+// PrintDemoMessage will print sandbox success message
+func PrintDemoMessage(flyteConsolePort int, kubeconfigLocation string, dryRun bool) {
+	kubeconfig := strings.Join([]string{
+		"$KUBECONFIG",
+		kubeconfigLocation,
+	}, ":")
+
+	var successMsg string
+	if dryRun {
+		successMsg = fmt.Sprintf("%v http://localhost:%v/console", ProgressSuccessMessagePending, flyteConsolePort)
+	} else {
+		successMsg = fmt.Sprintf("%v http://localhost:%v/console", ProgressSuccessMessage, flyteConsolePort)
+
+	}
+	fmt.Printf("%v %v %v %v %v \n", emoji.ManTechnologist, successMsg, emoji.Rocket, emoji.Rocket, emoji.PartyPopper)
+	fmt.Printf("%v Run the following command to export sandbox environment variables for accessing flytectl\n", emoji.Sparkle)
+	fmt.Printf("	export FLYTECTL_CONFIG=%v \n", configutil.FlytectlConfig)
+	if dryRun {
+		fmt.Printf("%v Run the following command to export kubeconfig variables for accessing flyte pods locally\n", emoji.Sparkle)
+		fmt.Printf("	export KUBECONFIG=%v \n", kubeconfig)
+	}
+	fmt.Printf("%s Flyte sandbox ships with a Docker registry. Tag and push custom workflow images to localhost:30000\n", emoji.Whale)
+	fmt.Printf("%s The Minio API is hosted on localhost:30002. Use http://localhost:30080/minio/login for Minio console\n", emoji.OpenFileFolder)
+}
+
 // PrintSandboxMessage will print sandbox success message
-func PrintSandboxMessage(flyteConsolePort int, dryRun bool) {
+func PrintSandboxMessage(flyteConsolePort int, kubeconfigLocation string, dryRun bool) {
 	kubeconfig := strings.Join([]string{
 		"$KUBECONFIG",
-		f.FilePathJoin(f.UserHomeDir(), ".kube", "config"),
-		docker.Kubeconfig,
+		kubeconfigLocation,
 	}, ":")
 
 	var successMsg string
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index 9bbb8996..492dc262 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -1,8 +1,12 @@
 package util
 
 import (
+	"os"
+	"path/filepath"
 	"testing"
 
+	"github.com/flyteorg/flytectl/pkg/docker"
+
 	"github.com/stretchr/testify/assert"
 )
 
@@ -25,7 +29,7 @@ func TestSetupFlyteDir(t *testing.T) {
 
 func TestPrintSandboxMessage(t *testing.T) {
 	t.Run("Print Sandbox Message", func(t *testing.T) {
-		PrintSandboxMessage(SandBoxConsolePort, false)
+		PrintSandboxMessage(SandBoxConsolePort, docker.SandboxKubeconfig, false)
 	})
 }
 
@@ -80,3 +84,15 @@ func TestIsVersionGreaterThan(t *testing.T) {
 		assert.NotNil(t, err)
 	})
 }
+
+func TestCreatePathAndFile(t *testing.T) {
+	dir, err := os.MkdirTemp("", "flytectl")
+	assert.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	testFile := filepath.Join(dir, "testfile.yaml")
+	err = CreatePathAndFile(testFile)
+	assert.NoError(t, err)
+	_, err = os.Stat(testFile)
+	assert.NoError(t, err)
+}