diff --git a/internal/flink/command.go b/internal/flink/command.go
index 23cf3b6d61..5038ec406c 100644
--- a/internal/flink/command.go
+++ b/internal/flink/command.go
@@ -46,6 +46,7 @@ func New(cfg *config.Config, prerunner pcmd.PreRunner) *cobra.Command {
 	cmd.AddCommand(c.newSystemInfoCommand())
 
 	// On-Prem and Cloud Commands
+	cmd.AddCommand(c.newArtifactCommand(cfg))
 	cmd.AddCommand(c.newComputePoolCommand(cfg))
 	if !cfg.IsOnPremLogin() {
 		cmd.AddCommand(c.newShellCommand(prerunner, cfg))
@@ -53,7 +54,6 @@ func New(cfg *config.Config, prerunner pcmd.PreRunner) *cobra.Command {
 	cmd.AddCommand(c.newStatementCommand(cfg))
 
 	// Cloud Specific Commands
-	cmd.AddCommand(c.newArtifactCommand())
 	cmd.AddCommand(c.newComputePoolConfigCommand())
 	cmd.AddCommand(c.newConnectionCommand())
 	cmd.AddCommand(c.newConnectivityTypeCommand())
diff --git a/internal/flink/command_artifact.go b/internal/flink/command_artifact.go
index 329568440f..c0726b35c4 100644
--- a/internal/flink/command_artifact.go
+++ b/internal/flink/command_artifact.go
@@ -6,6 +6,7 @@ import (
 	flinkartifactv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-artifact/v1"
 
 	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/config"
 	"github.com/confluentinc/cli/v4/pkg/output"
 )
 
@@ -21,17 +22,26 @@ type flinkArtifactOut struct {
 	DocumentationLink string `human:"Documentation Link" serialized:"documentation_link"`
 }
 
-func (c *command) newArtifactCommand() *cobra.Command {
+func (c *command) newArtifactCommand(cfg *config.Config) *cobra.Command {
 	cmd := &cobra.Command{
-		Use:         "artifact",
-		Short:       "Manage Flink UDF artifacts.",
-		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireNonAPIKeyCloudLogin},
+		Use:   "artifact",
+		Short: "Manage Flink UDF artifacts.",
 	}
 
-	cmd.AddCommand(c.newCreateCommand())
-	cmd.AddCommand(c.newDeleteCommand())
-	cmd.AddCommand(c.newDescribeCommand())
-	cmd.AddCommand(c.newListCommand())
+	if cfg.IsCloudLogin() {
+		cmd.Annotations = map[string]string{pcmd.RunRequirement: pcmd.RequireNonAPIKeyCloudLogin}
+		cmd.AddCommand(c.newCreateCommand())
+		cmd.AddCommand(c.newDeleteCommand())
+		cmd.AddCommand(c.newDescribeCommand())
+		cmd.AddCommand(c.newListCommand())
+	} else {
+		cmd.AddCommand(c.newArtifactCreateCommandOnPrem())
+		cmd.AddCommand(c.newArtifactDeleteCommandOnPrem())
+		cmd.AddCommand(c.newArtifactDescribeCommandOnPrem())
+		cmd.AddCommand(c.newArtifactListCommandOnPrem())
+		cmd.AddCommand(c.newArtifactUpdateCommandOnPrem())
+		cmd.AddCommand(c.newArtifactVersionCommandOnPrem())
+	}
 
 	return cmd
 }
diff --git a/internal/flink/command_artifact_create_onprem.go b/internal/flink/command_artifact_create_onprem.go
new file mode 100644
index 0000000000..c5a677cca3
--- /dev/null
+++ b/internal/flink/command_artifact_create_onprem.go
@@ -0,0 +1,74 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+)
+
+func (c *command) newArtifactCreateCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "create <name>",
+		Short:       "Create a Flink artifact in Confluent Platform.",
+		Long:        "Create a Flink artifact in Confluent Platform by uploading a JAR or ZIP file. This creates version 1 of the artifact.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactCreateOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Create Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact create my-artifact --artifact-file artifact.jar --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("artifact-file", "", "Path to the Flink artifact JAR or ZIP file.")
+	cmd.Flags().StringSlice("label", nil, `A comma-separated list of "key=value" label pairs.`)
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	cobra.CheckErr(cmd.MarkFlagRequired("artifact-file"))
+	cobra.CheckErr(cmd.MarkFlagFilename("artifact-file", "jar", "zip"))
+
+	return cmd
+}
+
+func (c *command) artifactCreateOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	artifactFile, err := cmd.Flags().GetString("artifact-file")
+	if err != nil {
+		return err
+	}
+
+	labels, err := getLabelsFlag(cmd)
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	file, err := openArtifactFile(artifactFile)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	outputArtifact, err := client.CreateArtifact(c.createContext(), environment, newSdkArtifact(name, labels), file)
+	if err != nil {
+		return err
+	}
+
+	return printArtifactOnPrem(cmd, outputArtifact)
+}
diff --git a/internal/flink/command_artifact_delete_onprem.go b/internal/flink/command_artifact_delete_onprem.go
new file mode 100644
index 0000000000..16c11e46f9
--- /dev/null
+++ b/internal/flink/command_artifact_delete_onprem.go
@@ -0,0 +1,58 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/deletion"
+	"github.com/confluentinc/cli/v4/pkg/errors"
+	"github.com/confluentinc/cli/v4/pkg/resource"
+)
+
+func (c *command) newArtifactDeleteCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "delete <name-1> [name-2] ... [name-n]",
+		Short:       "Delete one or more Flink artifacts in Confluent Platform.",
+		Long:        "Delete one or more Flink artifacts in Confluent Platform, including all of their versions.",
+		Args:        cobra.MinimumNArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactDeleteOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddForceFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactDeleteOnPrem(cmd *cobra.Command, args []string) error {
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	existenceFunc := func(name string) bool {
+		_, err := client.DescribeArtifact(c.createContext(), environment, name, "")
+		return err == nil
+	}
+
+	if err := deletion.ValidateAndConfirm(cmd, args, existenceFunc, resource.FlinkArtifact); err != nil {
+		return errors.NewErrorWithSuggestions(err.Error(), artifactLookupSuggestions)
+	}
+
+	// An empty version deletes the artifact and all of its versions.
+	deleteFunc := func(name string) error {
+		return client.DeleteArtifact(c.createContext(), environment, name, "")
+	}
+
+	_, err = deletion.Delete(cmd, args, deleteFunc, resource.FlinkArtifact)
+	return err
+}
diff --git a/internal/flink/command_artifact_describe_onprem.go b/internal/flink/command_artifact_describe_onprem.go
new file mode 100644
index 0000000000..ceaab020d5
--- /dev/null
+++ b/internal/flink/command_artifact_describe_onprem.go
@@ -0,0 +1,47 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+)
+
+func (c *command) newArtifactDescribeCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "describe <name>",
+		Short:       "Describe a Flink artifact in Confluent Platform.",
+		Long:        "Describe a Flink artifact in Confluent Platform. Details reflect the latest version of the artifact.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactDescribeOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactDescribeOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkArtifact, err := client.DescribeArtifact(c.createContext(), environment, name, "")
+	if err != nil {
+		return err
+	}
+
+	return printArtifactOnPrem(cmd, sdkArtifact)
+}
diff --git a/internal/flink/command_artifact_list_onprem.go b/internal/flink/command_artifact_list_onprem.go
new file mode 100644
index 0000000000..dc94fc1727
--- /dev/null
+++ b/internal/flink/command_artifact_list_onprem.go
@@ -0,0 +1,58 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *command) newArtifactListCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "list",
+		Short:       "List Flink artifacts in Confluent Platform.",
+		Args:        cobra.NoArgs,
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactListOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactListOnPrem(cmd *cobra.Command, _ []string) error {
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkArtifacts, err := client.ListArtifacts(c.createContext(), environment)
+	if err != nil {
+		return err
+	}
+
+	if output.GetFormat(cmd) == output.Human {
+		list := output.NewList(cmd)
+		for _, artifact := range sdkArtifacts {
+			list.Add(newArtifactOutOnPrem(artifact))
+		}
+		return list.Print()
+	}
+
+	localArtifacts := make([]LocalArtifact, 0, len(sdkArtifacts))
+	for _, sdkArtifact := range sdkArtifacts {
+		localArtifacts = append(localArtifacts, convertSdkArtifactToLocalArtifact(sdkArtifact))
+	}
+
+	return output.SerializedOutput(cmd, localArtifacts)
+}
diff --git a/internal/flink/command_artifact_onprem.go b/internal/flink/command_artifact_onprem.go
new file mode 100644
index 0000000000..a3ad192115
--- /dev/null
+++ b/internal/flink/command_artifact_onprem.go
@@ -0,0 +1,218 @@
+package flink
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/spf13/cobra"
+
+	cmfsdk "github.com/confluentinc/cmf-sdk-go/v1"
+
+	"github.com/confluentinc/cli/v4/pkg/output"
+	"github.com/confluentinc/cli/v4/pkg/properties"
+	"github.com/confluentinc/cli/v4/pkg/utils"
+)
+
+// CMF strictly validates these on create and update; the CLI always sends the standard values.
+const (
+	artifactApiVersion = "cmf.confluent.io/v1"
+	artifactKind       = "Artifact"
+)
+
+// artifactLookupSuggestions is shared by the delete handlers when an artifact can't be found (or CMF is unreachable).
+const artifactLookupSuggestions = "List available Flink artifacts with `confluent flink artifact list`." +
+	"\nCheck that CMF is running and accessible."
+
+// artifactOutOnPrem is the human-readable row used by the artifact list command (kept lean; no labels/annotations).
+type artifactOutOnPrem struct {
+	Name         string `human:"Name" serialized:"name"`
+	Version      string `human:"Version" serialized:"version"`
+	Phase        string `human:"Phase" serialized:"phase"`
+	Size         string `human:"Size" serialized:"size"`
+	CreationTime string `human:"Creation Time" serialized:"creation_time"`
+}
+
+// artifactDescribeOutOnPrem is the human-readable view for the single-artifact commands (describe, create, update). It
+// adds Labels and Annotations (omitted when empty) so users can see managed metadata without switching to -o json/yaml.
+type artifactDescribeOutOnPrem struct {
+	Name         string            `human:"Name"`
+	Version      string            `human:"Version"`
+	Phase        string            `human:"Phase"`
+	Size         string            `human:"Size"`
+	CreationTime string            `human:"Creation Time"`
+	Labels       map[string]string `human:"Labels,omitempty"`
+	Annotations  map[string]string `human:"Annotations,omitempty"`
+}
+
+// artifactVersionOutOnPrem is the human-readable view of a single artifact version (version create, version list, version describe).
+type artifactVersionOutOnPrem struct {
+	Version      string `human:"Version" serialized:"version"`
+	Phase        string `human:"Phase" serialized:"phase"`
+	Size         string `human:"Size" serialized:"size"`
+	Checksum     string `human:"Checksum" serialized:"checksum"`
+	CreationTime string `human:"Creation Time" serialized:"creation_time"`
+}
+
+func newArtifactOutOnPrem(artifact cmfsdk.Artifact) *artifactOutOnPrem {
+	out := &artifactOutOnPrem{Name: artifact.Metadata.Name}
+	// Artifact-level view: Creation Time is when the artifact was first created (Metadata), not the current version.
+	// The version-level view below deliberately uses Status.CreationTimestamp instead; keep the two sources distinct.
+	if artifact.Metadata.CreationTimestamp != nil {
+		out.CreationTime = *artifact.Metadata.CreationTimestamp
+	}
+	if artifact.Status != nil {
+		if artifact.Status.Version != nil {
+			out.Version = strconv.Itoa(int(*artifact.Status.Version))
+		}
+		if artifact.Status.Phase != nil {
+			out.Phase = *artifact.Status.Phase
+		}
+		if artifact.Status.Size != nil {
+			out.Size = strconv.FormatInt(*artifact.Status.Size, 10)
+		}
+	}
+	return out
+}
+
+func newArtifactVersionOutOnPrem(artifact cmfsdk.Artifact) *artifactVersionOutOnPrem {
+	out := &artifactVersionOutOnPrem{}
+	if artifact.Status != nil {
+		if artifact.Status.Version != nil {
+			out.Version = strconv.Itoa(int(*artifact.Status.Version))
+		}
+		if artifact.Status.Phase != nil {
+			out.Phase = *artifact.Status.Phase
+		}
+		if artifact.Status.Size != nil {
+			out.Size = strconv.FormatInt(*artifact.Status.Size, 10)
+		}
+		if artifact.Status.Checksum != nil {
+			out.Checksum = *artifact.Status.Checksum
+		}
+		// Version-level view: Creation Time is when this specific version was uploaded (Status), not the artifact.
+		// This intentionally differs from newArtifactOutOnPrem, which reads Metadata.CreationTimestamp.
+		if artifact.Status.CreationTimestamp != nil {
+			out.CreationTime = *artifact.Status.CreationTimestamp
+		}
+	}
+	return out
+}
+
+// newArtifactDescribeOutOnPrem builds the single-artifact human view, reusing the list row for the shared columns and
+// adding Labels/Annotations. They are left nil (and thus omitted from the table) when the artifact has none.
+func newArtifactDescribeOutOnPrem(artifact cmfsdk.Artifact) *artifactDescribeOutOnPrem {
+	base := newArtifactOutOnPrem(artifact)
+	out := &artifactDescribeOutOnPrem{
+		Name:         base.Name,
+		Version:      base.Version,
+		Phase:        base.Phase,
+		Size:         base.Size,
+		CreationTime: base.CreationTime,
+	}
+	// These are value-type maps with `human:"...,omitempty"`, so assigning a nil map is a no-op (the row is omitted).
+	// Unlike convertSdkArtifactToLocalArtifact, no nil guard is needed here.
+	out.Labels = artifact.Metadata.Labels
+	out.Annotations = artifact.Metadata.Annotations
+	return out
+}
+
+// printArtifactOnPrem prints a single artifact (artifact-level view).
+func printArtifactOnPrem(cmd *cobra.Command, artifact cmfsdk.Artifact) error {
+	if output.GetFormat(cmd) == output.Human {
+		table := output.NewTable(cmd)
+		table.Add(newArtifactDescribeOutOnPrem(artifact))
+		return table.Print()
+	}
+	return output.SerializedOutput(cmd, convertSdkArtifactToLocalArtifact(artifact))
+}
+
+// printArtifactVersionOnPrem prints a single artifact version (version-level view).
+func printArtifactVersionOnPrem(cmd *cobra.Command, artifact cmfsdk.Artifact) error {
+	if output.GetFormat(cmd) == output.Human {
+		table := output.NewTable(cmd)
+		table.Add(newArtifactVersionOutOnPrem(artifact))
+		return table.Print()
+	}
+	return output.SerializedOutput(cmd, convertSdkArtifactToLocalArtifact(artifact))
+}
+
+func convertSdkArtifactToLocalArtifact(sdkArtifact cmfsdk.Artifact) LocalArtifact {
+	localArtifact := LocalArtifact{
+		ApiVersion: sdkArtifact.ApiVersion,
+		Kind:       sdkArtifact.Kind,
+		Metadata: LocalArtifactMetadata{
+			Name:              sdkArtifact.Metadata.Name,
+			CreationTimestamp: sdkArtifact.Metadata.CreationTimestamp,
+			UpdateTimestamp:   sdkArtifact.Metadata.UpdateTimestamp,
+			Uid:               sdkArtifact.Metadata.Uid,
+		},
+		Spec: sdkArtifact.Spec,
+	}
+
+	if sdkArtifact.Metadata.Labels != nil {
+		localArtifact.Metadata.Labels = &sdkArtifact.Metadata.Labels
+	}
+	if sdkArtifact.Metadata.Annotations != nil {
+		localArtifact.Metadata.Annotations = &sdkArtifact.Metadata.Annotations
+	}
+
+	if sdkArtifact.Status != nil {
+		localArtifact.Status = &LocalArtifactStatus{
+			Version:           sdkArtifact.Status.Version,
+			CreationTimestamp: sdkArtifact.Status.CreationTimestamp,
+			Path:              sdkArtifact.Status.Path,
+			Size:              sdkArtifact.Status.Size,
+			Checksum:          sdkArtifact.Status.Checksum,
+			Phase:             sdkArtifact.Status.Phase,
+			Message:           sdkArtifact.Status.Message,
+		}
+	}
+
+	return localArtifact
+}
+
+// newSdkArtifact builds an Artifact payload for create and update requests. Labels are set only when non-nil so that
+// omitting them (nil) preserves any existing labels server-side.
+func newSdkArtifact(name string, labels map[string]string) cmfsdk.Artifact {
+	artifact := cmfsdk.Artifact{
+		ApiVersion: artifactApiVersion,
+		Kind:       artifactKind,
+		Metadata:   cmfsdk.ArtifactMetadata{Name: name},
+		Spec:       map[string]interface{}{},
+	}
+	if labels != nil {
+		artifact.Metadata.Labels = labels
+	}
+	return artifact
+}
+
+// getLabelsFlag parses the repeatable "--label key=value" flag into a map. It returns a nil map when the flag was not provided.
+func getLabelsFlag(cmd *cobra.Command) (map[string]string, error) {
+	labelSlice, err := cmd.Flags().GetStringSlice("label")
+	if err != nil {
+		return nil, err
+	}
+	if len(labelSlice) == 0 {
+		return nil, nil
+	}
+	return properties.ConfigSliceToMap(labelSlice)
+}
+
+// openArtifactFile validates the artifact file extension and opens it for upload. The caller is responsible for closing the returned file.
+func openArtifactFile(path string) (*os.File, error) {
+	extension := strings.TrimPrefix(filepath.Ext(path), ".")
+	if !slices.Contains(allowedFileExtensions, strings.ToLower(extension)) {
+		return nil, fmt.Errorf("only extensions allowed for `--artifact-file` are %s", utils.ArrayToCommaDelimitedString(allowedFileExtensions, "and"))
+	}
+
+	file, err := os.Open(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read artifact file: %w", err)
+	}
+
+	return file, nil
+}
diff --git a/internal/flink/command_artifact_update_onprem.go b/internal/flink/command_artifact_update_onprem.go
new file mode 100644
index 0000000000..22cd371b9f
--- /dev/null
+++ b/internal/flink/command_artifact_update_onprem.go
@@ -0,0 +1,62 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+)
+
+func (c *command) newArtifactUpdateCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "update <name>",
+		Short:       "Update a Flink artifact's metadata in Confluent Platform.",
+		Long:        "Update the labels of a Flink artifact in Confluent Platform without uploading new content.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactUpdateOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Replace the labels of Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact update my-artifact --label owner=team-a,tier=gold --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().StringSlice("label", nil, `A comma-separated list of "key=value" label pairs. Provide the complete set of labels to apply; omit the flag to leave existing labels unchanged.`)
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactUpdateOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	// getLabelsFlag returns a nil map when --label is omitted, which omits the field so existing labels are preserved server-side.
+	labels, err := getLabelsFlag(cmd)
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	// Metadata-only update: no file is uploaded, so no new version is created.
+	outputArtifact, err := client.UpdateArtifact(c.createContext(), environment, name, newSdkArtifact(name, labels), nil)
+	if err != nil {
+		return err
+	}
+
+	return printArtifactOnPrem(cmd, outputArtifact)
+}
diff --git a/internal/flink/command_artifact_version_create_onprem.go b/internal/flink/command_artifact_version_create_onprem.go
new file mode 100644
index 0000000000..45787e0879
--- /dev/null
+++ b/internal/flink/command_artifact_version_create_onprem.go
@@ -0,0 +1,69 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+)
+
+func (c *command) newArtifactVersionCreateCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "create <name>",
+		Short:       "Upload a new Flink artifact version in Confluent Platform.",
+		Long:        "Upload a new version of a Flink artifact in Confluent Platform. If the uploaded content is identical to the latest version, the artifact is left unchanged and no new version is created.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactVersionCreateOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Upload a new version of Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact version create my-artifact --artifact-file artifact-v2.jar --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("artifact-file", "", "Path to the Flink artifact JAR or ZIP file.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	cobra.CheckErr(cmd.MarkFlagRequired("artifact-file"))
+	cobra.CheckErr(cmd.MarkFlagFilename("artifact-file", "jar", "zip"))
+
+	return cmd
+}
+
+func (c *command) artifactVersionCreateOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	artifactFile, err := cmd.Flags().GetString("artifact-file")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	file, err := openArtifactFile(artifactFile)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	// Nil labels leave existing artifact labels unchanged; only the content (new version) is uploaded.
+	outputArtifact, err := client.UpdateArtifact(c.createContext(), environment, name, newSdkArtifact(name, nil), file)
+	if err != nil {
+		return err
+	}
+
+	return printArtifactVersionOnPrem(cmd, outputArtifact)
+}
diff --git a/internal/flink/command_artifact_version_delete_onprem.go b/internal/flink/command_artifact_version_delete_onprem.go
new file mode 100644
index 0000000000..fbd6096781
--- /dev/null
+++ b/internal/flink/command_artifact_version_delete_onprem.go
@@ -0,0 +1,84 @@
+package flink
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/deletion"
+	"github.com/confluentinc/cli/v4/pkg/errors"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *command) newArtifactVersionDeleteCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "delete <name>",
+		Short:       "Delete a version of a Flink artifact in Confluent Platform.",
+		Long:        "Delete a version of a Flink artifact in Confluent Platform. Deleting the last remaining version deletes the artifact.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactVersionDeleteOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Delete version 2 of Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact version delete my-artifact --version 2 --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("version", "", `Version of the artifact to delete, or "all" to delete every version.`)
+	addCmfFlagSet(cmd)
+	pcmd.AddForceFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	cobra.CheckErr(cmd.MarkFlagRequired("version"))
+
+	return cmd
+}
+
+func (c *command) artifactVersionDeleteOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	version, err := cmd.Flags().GetString("version")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	if _, err := client.DescribeArtifact(c.createContext(), environment, name, ""); err != nil {
+		return errors.NewErrorWithSuggestions(err.Error(), artifactLookupSuggestions)
+	}
+
+	var promptMsg string
+	if version == "all" {
+		promptMsg = fmt.Sprintf(`Are you sure you want to delete all versions of Flink artifact "%s"?`, name)
+	} else {
+		promptMsg = fmt.Sprintf(`Are you sure you want to delete version "%s" of Flink artifact "%s"?`, version, name)
+	}
+	if err := deletion.ConfirmPrompt(cmd, promptMsg); err != nil {
+		return err
+	}
+
+	if err := client.DeleteArtifact(c.createContext(), environment, name, version); err != nil {
+		return err
+	}
+
+	if version == "all" {
+		output.Printf(false, "Deleted all versions of Flink artifact %q.\n", name)
+	} else {
+		output.Printf(false, "Deleted version %q of Flink artifact %q.\n", version, name)
+	}
+	return nil
+}
diff --git a/internal/flink/command_artifact_version_describe_onprem.go b/internal/flink/command_artifact_version_describe_onprem.go
new file mode 100644
index 0000000000..cee9c8de9c
--- /dev/null
+++ b/internal/flink/command_artifact_version_describe_onprem.go
@@ -0,0 +1,60 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+)
+
+func (c *command) newArtifactVersionDescribeCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "describe <name>",
+		Short:       "Describe a Flink artifact version in Confluent Platform.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactVersionDescribeOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Describe version 2 of Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact version describe my-artifact --version 2 --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("version", "", "Version of the artifact to describe.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	cobra.CheckErr(cmd.MarkFlagRequired("version"))
+
+	return cmd
+}
+
+func (c *command) artifactVersionDescribeOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	version, err := cmd.Flags().GetString("version")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkArtifact, err := client.DescribeArtifact(c.createContext(), environment, name, version)
+	if err != nil {
+		return err
+	}
+
+	return printArtifactVersionOnPrem(cmd, sdkArtifact)
+}
diff --git a/internal/flink/command_artifact_version_download_onprem.go b/internal/flink/command_artifact_version_download_onprem.go
new file mode 100644
index 0000000000..9b63a081a3
--- /dev/null
+++ b/internal/flink/command_artifact_version_download_onprem.go
@@ -0,0 +1,96 @@
+package flink
+
+import (
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *command) newArtifactVersionDownloadCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "download <name>",
+		Short:       "Download a Flink artifact's content in Confluent Platform.",
+		Long:        "Download the binary content of a Flink artifact in Confluent Platform. Defaults to the latest version unless `--version` is specified.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactVersionDownloadOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Download the latest version of Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact version download my-artifact --output-file my-artifact.jar --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("output-file", "", "Path to write the downloaded artifact file.")
+	cmd.Flags().String("version", "", "Version of the artifact to download. Defaults to the latest version.")
+	addCmfFlagSet(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	cobra.CheckErr(cmd.MarkFlagRequired("output-file"))
+
+	return cmd
+}
+
+func (c *command) artifactVersionDownloadOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	version, err := cmd.Flags().GetString("version")
+	if err != nil {
+		return err
+	}
+
+	outputFile, err := cmd.Flags().GetString("output-file")
+	if err != nil {
+		return err
+	}
+
+	// Like `asyncapi export`, download overwrites the output file if it already exists.
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	downloadedFile, err := client.DownloadArtifactContent(c.createContext(), environment, name, version)
+	if err != nil {
+		return err
+	}
+	// A 2xx response with an empty body yields a nil file (the SDK does not allocate one); guard against it so the
+	// deferred cleanup below doesn't dereference a nil *os.File.
+	if downloadedFile == nil {
+		return fmt.Errorf(`no content was returned for Flink artifact "%s"`, name)
+	}
+	defer os.Remove(downloadedFile.Name())
+	defer downloadedFile.Close()
+
+	destination, err := os.Create(outputFile)
+	if err != nil {
+		return fmt.Errorf("failed to create output file: %w", err)
+	}
+
+	if _, err := io.Copy(destination, downloadedFile); err != nil {
+		destination.Close()
+		// Remove the partial output file so a failed download doesn't leave a truncated or corrupt artifact behind.
+		os.Remove(outputFile)
+		return fmt.Errorf("failed to write output file: %w", err)
+	}
+
+	if err := destination.Close(); err != nil {
+		return fmt.Errorf("failed to write output file: %w", err)
+	}
+
+	output.Printf(false, "Downloaded Flink artifact %q to %q.\n", name, outputFile)
+	return nil
+}
diff --git a/internal/flink/command_artifact_version_list_onprem.go b/internal/flink/command_artifact_version_list_onprem.go
new file mode 100644
index 0000000000..8d4d43a078
--- /dev/null
+++ b/internal/flink/command_artifact_version_list_onprem.go
@@ -0,0 +1,62 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *command) newArtifactVersionListCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "list <name>",
+		Short:       "List the versions of a Flink artifact in Confluent Platform.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactVersionListOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactVersionListOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkVersions, err := client.ListArtifactVersions(c.createContext(), environment, name)
+	if err != nil {
+		return err
+	}
+
+	if output.GetFormat(cmd) == output.Human {
+		list := output.NewList(cmd)
+		// Preserve the server's newest-first ordering instead of sorting version strings lexicographically.
+		list.Sort(false)
+		for _, version := range sdkVersions {
+			list.Add(newArtifactVersionOutOnPrem(version))
+		}
+		return list.Print()
+	}
+
+	localVersions := make([]LocalArtifact, 0, len(sdkVersions))
+	for _, sdkVersion := range sdkVersions {
+		localVersions = append(localVersions, convertSdkArtifactToLocalArtifact(sdkVersion))
+	}
+
+	return output.SerializedOutput(cmd, localVersions)
+}
diff --git a/internal/flink/command_artifact_version_onprem.go b/internal/flink/command_artifact_version_onprem.go
new file mode 100644
index 0000000000..39fe9087d5
--- /dev/null
+++ b/internal/flink/command_artifact_version_onprem.go
@@ -0,0 +1,20 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+)
+
+func (c *command) newArtifactVersionCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "version",
+		Short: "Manage versions of a Flink artifact in Confluent Platform.",
+	}
+
+	cmd.AddCommand(c.newArtifactVersionCreateCommandOnPrem())
+	cmd.AddCommand(c.newArtifactVersionDeleteCommandOnPrem())
+	cmd.AddCommand(c.newArtifactVersionDescribeCommandOnPrem())
+	cmd.AddCommand(c.newArtifactVersionDownloadCommandOnPrem())
+	cmd.AddCommand(c.newArtifactVersionListCommandOnPrem())
+
+	return cmd
+}
diff --git a/internal/flink/local_types.go b/internal/flink/local_types.go
index ea601f8e8b..0693b0dfb2 100644
--- a/internal/flink/local_types.go
+++ b/internal/flink/local_types.go
@@ -334,3 +334,30 @@ type LocalStatementTraits struct {
 	UpsertColumns *[]int32           `json:"upsertColumns,omitempty" yaml:"upsertColumns,omitempty"`
 	Schema        *LocalResultSchema `json:"schema,omitempty" yaml:"schema,omitempty"`
 }
