diff --git a/lib/actions/activity-actions.js b/lib/actions/activity-actions.js
--- a/lib/actions/activity-actions.js
+++ b/lib/actions/activity-actions.js
@@ -9,6 +9,10 @@
 import { useKeyserverCall } from '../keyserver-conn/keyserver-call.js';
 import type { CallKeyserverEndpoint } from '../keyserver-conn/keyserver-conn-types.js';
 import { useProcessAndSendDMOperation } from '../shared/dm-ops/process-dm-ops.js';
+import {
+  useStreamFarcasterDirectCastRead,
+  useMarkFarcasterDirectCastUnread,
+} from '../shared/farcaster/farcaster-api.js';
 import { threadSpecs } from '../shared/threads/thread-specs.js';
 import type {
   ActivityUpdate,
@@ -116,6 +120,8 @@
   );
   const processAndSendDMOperation = useProcessAndSendDMOperation();
   const keyserverCall = useKeyserverCall(setThreadUnreadStatus);
+  const streamFarcasterDirectCastRead = useStreamFarcasterDirectCastRead();
+  const markFarcasterDirectCastUnread = useMarkFarcasterDirectCastUnread();
 
   return React.useCallback(
     async (input: UseSetThreadUnreadStatusInput) =>
@@ -124,10 +130,18 @@
         {
           processAndSendDMOperation,
           keyserverSetThreadUnreadStatus: keyserverCall,
+          streamFarcasterDirectCastRead,
+          markFarcasterDirectCastUnread,
         },
       ),
 
-    [keyserverCall, viewerID, processAndSendDMOperation],
+    [
+      keyserverCall,
+      viewerID,
+      processAndSendDMOperation,
+      streamFarcasterDirectCastRead,
+      markFarcasterDirectCastUnread,
+    ],
   );
 }
 
diff --git a/lib/shared/farcaster/farcaster-api.js b/lib/shared/farcaster/farcaster-api.js
--- a/lib/shared/farcaster/farcaster-api.js
+++ b/lib/shared/farcaster/farcaster-api.js
@@ -317,6 +317,55 @@
   );
 }
 
+export type StreamFarcasterDirectCastReadInput = {
+  +conversationId: string,
+};
+
+function useStreamFarcasterDirectCastRead(): (
+  input: StreamFarcasterDirectCastReadInput,
+) => Promise<void> {
+  const { sendFarcasterRequest } = useTunnelbroker();
+  return React.useCallback(
+    async (input: StreamFarcasterDirectCastReadInput) => {
+      const { conversationId } = input;
+      const streamMessage = {
+        messageType: 'direct-cast-read',
+        payload: { conversationId },
+        data: conversationId,
+      };
+
+      await sendFarcasterRequest({
+        apiVersion: '',
+        endpoint: 'direct-cast-read',
+        method: { type: 'STREAM' },
+        payload: JSON.stringify(streamMessage),
+      });
+    },
+    [sendFarcasterRequest],
+  );
+}
+
+export type MarkFarcasterDirectCastUnreadInput = {
+  +conversationId: string,
+};
+
+function useMarkFarcasterDirectCastUnread(): (
+  input: MarkFarcasterDirectCastUnreadInput,
+) => Promise<void> {
+  const { sendFarcasterRequest } = useTunnelbroker();
+  return React.useCallback(
+    async (input: MarkFarcasterDirectCastUnreadInput) => {
+      await sendFarcasterRequest({
+        apiVersion: 'v2',
+        endpoint: 'direct-cast-manually-mark-unread',
+        method: { type: 'PUT' },
+        payload: JSON.stringify(input),
+      });
+    },
+    [sendFarcasterRequest],
+  );
+}
+
 export {
   useSendFarcasterTextMessage,
   useFetchFarcasterMessages,
@@ -324,4 +373,6 @@
   useFetchFarcasterInbox,
   useUpdateFarcasterGroupNameAndDescription,
   useUpdateFarcasterSubscription,
+  useStreamFarcasterDirectCastRead,
+  useMarkFarcasterDirectCastUnread,
 };
diff --git a/lib/shared/threads/protocols/farcaster-thread-protocol.js b/lib/shared/threads/protocols/farcaster-thread-protocol.js
--- a/lib/shared/threads/protocols/farcaster-thread-protocol.js
+++ b/lib/shared/threads/protocols/farcaster-thread-protocol.js
@@ -49,8 +49,10 @@
   FetchMessageUtils,
   ProtocolFetchMessageInput,
   ProtocolSendTextMessageInput,
+  ProtocolSetThreadUnreadStatusInput,
   ProtocolUpdateSubscriptionInput,
   SendTextMessageUtils,
+  SetThreadUnreadStatusUtils,
   ThreadProtocol,
   OnOpenThreadUtils,
   ProtocolOnOpenThreadInput,
@@ -206,8 +208,27 @@
     throw new Error('editCalendarEntry method is not yet implemented');
   },
 
-  setThreadUnreadStatus: async (): Promise<SetThreadUnreadStatusPayload> => {
-    throw new Error('setThreadUnreadStatus method is not yet implemented');
+  setThreadUnreadStatus: async (
+    protocolInput: ProtocolSetThreadUnreadStatusInput,
+    utils: SetThreadUnreadStatusUtils,
+  ): Promise<SetThreadUnreadStatusPayload> => {
+    const { input } = protocolInput;
+    const { threadInfo, unread } = input;
+    const { streamFarcasterDirectCastRead, markFarcasterDirectCastUnread } =
+      utils;
+
+    const conversationId = conversationIDFromFarcasterThreadID(threadInfo.id);
+
+    if (unread) {
+      await markFarcasterDirectCastUnread({ conversationId });
+    } else {
+      await streamFarcasterDirectCastRead({ conversationId });
+    }
+
+    return {
+      threadID: threadInfo.id,
+      resetToUnread: false,
+    };
   },
 
   sendReaction: async (): Promise<mixed> => {
diff --git a/lib/shared/threads/thread-spec.js b/lib/shared/threads/thread-spec.js
--- a/lib/shared/threads/thread-spec.js
+++ b/lib/shared/threads/thread-spec.js
@@ -93,6 +93,8 @@
   FetchFarcasterMessageInput,
   FetchFarcasterMessageResult,
   UpdateFarcasterSubscriptionInput,
+  StreamFarcasterDirectCastReadInput,
+  MarkFarcasterDirectCastUnreadInput,
 } from '../farcaster/farcaster-api.js';
 import type { FetchMessagesFromDBType } from '../message-utils.js';
 import type {
@@ -198,6 +200,8 @@
 export type SetThreadUnreadStatusUtils = {
   +processAndSendDMOperation: OutboundDMOperationSpecification => Promise<void>,
   +keyserverSetThreadUnreadStatus: SetThreadUnreadStatusRequest => Promise<SetThreadUnreadStatusPayload>,
+  +streamFarcasterDirectCastRead: StreamFarcasterDirectCastReadInput => Promise<void>,
+  +markFarcasterDirectCastUnread: MarkFarcasterDirectCastUnreadInput => Promise<void>,
 };
 
 export type ProtocolSendReactionInput = {