diff --git a/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt b/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt
index 8f3094dc9..985184606 100644
--- a/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt
+++ b/library/src/androidTest/java/org/xmtp/android/library/LocalInstrumentedTest.kt
@@ -345,4 +345,48 @@ class LocalInstrumentedTest {
     private fun delayToPropagate() {
         Thread.sleep(500)
     }
+
+    @Test
+    fun testStreamEphemeralInV1Conversation() {
+        val bob = PrivateKeyBuilder()
+        val alice = PrivateKeyBuilder()
+        val clientOptions =
+            ClientOptions(api = ClientOptions.Api(env = XMTPEnvironment.LOCAL, isSecure = false))
+        val bobClient = Client().create(bob, clientOptions)
+        val aliceClient = Client().create(account = alice, options = clientOptions)
+        aliceClient.publishUserContact(legacy = true)
+        bobClient.publishUserContact(legacy = true)
+        val convo = ConversationV1(client = bobClient, peerAddress = alice.address, sentAt = Date())
+        convo.streamEphemeral().mapLatest {
+            assertEquals("hi", it.message.toStringUtf8())
+        }
+        convo.send(content = "hi", options = SendOptions(ephemeral = true))
+        val messages = convo.messages()
+        assertEquals(0, messages.size)
+    }
+
+    @Test
+    fun testStreamEphemeralInV2Conversation() {
+        val bob = PrivateKeyBuilder()
+        val alice = PrivateKeyBuilder()
+        val clientOptions =
+            ClientOptions(api = ClientOptions.Api(env = XMTPEnvironment.LOCAL, isSecure = false))
+        val bobClient = Client().create(bob, clientOptions)
+        val aliceClient = Client().create(account = alice, options = clientOptions)
+        val aliceConversation = aliceClient.conversations.newConversation(
+            bob.address,
+            context = InvitationV1ContextBuilder.buildFromConversation("https://example.com/3")
+        )
+        val bobConversation = bobClient.conversations.newConversation(
+            alice.address,
+            context = InvitationV1ContextBuilder.buildFromConversation("https://example.com/3")
+        )
+
+        bobConversation.streamEphemeral().mapLatest {
+            assertEquals("hi", it.message.toStringUtf8())
+        }
+        aliceConversation.send(content = "hi", options = SendOptions(ephemeral = true))
+        val messages = aliceConversation.messages()
+        assertEquals(0, messages.size)
+    }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/Conversation.kt b/library/src/main/java/org/xmtp/android/library/Conversation.kt
index 35460b96e..05fbbb8ef 100644
--- a/library/src/main/java/org/xmtp/android/library/Conversation.kt
+++ b/library/src/main/java/org/xmtp/android/library/Conversation.kt
@@ -127,12 +127,18 @@ sealed class Conversation {
         }
     }
 