+
+type LocalArtifact struct {
+	ApiVersion string                 `json:"apiVersion" yaml:"apiVersion"`
+	Kind       string                 `json:"kind" yaml:"kind"`
+	Metadata   LocalArtifactMetadata  `json:"metadata" yaml:"metadata"`
+	Spec       map[string]interface{} `json:"spec" yaml:"spec"`
+	Status     *LocalArtifactStatus   `json:"status,omitempty" yaml:"status,omitempty"`
+}
+
+type LocalArtifactMetadata struct {
+	Name              string             `json:"name" yaml:"name"`
+	CreationTimestamp *string            `json:"creationTimestamp,omitempty" yaml:"creationTimestamp,omitempty"`
+	UpdateTimestamp   *string            `json:"updateTimestamp,omitempty" yaml:"updateTimestamp,omitempty"`
+	Uid               *string            `json:"uid,omitempty" yaml:"uid,omitempty"`
+	Labels            *map[string]string `json:"labels,omitempty" yaml:"labels,omitempty"`
+	Annotations       *map[string]string `json:"annotations,omitempty" yaml:"annotations,omitempty"`
+}
+
+type LocalArtifactStatus struct {
+	Version           *int32  `json:"version,omitempty" yaml:"version,omitempty"`
+	CreationTimestamp *string `json:"creationTimestamp,omitempty" yaml:"creationTimestamp,omitempty"`
+	Path              *string `json:"path,omitempty" yaml:"path,omitempty"`
+	Size              *int64  `json:"size,omitempty" yaml:"size,omitempty"`
+	Checksum          *string `json:"checksum,omitempty" yaml:"checksum,omitempty"`
+	Phase             *string `json:"phase,omitempty" yaml:"phase,omitempty"`
+	Message           *string `json:"message,omitempty" yaml:"message,omitempty"`
+}
diff --git a/pkg/flink/cmf_rest_client.go b/pkg/flink/cmf_rest_client.go
index 8518ad1423..5031a6747b 100644
--- a/pkg/flink/cmf_rest_client.go
+++ b/pkg/flink/cmf_rest_client.go
@@ -1,14 +1,19 @@
 package flink
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
+	"mime/multipart"
 	"net/http"
 	_nethttp "net/http"
