diff --git a/cli/app.go b/cli/app.go
index 67519650ec0..e042d42f314 100644
--- a/cli/app.go
+++ b/cli/app.go
@@ -20,11 +20,16 @@ const (
 	disableProfilesFlag = "disable-profiles"
 	profileFlagName     = "profile-name"
 
+	generalFlagStart = "start"
+	generalFlagEnd   = "end"
+
 	// TODO: RSDK-6683.
 	quietFlag = "quiet"
 
 	logsFlagFormat     = "format"
 	logsFlagOutputFile = "output"
+	logsFlagKeyword    = "keyword"
+	logsFlagLevels     = "levels"
 	logsFlagErrors     = "errors"
 	logsFlagTail       = "tail"
 	logsFlagCount      = "count"
@@ -114,8 +119,6 @@ const (
 	dataFlagResourceName                   = "resource-name"
 	dataFlagMethod                         = "method"
 	dataFlagMimeTypes                      = "mime-types"
-	dataFlagStart                          = "start"
-	dataFlagEnd                            = "end"
 	dataFlagParallelDownloads              = "parallel"
 	dataFlagTags                           = "tags"
 	dataFlagBboxLabels                     = "bbox-labels"
@@ -211,11 +214,11 @@ var commonFilterFlags = []cli.Flag{
 		Usage: "mime types filter",
 	},
 	&cli.StringFlag{
-		Name:  dataFlagStart,
+		Name:  generalFlagStart,
 		Usage: "ISO-8601 timestamp in RFC3339 format indicating the start of the interval filter",
 	},
 	&cli.StringFlag{
-		Name:  dataFlagEnd,
+		Name:  generalFlagEnd,
 		Usage: "ISO-8601 timestamp in RFC3339 format indicating the end of the interval filter",
 	},
 	&cli.StringSliceFlag{
@@ -1060,12 +1063,12 @@ var app = &cli.App{
 									Usage:    "orgs filter",
 								},
 								&cli.StringFlag{
-									Name:     dataFlagStart,
+									Name:     generalFlagStart,
 									Required: true,
 									Usage:    "ISO-8601 timestamp in RFC3339 format indicating the start of the interval filter",
 								},
 								&cli.StringFlag{
-									Name:     dataFlagEnd,
+									Name:     generalFlagEnd,
 									Required: true,
 									Usage:    "ISO-8601 timestamp in RFC3339 format indicating the end of the interval filter",
 								},
@@ -1792,9 +1795,21 @@ var app = &cli.App{
 							Name:  logsFlagFormat,
 							Usage: "file format (text or json)",
 						},
-						&cli.BoolFlag{
-							Name:  logsFlagErrors,
-							Usage: "show only errors",
+						&cli.StringFlag{
+							Name:  logsFlagKeyword,
+							Usage: "filter logs by keyword",
+						},
+						&cli.StringSliceFlag{
+							Name:  logsFlagLevels,
+							Usage: "filter logs by levels (e.g., info, warn, error)",
+						},
+						&cli.StringFlag{
+							Name:  generalFlagStart,
+							Usage: "ISO-8601 timestamp in RFC3339 format indicating the start of the interval filter (e.g., 2025-01-15T14:00:00Z)",
+						},
+						&cli.StringFlag{
+							Name:  generalFlagEnd,
+							Usage: "ISO-8601 timestamp in RFC3339 format indicating the end of the interval filter (e.g., 2025-01-15T15:00:00Z)",
 						},
 						&cli.IntFlag{
 							Name:        logsFlagCount,
diff --git a/cli/client.go b/cli/client.go
index d0d0d175d44..64ea8c25f84 100644
--- a/cli/client.go
+++ b/cli/client.go
@@ -747,7 +747,10 @@ type robotsLogsArgs struct {
 	Machine      string
 	Output       string
 	Format       string
-	Errors       bool
+	Keyword      string
+	Levels       []string
+	Start        string
+	End          string
 	Count        int
 }
 
@@ -758,6 +761,18 @@ func RobotsLogsAction(c *cli.Context, args robotsLogsArgs) error {
 		return err
 	}
 
+	// Check if both start time and count are provided
+	// TODO: [APP-7415] Enhance LogsForPart API to Support Sorting Options for Log Display Order
+	// TODO: [APP-7450] Implement "Start Time with Count without End Time" Functionality in LogsForPart
+	if args.Start != "" && args.Count > 0 && args.End == "" {
+		return errors.New("unsupported functionality: specifying both a start time and a count without an end time is not supported. " +
+			"This behavior can be counterintuitive because logs are currently only sorted in descending order. " +
+			"For example, if there are 200 logs after the specified start time and you request 10 logs, it will return the 10 most recent logs, " +
+			"rather than the 10 logs closest to the start time. " +
+			"Please provide either a start time and an end time to define a clear range, or a count without a start time for recent logs",
+		)
+	}
+
 	orgStr := args.Organization
 	locStr := args.Location
 	robotStr := args.Machine
@@ -818,33 +833,58 @@ func (c *viamClient) fetchAndSaveLogs(robot *apppb.Robot, parts []*apppb.RobotPa
 
 // streamLogsForPart streams logs for a specific part directly to a file.
 func (c *viamClient) streamLogsForPart(part *apppb.RobotPart, args robotsLogsArgs, writer io.Writer) error {
-	numLogs, err := getNumLogs(c.c, args.Count)
+	maxLogsToFetch, err := getNumLogs(c.c, args.Count)
 	if err != nil {
 		return err
 	}
 
-	// Write logs for this part
+	startTime, err := parseTimeString(args.Start)
+	if err != nil {
+		return errors.Wrap(err, "invalid start time format")
+	}
+	endTime, err := parseTimeString(args.End)
+	if err != nil {
+		return errors.Wrap(err, "invalid end time format")
+	}
+
+	keyword := &args.Keyword
+
+	// Tracks the token for the next page of logs to fetch, allowing pagination through log results.
 	var pageToken string
-	for logsFetched := 0; logsFetched < numLogs; {
+
+	// Fetch logs in batches and write them to the output.
+	for fetchedLogCount := 0; fetchedLogCount < maxLogsToFetch; {
+		// We do not request the exact limit specified by the user in the `count` argument because the API enforces a maximum
+		// limit of 100 logs per batch fetch. To keep the RDK independent of specific limits imposed by the app API,
+		// we always request the next full batch of logs as allowed by the API (currently 100). This approach
+		// ensures that if the API limit changes in the future, only the app API logic needs to be updated without requiring
+		// changes in the RDK.
 		resp, err := c.client.GetRobotPartLogs(c.c.Context, &apppb.GetRobotPartLogsRequest{
-			Id:         part.Id,
-			ErrorsOnly: args.Errors,
-			PageToken:  &pageToken,
+			Id:        part.Id,
+			Filter:    keyword,
+			PageToken: &pageToken,
+			Levels:    args.Levels,
+			Start:     startTime,
+			End:       endTime,
 		})
 		if err != nil {
 			return errors.Wrap(err, "failed to fetch logs")
 		}
 
-		pageToken = resp.NextPageToken
-		// Break in the event of no logs in GetRobotPartLogsResponse or when
-		// page token is empty (no more pages).
-		if resp.Logs == nil || pageToken == "" {
+		// End of pagination if no logs are returned.
+		if len(resp.Logs) == 0 {
 			break
 		}
 
-		// Truncate this intermediate slice of resp.Logs based on how many logs
-		// are still required by numLogs.
-		remainingLogsNeeded := numLogs - logsFetched
+		// The API may return more logs than the user requested via the `count` argument.
+		// This is because the API uses pagination internally and fetches logs in batches.
+		// To ensure we do not append more logs than the user requested, we calculate the
+		// `remainingLogsNeeded` by subtracting the logs we have already fetched (`logsFetched`)
+		// from the total number of logs the user asked for (`numLogs`).
+		// If the current batch contains more logs than the remaining needed, we truncate the
+		// batch to include only the necessary number of logs.
+		// This ensures the output strictly adheres to the `count` limit specified by the user.
+		remainingLogsNeeded := maxLogsToFetch - fetchedLogCount
 		if remainingLogsNeeded < len(resp.Logs) {
 			resp.Logs = resp.Logs[:remainingLogsNeeded]
 		}
@@ -860,7 +900,12 @@ func (c *viamClient) streamLogsForPart(part *apppb.RobotPart, args robotsLogsArg
 			}
 		}
 
-		logsFetched += len(resp.Logs)
+		fetchedLogCount += len(resp.Logs)
+
+		// End of pagination if there is no next page token.
+		if pageToken = resp.NextPageToken; pageToken == "" {
+			break
+		}
 	}
 
 	return nil
diff --git a/cli/data.go b/cli/data.go
index bcfedb62f7f..b8485cb186b 100644
--- a/cli/data.go
+++ b/cli/data.go
@@ -23,7 +23,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/encoding/protojson"
-	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"go.viam.com/rdk/data"
 )
@@ -308,23 +307,14 @@ func createExportTabularRequest(c *cli.Context) (*datapb.ExportTabularDataReques
 }
 
 func createCaptureInterval(startStr, endStr string) (*datapb.CaptureInterval, error) {
-	var start *timestamppb.Timestamp
-	var end *timestamppb.Timestamp
-	timeLayout := time.RFC3339
-
-	if startStr != "" {
-		t, err := time.Parse(timeLayout, startStr)
-		if err != nil {
-			return nil, errors.Wrap(err, "could not parse start flag")
-		}
-		start = timestamppb.New(t)
+	start, err := parseTimeString(startStr)
+	if err != nil {
+		return nil, err
 	}
-	if endStr != "" {
-		t, err := time.Parse(timeLayout, endStr)
-		if err != nil {
-			return nil, errors.Wrap(err, "could not parse end flag")
-		}
-		end = timestamppb.New(t)
+
+	end, err := parseTimeString(endStr)
+	if err != nil {
+		return nil, err
 	}
 
 	return &datapb.CaptureInterval{
diff --git a/cli/utils.go b/cli/utils.go
index 67536aff5c6..477e1c5ad9a 100644
--- a/cli/utils.go
+++ b/cli/utils.go
@@ -5,9 +5,11 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/pkg/errors"
 	apppb "go.viam.com/api/app/v1"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // samePath returns true if abs(path1) and abs(path2) are the same.
@@ -93,6 +95,19 @@ func parseBillingAddress(address string) (*apppb.BillingAddress, error) {
 	}, nil
 }
 
+func parseTimeString(timeStr string) (*timestamppb.Timestamp, error) {
+	if timeStr == "" {
+		return nil, nil
+	}
+
+	t, err := time.Parse(time.RFC3339, timeStr)
+	if err != nil {
+		return nil, errors.Wrapf(err, "could not parse time string: %s", timeStr)
+	}
+
+	return timestamppb.New(t), nil
+}
+
 func formatStringForOutput(protoString, prefixToTrim string) string {
 	return strings.ToLower(strings.TrimPrefix(protoString, prefixToTrim))
 }