-    fun send(encodedContent: EncodedContent): String {
+    fun send(encodedContent: EncodedContent, options: SendOptions? = null): String {
         return when (this) {
-            is V1 -> conversationV1.send(encodedContent = encodedContent)
-            is V2 -> conversationV2.send(encodedContent = encodedContent)
+            is V1 -> conversationV1.send(encodedContent = encodedContent, options = options)
+            is V2 -> conversationV2.send(encodedContent = encodedContent, options = options)
         }
     }
+
+    val clientAddress: String
+        get() {
+            return client.address
+        }
+
     val topic: String
         get() {
             return when (this) {
@@ -176,4 +182,11 @@ sealed class Conversation {
             is V2 -> conversationV2.streamMessages()
         }
     }
+
+    fun streamEphemeral(): Flow<Envelope> {
+        return when (this) {
+            is V1 -> return conversationV1.streamEphemeral()
+            is V2 -> return conversationV2.streamEphemeral()
+        }
+    }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/ConversationV1.kt b/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
index f80132e88..d0441c36f 100644
--- a/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
+++ b/library/src/main/java/org/xmtp/android/library/ConversationV1.kt
@@ -2,6 +2,7 @@ package org.xmtp.android.library
 
 import android.util.Log
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.runBlocking
 import org.web3j.crypto.Hash
@@ -97,8 +98,8 @@ data class ConversationV1(
         return preparedMessage.messageId
     }
 
-    fun send(encodedContent: EncodedContent): String {
-        val preparedMessage = prepareMessage(encodedContent = encodedContent)
+    fun send(encodedContent: EncodedContent, options: SendOptions? = null): String {
+        val preparedMessage = prepareMessage(encodedContent = encodedContent, options = options)
         preparedMessage.send()
         return preparedMessage.messageId
     }
@@ -123,10 +124,13 @@ data class ConversationV1(
         if (compression != null) {
             encoded = encoded.compress(compression)
         }
-        return prepareMessage(encodedContent = encoded)
+        return prepareMessage(encodedContent = encoded, options = options)
     }
 
-    fun prepareMessage(encodedContent: EncodedContent): PreparedMessage {
+    fun prepareMessage(
+        encodedContent: EncodedContent,
+        options: SendOptions? = null,
+    ): PreparedMessage {
         val contact = client.contacts.find(peerAddress) ?: throw XMTPException("address not found")
         val recipient = contact.toPublicKeyBundle()
         if (!recipient.identityKey.hasSignature()) {
@@ -139,9 +143,12 @@ data class ConversationV1(
             message = encodedContent.toByteArray(),
             timestamp = date
         )
+
+        val isEphemeral: Boolean = options != null && options.ephemeral
+
         val messageEnvelope =
-            EnvelopeBuilder.buildFromTopic(
-                topic = Topic.directMessageV1(client.address, peerAddress),
+            EnvelopeBuilder.buildFromString(
+                topic = if (isEphemeral) ephemeralTopic else topic.description,
                 timestamp = date,
                 message = MessageBuilder.buildFromMessageV1(v1 = message).toByteArray()
             )
@@ -150,7 +157,7 @@ data class ConversationV1(
             conversation = Conversation.V1(this)
         ) {
             val envelopes = mutableListOf(messageEnvelope)
-            if (client.contacts.needsIntroduction(peerAddress)) {
+            if (client.contacts.needsIntroduction(peerAddress) && !isEphemeral) {
                 envelopes.addAll(
                     listOf(
                         EnvelopeBuilder.buildFromTopic(
@@ -173,4 +180,13 @@ data class ConversationV1(
 
     private fun generateId(envelope: Envelope): String =
         Hash.sha256(envelope.message.toByteArray()).toHex()
+
+    val ephemeralTopic: String
+        get() = topic.description.replace("/xmtp/0/dm-", "/xmtp/0/dmE-")
+
+    fun streamEphemeral(): Flow<Envelope> = flow {
+        client.subscribe(topics = listOf(ephemeralTopic)).collect {
+            emit(it)
+        }
+    }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/ConversationV2.kt b/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
index a80ab79fc..71165aa50 100644
--- a/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
+++ b/library/src/main/java/org/xmtp/android/library/ConversationV2.kt
@@ -113,8 +113,8 @@ data class ConversationV2(
         return preparedMessage.messageId
     }
 
-    fun send(encodedContent: EncodedContent): String {
-        val preparedMessage = prepareMessage(encodedContent = encodedContent)
+    fun send(encodedContent: EncodedContent, options: SendOptions?): String {
+        val preparedMessage = prepareMessage(encodedContent = encodedContent, options = options)
         preparedMessage.send()
         return preparedMessage.messageId
     }
@@ -155,18 +155,21 @@ data class ConversationV2(
         if (compression != null) {
             encoded = encoded.compress(compression)
         }
-        return prepareMessage(encoded)
+        return prepareMessage(encoded, options = options)
     }
 
-    fun prepareMessage(encodedContent: EncodedContent): PreparedMessage {
+    fun prepareMessage(encodedContent: EncodedContent, options: SendOptions?): PreparedMessage {
         val message = MessageV2Builder.buildEncode(
             client = client,
             encodedContent = encodedContent,
             topic = topic,
             keyMaterial = keyMaterial
         )
+
+        val newTopic = if (options?.ephemeral == true) ephemeralTopic else topic
+
         val envelope = EnvelopeBuilder.buildFromString(
-            topic = topic,
+            topic = newTopic,
             timestamp = Date(),
             message = MessageBuilder.buildFromMessageV2(v2 = message).toByteArray()
         )
@@ -177,4 +180,13 @@ data class ConversationV2(
 
     private fun generateId(envelope: Envelope): String =
         Hash.sha256(envelope.message.toByteArray()).toHex()
+
+    val ephemeralTopic: String
+        get() = topic.replace("/xmtp/0/m", "/xmtp/0/mE")
+
+    fun streamEphemeral(): Flow<Envelope> = flow {
+        client.subscribe(topics = listOf(ephemeralTopic)).collect {
+            emit(it)
+        }
+    }
 }
diff --git a/library/src/main/java/org/xmtp/android/library/SendOptions.kt b/library/src/main/java/org/xmtp/android/library/SendOptions.kt
index 71e183163..e2077d350 100644
--- a/library/src/main/java/org/xmtp/android/library/SendOptions.kt
+++ b/library/src/main/java/org/xmtp/android/library/SendOptions.kt
@@ -5,5 +5,6 @@ import org.xmtp.proto.message.contents.Content
 data class SendOptions(
     var compression: EncodedContentCompression? = null,
     var contentType: Content.ContentTypeId? = null,
-    var contentFallback: String? = null
+    var contentFallback: String? = null,
+    var ephemeral: Boolean = false
 )