diff --git a/cmd/products.gen.go b/cmd/products.gen.go
index 8a02dabd04..ddb7c70baf 100644
--- a/cmd/products.gen.go
+++ b/cmd/products.gen.go
@@ -18,6 +18,7 @@ import (
 	"github.com/ucloud/ucloud-cli/products/udpn"
 	"github.com/ucloud/ucloud-cli/products/ufs"
 	"github.com/ucloud/ucloud-cli/products/uhost"
+	"github.com/ucloud/ucloud-cli/products/ukafka"
 	"github.com/ucloud/ucloud-cli/products/ulb"
 	"github.com/ucloud/ucloud-cli/products/umodelverse"
 	"github.com/ucloud/ucloud-cli/products/upfs"
@@ -45,6 +46,7 @@ func registeredProducts() []cli.Product {
 		udpn.New(),
 		ufs.New(),
 		uhost.New(),
+		ukafka.New(),
 		ulb.New(),
 		umodelverse.New(),
 		upfs.New(),
diff --git a/products/ukafka/internal/ukafka/add_node.go b/products/ukafka/internal/ukafka/add_node.go
new file mode 100644
index 0000000000..d51062290f
--- /dev/null
+++ b/products/ukafka/internal/ukafka/add_node.go
@@ -0,0 +1,64 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+	sdk "github.com/ucloud/ucloud-sdk-go/ucloud"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newAddNode ucloud ukafka add-node
+func newAddNode(ctx *cli.Context) *cobra.Command {
+	var async *bool
+	var nodeCount *int
+	var nodeType *string
+	var instanceID *string
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewAddUKafkaInstanceNodeRequest()
+
+	cmd := &cobra.Command{
+		Use:   "add-node",
+		Short: "Add nodes to UKafka instance",
+		Long:  "Add nodes to UKafka instance",
+		Run: func(cmd *cobra.Command, args []string) {
+			w := ctx.ProgressWriter()
+			req.InstanceId = sdk.String(*instanceID)
+			req.NodeCount = sdk.String(fmt.Sprintf("%d", *nodeCount))
+			req.NodeType = sdk.String(*nodeType)
+
+			_, err := client.AddUKafkaInstanceNode(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			text := fmt.Sprintf("ukafka[%s] adding %d node(s)", *instanceID, *nodeCount)
+			if *async {
+				fmt.Fprintln(w, text)
+			} else {
+				ctx.PollerTo(w, describeUKafkaInstanceByID(ctx)).Spoll(*instanceID, text, []string{StateRunning, StateAbnormal})
+			}
+			ctx.EmitResult(cli.OpResultRow{ResourceID: *instanceID, Action: "add-node", Status: "Running"})
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	nodeCount = flags.Int("node-count", 1, "Required. Number of nodes to add")
+	nodeType = flags.String("node-type", "", "Required. Node type")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+	async = flags.Bool("async", false, "Optional. Do not wait for operation to finish")
+
+	cmd.MarkFlagRequired("ukafka-id")
+	cmd.MarkFlagRequired("node-type")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/appversion.go b/products/ukafka/internal/ukafka/appversion.go
new file mode 100644
index 0000000000..d4bd810165
--- /dev/null
+++ b/products/ukafka/internal/ukafka/appversion.go
@@ -0,0 +1,42 @@
+package ukafka
+
+import (
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newAppVersion ucloud ukafka app-version
+func newAppVersion(ctx *cli.Context) *cobra.Command {
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewListUKafkaFrameworkVersionRequest()
+	cmd := &cobra.Command{
+		Use:   "app-version",
+		Short: "List available Kafka versions",
+		Long:  "List available Kafka versions",
+		Run: func(cmd *cobra.Command, args []string) {
+			resp, err := client.ListUKafkaFrameworkVersion(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+			list := []VersionRow{}
+			for _, v := range resp.FrameworkVersions {
+				row := VersionRow{
+					Version: v.Version,
+					Label:   v.Label,
+				}
+				list = append(list, row)
+			}
+			ctx.PrintList(list)
+		},
+	}
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/check_topic.go b/products/ukafka/internal/ukafka/check_topic.go
new file mode 100644
index 0000000000..8ff559b799
--- /dev/null
+++ b/products/ukafka/internal/ukafka/check_topic.go
@@ -0,0 +1,95 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// IsUKafkaTopicNameExistResponse 自定义响应结构
+type IsUKafkaTopicNameExistResponse struct {
+	RetCode int    `json:"RetCode"`
+	Action  string `json:"Action"`
+	IsExist string `json:"IsExist"`
+	Message string `json:"Message"`
+}
+
+// newCheckTopic ucloud ukafka check-topic
+func newCheckTopic(ctx *cli.Context) *cobra.Command {
+	var instanceID *string
+	var topicName *string
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewIsUKafkaTopicNameExistRequest()
+
+	cmd := &cobra.Command{
+		Use:   "check-topic",
+		Short: "Check if a topic name exists in UKafka instance",
+		Long:  "Check if a topic name exists in UKafka instance",
+		Run: func(cmd *cobra.Command, args []string) {
+			// 使用 GenericInvoke 调用 API
+			genReq := client.Client.NewGenericRequest()
+			genReq.SetAction("IsUKafkaTopicNameExist")
+			genReq.SetRegion(*req.Region)
+			genReq.SetZone(*req.Zone)
+			if req.ProjectId != nil && *req.ProjectId != "" {
+				genReq.SetProjectId(*req.ProjectId)
+			}
+
+			payload := map[string]interface{}{
+				"ClusterInstanceId": *instanceID,
+				"TopicName":         *topicName,
+			}
+			genReq.SetPayload(payload)
+
+			genResp, err := client.Client.GenericInvoke(genReq)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			var resp IsUKafkaTopicNameExistResponse
+			if err := genResp.Unmarshal(&resp); err != nil {
+				ctx.HandleError(fmt.Errorf("parse response: %w", err))
+				return
+			}
+
+			if resp.RetCode != 0 {
+				ctx.HandleError(fmt.Errorf("API error: RetCode=%d, Message=%s", resp.RetCode, resp.Message))
+				return
+			}
+
+			// 输出结构化结果
+			rows := []cli.DescribeRow{
+				{Attribute: "InstanceID", Content: *instanceID},
+				{Attribute: "TopicName", Content: *topicName},
+				{Attribute: "Exists", Content: resp.IsExist},
+			}
+			ctx.PrintList(rows)
+
+			// 同时输出事件结果
+			ctx.EmitResult(cli.OpResultRow{
+				ResourceID: *instanceID,
+				Action:     "check-topic",
+				Status:     resp.IsExist,
+			})
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	topicName = flags.String("topic-name", "", "Required. Topic name to check")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+
+	cmd.MarkFlagRequired("ukafka-id")
+	cmd.MarkFlagRequired("topic-name")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/cmd.go b/products/ukafka/internal/ukafka/cmd.go
new file mode 100644
index 0000000000..a5ed0bc893
--- /dev/null
+++ b/products/ukafka/internal/ukafka/cmd.go
@@ -0,0 +1,30 @@
+package ukafka
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// NewCommand builds the `ukafka` root command and mounts the subcommands.
+func NewCommand(ctx *cli.Context) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "ukafka",
+		Short: "Manage UKafka instances",
+		Long:  "Manage UKafka instances",
+	}
+	cmd.AddCommand(newCreate(ctx))
+	cmd.AddCommand(newDelete(ctx))
+	cmd.AddCommand(newDescribe(ctx))
+	cmd.AddCommand(newList(ctx))
+	cmd.AddCommand(newNodeConf(ctx))
+	cmd.AddCommand(newAppVersion(ctx))
+	cmd.AddCommand(newAddNode(ctx))
+	cmd.AddCommand(newDescribeConsumer(ctx))
+	cmd.AddCommand(newCheckTopic(ctx))
+	cmd.AddCommand(newListConsumers(ctx))
+	cmd.AddCommand(newListTopics(ctx))
+	cmd.AddCommand(newModifyType(ctx))
+	cmd.AddCommand(newResizeDisk(ctx))
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/create.go b/products/ukafka/internal/ukafka/create.go
new file mode 100644
index 0000000000..387d2ce4f1
--- /dev/null
+++ b/products/ukafka/internal/ukafka/create.go
@@ -0,0 +1,72 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+	"github.com/ucloud/ucloud-cli/pkg/command"
+)
+
+// newCreate ucloud ukafka create
+func newCreate(ctx *cli.Context) *cobra.Command {
+	var async *bool
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewCreateUKafkaInstanceRequest()
+	cmd := &cobra.Command{
+		Use:   "create",
+		Short: "Create UKafka instance",
+		Long:  "Create UKafka instance",
+		Run: func(cmd *cobra.Command, args []string) {
+			w := ctx.ProgressWriter()
+			resp, err := client.CreateUKafkaInstance(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+			text := fmt.Sprintf("ukafka[%s] is creating", resp.InstanceId)
+			if *async {
+				fmt.Fprintln(w, text)
+			} else {
+				ctx.PollerTo(w, describeUKafkaInstanceByID(ctx)).Spoll(resp.InstanceId, text, []string{StateRunning, StateAbnormal})
+			}
+			ctx.EmitResult(cli.OpResultRow{ResourceID: resp.InstanceId, Action: "create", Status: "Creating"})
+		},
+	}
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	req.InstanceName = flags.String("name", "", "Required. Instance name")
+	req.FrameworkVersion = flags.String("kafka-version", "", "Required. Kafka version, e.g. 2.12-2.4.1")
+	req.NodeType = flags.String("node-type", "", "Required. Node type")
+	req.DiskSize = flags.Int("disk-size-gb", 0, "Required. Disk size in GB")
+	req.NodeCount = flags.Int("node-count", 3, "Optional. Node count, default 3")
+	req.LogRetentionHours = flags.String("log-retention-hours", "72", "Optional. Log retention hours (1-240), default 72")
+	req.VPCId = flags.String("vpc-id", "", "Optional. VPC ID")
+	req.SubnetId = flags.String("subnet-id", "", "Optional. Subnet ID")
+	req.BusinessId = flags.String("business-id", "", "Optional. Business group ID")
+	req.Quantity = flags.String("quantity", "1", "Optional. Instance quantity, default 1")
+	req.DiskControllerType = flags.String("disk-controller-type", "NONE", "Optional. Disk controller type: NONE or CLEAN")
+	req.DiskThreshold = flags.String("disk-threshold", "90", "Optional. Disk cleanup threshold (70-90), default 90")
+	req.IsSecurityEnabled = flags.String("enable-security", "false", "Optional. Enable security group: true or false")
+	async = flags.Bool("async", false, "Optional. Do not wait for creation to finish")
+
+	// Bind common params with Tab completion
+	// Note: UKafka SDK uses *string for ChargeType/Quantity (not *int),
+	// so we cannot use ctx.BindCommonParams which assumes standard types.
+	// Instead we bind region/zone/project-id individually with completion.
+	ctx.BindRegion(cmd, req)
+	ctx.BindZone(cmd, req)
+	ctx.BindProjectID(cmd, req)
+
+	command.SetFlagValues(cmd, "charge-type", "Month", "Year", "Dynamic")
+
+	cmd.MarkFlagRequired("name")
+	cmd.MarkFlagRequired("kafka-version")
+	cmd.MarkFlagRequired("node-type")
+	cmd.MarkFlagRequired("disk-size-gb")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/delete.go b/products/ukafka/internal/ukafka/delete.go
new file mode 100644
index 0000000000..d9f1eba566
--- /dev/null
+++ b/products/ukafka/internal/ukafka/delete.go
@@ -0,0 +1,59 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newDelete ucloud ukafka delete
+func newDelete(ctx *cli.Context) *cobra.Command {
+	var yes *bool
+	var instanceIDs *[]string
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewDeleteUKafkaInstanceRequest()
+	cmd := &cobra.Command{
+		Use:   "delete",
+		Short: "Delete UKafka instances",
+		Long:  "Delete UKafka instances",
+		Run: func(cmd *cobra.Command, args []string) {
+			ok, err := ctx.Confirm(*yes, "Are you sure to delete UKafka instance(s)?")
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+			if !ok {
+				return
+			}
+			w := ctx.ProgressWriter()
+			results := []cli.OpResultRow{}
+			for _, idName := range *instanceIDs {
+				id := ctx.PickResourceID(idName)
+				req.InstanceId = &id
+				_, err := client.DeleteUKafkaInstance(req)
+				if err != nil {
+					ctx.HandleError(err)
+					continue
+				}
+				fmt.Fprintf(w, "ukafka[%s] deleted\n", id)
+				results = append(results, cli.OpResultRow{ResourceID: id, Action: "delete", Status: "Deleted"})
+			}
+			ctx.EmitResult(results...)
+		},
+	}
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceIDs = flags.StringSlice("ukafka-id", nil, "Required. Instance ID(s) to delete")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+	yes = flags.BoolP("yes", "y", false, "Optional. Skip confirmation prompt")
+
+	cmd.MarkFlagRequired("ukafka-id")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/describe.go b/products/ukafka/internal/ukafka/describe.go
new file mode 100644
index 0000000000..b0f9d0a9f4
--- /dev/null
+++ b/products/ukafka/internal/ukafka/describe.go
@@ -0,0 +1,182 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// DescribeUKafkaInstanceResponse 自定义响应结构
+type DescribeUKafkaInstanceResponse struct {
+	RetCode    int           `json:"RetCode"`
+	Action     string        `json:"Action"`
+	ClusterSet []ClusterInfo `json:"ClusterSet"`
+	Message    string        `json:"Message"`
+}
+
+// ClusterInfo 实例信息
+type ClusterInfo struct {
+	Zone                string   `json:"Zone"`
+	ClusterInstanceId   string   `json:"ClusterInstanceId"`
+	ClusterInstanceName string   `json:"ClusterInstanceName"`
+	Remark              string   `json:"Remark"`
+	Tag                 string   `json:"Tag"`
+	Framework           string   `json:"Framework"`
+	FrameworkVersion    string   `json:"FrameworkVersion"`
+	NetworkId           string   `json:"NetworkId"`
+	VPCId               string   `json:"VPCId"`
+	SubnetId            string   `json:"SubnetId"`
+	BusinessId          string   `json:"BusinessId"`
+	UHostSet            []Broker `json:"UHostSet"`
+	IsOpenSecgroup      bool     `json:"IsOpenSecgroup"`
+	ChargeType          string   `json:"ChargeType"`
+	AutoRenew           string   `json:"AutoRenew"`
+	ValidBrokerNum      int      `json:"ValidBrokerNum"`
+	UHostCount          int      `json:"UHostCount"`
+	ExpireTime          int      `json:"ExpireTime"`
+	CreateTime          int      `json:"CreateTime"`
+	RunningTime         int      `json:"RunningTime"`
+	State               string   `json:"State"`
+}
+
+// Broker 节点信息
+type Broker struct {
+	BrokerId          string      `json:"BrokerId"`
+	UHostId           string      `json:"UHostId"`
+	ResourceId        string      `json:"ResourceId"`
+	UHostRole         string      `json:"UHostRole"`
+	UHostName         string      `json:"UHostName"`
+	DomainName        string      `json:"DomainName"`
+	Remark            string      `json:"Remark"`
+	CreateTime        int         `json:"CreateTime"`
+	ExpireTime        int         `json:"ExpireTime"`
+	InstanceGroupType string      `json:"InstanceGroupType"`
+	SecurityGroupId   string      `json:"SecurityGroupId"`
+	State             string      `json:"State"`
+	ZooKeeper         string      `json:"ZooKeeper"`
+	UHostConfig       UHostConfig `json:"UHostConfig"`
+	IPSet             []IPInfo    `json:"IPSet"`
+	KafkaPort         int         `json:"KafkaPort"`
+	ZooKeeperPort     int         `json:"ZooKeeperPort"`
+}
+
+// UHostConfig 节点配置
+type UHostConfig struct {
+	CPU          int    `json:"CPU"`
+	Memory       int    `json:"Memory"`
+	DataDiskSize int    `json:"DataDiskSize"`
+	DiskType     string `json:"DiskType"`
+}
+
+// IPInfo IP信息
+type IPInfo struct {
+	Type string `json:"Type"`
+	IP   string `json:"IP"`
+}
+
+// newDescribe ucloud ukafka describe
+func newDescribe(ctx *cli.Context) *cobra.Command {
+	var instanceID *string
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewDescribeUKafkaInstanceRequest()
+	cmd := &cobra.Command{
+		Use:   "describe",
+		Short: "Describe UKafka instance details",
+		Long:  "Describe UKafka instance details",
+		Run: func(cmd *cobra.Command, args []string) {
+			id := ctx.PickResourceID(*instanceID)
+
+			// 使用 GenericInvoke 绕过 SDK 类型问题
+			genReq := client.Client.NewGenericRequest()
+			genReq.SetAction("DescribeUKafkaInstance")
+			genReq.SetRegion(*req.Region)
+			genReq.SetZone(*req.Zone)
+			if req.ProjectId != nil && *req.ProjectId != "" {
+				genReq.SetProjectId(*req.ProjectId)
+			}
+
+			payload := map[string]interface{}{
+				"ClusterInstanceId": id,
+			}
+			genReq.SetPayload(payload)
+
+			genResp, err := client.Client.GenericInvoke(genReq)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			var resp DescribeUKafkaInstanceResponse
+			if err := genResp.Unmarshal(&resp); err != nil {
+				ctx.HandleError(fmt.Errorf("parse response: %w", err))
+				return
+			}
+
+			if resp.RetCode != 0 {
+				ctx.HandleError(fmt.Errorf("API error: RetCode=%d, Message=%s", resp.RetCode, resp.Message))
+				return
+			}
+
+			if len(resp.ClusterSet) == 0 {
+				ctx.EmitResult(cli.OpResultRow{ResourceID: id, Action: "describe", Status: "NotFound"})
+				return
+			}
+
+			cluster := resp.ClusterSet[0]
+			rows := []cli.DescribeRow{
+				{Attribute: "InstanceID", Content: cluster.ClusterInstanceId},
+				{Attribute: "InstanceName", Content: cluster.ClusterInstanceName},
+				{Attribute: "Region", Content: *req.Region},
+				{Attribute: "Zone", Content: cluster.Zone},
+				{Attribute: "State", Content: cluster.State},
+				{Attribute: "Framework", Content: cluster.Framework},
+				{Attribute: "FrameworkVersion", Content: cluster.FrameworkVersion},
+				{Attribute: "VPCId", Content: cluster.VPCId},
+				{Attribute: "SubnetId", Content: cluster.SubnetId},
+				{Attribute: "BusinessId", Content: cluster.BusinessId},
+				{Attribute: "ChargeType", Content: cluster.ChargeType},
+				{Attribute: "AutoRenew", Content: cluster.AutoRenew},
+				{Attribute: "Remark", Content: cluster.Remark},
+			}
+
+			// Add node information
+			if len(cluster.UHostSet) > 0 {
+				rows = append(rows, cli.DescribeRow{Attribute: "--- Nodes ---", Content: fmt.Sprintf("%d nodes", len(cluster.UHostSet))})
+				for i, node := range cluster.UHostSet {
+					prefix := fmt.Sprintf("Node[%d]", i)
+					var ip string
+					if len(node.IPSet) > 0 {
+						ip = node.IPSet[0].IP
+					}
+					rows = append(rows,
+						cli.DescribeRow{Attribute: prefix + ".NodeID", Content: node.UHostId},
+						cli.DescribeRow{Attribute: prefix + ".NodeName", Content: node.UHostName},
+						cli.DescribeRow{Attribute: prefix + ".NodeRole", Content: node.UHostRole},
+						cli.DescribeRow{Attribute: prefix + ".State", Content: node.State},
+						cli.DescribeRow{Attribute: prefix + ".IP", Content: ip},
+						cli.DescribeRow{Attribute: prefix + ".CPU", Content: fmt.Sprintf("%d", node.UHostConfig.CPU)},
+						cli.DescribeRow{Attribute: prefix + ".Memory", Content: fmt.Sprintf("%dMB", node.UHostConfig.Memory)},
+						cli.DescribeRow{Attribute: prefix + ".DiskSize", Content: fmt.Sprintf("%dGB", node.UHostConfig.DataDiskSize)},
+						cli.DescribeRow{Attribute: prefix + ".DiskType", Content: node.UHostConfig.DiskType},
+						cli.DescribeRow{Attribute: prefix + ".KafkaPort", Content: fmt.Sprintf("%d", node.KafkaPort)},
+					)
+				}
+			}
+			ctx.PrintList(rows)
+		},
+	}
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID to describe")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+
+	cmd.MarkFlagRequired("ukafka-id")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/describe_consumer.go b/products/ukafka/internal/ukafka/describe_consumer.go
new file mode 100644
index 0000000000..6f8c1344ab
--- /dev/null
+++ b/products/ukafka/internal/ukafka/describe_consumer.go
@@ -0,0 +1,68 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+	sdk "github.com/ucloud/ucloud-sdk-go/ucloud"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newDescribeConsumer ucloud ukafka describe-consumer
+func newDescribeConsumer(ctx *cli.Context) *cobra.Command {
+	var instanceID *string
+	var consumerGroup *string
+	var consumerType *string
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewDescribeUKafkaConsumerRequest()
+
+	cmd := &cobra.Command{
+		Use:   "describe-consumer",
+		Short: "Describe Kafka consumer group details",
+		Long:  "Describe Kafka consumer group details",
+		Run: func(cmd *cobra.Command, args []string) {
+			req.ClusterInstanceId = sdk.String(*instanceID)
+			req.ConsumerGroup = sdk.String(*consumerGroup)
+			req.Type = sdk.String(*consumerType)
+
+			resp, err := client.DescribeUKafkaConsumer(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			rows := []cli.DescribeRow{
+				{Attribute: "GroupName", Content: resp.GroupName},
+				{Attribute: "Type", Content: resp.Type},
+			}
+			if len(resp.Topics) > 0 {
+				for i, topic := range resp.Topics {
+					rows = append(rows, cli.DescribeRow{
+						Attribute: fmt.Sprintf("Topic[%d]", i),
+						Content:   topic,
+					})
+				}
+			}
+			ctx.PrintList(rows)
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	consumerGroup = flags.String("consumer-group", "", "Required. Consumer group name")
+	consumerType = flags.String("type", "", "Required. Consumer group type (e.g. ZK, KF)")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+
+	cmd.MarkFlagRequired("ukafka-id")
+	cmd.MarkFlagRequired("consumer-group")
+	cmd.MarkFlagRequired("type")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/list.go b/products/ukafka/internal/ukafka/list.go
new file mode 100644
index 0000000000..e6645bc518
--- /dev/null
+++ b/products/ukafka/internal/ukafka/list.go
@@ -0,0 +1,133 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/internal/common"
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// ListUKafkaInstanceResponse 自定义响应结构，修复 SDK 类型问题
+type ListUKafkaInstanceResponse struct {
+	RetCode    int             `json:"RetCode"`
+	Action     string          `json:"Action"`
+	TotalCount int             `json:"TotalCount"`
+	ClusterSet []ClusterSetRaw `json:"ClusterSet"`
+	Message    string          `json:"Message"`
+}
+
+// ClusterSetRaw 实例信息
+type ClusterSetRaw struct {
+	Zone                string `json:"Zone"`
+	ClusterInstanceId   string `json:"ClusterInstanceId"`
+	ClusterInstanceName string `json:"ClusterInstanceName"`
+	Framework           string `json:"Framework"`
+	FrameworkVersion    string `json:"FrameworkVersion"`
+	Remark              string `json:"Remark"`
+	CreateTime          int    `json:"CreateTime"`
+	RunningTime         int    `json:"RunningTime"`
+	ExpireTime          int    `json:"ExpireTime"`
+	AutoRenew           string `json:"AutoRenew"`
+	ChargeType          string `json:"ChargeType"`
+	UHostCount          int    `json:"UHostCount"`
+	State               string `json:"State"`
+	Tag                 string `json:"Tag"`
+	InstanceGroupType   string `json:"InstanceGroupType"`
+	VPCId               string `json:"VPCId"`
+	SubnetId            string `json:"SubnetId"`
+	BusinessId          string `json:"BusinessId"`
+}
+
+// newList ucloud ukafka list
+func newList(ctx *cli.Context) *cobra.Command {
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewListUKafkaInstanceRequest()
+	cmd := &cobra.Command{
+		Use:   "list",
+		Short: "List UKafka instances",
+		Long:  "List UKafka instances",
+		Run: func(cmd *cobra.Command, args []string) {
+			// 使用 GenericInvoke 绕过 SDK 类型问题
+			genReq := client.Client.NewGenericRequest()
+			genReq.SetAction("ListUKafkaInstance")
+			genReq.SetRegion(*req.Region)
+			genReq.SetZone(*req.Zone)
+			if req.ProjectId != nil && *req.ProjectId != "" {
+				genReq.SetProjectId(*req.ProjectId)
+			}
+
+			// 构建额外参数
+			payload := map[string]interface{}{}
+			if req.Offset != nil && *req.Offset != "0" {
+				payload["Offset"] = *req.Offset
+			}
+			if req.Limit != nil && *req.Limit != "60" {
+				payload["Limit"] = *req.Limit
+			}
+			if req.VPCId != nil && *req.VPCId != "" {
+				payload["VPCId"] = *req.VPCId
+			}
+			if req.SubnetId != nil && *req.SubnetId != "" {
+				payload["SubnetId"] = *req.SubnetId
+			}
+			if req.BusinessId != nil && *req.BusinessId != "" {
+				payload["BusinessId"] = *req.BusinessId
+			}
+			if len(payload) > 0 {
+				genReq.SetPayload(payload)
+			}
+
+			genResp, err := client.Client.GenericInvoke(genReq)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			var resp ListUKafkaInstanceResponse
+			if err := genResp.Unmarshal(&resp); err != nil {
+				ctx.HandleError(fmt.Errorf("parse response: %w", err))
+				return
+			}
+
+			if resp.RetCode != 0 {
+				ctx.HandleError(fmt.Errorf("API error: RetCode=%d, Message=%s", resp.RetCode, resp.Message))
+				return
+			}
+
+			list := []InstanceRow{}
+			for _, ins := range resp.ClusterSet {
+				row := InstanceRow{
+					InstanceID:   ins.ClusterInstanceId,
+					InstanceName: ins.ClusterInstanceName,
+					Framework:    ins.Framework,
+					Version:      ins.FrameworkVersion,
+					Zone:         ins.Zone,
+					State:        ins.State,
+					NodeCount:    fmt.Sprintf("%d", ins.UHostCount),
+					VPCId:        ins.VPCId,
+					SubnetId:     ins.SubnetId,
+					ChargeType:   ins.ChargeType,
+					CreateTime:   common.FormatDate(ins.CreateTime),
+					ExpireTime:   common.FormatDate(ins.ExpireTime),
+				}
+				list = append(list, row)
+			}
+			ctx.PrintList(list)
+		},
+	}
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", "", "Optional. Assign availability zone")
+	req.Offset = flags.String("offset", "0", "Optional. Offset")
+	req.Limit = flags.String("limit", "60", "Optional. Limit, default 60")
+	req.VPCId = flags.String("vpc-id", "", "Optional. VPC ID")
+	req.SubnetId = flags.String("subnet-id", "", "Optional. Subnet ID")
+	req.BusinessId = flags.String("business-id", "", "Optional. Business group ID")
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/list_consumers.go b/products/ukafka/internal/ukafka/list_consumers.go
new file mode 100644
index 0000000000..80e1e2cf0a
--- /dev/null
+++ b/products/ukafka/internal/ukafka/list_consumers.go
@@ -0,0 +1,58 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+	sdk "github.com/ucloud/ucloud-sdk-go/ucloud"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newListConsumers ucloud ukafka list-consumers
+func newListConsumers(ctx *cli.Context) *cobra.Command {
+	var instanceID *string
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewListUKafkaConsumersRequest()
+
+	cmd := &cobra.Command{
+		Use:   "list-consumers",
+		Short: "List Kafka consumer groups",
+		Long:  "List Kafka consumer groups in UKafka instance",
+		Run: func(cmd *cobra.Command, args []string) {
+			req.ClusterInstanceId = sdk.String(*instanceID)
+
+			resp, err := client.ListUKafkaConsumers(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			list := []ConsumerGroupRow{}
+			for _, g := range resp.Groups {
+				row := ConsumerGroupRow{
+					GroupName:   g.GroupName,
+					Type:        g.Type,
+					NumOfTopics: fmt.Sprintf("%d", g.NumOfTopics),
+					GroupID:     g.GroupId,
+				}
+				list = append(list, row)
+			}
+			ctx.PrintList(list)
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+
+	cmd.MarkFlagRequired("ukafka-id")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/list_topics.go b/products/ukafka/internal/ukafka/list_topics.go
new file mode 100644
index 0000000000..7198afc3f4
--- /dev/null
+++ b/products/ukafka/internal/ukafka/list_topics.go
@@ -0,0 +1,101 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// ListUKafkaTopicsResponse 自定义响应结构
+type ListUKafkaTopicsResponse struct {
+	RetCode   int         `json:"RetCode"`
+	Action    string      `json:"Action"`
+	TopicList []TopicInfo `json:"TopicList"`
+	Length    int         `json:"Length"`
+	Message   string      `json:"Message"`
+}
+
+// TopicInfo topic信息
+type TopicInfo struct {
+	Topic             string `json:"Topic"`
+	NumOfPartition    int    `json:"NumOfPartition"`
+	NumOfOccupyBroker int    `json:"NumOfOccupyBroker"`
+	NumOfReplica      int    `json:"NumOfReplica"`
+	Status            string `json:"Status"`
+	UnderReplicasPer  string `json:"UnderReplicasPer"`
+}
+
+// newListTopics ucloud ukafka list-topics
+func newListTopics(ctx *cli.Context) *cobra.Command {
+	var instanceID *string
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewListUKafkaTopicsRequest()
+
+	cmd := &cobra.Command{
+		Use:   "list-topics",
+		Short: "List Kafka topics in UKafka instance",
+		Long:  "List Kafka topics in UKafka instance",
+		Run: func(cmd *cobra.Command, args []string) {
+			// 使用 GenericInvoke 绕过 SDK 类型问题
+			genReq := client.Client.NewGenericRequest()
+			genReq.SetAction("ListUKafkaTopics")
+			genReq.SetRegion(*req.Region)
+			genReq.SetZone(*req.Zone)
+			if req.ProjectId != nil && *req.ProjectId != "" {
+				genReq.SetProjectId(*req.ProjectId)
+			}
+
+			payload := map[string]interface{}{
+				"ClusterInstanceId": *instanceID,
+			}
+			genReq.SetPayload(payload)
+
+			genResp, err := client.Client.GenericInvoke(genReq)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			var resp ListUKafkaTopicsResponse
+			if err := genResp.Unmarshal(&resp); err != nil {
+				ctx.HandleError(fmt.Errorf("parse response: %w", err))
+				return
+			}
+
+			if resp.RetCode != 0 {
+				ctx.HandleError(fmt.Errorf("API error: RetCode=%d, Message=%s", resp.RetCode, resp.Message))
+				return
+			}
+
+			list := []TopicRow{}
+			for _, t := range resp.TopicList {
+				row := TopicRow{
+					Topic:             t.Topic,
+					NumOfPartition:    fmt.Sprintf("%d", t.NumOfPartition),
+					NumOfReplica:      fmt.Sprintf("%d", t.NumOfReplica),
+					NumOfOccupyBroker: fmt.Sprintf("%d", t.NumOfOccupyBroker),
+					UnderReplicasPer:  t.UnderReplicasPer,
+					Status:            t.Status,
+				}
+				list = append(list, row)
+			}
+			ctx.PrintList(list)
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+
+	cmd.MarkFlagRequired("ukafka-id")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/modify_type.go b/products/ukafka/internal/ukafka/modify_type.go
new file mode 100644
index 0000000000..711fe418b4
--- /dev/null
+++ b/products/ukafka/internal/ukafka/modify_type.go
@@ -0,0 +1,61 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+	sdk "github.com/ucloud/ucloud-sdk-go/ucloud"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newModifyType ucloud ukafka modify-type
+func newModifyType(ctx *cli.Context) *cobra.Command {
+	var async *bool
+	var instanceID *string
+	var nodeType *string
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewModifyUKafkaInstanceTypeRequest()
+
+	cmd := &cobra.Command{
+		Use:   "modify-type",
+		Short: "Modify UKafka instance type (CPU and memory)",
+		Long:  "Modify UKafka instance type, only upgrade CPU and memory",
+		Run: func(cmd *cobra.Command, args []string) {
+			w := ctx.ProgressWriter()
+			req.InstanceId = sdk.String(*instanceID)
+			req.NodeType = sdk.String(*nodeType)
+
+			_, err := client.ModifyUKafkaInstanceType(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			text := fmt.Sprintf("ukafka[%s] is modifying type to %s", *instanceID, *nodeType)
+			if *async {
+				fmt.Fprintln(w, text)
+			} else {
+				ctx.PollerTo(w, describeUKafkaInstanceByID(ctx)).Spoll(*instanceID, text, []string{StateRunning, StateAbnormal})
+			}
+			ctx.EmitResult(cli.OpResultRow{ResourceID: *instanceID, Action: "modify-type", Status: "Running"})
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	nodeType = flags.String("node-type", "", "Required. Target node type")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+	async = flags.Bool("async", false, "Optional. Do not wait for operation to finish")
+
+	cmd.MarkFlagRequired("ukafka-id")
+	cmd.MarkFlagRequired("node-type")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/nodeconf.go b/products/ukafka/internal/ukafka/nodeconf.go
new file mode 100644
index 0000000000..5a32920eac
--- /dev/null
+++ b/products/ukafka/internal/ukafka/nodeconf.go
@@ -0,0 +1,105 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// GetUKafkaNodeTypeResponse 自定义响应结构
+type GetUKafkaNodeTypeResponse struct {
+	RetCode     int            `json:"RetCode"`
+	Action      string         `json:"Action"`
+	NodeTypeSet []InstanceType `json:"NodeTypeSet"`
+	TotalCount  int            `json:"TotalCount"`
+	Message     string         `json:"Message"`
+}
+
+// InstanceType 机型信息
+type InstanceType struct {
+	NodeTypeName   string    `json:"NodeTypeName"`
+	CPU            int       `json:"CPU"`
+	Memory         int       `json:"Memory"`
+	DiskType       string    `json:"DiskType"`
+	DiskSet        []DiskSet `json:"DiskSet"`
+	MaxDiskSize    int       `json:"MaxDiskSize"`
+	MinDiskSize    int       `json:"MinDiskSize"`
+	IsOpenSecGroup bool      `json:"IsOpenSecGroup"`
+	UHostFamily    string    `json:"UHostFamily"`
+}
+
+// DiskSet 磁盘配置
+type DiskSet struct {
+	Type string `json:"Type"`
+	Size int    `json:"Size"`
+}
+
+// newNodeConf ucloud ukafka node-conf
+func newNodeConf(ctx *cli.Context) *cobra.Command {
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewGetUKafkaNodeTypeRequest()
+	cmd := &cobra.Command{
+		Use:   "node-conf",
+		Short: "List available UKafka node configurations",
+		Long:  "List available UKafka node configurations",
+		Run: func(cmd *cobra.Command, args []string) {
+			// 使用 GenericInvoke 绕过 SDK 类型问题
+			genReq := client.Client.NewGenericRequest()
+			genReq.SetAction("GetUKafkaNodeType")
+			genReq.SetRegion(*req.Region)
+			genReq.SetZone(*req.Zone)
+			if req.ProjectId != nil && *req.ProjectId != "" {
+				genReq.SetProjectId(*req.ProjectId)
+			}
+			if req.NodeType != nil && *req.NodeType != "" {
+				payload := map[string]interface{}{
+					"NodeType": *req.NodeType,
+				}
+				genReq.SetPayload(payload)
+			}
+
+			genResp, err := client.Client.GenericInvoke(genReq)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			var resp GetUKafkaNodeTypeResponse
+			if err := genResp.Unmarshal(&resp); err != nil {
+				ctx.HandleError(fmt.Errorf("parse response: %w", err))
+				return
+			}
+
+			if resp.RetCode != 0 {
+				ctx.HandleError(fmt.Errorf("API error: RetCode=%d, Message=%s", resp.RetCode, resp.Message))
+				return
+			}
+
+			list := []NodeConfRow{}
+			for _, t := range resp.NodeTypeSet {
+				row := NodeConfRow{
+					NodeType:    t.NodeTypeName,
+					CPU:         fmt.Sprintf("%d", t.CPU),
+					Memory:      fmt.Sprintf("%dMB", t.Memory),
+					DiskType:    t.DiskType,
+					MinDiskSize: fmt.Sprintf("%d", t.MinDiskSize),
+					MaxDiskSize: fmt.Sprintf("%d", t.MaxDiskSize),
+					SecGroup:    fmt.Sprintf("%v", t.IsOpenSecGroup),
+				}
+				list = append(list, row)
+			}
+			ctx.PrintList(list)
+		},
+	}
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+	req.NodeType = flags.String("node-type", "", "Optional. Specify node type")
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/poll.go b/products/ukafka/internal/ukafka/poll.go
new file mode 100644
index 0000000000..d5dd12e05e
--- /dev/null
+++ b/products/ukafka/internal/ukafka/poll.go
@@ -0,0 +1,59 @@
+package ukafka
+
+import (
+	"fmt"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+	"github.com/ucloud/ucloud-sdk-go/ucloud/request"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// describeUKafkaInstanceByID returns the poller's describe func
+// It uses commonBase to get region and zone for the request
+func describeUKafkaInstanceByID(ctx *cli.Context) func(instanceID string, commonBase *request.CommonBase) (interface{}, error) {
+	return func(instanceID string, commonBase *request.CommonBase) (interface{}, error) {
+		client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+		genReq := client.Client.NewGenericRequest()
+		genReq.SetAction("DescribeUKafkaInstance")
+
+		// Use region/zone from commonBase if available, otherwise use defaults
+		if commonBase != nil {
+			if commonBase.Region != nil && *commonBase.Region != "" {
+				genReq.SetRegion(*commonBase.Region)
+			} else {
+				genReq.SetRegion(ctx.DefaultRegion())
+			}
+			if commonBase.Zone != nil && *commonBase.Zone != "" {
+				genReq.SetZone(*commonBase.Zone)
+			} else {
+				genReq.SetZone(ctx.DefaultZone())
+			}
+			if commonBase.ProjectId != nil && *commonBase.ProjectId != "" {
+				genReq.SetProjectId(*commonBase.ProjectId)
+			}
+		} else {
+			genReq.SetRegion(ctx.DefaultRegion())
+			genReq.SetZone(ctx.DefaultZone())
+		}
+
+		payload := map[string]interface{}{
+			"ClusterInstanceId": instanceID,
+		}
+		genReq.SetPayload(payload)
+
+		genResp, err := client.Client.GenericInvoke(genReq)
+		if err != nil {
+			return nil, err
+		}
+
+		var resp DescribeUKafkaInstanceResponse
+		if err := genResp.Unmarshal(&resp); err != nil {
+			return nil, err
+		}
+		if len(resp.ClusterSet) == 0 {
+			return nil, fmt.Errorf("instance not found")
+		}
+		return &resp.ClusterSet[0], nil
+	}
+}
diff --git a/products/ukafka/internal/ukafka/resize_disk.go b/products/ukafka/internal/ukafka/resize_disk.go
new file mode 100644
index 0000000000..baa9143323
--- /dev/null
+++ b/products/ukafka/internal/ukafka/resize_disk.go
@@ -0,0 +1,61 @@
+package ukafka
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	ukafkasdk "github.com/ucloud/ucloud-sdk-go/services/ukafka"
+	sdk "github.com/ucloud/ucloud-sdk-go/ucloud"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+)
+
+// newResizeDisk ucloud ukafka resize-disk
+func newResizeDisk(ctx *cli.Context) *cobra.Command {
+	var async *bool
+	var instanceID *string
+	var diskSize *int
+
+	client := cli.NewServiceClient(ctx, ukafkasdk.NewClient)
+	req := client.NewResizeUKafkaDiskRequest()
+
+	cmd := &cobra.Command{
+		Use:   "resize-disk",
+		Short: "Resize UKafka instance disk",
+		Long:  "Resize UKafka instance disk size",
+		Run: func(cmd *cobra.Command, args []string) {
+			w := ctx.ProgressWriter()
+			req.InstanceId = sdk.String(*instanceID)
+			req.DiskSize = sdk.Int(*diskSize)
+
+			_, err := client.ResizeUKafkaDisk(req)
+			if err != nil {
+				ctx.HandleError(err)
+				return
+			}
+
+			text := fmt.Sprintf("ukafka[%s] is resizing disk to %dGB", *instanceID, *diskSize)
+			if *async {
+				fmt.Fprintln(w, text)
+			} else {
+				ctx.PollerTo(w, describeUKafkaInstanceByID(ctx)).Spoll(*instanceID, text, []string{StateRunning, StateAbnormal})
+			}
+			ctx.EmitResult(cli.OpResultRow{ResourceID: *instanceID, Action: "resize-disk", Status: "Running"})
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.SortFlags = false
+	instanceID = flags.String("ukafka-id", "", "Required. Instance ID")
+	diskSize = flags.Int("disk-size-gb", 0, "Required. Target disk size in GB")
+	req.ProjectId = flags.String("project-id", ctx.DefaultProjectID(), "Optional. Assign project-id")
+	req.Region = flags.String("region", ctx.DefaultRegion(), "Optional. Assign region")
+	req.Zone = flags.String("zone", ctx.DefaultZone(), "Optional. Assign availability zone")
+	async = flags.Bool("async", false, "Optional. Do not wait for operation to finish")
+
+	cmd.MarkFlagRequired("ukafka-id")
+	cmd.MarkFlagRequired("disk-size-gb")
+
+	return cmd
+}
diff --git a/products/ukafka/internal/ukafka/rows.go b/products/ukafka/internal/ukafka/rows.go
new file mode 100644
index 0000000000..0333924e31
--- /dev/null
+++ b/products/ukafka/internal/ukafka/rows.go
@@ -0,0 +1,52 @@
+package ukafka
+
+// InstanceRow represents a UKafka instance in list output
+type InstanceRow struct {
+	InstanceID   string
+	InstanceName string
+	Framework    string
+	Version      string
+	Zone         string
+	State        string
+	NodeCount    string
+	VPCId        string
+	SubnetId     string
+	ChargeType   string
+	CreateTime   string
+	ExpireTime   string
+}
+
+// NodeConfRow represents a node configuration in node-conf output
+type NodeConfRow struct {
+	NodeType    string
+	CPU         string
+	Memory      string
+	DiskType    string
+	MinDiskSize string
+	MaxDiskSize string
+	SecGroup    string
+}
+
+// VersionRow represents a version in app-version output
+type VersionRow struct {
+	Version string
+	Label   string
+}
+
+// ConsumerGroupRow represents a consumer group in list-consumers output
+type ConsumerGroupRow struct {
+	GroupName   string
+	Type        string
+	NumOfTopics string
+	GroupID     string
+}
+
+// TopicRow represents a topic in list-topics output
+type TopicRow struct {
+	Topic             string
+	NumOfPartition    string
+	NumOfReplica      string
+	NumOfOccupyBroker string
+	UnderReplicasPer  string
+	Status            string
+}
diff --git a/products/ukafka/internal/ukafka/status.go b/products/ukafka/internal/ukafka/status.go
new file mode 100644
index 0000000000..3df0605b4a
--- /dev/null
+++ b/products/ukafka/internal/ukafka/status.go
@@ -0,0 +1,12 @@
+package ukafka
+
+// Instance state constants
+const (
+	StateRunning   = "Running"
+	StateAbnormal  = "Abnormal"
+	StateCreating  = "Creating"
+	StateDeleting  = "Deleting"
+	StateDeleted   = "Deleted"
+	StateUpdating  = "Updating"
+	StateDeploying = "Deploying"
+)
diff --git a/products/ukafka/product.go b/products/ukafka/product.go
new file mode 100644
index 0000000000..a3a277b1f5
--- /dev/null
+++ b/products/ukafka/product.go
@@ -0,0 +1,21 @@
+package ukafka
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/ucloud/ucloud-cli/pkg/cli"
+	internalukafka "github.com/ucloud/ucloud-cli/products/ukafka/internal/ukafka"
+)
+
+type product struct{}
+
+// New returns the ukafka product (registered via hack/gen-products)
+func New() cli.Product { return product{} }
+
+func (product) Metadata() cli.Metadata {
+	return cli.Metadata{Name: "ukafka", Commands: []string{"ukafka"}}
+}
+
+func (product) NewCommand(ctx *cli.Context) []*cobra.Command {
+	return []*cobra.Command{internalukafka.NewCommand(ctx)}
+}
diff --git a/products/ukafka/product.yaml b/products/ukafka/product.yaml
new file mode 100644
index 0000000000..fd1ae115d3
--- /dev/null
+++ b/products/ukafka/product.yaml
@@ -0,0 +1,7 @@
+# products/ukafka/product.yaml — ukafka 产品元数据
+name: ukafka
+owners:
+  - zxl-wangwang
+commands:
+  - ukafka
+enabled: true
diff --git a/products/ukafka/testdata/cmdtree.golden b/products/ukafka/testdata/cmdtree.golden
new file mode 100644
index 0000000000..a3be5b9cb6
--- /dev/null
+++ b/products/ukafka/testdata/cmdtree.golden
@@ -0,0 +1,93 @@
+ucloud ukafka	use=ukafka	short=Manage UKafka instances
+ucloud ukafka add-node	use=add-node	short=Add nodes to UKafka instance
+  flag=async	short=	default=false	required=
+  flag=node-count	short=	default=1	required=
+  flag=node-type	short=	default=	required=true
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka app-version	use=app-version	short=List available Kafka versions
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=zone	short=	default=	required=
+ucloud ukafka check-topic	use=check-topic	short=Check if a topic name exists in UKafka instance
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=topic-name	short=	default=	required=true
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka create	use=create	short=Create UKafka instance
+  flag=async	short=	default=false	required=
+  flag=business-id	short=	default=	required=
+  flag=disk-controller-type	short=	default=NONE	required=
+  flag=disk-size-gb	short=	default=0	required=true
+  flag=disk-threshold	short=	default=90	required=
+  flag=enable-security	short=	default=false	required=
+  flag=kafka-version	short=	default=	required=true
+  flag=log-retention-hours	short=	default=72	required=
+  flag=name	short=	default=	required=true
+  flag=node-count	short=	default=3	required=
+  flag=node-type	short=	default=	required=true
+  flag=project-id	short=	default=	required=
+  flag=quantity	short=	default=1	required=
+  flag=region	short=	default=	required=
+  flag=subnet-id	short=	default=	required=
+  flag=vpc-id	short=	default=	required=
+  flag=zone	short=	default=	required=
+ucloud ukafka delete	use=delete	short=Delete UKafka instances
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=[]	required=true
+  flag=yes	short=y	default=false	required=
+  flag=zone	short=	default=	required=
+ucloud ukafka describe	use=describe	short=Describe UKafka instance details
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka describe-consumer	use=describe-consumer	short=Describe Kafka consumer group details
+  flag=consumer-group	short=	default=	required=true
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=type	short=	default=	required=true
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka list	use=list	short=List UKafka instances
+  flag=business-id	short=	default=	required=
+  flag=limit	short=	default=60	required=
+  flag=offset	short=	default=0	required=
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=subnet-id	short=	default=	required=
+  flag=vpc-id	short=	default=	required=
+  flag=zone	short=	default=	required=
+ucloud ukafka list-consumers	use=list-consumers	short=List Kafka consumer groups
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka list-topics	use=list-topics	short=List Kafka topics in UKafka instance
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka modify-type	use=modify-type	short=Modify UKafka instance type (CPU and memory)
+  flag=async	short=	default=false	required=
+  flag=node-type	short=	default=	required=true
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
+ucloud ukafka node-conf	use=node-conf	short=List available UKafka node configurations
+  flag=node-type	short=	default=	required=
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=zone	short=	default=	required=
+ucloud ukafka resize-disk	use=resize-disk	short=Resize UKafka instance disk
+  flag=async	short=	default=false	required=
+  flag=disk-size-gb	short=	default=0	required=true
+  flag=project-id	short=	default=	required=
+  flag=region	short=	default=	required=
+  flag=ukafka-id	short=	default=	required=true
+  flag=zone	short=	default=	required=
diff --git a/products/ukafka/testdata/completion.golden b/products/ukafka/testdata/completion.golden
new file mode 100644
index 0000000000..5ebec444a9
--- /dev/null
+++ b/products/ukafka/testdata/completion.golden
@@ -0,0 +1,3 @@
+ucloud ukafka create	project-id	dynamic
+ucloud ukafka create	region	dynamic
+ucloud ukafka create	zone	dynamic