+	"net/textproto"
+	neturl "net/url"
 	"os"
+	"path/filepath"
 	"strings"
 
 	"github.com/spf13/cobra"
@@ -550,17 +555,14 @@ func (cmfClient *CmfRestClient) GetStatementResults(ctx context.Context, environ
 }
 
 func (cmfClient *CmfRestClient) GetSystemInformation(ctx context.Context) (map[string]interface{}, error) {
-	baseURL := strings.TrimRight(cmfClient.GetConfig().Servers[0].URL, "/")
-	url := baseURL + "/cmf/api/v1/system-information"
+	url := cmfClient.cmfBaseURL() + "/cmf/api/v1/system-information"
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create system information request: %s", err)
 	}
 
-	if token, ok := ctx.Value(cmfsdk.ContextAccessToken).(string); ok && token != "" {
-		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
-	}
+	setCmfAuthHeader(ctx, req)
 
 	resp, err := cmfClient.GetConfig().HTTPClient.Do(req)
 	if err != nil {
@@ -574,11 +576,7 @@ func (cmfClient *CmfRestClient) GetSystemInformation(ctx context.Context) (map[s
 	}
 
 	if resp.StatusCode != http.StatusOK {
-		trimmed := strings.TrimSpace(string(body))
-		if trimmed != "" {
-			return nil, errors.New(trimmed)
-		}
-		return nil, errors.New(resp.Status)
+		return nil, cmfErrorFromBody(resp.Status, body)
 	}
 
 	var result map[string]interface{}
@@ -815,6 +813,195 @@ func (cmfClient *CmfRestClient) ListDatabases(ctx context.Context, catalogName s
 	return databases, nil
 }
 
+// CreateArtifact uploads a new artifact (version 1) to the specified environment.
+// Both the artifact metadata and the binary file are required by the CMF API.
+func (cmfClient *CmfRestClient) CreateArtifact(ctx context.Context, environment string, artifact cmfsdk.Artifact, file *os.File) (cmfsdk.Artifact, error) {
+	url := fmt.Sprintf("%s/cmf/api/v1/environments/%s/artifacts", cmfClient.cmfBaseURL(), neturl.PathEscape(environment))
+	outputArtifact, err := cmfClient.uploadArtifact(ctx, http.MethodPost, url, artifact, file)
+	if err != nil {
+		return cmfsdk.Artifact{}, fmt.Errorf(`failed to create artifact "%s" in the environment "%s": %s`, artifact.Metadata.Name, environment, err)
+	}
+	return outputArtifact, nil
+}
+
+// UpdateArtifact updates an artifact in the specified environment.
+// When file is nil, only the metadata is updated; when a file is provided, a new version is created (or deduplicated if identical to the latest).
+func (cmfClient *CmfRestClient) UpdateArtifact(ctx context.Context, environment, name string, artifact cmfsdk.Artifact, file *os.File) (cmfsdk.Artifact, error) {
+	url := fmt.Sprintf("%s/cmf/api/v1/environments/%s/artifacts/%s", cmfClient.cmfBaseURL(), neturl.PathEscape(environment), neturl.PathEscape(name))
+	outputArtifact, err := cmfClient.uploadArtifact(ctx, http.MethodPut, url, artifact, file)
+	if err != nil {
+		return cmfsdk.Artifact{}, fmt.Errorf(`failed to update artifact "%s" in the environment "%s": %s`, name, environment, err)
+	}
+	return outputArtifact, nil
+}
+
+// cmfBaseURL returns the CMF server base URL with any trailing slash trimmed. Used by the handful of methods that
+// build CMF requests manually rather than through the generated SDK.
+func (cmfClient *CmfRestClient) cmfBaseURL() string {
+	return strings.TrimRight(cmfClient.GetConfig().Servers[0].URL, "/")
+}
+
+// setCmfAuthHeader copies the bearer token from the context onto a manually built CMF request, if one is present.
+func setCmfAuthHeader(ctx context.Context, request *http.Request) {
+	if token, ok := ctx.Value(cmfsdk.ContextAccessToken).(string); ok && token != "" {
+		request.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
+	}
+}
+
+// cmfErrorFromBody builds an error from a failed CMF response, preferring the response body over the status line.
+func cmfErrorFromBody(status string, body []byte) error {
+	if trimmed := strings.TrimSpace(string(body)); trimmed != "" {
+		return errors.New(trimmed)
+	}
+	return errors.New(status)
+}
+
+// uploadArtifact sends a multipart/form-data request for the artifact create and update endpoints.
+// The generated SDK serializes the "artifact" object part with fmt "%v" (Go struct representation) rather than JSON,
+// so the request is built here (mirroring GetSystemInformation's manual CMF request handling) to send a proper JSON part.
+// Like GetSystemInformation, this bypasses the SDK's request pipeline, so `--unsafe-trace` request logging and the
+// configured User-Agent do not apply to the manually built artifact-upload and system-information requests.
+func (cmfClient *CmfRestClient) uploadArtifact(ctx context.Context, method, url string, artifact cmfsdk.Artifact, file *os.File) (cmfsdk.Artifact, error) {
+	body := new(bytes.Buffer)
+	writer := multipart.NewWriter(body)
+
+	artifactJson, err := json.Marshal(artifact)
+	if err != nil {
+		return cmfsdk.Artifact{}, fmt.Errorf("failed to serialize artifact: %s", err)
+	}
+
+	partHeader := textproto.MIMEHeader{}
+	partHeader.Set("Content-Disposition", `form-data; name="artifact"`)
+	partHeader.Set("Content-Type", "application/json")
+	artifactPart, err := writer.CreatePart(partHeader)
+	if err != nil {
+		return cmfsdk.Artifact{}, err
+	}
+	if _, err := artifactPart.Write(artifactJson); err != nil {
+		return cmfsdk.Artifact{}, err
+	}
+
+	if file != nil {
+		filePart, err := writer.CreateFormFile("file", filepath.Base(file.Name()))
+		if err != nil {
+			return cmfsdk.Artifact{}, err
+		}
+		if _, err := io.Copy(filePart, file); err != nil {
+			return cmfsdk.Artifact{}, err
+		}
+	}
+
+	if err := writer.Close(); err != nil {
+		return cmfsdk.Artifact{}, err
+	}
+
+	request, err := http.NewRequestWithContext(ctx, method, url, body)
+	if err != nil {
+		return cmfsdk.Artifact{}, err
+	}
+	request.Header.Set("Content-Type", writer.FormDataContentType())
+	setCmfAuthHeader(ctx, request)
+
+	response, err := cmfClient.GetConfig().HTTPClient.Do(request)
+	if err != nil {
+		return cmfsdk.Artifact{}, err
+	}
+	defer response.Body.Close()
+
+	responseBody, err := io.ReadAll(response.Body)
+	if err != nil {
+		return cmfsdk.Artifact{}, err
+	}
+
+	// Create returns 201 and update 200, so treat any non-2xx status as a failure.
+	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
+		return cmfsdk.Artifact{}, cmfErrorFromBody(response.Status, responseBody)
+	}
+
+	var outputArtifact cmfsdk.Artifact
+	if err := json.Unmarshal(responseBody, &outputArtifact); err != nil {
+		return cmfsdk.Artifact{}, fmt.Errorf("failed to parse artifact response: %s", err)
+	}
+	return outputArtifact, nil
+}
+
+// DescribeArtifact fetches an artifact from the specified environment. When version is empty, the latest version is returned.
+func (cmfClient *CmfRestClient) DescribeArtifact(ctx context.Context, environment, name, version string) (cmfsdk.Artifact, error) {
+	request := cmfClient.ArtifactsApi.GetArtifact(ctx, environment, name)
+	if version != "" {
+		request = request.Version(version)
+	}
+	outputArtifact, httpResponse, err := request.Execute()
+	if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil {
+		return cmfsdk.Artifact{}, fmt.Errorf(`failed to describe artifact "%s" in the environment "%s": %s`, name, environment, parsedErr)
+	}
+	return outputArtifact, nil
+}
+
+func (cmfClient *CmfRestClient) ListArtifacts(ctx context.Context, environment string) ([]cmfsdk.Artifact, error) {
+	artifacts := make([]cmfsdk.Artifact, 0)
+	done := false
+	// 100 is an arbitrary page size we've chosen.
+	const pageSize = 100
+	var currentPageNumber int32 = 0
+
+	for !done {
+		artifactsPage, httpResponse, err := cmfClient.ArtifactsApi.ListArtifacts(ctx, environment).Page(currentPageNumber).Size(pageSize).Execute()
+		if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil {
+			return nil, fmt.Errorf(`failed to list artifacts in the environment "%s": %s`, environment, parsedErr)
+		}
+		artifacts = append(artifacts, artifactsPage.GetItems()...)
+		currentPageNumber, done = extractPageOptions(len(artifactsPage.GetItems()), currentPageNumber)
+	}
+
+	return artifacts, nil
+}
+
+// ListArtifactVersions lists all versions of an artifact, newest-first.
+func (cmfClient *CmfRestClient) ListArtifactVersions(ctx context.Context, environment, name string) ([]cmfsdk.Artifact, error) {
+	versions := make([]cmfsdk.Artifact, 0)
+	done := false
+	// 100 is an arbitrary page size we've chosen.
+	const pageSize = 100
+	var currentPageNumber int32 = 0
+
+	for !done {
+		versionsPage, httpResponse, err := cmfClient.ArtifactsApi.ListArtifactVersions(ctx, environment, name).Page(currentPageNumber).Size(pageSize).Execute()
+		if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil {
+			return nil, fmt.Errorf(`failed to list versions of artifact "%s" in the environment "%s": %s`, name, environment, parsedErr)
+		}
+		versions = append(versions, versionsPage.GetItems()...)
+		currentPageNumber, done = extractPageOptions(len(versionsPage.GetItems()), currentPageNumber)
+	}
+
+	return versions, nil
+}
+
+// DeleteArtifact deletes an artifact from the specified environment. When version is empty the entire artifact is removed;
+// otherwise the given version ("N" or "all") is removed.
+func (cmfClient *CmfRestClient) DeleteArtifact(ctx context.Context, environment, name, version string) error {
+	request := cmfClient.ArtifactsApi.DeleteArtifact(ctx, environment, name)
+	if version != "" {
+		request = request.Version(version)
+	}
+	httpResp, err := request.Execute()
+	return parseSdkError(httpResp, err)
+}
+
+// DownloadArtifactContent downloads the binary content of an artifact. When version is empty, the latest version is downloaded.
+// The returned *os.File is a temporary file the SDK created; the caller owns it and must Close and os.Remove it.
+func (cmfClient *CmfRestClient) DownloadArtifactContent(ctx context.Context, environment, name, version string) (*os.File, error) {
+	request := cmfClient.ArtifactsApi.DownloadArtifactContent(ctx, environment, name)
+	if version != "" {
+		request = request.Version(version)
+	}
+	file, httpResponse, err := request.Execute()
+	if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil {
+		return nil, fmt.Errorf(`failed to download content of artifact "%s" in the environment "%s": %s`, name, environment, parsedErr)
+	}
+	return file, nil
+}
+
 // Returns the next page number and whether we need to fetch more pages or not.
 func extractPageOptions(receivedItemsLength int, currentPageNumber int32) (int32, bool) {
 	if receivedItemsLength == 0 {
diff --git a/test/fixtures/input/flink/artifact/artifact-v2.jar b/test/fixtures/input/flink/artifact/artifact-v2.jar
new file mode 100644
index 0000000000..4f20fd7d5b
--- /dev/null
+++ b/test/fixtures/input/flink/artifact/artifact-v2.jar
@@ -0,0 +1 @@
+dummy jar content v2 for flink artifact integration tests
diff --git a/test/fixtures/input/flink/artifact/artifact.jar b/test/fixtures/input/flink/artifact/artifact.jar
new file mode 100644
index 0000000000..a59e400019
--- /dev/null
+++ b/test/fixtures/input/flink/artifact/artifact.jar
@@ -0,0 +1 @@
+dummy jar content for flink artifact integration tests
diff --git a/test/fixtures/output/flink/artifact/create-existing-failure.golden b/test/fixtures/output/flink/artifact/create-existing-failure.golden
new file mode 100644
index 0000000000..c6c9c8bc1a
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-existing-failure.golden
@@ -0,0 +1 @@
+Error: failed to create artifact "existing-artifact" in the environment "test-env": The artifact name already exists, please try with another artifact name
diff --git a/test/fixtures/output/flink/artifact/create-help-onprem.golden b/test/fixtures/output/flink/artifact/create-help-onprem.golden
new file mode 100644
index 0000000000..2bbde71bf1
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-help-onprem.golden
@@ -0,0 +1,24 @@
+Create a Flink artifact in Confluent Platform by uploading a JAR or ZIP file. This creates version 1 of the artifact.
+
+Usage:
+  confluent flink artifact create <name> [flags]
+
+Examples:
+Create Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact create my-artifact --artifact-file artifact.jar --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --artifact-file string                REQUIRED: Path to the Flink artifact JAR or ZIP file.
+      --label strings                       A comma-separated list of "key=value" label pairs.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/create-invalid-extension-failure.golden b/test/fixtures/output/flink/artifact/create-invalid-extension-failure.golden
new file mode 100644
index 0000000000..88385be0f5
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-invalid-extension-failure.golden
@@ -0,0 +1 @@
+Error: only extensions allowed for `--artifact-file` are "jar" and "zip"
diff --git a/test/fixtures/output/flink/artifact/create-missing-file-flag-failure.golden b/test/fixtures/output/flink/artifact/create-missing-file-flag-failure.golden
new file mode 100644
index 0000000000..2fa76f6944
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-missing-file-flag-failure.golden
@@ -0,0 +1,24 @@
+Error: required flag(s) "artifact-file" not set
+Usage:
+  confluent flink artifact create <name> [flags]
+
+Examples:
+Create Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact create my-artifact --artifact-file artifact.jar --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --artifact-file string                REQUIRED: Path to the Flink artifact JAR or ZIP file.
+      --label strings                       A comma-separated list of "key=value" label pairs.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
+
diff --git a/test/fixtures/output/flink/artifact/create-success-json.golden b/test/fixtures/output/flink/artifact/create-success-json.golden
new file mode 100644
index 0000000000..48a6586d07
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-success-json.golden
@@ -0,0 +1,19 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111"
+  },
+  "spec": {},
+  "status": {
+    "version": 1,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/1",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/create-success-yaml.golden b/test/fixtures/output/flink/artifact/create-success-yaml.golden
new file mode 100644
index 0000000000..44c7ea4765
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-success-yaml.golden
@@ -0,0 +1,15 @@
+apiVersion: cmf.confluent.io/v1
+kind: Artifact
+metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+spec: {}
+status:
+    version: 1
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/1
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/create-success.golden b/test/fixtures/output/flink/artifact/create-success.golden
new file mode 100644
index 0000000000..f639323b05
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-success.golden
@@ -0,0 +1,7 @@
++---------------+-------------------------------+
+| Name          | test-artifact                 |
+| Version       | 1                             |
+| Phase         | READY                         |
+| Size          | 1024                          |
+| Creation Time | 2025-03-12 23:42:00 +0000 UTC |
++---------------+-------------------------------+
diff --git a/test/fixtures/output/flink/artifact/create-with-labels-json.golden b/test/fixtures/output/flink/artifact/create-with-labels-json.golden
new file mode 100644
index 0000000000..5a6140cc68
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/create-with-labels-json.golden
@@ -0,0 +1,23 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111",
+    "labels": {
+      "owner": "team-a",
+      "tier": "gold"
+    }
+  },
+  "spec": {},
+  "status": {
+    "version": 1,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/1",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/delete-help-onprem.golden b/test/fixtures/output/flink/artifact/delete-help-onprem.golden
new file mode 100644
index 0000000000..de87321f0b
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/delete-help-onprem.golden
@@ -0,0 +1,17 @@
+Delete one or more Flink artifacts in Confluent Platform, including all of their versions.
+
+Usage:
+  confluent flink artifact delete <name-1> [name-2] ... [name-n] [flags]
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+      --force                               Skip the deletion confirmation prompt.
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/delete-multiple-successful.golden b/test/fixtures/output/flink/artifact/delete-multiple-successful.golden
new file mode 100644
index 0000000000..be57d049d8
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/delete-multiple-successful.golden
@@ -0,0 +1 @@
+Are you sure you want to delete Flink artifacts "test-artifact-1" and "test-artifact-2"? (y/n): Deleted Flink artifacts "test-artifact-1" and "test-artifact-2".
diff --git a/test/fixtures/output/flink/artifact/delete-non-exist-failure.golden b/test/fixtures/output/flink/artifact/delete-non-exist-failure.golden
new file mode 100644
index 0000000000..3f427a6ed3
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/delete-non-exist-failure.golden
@@ -0,0 +1,5 @@
+Error: Flink artifact "non-exist-artifact" not found
+
+Suggestions:
+    List available Flink artifacts with `confluent flink artifact list`.
+    Check that CMF is running and accessible.
diff --git a/test/fixtures/output/flink/artifact/delete-single-force.golden b/test/fixtures/output/flink/artifact/delete-single-force.golden
new file mode 100644
index 0000000000..9e8ef5f635
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/delete-single-force.golden
@@ -0,0 +1 @@
+Deleted Flink artifact "test-artifact".
diff --git a/test/fixtures/output/flink/artifact/delete-single-successful.golden b/test/fixtures/output/flink/artifact/delete-single-successful.golden
new file mode 100644
index 0000000000..811b612e5a
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/delete-single-successful.golden
@@ -0,0 +1 @@
+Are you sure you want to delete Flink artifact "test-artifact"? (y/n): Deleted Flink artifact "test-artifact".
diff --git a/test/fixtures/output/flink/artifact/describe-help-onprem.golden b/test/fixtures/output/flink/artifact/describe-help-onprem.golden
new file mode 100644
index 0000000000..4c9c265f08
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-help-onprem.golden
@@ -0,0 +1,17 @@
+Describe a Flink artifact in Confluent Platform. Details reflect the latest version of the artifact.
+
+Usage:
+  confluent flink artifact describe <name> [flags]
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/describe-labeled-json.golden b/test/fixtures/output/flink/artifact/describe-labeled-json.golden
new file mode 100644
index 0000000000..c1b7e2ab3c
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-labeled-json.golden
@@ -0,0 +1,26 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "labeled-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111",
+    "labels": {
+      "owner": "team-a",
+      "tier": "gold"
+    },
+    "annotations": {
+      "note": "managed-externally"
+    }
+  },
+  "spec": {},
+  "status": {
+    "version": 3,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/labeled-artifact/3",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/describe-labeled.golden b/test/fixtures/output/flink/artifact/describe-labeled.golden
new file mode 100644
index 0000000000..9ec59c9bb1
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-labeled.golden
@@ -0,0 +1,10 @@
++---------------+-------------------------------+
+| Name          | labeled-artifact              |
+| Version       | 3                             |
+| Phase         | READY                         |
+| Size          | 1024                          |
+| Creation Time | 2025-03-12 23:42:00 +0000 UTC |
+| Labels        | owner=team-a                  |
+|               | tier=gold                     |
+| Annotations   | note=managed-externally       |
++---------------+-------------------------------+
diff --git a/test/fixtures/output/flink/artifact/describe-non-exist-failure.golden b/test/fixtures/output/flink/artifact/describe-non-exist-failure.golden
new file mode 100644
index 0000000000..520294b0b3
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-non-exist-failure.golden
@@ -0,0 +1 @@
+Error: failed to describe artifact "invalid-artifact" in the environment "test-env": The artifact name is invalid
diff --git a/test/fixtures/output/flink/artifact/describe-success-json.golden b/test/fixtures/output/flink/artifact/describe-success-json.golden
new file mode 100644
index 0000000000..0af9db043a
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-success-json.golden
@@ -0,0 +1,19 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111"
+  },
+  "spec": {},
+  "status": {
+    "version": 3,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/3",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/describe-success-yaml.golden b/test/fixtures/output/flink/artifact/describe-success-yaml.golden
new file mode 100644
index 0000000000..a75f11d473
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-success-yaml.golden
@@ -0,0 +1,15 @@
+apiVersion: cmf.confluent.io/v1
+kind: Artifact
+metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+spec: {}
+status:
+    version: 3
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/3
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/describe-success.golden b/test/fixtures/output/flink/artifact/describe-success.golden
new file mode 100644
index 0000000000..f56f8457eb
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/describe-success.golden
@@ -0,0 +1,7 @@
++---------------+-------------------------------+
+| Name          | test-artifact                 |
+| Version       | 3                             |
+| Phase         | READY                         |
+| Size          | 1024                          |
+| Creation Time | 2025-03-12 23:42:00 +0000 UTC |
++---------------+-------------------------------+
diff --git a/test/fixtures/output/flink/artifact/help-onprem.golden b/test/fixtures/output/flink/artifact/help-onprem.golden
new file mode 100644
index 0000000000..00c4a89f28
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/help-onprem.golden
@@ -0,0 +1,19 @@
+Manage Flink UDF artifacts.
+
+Usage:
+  confluent flink artifact [command]
+
+Available Commands:
+  create      Create a Flink artifact in Confluent Platform.
+  delete      Delete one or more Flink artifacts in Confluent Platform.
+  describe    Describe a Flink artifact in Confluent Platform.
+  list        List Flink artifacts in Confluent Platform.
+  update      Update a Flink artifact's metadata in Confluent Platform.
+  version     Manage versions of a Flink artifact in Confluent Platform.
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
+
+Use "confluent flink artifact [command] --help" for more information about a command.
diff --git a/test/fixtures/output/flink/artifact/list-help-onprem.golden b/test/fixtures/output/flink/artifact/list-help-onprem.golden
new file mode 100644
index 0000000000..86c5104a13
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/list-help-onprem.golden
@@ -0,0 +1,17 @@
+List Flink artifacts in Confluent Platform.
+
+Usage:
+  confluent flink artifact list [flags]
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/list-missing-env-flag-failure.golden b/test/fixtures/output/flink/artifact/list-missing-env-flag-failure.golden
new file mode 100644
index 0000000000..083823b47e
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/list-missing-env-flag-failure.golden
@@ -0,0 +1,17 @@
+Error: required flag(s) "environment" not set
+Usage:
+  confluent flink artifact list [flags]
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
+
diff --git a/test/fixtures/output/flink/artifact/list-non-exist-environment-failure.golden b/test/fixtures/output/flink/artifact/list-non-exist-environment-failure.golden
new file mode 100644
index 0000000000..057f9bb06a
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/list-non-exist-environment-failure.golden
@@ -0,0 +1 @@
+Error: failed to list artifacts in the environment "non-exist": Environment not found
diff --git a/test/fixtures/output/flink/artifact/list-success-json.golden b/test/fixtures/output/flink/artifact/list-success-json.golden
new file mode 100644
index 0000000000..17157bffef
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/list-success-json.golden
@@ -0,0 +1,40 @@
+[
+  {
+    "apiVersion": "cmf.confluent.io/v1",
+    "kind": "Artifact",
+    "metadata": {
+      "name": "test-artifact-1",
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "uid": "11111111-1111-1111-1111-111111111111"
+    },
+    "spec": {},
+    "status": {
+      "version": 1,
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "path": "artifacts/test-artifact-1/1",
+      "size": 1024,
+      "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+      "phase": "READY"
+    }
+  },
+  {
+    "apiVersion": "cmf.confluent.io/v1",
+    "kind": "Artifact",
+    "metadata": {
+      "name": "test-artifact-2",
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "uid": "11111111-1111-1111-1111-111111111111"
+    },
+    "spec": {},
+    "status": {
+      "version": 3,
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "path": "artifacts/test-artifact-2/3",
+      "size": 1024,
+      "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+      "phase": "READY"
+    }
+  }
+]
diff --git a/test/fixtures/output/flink/artifact/list-success-yaml.golden b/test/fixtures/output/flink/artifact/list-success-yaml.golden
new file mode 100644
index 0000000000..54a0adec9a
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/list-success-yaml.golden
@@ -0,0 +1,30 @@
+- apiVersion: cmf.confluent.io/v1
+  kind: Artifact
+  metadata:
+    name: test-artifact-1
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+  spec: {}
+  status:
+    version: 1
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact-1/1
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
+- apiVersion: cmf.confluent.io/v1
+  kind: Artifact
+  metadata:
+    name: test-artifact-2
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+  spec: {}
+  status:
+    version: 3
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact-2/3
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/list-success.golden b/test/fixtures/output/flink/artifact/list-success.golden
new file mode 100644
index 0000000000..0c027ac6d7
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/list-success.golden
@@ -0,0 +1,4 @@
+       Name       | Version | Phase | Size |         Creation Time          
+------------------+---------+-------+------+--------------------------------
+  test-artifact-1 | 1       | READY | 1024 | 2025-03-12 23:42:00 +0000 UTC  
+  test-artifact-2 | 3       | READY | 1024 | 2025-03-12 23:42:00 +0000 UTC  
diff --git a/test/fixtures/output/flink/artifact/update-help-onprem.golden b/test/fixtures/output/flink/artifact/update-help-onprem.golden
new file mode 100644
index 0000000000..42a98bd88a
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/update-help-onprem.golden
@@ -0,0 +1,23 @@
+Update the labels of a Flink artifact in Confluent Platform without uploading new content.
+
+Usage:
+  confluent flink artifact update <name> [flags]
+
+Examples:
+Replace the labels of Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact update my-artifact --label owner=team-a,tier=gold --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --label strings                       A comma-separated list of "key=value" label pairs. Provide the complete set of labels to apply; omit the flag to leave existing labels unchanged.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/update-metadata-only-json.golden b/test/fixtures/output/flink/artifact/update-metadata-only-json.golden
new file mode 100644
index 0000000000..48a6586d07
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/update-metadata-only-json.golden
@@ -0,0 +1,19 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111"
+  },
+  "spec": {},
+  "status": {
+    "version": 1,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/1",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/update-metadata-only-success.golden b/test/fixtures/output/flink/artifact/update-metadata-only-success.golden
new file mode 100644
index 0000000000..f639323b05
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/update-metadata-only-success.golden
@@ -0,0 +1,7 @@
++---------------+-------------------------------+
+| Name          | test-artifact                 |
+| Version       | 1                             |
+| Phase         | READY                         |
+| Size          | 1024                          |
+| Creation Time | 2025-03-12 23:42:00 +0000 UTC |
++---------------+-------------------------------+
diff --git a/test/fixtures/output/flink/artifact/update-non-exist-failure.golden b/test/fixtures/output/flink/artifact/update-non-exist-failure.golden
new file mode 100644
index 0000000000..24dda7acf8
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/update-non-exist-failure.golden
@@ -0,0 +1 @@
+Error: failed to update artifact "invalid-artifact" in the environment "test-env": The artifact name is invalid
diff --git a/test/fixtures/output/flink/artifact/update-success-json.golden b/test/fixtures/output/flink/artifact/update-success-json.golden
new file mode 100644
index 0000000000..5a6140cc68
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/update-success-json.golden
@@ -0,0 +1,23 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111",
+    "labels": {
+      "owner": "team-a",
+      "tier": "gold"
+    }
+  },
+  "spec": {},
+  "status": {
+    "version": 1,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/1",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/update-success-yaml.golden b/test/fixtures/output/flink/artifact/update-success-yaml.golden
new file mode 100644
index 0000000000..c14cb37499
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/update-success-yaml.golden
@@ -0,0 +1,18 @@
+apiVersion: cmf.confluent.io/v1
+kind: Artifact
+metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+    labels:
+        owner: team-a
+        tier: gold
+spec: {}
+status:
+    version: 1
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/1
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/version/create-help-onprem.golden b/test/fixtures/output/flink/artifact/version/create-help-onprem.golden
new file mode 100644
index 0000000000..f2223456c8
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/create-help-onprem.golden
@@ -0,0 +1,23 @@
+Upload a new version of a Flink artifact in Confluent Platform. If the uploaded content is identical to the latest version, the artifact is left unchanged and no new version is created.
+
+Usage:
+  confluent flink artifact version create <name> [flags]
+
+Examples:
+Upload a new version of Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact version create my-artifact --artifact-file artifact-v2.jar --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --artifact-file string                REQUIRED: Path to the Flink artifact JAR or ZIP file.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/version/create-invalid-extension-failure.golden b/test/fixtures/output/flink/artifact/version/create-invalid-extension-failure.golden
new file mode 100644
index 0000000000..88385be0f5
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/create-invalid-extension-failure.golden
@@ -0,0 +1 @@
+Error: only extensions allowed for `--artifact-file` are "jar" and "zip"
diff --git a/test/fixtures/output/flink/artifact/version/create-non-exist-failure.golden b/test/fixtures/output/flink/artifact/version/create-non-exist-failure.golden
new file mode 100644
index 0000000000..bb755b7cba
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/create-non-exist-failure.golden
@@ -0,0 +1 @@
+Error: failed to update artifact "non-exist-artifact" in the environment "test-env": The artifact name is invalid
diff --git a/test/fixtures/output/flink/artifact/version/create-success-json.golden b/test/fixtures/output/flink/artifact/version/create-success-json.golden
new file mode 100644
index 0000000000..9e6606e138
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/create-success-json.golden
@@ -0,0 +1,19 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111"
+  },
+  "spec": {},
+  "status": {
+    "version": 2,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/2",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/version/create-success-yaml.golden b/test/fixtures/output/flink/artifact/version/create-success-yaml.golden
new file mode 100644
index 0000000000..f204c45e4d
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/create-success-yaml.golden
@@ -0,0 +1,15 @@
+apiVersion: cmf.confluent.io/v1
+kind: Artifact
+metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+spec: {}
+status:
+    version: 2
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/2
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/version/create-success.golden b/test/fixtures/output/flink/artifact/version/create-success.golden
new file mode 100644
index 0000000000..fa454432a6
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/create-success.golden
@@ -0,0 +1,7 @@
++---------------+----------------------------------+
+| Version       | 2                                |
+| Phase         | READY                            |
+| Size          | 1024                             |
+| Checksum      | d41d8cd98f00b204e9800998ecf8427e |
+| Creation Time | 2025-03-12 23:42:00 +0000 UTC    |
++---------------+----------------------------------+
diff --git a/test/fixtures/output/flink/artifact/version/delete-all-force.golden b/test/fixtures/output/flink/artifact/version/delete-all-force.golden
new file mode 100644
index 0000000000..be50f24560
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/delete-all-force.golden
@@ -0,0 +1 @@
+Deleted all versions of Flink artifact "delete-version-all".
diff --git a/test/fixtures/output/flink/artifact/version/delete-force.golden b/test/fixtures/output/flink/artifact/version/delete-force.golden
new file mode 100644
index 0000000000..e97d92693d
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/delete-force.golden
@@ -0,0 +1 @@
+Deleted version "2" of Flink artifact "delete-version-2".
diff --git a/test/fixtures/output/flink/artifact/version/delete-help-onprem.golden b/test/fixtures/output/flink/artifact/version/delete-help-onprem.golden
new file mode 100644
index 0000000000..e4af2fe423
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/delete-help-onprem.golden
@@ -0,0 +1,23 @@
+Delete a version of a Flink artifact in Confluent Platform. Deleting the last remaining version deletes the artifact.
+
+Usage:
+  confluent flink artifact version delete <name> [flags]
+
+Examples:
+Delete version 2 of Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact version delete my-artifact --version 2 --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --version string                      REQUIRED: Version of the artifact to delete, or "all" to delete every version.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+      --force                               Skip the deletion confirmation prompt.
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/version/delete-non-exist-failure.golden b/test/fixtures/output/flink/artifact/version/delete-non-exist-failure.golden
new file mode 100644
index 0000000000..dd5bbc9b7b
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/delete-non-exist-failure.golden
@@ -0,0 +1,5 @@
+Error: failed to describe artifact "non-exist-artifact" in the environment "test-env": The artifact name is invalid
+
+Suggestions:
+    List available Flink artifacts with `confluent flink artifact list`.
+    Check that CMF is running and accessible.
diff --git a/test/fixtures/output/flink/artifact/version/delete-successful.golden b/test/fixtures/output/flink/artifact/version/delete-successful.golden
new file mode 100644
index 0000000000..515cec3beb
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/delete-successful.golden
@@ -0,0 +1 @@
+Are you sure you want to delete version "2" of Flink artifact "delete-version-2"? (y/n): Deleted version "2" of Flink artifact "delete-version-2".
diff --git a/test/fixtures/output/flink/artifact/version/describe-help-onprem.golden b/test/fixtures/output/flink/artifact/version/describe-help-onprem.golden
new file mode 100644
index 0000000000..b5629f75d9
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/describe-help-onprem.golden
@@ -0,0 +1,23 @@
+Describe a Flink artifact version in Confluent Platform.
+
+Usage:
+  confluent flink artifact version describe <name> [flags]
+
+Examples:
+Describe version 2 of Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact version describe my-artifact --version 2 --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --version string                      REQUIRED: Version of the artifact to describe.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/version/describe-non-exist-failure.golden b/test/fixtures/output/flink/artifact/version/describe-non-exist-failure.golden
new file mode 100644
index 0000000000..520294b0b3
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/describe-non-exist-failure.golden
@@ -0,0 +1 @@
+Error: failed to describe artifact "invalid-artifact" in the environment "test-env": The artifact name is invalid
diff --git a/test/fixtures/output/flink/artifact/version/describe-success-json.golden b/test/fixtures/output/flink/artifact/version/describe-success-json.golden
new file mode 100644
index 0000000000..9e6606e138
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/describe-success-json.golden
@@ -0,0 +1,19 @@
+{
+  "apiVersion": "cmf.confluent.io/v1",
+  "kind": "Artifact",
+  "metadata": {
+    "name": "test-artifact",
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "uid": "11111111-1111-1111-1111-111111111111"
+  },
+  "spec": {},
+  "status": {
+    "version": 2,
+    "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+    "path": "artifacts/test-artifact/2",
+    "size": 1024,
+    "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+    "phase": "READY"
+  }
+}
diff --git a/test/fixtures/output/flink/artifact/version/describe-success-yaml.golden b/test/fixtures/output/flink/artifact/version/describe-success-yaml.golden
new file mode 100644
index 0000000000..f204c45e4d
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/describe-success-yaml.golden
@@ -0,0 +1,15 @@
+apiVersion: cmf.confluent.io/v1
+kind: Artifact
+metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+spec: {}
+status:
+    version: 2
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/2
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/version/describe-success.golden b/test/fixtures/output/flink/artifact/version/describe-success.golden
new file mode 100644
index 0000000000..fa454432a6
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/describe-success.golden
@@ -0,0 +1,7 @@
++---------------+----------------------------------+
+| Version       | 2                                |
+| Phase         | READY                            |
+| Size          | 1024                             |
+| Checksum      | d41d8cd98f00b204e9800998ecf8427e |
+| Creation Time | 2025-03-12 23:42:00 +0000 UTC    |
++---------------+----------------------------------+
diff --git a/test/fixtures/output/flink/artifact/version/download-help-onprem.golden b/test/fixtures/output/flink/artifact/version/download-help-onprem.golden
new file mode 100644
index 0000000000..409644070d
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/download-help-onprem.golden
@@ -0,0 +1,23 @@
+Download the binary content of a Flink artifact in Confluent Platform. Defaults to the latest version unless `--version` is specified.
+
+Usage:
+  confluent flink artifact version download <name> [flags]
+
+Examples:
+Download the latest version of Flink artifact "my-artifact" in the environment "my-environment".
+
+  $ confluent flink artifact version download my-artifact --output-file my-artifact.jar --environment my-environment
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --output-file string                  REQUIRED: Path to write the downloaded artifact file.
+      --version string                      Version of the artifact to download. Defaults to the latest version.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/version/download-success.golden b/test/fixtures/output/flink/artifact/version/download-success.golden
new file mode 100644
index 0000000000..359952d4e2
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/download-success.golden
@@ -0,0 +1 @@
+Downloaded Flink artifact "test-artifact" to "downloaded-artifact.jar".
diff --git a/test/fixtures/output/flink/artifact/version/download-version-success.golden b/test/fixtures/output/flink/artifact/version/download-version-success.golden
new file mode 100644
index 0000000000..359952d4e2
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/download-version-success.golden
@@ -0,0 +1 @@
+Downloaded Flink artifact "test-artifact" to "downloaded-artifact.jar".
diff --git a/test/fixtures/output/flink/artifact/version/help-onprem.golden b/test/fixtures/output/flink/artifact/version/help-onprem.golden
new file mode 100644
index 0000000000..1673d0a00f
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/help-onprem.golden
@@ -0,0 +1,18 @@
+Manage versions of a Flink artifact in Confluent Platform.
+
+Usage:
+  confluent flink artifact version [command]
+
+Available Commands:
+  create      Upload a new Flink artifact version in Confluent Platform.
+  delete      Delete a version of a Flink artifact in Confluent Platform.
+  describe    Describe a Flink artifact version in Confluent Platform.
+  download    Download a Flink artifact's content in Confluent Platform.
+  list        List the versions of a Flink artifact in Confluent Platform.
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
+
+Use "confluent flink artifact version [command] --help" for more information about a command.
diff --git a/test/fixtures/output/flink/artifact/version/list-help-onprem.golden b/test/fixtures/output/flink/artifact/version/list-help-onprem.golden
new file mode 100644
index 0000000000..7ee02c8df8
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/list-help-onprem.golden
@@ -0,0 +1,17 @@
+List the versions of a Flink artifact in Confluent Platform.
+
+Usage:
+  confluent flink artifact version list <name> [flags]
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
diff --git a/test/fixtures/output/flink/artifact/version/list-missing-env-flag-failure.golden b/test/fixtures/output/flink/artifact/version/list-missing-env-flag-failure.golden
new file mode 100644
index 0000000000..841f5574ad
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/list-missing-env-flag-failure.golden
@@ -0,0 +1,17 @@
+Error: required flag(s) "environment" not set
+Usage:
+  confluent flink artifact version list <name> [flags]
+
+Flags:
+      --environment string                  REQUIRED: Name of the Flink environment.
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
+
diff --git a/test/fixtures/output/flink/artifact/version/list-non-exist-failure.golden b/test/fixtures/output/flink/artifact/version/list-non-exist-failure.golden
new file mode 100644
index 0000000000..1178dd93e5
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/list-non-exist-failure.golden
@@ -0,0 +1 @@
+Error: failed to list versions of artifact "non-exist-artifact" in the environment "test-env": The artifact name is invalid
diff --git a/test/fixtures/output/flink/artifact/version/list-success-json.golden b/test/fixtures/output/flink/artifact/version/list-success-json.golden
new file mode 100644
index 0000000000..7292ba5d77
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/list-success-json.golden
@@ -0,0 +1,59 @@
+[
+  {
+    "apiVersion": "cmf.confluent.io/v1",
+    "kind": "Artifact",
+    "metadata": {
+      "name": "test-artifact",
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "uid": "11111111-1111-1111-1111-111111111111"
+    },
+    "spec": {},
+    "status": {
+      "version": 3,
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "path": "artifacts/test-artifact/3",
+      "size": 1024,
+      "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+      "phase": "READY"
+    }
+  },
+  {
+    "apiVersion": "cmf.confluent.io/v1",
+    "kind": "Artifact",
+    "metadata": {
+      "name": "test-artifact",
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "uid": "11111111-1111-1111-1111-111111111111"
+    },
+    "spec": {},
+    "status": {
+      "version": 2,
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "path": "artifacts/test-artifact/2",
+      "size": 1024,
+      "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+      "phase": "READY"
+    }
+  },
+  {
+    "apiVersion": "cmf.confluent.io/v1",
+    "kind": "Artifact",
+    "metadata": {
+      "name": "test-artifact",
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "updateTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "uid": "11111111-1111-1111-1111-111111111111"
+    },
+    "spec": {},
+    "status": {
+      "version": 1,
+      "creationTimestamp": "2025-03-12 23:42:00 +0000 UTC",
+      "path": "artifacts/test-artifact/1",
+      "size": 1024,
+      "checksum": "d41d8cd98f00b204e9800998ecf8427e",
+      "phase": "READY"
+    }
+  }
+]
diff --git a/test/fixtures/output/flink/artifact/version/list-success-yaml.golden b/test/fixtures/output/flink/artifact/version/list-success-yaml.golden
new file mode 100644
index 0000000000..5def10ef2e
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/list-success-yaml.golden
@@ -0,0 +1,45 @@
+- apiVersion: cmf.confluent.io/v1
+  kind: Artifact
+  metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+  spec: {}
+  status:
+    version: 3
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/3
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
+- apiVersion: cmf.confluent.io/v1
+  kind: Artifact
+  metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+  spec: {}
+  status:
+    version: 2
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/2
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
+- apiVersion: cmf.confluent.io/v1
+  kind: Artifact
+  metadata:
+    name: test-artifact
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    updateTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    uid: 11111111-1111-1111-1111-111111111111
+  spec: {}
+  status:
+    version: 1
+    creationTimestamp: 2025-03-12 23:42:00 +0000 UTC
+    path: artifacts/test-artifact/1
+    size: 1024
+    checksum: d41d8cd98f00b204e9800998ecf8427e
+    phase: READY
diff --git a/test/fixtures/output/flink/artifact/version/list-success.golden b/test/fixtures/output/flink/artifact/version/list-success.golden
new file mode 100644
index 0000000000..0318d31cf3
--- /dev/null
+++ b/test/fixtures/output/flink/artifact/version/list-success.golden
@@ -0,0 +1,5 @@
+  Version | Phase | Size |             Checksum             |         Creation Time          
+----------+-------+------+----------------------------------+--------------------------------
+  3       | READY | 1024 | d41d8cd98f00b204e9800998ecf8427e | 2025-03-12 23:42:00 +0000 UTC  
+  2       | READY | 1024 | d41d8cd98f00b204e9800998ecf8427e | 2025-03-12 23:42:00 +0000 UTC  
+  1       | READY | 1024 | d41d8cd98f00b204e9800998ecf8427e | 2025-03-12 23:42:00 +0000 UTC  
diff --git a/test/fixtures/output/flink/help-onprem.golden b/test/fixtures/output/flink/help-onprem.golden
index 0c5b861653..ab401da490 100644
--- a/test/fixtures/output/flink/help-onprem.golden
+++ b/test/fixtures/output/flink/help-onprem.golden
@@ -5,6 +5,7 @@ Usage:
 
 Available Commands:
   application         Manage Flink applications.
+  artifact            Manage Flink UDF artifacts.
   catalog             Manage Flink catalogs in Confluent Platform.
   compute-pool        Manage Flink compute pools.
   detached-savepoint  Manage Flink detached savepoints.
diff --git a/test/flink_onprem_test.go b/test/flink_onprem_test.go
index c98ca8c4b6..1f5fb1c3d5 100644
--- a/test/flink_onprem_test.go
+++ b/test/flink_onprem_test.go
@@ -2,6 +2,7 @@ package test
 
 import (
 	"os"
+	"testing"
 
 	"github.com/stretchr/testify/require"
 
@@ -409,6 +410,154 @@ func (s *CLITestSuite) TestFlinkComputePoolListOnPrem() {
 	runIntegrationTestsWithMultipleAuth(s, tests)
 }
 
+func (s *CLITestSuite) TestFlinkArtifactCreateOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.jar --environment test-env", fixture: "flink/artifact/create-success.golden"},
+		{args: "flink artifact create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.jar --environment test-env --output json", fixture: "flink/artifact/create-success-json.golden"},
+		{args: "flink artifact create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.jar --environment test-env --output yaml", fixture: "flink/artifact/create-success-yaml.golden"},
+		{args: "flink artifact create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.jar --label owner=team-a,tier=gold --environment test-env --output json", fixture: "flink/artifact/create-with-labels-json.golden"},
+		// failure
+		{args: "flink artifact create existing-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.jar --environment test-env", fixture: "flink/artifact/create-existing-failure.golden", exitCode: 1},
+		{args: "flink artifact create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.txt --environment test-env", fixture: "flink/artifact/create-invalid-extension-failure.golden", exitCode: 1},
+		{args: "flink artifact create test-artifact --environment test-env", fixture: "flink/artifact/create-missing-file-flag-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactListOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact list --environment test-env", fixture: "flink/artifact/list-success.golden"},
+		{args: "flink artifact list --environment test-env --output json", fixture: "flink/artifact/list-success-json.golden"},
+		{args: "flink artifact list --environment test-env --output yaml", fixture: "flink/artifact/list-success-yaml.golden"},
+		// failure
+		{args: "flink artifact list", fixture: "flink/artifact/list-missing-env-flag-failure.golden", exitCode: 1},
+		{args: "flink artifact list --environment non-exist", fixture: "flink/artifact/list-non-exist-environment-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactDescribeOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact describe test-artifact --environment test-env", fixture: "flink/artifact/describe-success.golden"},
+		{args: "flink artifact describe test-artifact --environment test-env --output json", fixture: "flink/artifact/describe-success-json.golden"},
+		{args: "flink artifact describe test-artifact --environment test-env --output yaml", fixture: "flink/artifact/describe-success-yaml.golden"},
+		// labels + annotations are surfaced in the human describe table (Labels/Annotations rows)
+		{args: "flink artifact describe labeled-artifact --environment test-env", fixture: "flink/artifact/describe-labeled.golden"},
+		{args: "flink artifact describe labeled-artifact --environment test-env --output json", fixture: "flink/artifact/describe-labeled-json.golden"},
+		// failure
+		{args: "flink artifact describe invalid-artifact --environment test-env", fixture: "flink/artifact/describe-non-exist-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactUpdateOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact update test-artifact --label owner=team-a,tier=gold --environment test-env --output json", fixture: "flink/artifact/update-success-json.golden"},
+		{args: "flink artifact update test-artifact --label owner=team-a,tier=gold --environment test-env --output yaml", fixture: "flink/artifact/update-success-yaml.golden"},
+		// metadata-only update: no --label, so existing labels are preserved (the label-preserving branch) and output is human-readable
+		{args: "flink artifact update test-artifact --environment test-env", fixture: "flink/artifact/update-metadata-only-success.golden"},
+		// metadata-only update in JSON: the response carries no "labels" field, confirming the CLI omitted labels (preserve) rather than sending an empty object (clear)
+		{args: "flink artifact update test-artifact --environment test-env --output json", fixture: "flink/artifact/update-metadata-only-json.golden"},
+		// failure
+		{args: "flink artifact update invalid-artifact --label owner=team-a --environment test-env", fixture: "flink/artifact/update-non-exist-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactDeleteOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact delete test-artifact --environment test-env", input: "y\n", fixture: "flink/artifact/delete-single-successful.golden"},
+		{args: "flink artifact delete test-artifact-1 test-artifact-2 --environment test-env", input: "y\n", fixture: "flink/artifact/delete-multiple-successful.golden"},
+		{args: "flink artifact delete test-artifact --force --environment test-env", fixture: "flink/artifact/delete-single-force.golden"},
+		// failure
+		{args: "flink artifact delete non-exist-artifact --environment test-env", input: "y\n", fixture: "flink/artifact/delete-non-exist-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactVersionCreateOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact version create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact-v2.jar --environment test-env", fixture: "flink/artifact/version/create-success.golden"},
+		{args: "flink artifact version create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact-v2.jar --environment test-env --output json", fixture: "flink/artifact/version/create-success-json.golden"},
+		{args: "flink artifact version create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact-v2.jar --environment test-env --output yaml", fixture: "flink/artifact/version/create-success-yaml.golden"},
+		// failure
+		{args: "flink artifact version create non-exist-artifact --artifact-file test/fixtures/input/flink/artifact/artifact-v2.jar --environment test-env", fixture: "flink/artifact/version/create-non-exist-failure.golden", exitCode: 1},
+		{args: "flink artifact version create test-artifact --artifact-file test/fixtures/input/flink/artifact/artifact.txt --environment test-env", fixture: "flink/artifact/version/create-invalid-extension-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactVersionListOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact version list test-artifact --environment test-env", fixture: "flink/artifact/version/list-success.golden"},
+		{args: "flink artifact version list test-artifact --environment test-env --output json", fixture: "flink/artifact/version/list-success-json.golden"},
+		{args: "flink artifact version list test-artifact --environment test-env --output yaml", fixture: "flink/artifact/version/list-success-yaml.golden"},
+		// failure
+		{args: "flink artifact version list non-exist-artifact --environment test-env", fixture: "flink/artifact/version/list-non-exist-failure.golden", exitCode: 1},
+		{args: "flink artifact version list test-artifact", fixture: "flink/artifact/version/list-missing-env-flag-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactVersionDescribeOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact version describe test-artifact --version 2 --environment test-env", fixture: "flink/artifact/version/describe-success.golden"},
+		{args: "flink artifact version describe test-artifact --version 2 --environment test-env --output json", fixture: "flink/artifact/version/describe-success-json.golden"},
+		{args: "flink artifact version describe test-artifact --version 2 --environment test-env --output yaml", fixture: "flink/artifact/version/describe-success-yaml.golden"},
+		// failure
+		{args: "flink artifact version describe invalid-artifact --version 2 --environment test-env", fixture: "flink/artifact/version/describe-non-exist-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactVersionDeleteOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink artifact version delete delete-version-2 --version 2 --environment test-env", input: "y\n", fixture: "flink/artifact/version/delete-successful.golden"},
+		{args: "flink artifact version delete delete-version-2 --version 2 --force --environment test-env", fixture: "flink/artifact/version/delete-force.golden"},
+		{args: "flink artifact version delete delete-version-all --version all --force --environment test-env", fixture: "flink/artifact/version/delete-all-force.golden"},
+		// failure
+		{args: "flink artifact version delete non-exist-artifact --version 2 --environment test-env", input: "y\n", fixture: "flink/artifact/version/delete-non-exist-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
+func (s *CLITestSuite) TestFlinkArtifactVersionDownloadOnPrem() {
+	defer os.Remove("downloaded-artifact.jar")
+
+	// verifyDownloadedContent asserts the downloaded bytes match the mock body. The output file is reused across cases
+	// and auth passes, which also exercises that download overwrites an existing file (no --force needed).
+	verifyDownloadedContent := func(t *testing.T) {
+		content, err := os.ReadFile("downloaded-artifact.jar")
+		require.NoError(t, err)
+		require.Equal(t, "dummy artifact content", string(content))
+	}
+
+	tests := []CLITest{
+		{args: "flink artifact version download test-artifact --output-file downloaded-artifact.jar --environment test-env", fixture: "flink/artifact/version/download-success.golden", wantFunc: verifyDownloadedContent},
+		{args: "flink artifact version download test-artifact --version 1 --output-file downloaded-artifact.jar --environment test-env", fixture: "flink/artifact/version/download-version-success.golden", wantFunc: verifyDownloadedContent},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
 func (s *CLITestSuite) TestFlinkCatalogCreateOnPrem() {
 	tests := []CLITest{
 		// success
diff --git a/test/test-server/flink_onprem_handler.go b/test/test-server/flink_onprem_handler.go
index d58b006ee8..71480adfc7 100644
--- a/test/test-server/flink_onprem_handler.go
+++ b/test/test-server/flink_onprem_handler.go
@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -1861,3 +1862,233 @@ func handleCmfSecret(t *testing.T) http.HandlerFunc {
 		}
 	}
 }
+
+// assertArtifactLabelContract enforces the CLI's label contract on the submitted artifact JSON: the "labels" field must
+// be omitted entirely to preserve existing labels (nil map / --label not passed), and the CLI must never send an empty
+// object, which CMF interprets as "clear all labels". When present, labels must carry the caller's entries.
+func assertArtifactLabelContract(t *testing.T, rawArtifact string) {
+	var probe struct {
+		Metadata struct {
+			Labels *map[string]string `json:"labels"`
+		} `json:"metadata"`
+	}
+	require.NoError(t, json.Unmarshal([]byte(rawArtifact), &probe))
+	if probe.Metadata.Labels != nil {
+		require.NotEmpty(t, *probe.Metadata.Labels, "CLI must omit the labels field to preserve labels, never send an empty object")
+	}
+}
+
+// createArtifactObject builds a fully-populated Artifact for the given name and version with deterministic field values.
+func createArtifactObject(name string, version int32) cmfsdk.Artifact {
+	timestamp := time.Date(2025, time.March, 12, 23, 42, 0, 0, time.UTC).String()
+	return cmfsdk.Artifact{
+		ApiVersion: "cmf.confluent.io/v1",
+		Kind:       "Artifact",
+		Metadata: cmfsdk.ArtifactMetadata{
+			Name:              name,
+			Uid:               cmfsdk.PtrString("11111111-1111-1111-1111-111111111111"),
+			CreationTimestamp: &timestamp,
+			UpdateTimestamp:   &timestamp,
+		},
+		Spec: map[string]interface{}{},
+		Status: &cmfsdk.ArtifactStatus{
+			Version:           cmfsdk.PtrInt32(version),
+			CreationTimestamp: &timestamp,
+			Path:              cmfsdk.PtrString(fmt.Sprintf("artifacts/%s/%d", name, version)),
+			Size:              cmfsdk.PtrInt64(1024),
+			Checksum:          cmfsdk.PtrString("d41d8cd98f00b204e9800998ecf8427e"),
+			Phase:             cmfsdk.PtrString("READY"),
+		},
+	}
+}
+
+// buildArtifactResponse echoes the submitted name, labels, and annotations while attaching a deterministic server-side status.
+func buildArtifactResponse(submitted cmfsdk.Artifact, version int32) cmfsdk.Artifact {
+	artifact := createArtifactObject(submitted.Metadata.Name, version)
+	if submitted.Metadata.Labels != nil {
+		artifact.Metadata.Labels = submitted.Metadata.Labels
+	}
+	if submitted.Metadata.Annotations != nil {
+		artifact.Metadata.Annotations = submitted.Metadata.Annotations
+	}
+	return artifact
+}
+
+// Handler for "/cmf/api/v1/environments/{environment}/artifacts"
+// Used by list and create Flink artifacts.
+func handleCmfArtifacts(t *testing.T) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		handleLoginType(t, r)
+
+		if mux.Vars(r)["environment"] == "non-exist" {
+			http.Error(w, "Environment not found", http.StatusNotFound)
+			return
+		}
+
+		switch r.Method {
+		case http.MethodGet:
+			artifactsPage := cmfsdk.ArtifactsPage{}
+			if r.URL.Query().Get("page") == "0" {
+				artifactsPage.SetItems([]cmfsdk.Artifact{
+					createArtifactObject("test-artifact-1", 1),
+					createArtifactObject("test-artifact-2", 3),
+				})
+			}
+			err := json.NewEncoder(w).Encode(artifactsPage)
+			require.NoError(t, err)
+			return
+		case http.MethodPost:
+			require.NoError(t, r.ParseMultipartForm(32<<20))
+			rawArtifact := r.FormValue("artifact")
+			assertArtifactLabelContract(t, rawArtifact)
+			var artifact cmfsdk.Artifact
+			require.NoError(t, json.Unmarshal([]byte(rawArtifact), &artifact))
+
+			if artifact.Metadata.Name == "existing-artifact" {
+				http.Error(w, "The artifact name already exists, please try with another artifact name", http.StatusConflict)
+				return
+			}
+
+			w.WriteHeader(http.StatusCreated)
+			err := json.NewEncoder(w).Encode(buildArtifactResponse(artifact, 1))
+			require.NoError(t, err)
+			return
+		default:
+			require.Fail(t, fmt.Sprintf("Unexpected method %s", r.Method))
+		}
+	}
+}
+
+// Handler for "/cmf/api/v1/environments/{environment}/artifacts/{artifactName}"
+// Used by describe, update (metadata-only), version create (with file), and delete a Flink artifact.
+func handleCmfArtifact(t *testing.T) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		handleLoginType(t, r)
+
+		vars := mux.Vars(r)
+		artifactName := vars["artifactName"]
+
+		if vars["environment"] == "non-exist" {
+			http.Error(w, "Environment not found", http.StatusNotFound)
+			return
+		}
+		if artifactName == "invalid-artifact" || artifactName == "non-exist-artifact" {
+			http.Error(w, "The artifact name is invalid", http.StatusNotFound)
+			return
+		}
+
+		switch r.Method {
+		case http.MethodGet:
+			version := int32(3)
+			if requested := r.URL.Query().Get("version"); requested != "" {
+				parsed, err := strconv.Atoi(requested)
+				if err != nil {
+					http.Error(w, "invalid version", http.StatusBadRequest)
+					return
+				}
+				version = int32(parsed)
+			}
+			artifact := createArtifactObject(artifactName, version)
+			// Exercise the human describe view's Labels/Annotations rows for this fixture name.
+			if artifactName == "labeled-artifact" {
+				artifact.Metadata.Labels = map[string]string{"owner": "team-a", "tier": "gold"}
+				artifact.Metadata.Annotations = map[string]string{"note": "managed-externally"}
+			}
+			err := json.NewEncoder(w).Encode(artifact)
+			require.NoError(t, err)
+			return
+		case http.MethodPut:
+			require.NoError(t, r.ParseMultipartForm(32<<20))
+			rawArtifact := r.FormValue("artifact")
+			assertArtifactLabelContract(t, rawArtifact)
+			var artifact cmfsdk.Artifact
+			require.NoError(t, json.Unmarshal([]byte(rawArtifact), &artifact))
+
+			// A "file" part indicates a new version upload; its absence is a metadata-only update.
+			version := int32(1)
+			if r.MultipartForm != nil && len(r.MultipartForm.File["file"]) > 0 {
+				version = 2
+			}
+
+			err := json.NewEncoder(w).Encode(buildArtifactResponse(artifact, version))
+			require.NoError(t, err)
+			return
+		case http.MethodDelete:
+			// Assert the CLI forwarded the --version flag verbatim: a version-scoped delete sends ?version=<value>,
+			// a whole-artifact delete sends none. (These fixture names are used only by the version-delete tests.)
+			version := r.URL.Query().Get("version")
+			switch artifactName {
+			case "delete-version-2":
+				require.Equal(t, "2", version)
+			case "delete-version-all":
+				require.Equal(t, "all", version)
+			default:
+				require.Empty(t, version, "whole-artifact delete must not send a version query param")
+			}
+			w.WriteHeader(http.StatusNoContent)
+			return
+		default:
+			require.Fail(t, fmt.Sprintf("Unexpected method %s", r.Method))
+		}
+	}
+}
+
+// Handler for "/cmf/api/v1/environments/{environment}/artifacts/{artifactName}/versions"
+// Used by list the versions of a Flink artifact.
+func handleCmfArtifactVersions(t *testing.T) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		handleLoginType(t, r)
+
+		vars := mux.Vars(r)
+		artifactName := vars["artifactName"]
+
+		if artifactName == "invalid-artifact" || artifactName == "non-exist-artifact" {
+			http.Error(w, "The artifact name is invalid", http.StatusNotFound)
+			return
+		}
+
+		switch r.Method {
+		case http.MethodGet:
+			versionsPage := cmfsdk.ArtifactsPage{}
+			if r.URL.Query().Get("page") == "0" {
+				versionsPage.SetItems([]cmfsdk.Artifact{
+					createArtifactObject(artifactName, 3),
+					createArtifactObject(artifactName, 2),
+					createArtifactObject(artifactName, 1),
+				})
+			}
+			err := json.NewEncoder(w).Encode(versionsPage)
+			require.NoError(t, err)
+			return
+		default:
+			require.Fail(t, fmt.Sprintf("Unexpected method %s", r.Method))
+		}
+	}
+}
+
+// Handler for "/cmf/api/v1/environments/{environment}/artifacts/{artifactName}/content"
+// Used by download the content of a Flink artifact.
+func handleCmfArtifactContent(t *testing.T) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		handleLoginType(t, r)
+
+		vars := mux.Vars(r)
+		artifactName := vars["artifactName"]
+
+		if artifactName == "invalid-artifact" || artifactName == "non-exist-artifact" {
+			http.Error(w, "The artifact name is invalid", http.StatusNotFound)
+			return
+		}
+
+		switch r.Method {
+		case http.MethodGet:
+			w.Header().Set("Content-Type", "application/octet-stream")
+			w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s.jar"`, artifactName))
+			_, err := w.Write([]byte("dummy artifact content"))
+			require.NoError(t, err)
+			return
+		default:
+			require.Fail(t, fmt.Sprintf("Unexpected method %s", r.Method))
+		}
+	}
+}
diff --git a/test/test-server/flink_onprem_router.go b/test/test-server/flink_onprem_router.go
index 49360810f6..84dc4b0330 100644
--- a/test/test-server/flink_onprem_router.go
+++ b/test/test-server/flink_onprem_router.go
@@ -11,6 +11,10 @@ var flinkRoutes = []route{
 	{"/cmf/api/v1/catalogs/kafka/{catName}", handleCmfCatalog},
 	{"/cmf/api/v1/catalogs/kafka/{catName}/databases", handleCmfCatalogDatabases},
 	{"/cmf/api/v1/catalogs/kafka/{catName}/databases/{dbName}", handleCmfCatalogDatabase},
+	{"/cmf/api/v1/environments/{environment}/artifacts", handleCmfArtifacts},
+	{"/cmf/api/v1/environments/{environment}/artifacts/{artifactName}", handleCmfArtifact},
+	{"/cmf/api/v1/environments/{environment}/artifacts/{artifactName}/versions", handleCmfArtifactVersions},
+	{"/cmf/api/v1/environments/{environment}/artifacts/{artifactName}/content", handleCmfArtifactContent},
 	{"/cmf/api/v1/environments/{environment}/applications", handleCmfApplications},
 	{"/cmf/api/v1/environments/{environment}/applications/{application}", handleCmfApplication},
 	{"/cmf/api/v1/environments/{environment}/applications/{application}/instances", handleCmfApplicationInstances},