diff --git a/changelog.d/23509_allow_headers_in_nats_jetstream_sink.enhancement.md b/changelog.d/23509_allow_headers_in_nats_jetstream_sink.enhancement.md
new file mode 100644
index 0000000000000..a709c4c9f5c2a
--- /dev/null
+++ b/changelog.d/23509_allow_headers_in_nats_jetstream_sink.enhancement.md
@@ -0,0 +1,5 @@
+The `nats` sink now supports message headers when publishing to JetStream.
+
+It introduces a configurable, templated Nats-Msg-Id header that ensures a unique ID for each message. This enables broker-level deduplication, resulting in stronger delivery guarantees and exactly-once semantics when combined with idempotent consumers.
+
+authors: benjamin-awd
diff --git a/scripts/integration/nats/compose.yaml b/scripts/integration/nats/compose.yaml
index fb36e78d819d4..9be71dc529380 100644
--- a/scripts/integration/nats/compose.yaml
+++ b/scripts/integration/nats/compose.yaml
@@ -43,3 +43,10 @@ services:
     - /usr/share/nats/config/nats-jwt.conf
     volumes:
     - ../../../tests/data/nats:/usr/share/nats/config
+  nats-jetstream-test:
+    image: docker.io/library/nats:${CONFIG_VERSION}
+    command:
+    - --config
+    - /usr/share/nats/config/nats-jetstream.conf
+    volumes:
+    - ../../../tests/data/nats:/usr/share/nats/config
diff --git a/scripts/integration/nats/test.yaml b/scripts/integration/nats/test.yaml
index 553c9efa25257..171a1c4ae6a3e 100644
--- a/scripts/integration/nats/test.yaml
+++ b/scripts/integration/nats/test.yaml
@@ -1,7 +1,7 @@
 features:
-- nats-integration-tests
+  - nats-integration-tests
 
-test_filter: '::nats::'
+test_filter: "::nats::"
 
 env:
   NATS_ADDRESS: nats://nats:4222
@@ -11,6 +11,7 @@ env:
   NATS_TLS_CLIENT_CERT_ADDRESS: nats://nats-tls-client-cert:4222
   NATS_TOKEN_ADDRESS: nats://nats-token:4222
   NATS_USERPASS_ADDRESS: nats://nats-userpass:4222
+  NATS_JETSTREAM_ADDRESS: nats://nats-jetstream-test:4222
 
 matrix:
   version: [latest]
@@ -18,10 +19,10 @@ matrix:
 # changes to these files/paths will invoke the integration test in CI
 # expressions are evaluated using https://github.com/micromatch/picomatch
 paths:
-- "src/internal_events/nats.rs"
-- "src/sources/nats.rs"
-- "src/sources/util/**"
-- "src/sinks/nats/**"
-- "src/sinks/util/**"
-- "src/nats.rs"
-- "scripts/integration/nats/**"
+  - "src/internal_events/nats.rs"
+  - "src/sources/nats.rs"
+  - "src/sources/util/**"
+  - "src/sinks/nats/**"
+  - "src/sinks/util/**"
+  - "src/nats.rs"
+  - "scripts/integration/nats/**"
diff --git a/src/sinks/nats/config.rs b/src/sinks/nats/config.rs
index c5209c54041bf..9082e5123714d 100644
--- a/src/sinks/nats/config.rs
+++ b/src/sinks/nats/config.rs
@@ -1,15 +1,16 @@
+use async_nats::header;
 use bytes::Bytes;
 use futures_util::TryFutureExt;
 use snafu::ResultExt;
 use vector_lib::codecs::JsonSerializerConfig;
 use vector_lib::tls::TlsEnableableConfig;
 
+use super::{sink::NatsSink, ConfigSnafu, ConnectSnafu, NatsError};
 use crate::{
     nats::{from_tls_auth_config, NatsAuthConfig, NatsConfigError},
     sinks::{prelude::*, util::service::TowerRequestConfigDefaults},
 };
-
-use super::{sink::NatsSink, ConfigSnafu, ConnectSnafu, NatsError};
+use async_nats::HeaderMap;
 
 #[derive(Clone, Copy, Debug)]
 pub struct NatsTowerRequestConfigDefaults;
@@ -18,6 +19,61 @@ impl TowerRequestConfigDefaults for NatsTowerRequestConfigDefaults {
     const CONCURRENCY: Concurrency = Concurrency::None;
 }
 
+/// A set of NATS headers that can be added to each message.
+#[configurable_component]
+#[serde_with::serde_as]
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct NatsHeaderConfig {
+    /// A unique identifier for the message. Useful for deduplication.
+    ///
+    /// Can be a template that references fields in the event, e.g., `{{ event_id }}`.
+    #[configurable(metadata(docs::templateable))]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[configurable(metadata(docs::examples = "{{ event_id }}"))]
+    pub(super) message_id: Option<Template>,
+}
+
+impl NatsHeaderConfig {
+    pub fn build_headers(&self, event: &Event) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+
+        if let Some(template) = &self.message_id {
+            if let Ok(value) = template.render_string(event) {
+                headers.insert(header::NATS_MESSAGE_ID, value.as_str());
+            }
+        }
+
+        headers
+    }
+}
+
+/// Configuration for sending messages using NATS JetStream.
+#[configurable_component]
+#[serde_with::serde_as]
+#[derive(Clone, Debug, Eq, PartialEq, Default)]
+#[serde(deny_unknown_fields)]
+pub struct JetStreamConfig {
+    /// Whether to enable Jetstream.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub enabled: bool,
+
+    /// A map of NATS headers to be included in each message.
+    #[configurable(metadata(docs::templateable))]
+    #[serde(default)]
+    pub(super) headers: Option<NatsHeaderConfig>,
+}
+
+impl From<bool> for JetStreamConfig {
+    fn from(enabled: bool) -> Self {
+        Self {
+            enabled,
+            ..Default::default()
+        }
+    }
+}
+
 /// Configuration for the `nats` sink.
 #[configurable_component(sink(
     "nats",
@@ -86,8 +142,13 @@ pub struct NatsSinkConfig {
     /// If set, the `subject` must belong to an existing JetStream stream.
     ///
     /// [jetstream]: https://docs.nats.io/nats-concepts/jetstream
-    #[serde(default)]
-    pub(super) jetstream: bool,
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::is_default"
+    )]
+    pub(super) jetstream: JetStreamConfig,
 }
 
 fn default_name() -> String {
@@ -105,7 +166,12 @@ impl GenerateConfig for NatsSinkConfig {
             tls: None,
             url: "nats://127.0.0.1:4222".into(),
             request: Default::default(),
-            jetstream: Default::default(),
+            jetstream: JetStreamConfig {
+                enabled: true,
+                headers: Some(NatsHeaderConfig {
+                    message_id: Some(Template::try_from("{{ event_id }}").unwrap()),
+                }),
+            },
         })
         .unwrap()
     }
@@ -175,7 +241,7 @@ impl NatsSinkConfig {
         let options = self.create_connect_options()?;
         let connection = self.connect(options).await?;
 
-        if self.jetstream {
+        if self.jetstream.enabled {
             Ok(NatsPublisher::JetStream(async_nats::jetstream::new(
                 connection,
             )))
@@ -203,6 +269,7 @@ impl NatsPublisher {
     pub(super) async fn publish<S: async_nats::subject::ToSubject>(
         &self,
         subject: S,
+        headers: HeaderMap,
         payload: Bytes,
     ) -> Result<(), NatsError> {
         match self {
@@ -222,11 +289,12 @@ impl NatsPublisher {
                     .await
             }
             NatsPublisher::JetStream(jetstream) => {
-                let ack = jetstream.publish(subject, payload).await.map_err(|e| {
-                    NatsError::PublishError {
+                let ack = jetstream
+                    .publish_with_headers(subject, headers, payload)
+                    .await
+                    .map_err(|e| NatsError::PublishError {
                         source: Box::new(e),
-                    }
-                })?;
+                    })?;
                 ack.await.map(|_| ()).map_err(|e| NatsError::PublishError {
                     source: Box::new(e),
                 })
diff --git a/src/sinks/nats/integration_tests.rs b/src/sinks/nats/integration_tests.rs
index 3555c55254d98..6e76d362e75c5 100644
--- a/src/sinks/nats/integration_tests.rs
+++ b/src/sinks/nats/integration_tests.rs
@@ -1,11 +1,13 @@
-use std::time::Duration;
-use vector_lib::codecs::TextSerializerConfig;
-
-use super::{config::NatsSinkConfig, sink::NatsSink, ConfigSnafu, NatsError};
+use super::{
+    config::{NatsHeaderConfig, NatsSinkConfig},
+    sink::NatsSink,
+    ConfigSnafu, NatsError,
+};
 use crate::{
     nats::{
         NatsAuthConfig, NatsAuthCredentialsFile, NatsAuthNKey, NatsAuthToken, NatsAuthUserPassword,
     },
+    sinks::nats::config::JetStreamConfig,
     sinks::prelude::*,
     test_util::{
         components::{run_and_assert_sink_compliance, SINK_TAGS},
@@ -13,7 +15,28 @@ use crate::{
     },
     tls::TlsEnableableConfig,
 };
+use async_nats::jetstream::stream::StorageType;
+use futures_util::StreamExt;
+use serde::Deserialize;
 use snafu::ResultExt;
+use std::time::Duration;
+use vector_lib::codecs::{JsonSerializerConfig, TextSerializerConfig};
+use vector_lib::event::{EventArray, LogEvent};
+use vrl::value;
+
+fn generate_sink_config(url: &str, subject: &str) -> NatsSinkConfig {
+    NatsSinkConfig {
+        acknowledgements: Default::default(),
+        encoding: TextSerializerConfig::default().into(),
+        connection_name: "".to_owned(),
+        subject: Template::try_from(subject).unwrap(),
+        url: url.to_string(),
+        tls: None,
+        auth: None,
+        request: Default::default(),
+        jetstream: false.into(),
+    }
+}
 
 async fn publish_and_check(conf: NatsSinkConfig) -> Result<(), NatsError> {
     // Publish `N` messages to NATS.
@@ -71,17 +94,7 @@ async fn nats_no_auth() {
     let url =
         std::env::var("NATS_ADDRESS").unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: None,
-        request: Default::default(),
-        jetstream: false,
-    };
+    let conf = generate_sink_config(&url, &subject);
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -98,22 +111,13 @@ async fn nats_userpass_auth_valid() {
     let url = std::env::var("NATS_USERPASS_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: Some(NatsAuthConfig::UserPassword {
-            user_password: NatsAuthUserPassword {
-                user: "natsuser".to_string(),
-                password: "natspass".to_string().into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.auth = Some(NatsAuthConfig::UserPassword {
+        user_password: NatsAuthUserPassword {
+            user: "natsuser".to_string(),
+            password: "natspass".to_string().into(),
+        },
+    });
 
     publish_and_check(conf)
         .await
@@ -128,22 +132,13 @@ async fn nats_userpass_auth_invalid() {
     let url = std::env::var("NATS_USERPASS_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: Some(NatsAuthConfig::UserPassword {
-            user_password: NatsAuthUserPassword {
-                user: "natsuser".to_string(),
-                password: "wrongpass".to_string().into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.auth = Some(NatsAuthConfig::UserPassword {
+        user_password: NatsAuthUserPassword {
+            user: "natsuser".to_string(),
+            password: "wrongpass".to_string().into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -160,21 +155,12 @@ async fn nats_token_auth_valid() {
     let url = std::env::var("NATS_TOKEN_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: Some(NatsAuthConfig::Token {
-            token: NatsAuthToken {
-                value: "secret".to_string().into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.auth = Some(NatsAuthConfig::Token {
+        token: NatsAuthToken {
+            value: "secret".to_string().into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -191,21 +177,12 @@ async fn nats_token_auth_invalid() {
     let url = std::env::var("NATS_TOKEN_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: Some(NatsAuthConfig::Token {
-            token: NatsAuthToken {
-                value: "wrongsecret".to_string().into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.auth = Some(NatsAuthConfig::Token {
+        token: NatsAuthToken {
+            value: "wrongsecret".to_string().into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -222,22 +199,13 @@ async fn nats_nkey_auth_valid() {
     let url = std::env::var("NATS_NKEY_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: Some(NatsAuthConfig::Nkey {
-            nkey: NatsAuthNKey {
-                nkey: "UD345ZYSUJQD7PNCTWQPINYSO3VH4JBSADBSYUZOBT666DRASFRAWAWT".into(),
-                seed: "SUANIRXEZUROTXNFN3TJYMT27K7ZZVMD46FRIHF6KXKS4KGNVBS57YAFGY".into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.auth = Some(NatsAuthConfig::Nkey {
+        nkey: NatsAuthNKey {
+            nkey: "UD345ZYSUJQD7PNCTWQPINYSO3VH4JBSADBSYUZOBT666DRASFRAWAWT".into(),
+            seed: "SUANIRXEZUROTXNFN3TJYMT27K7ZZVMD46FRIHF6KXKS4KGNVBS57YAFGY".into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -254,22 +222,13 @@ async fn nats_nkey_auth_invalid() {
     let url = std::env::var("NATS_NKEY_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: Some(NatsAuthConfig::Nkey {
-            nkey: NatsAuthNKey {
-                nkey: "UAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".into(),
-                seed: "SBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB".into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.auth = Some(NatsAuthConfig::Nkey {
+        nkey: NatsAuthNKey {
+            nkey: "UAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".into(),
+            seed: "SBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB".into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -286,23 +245,14 @@ async fn nats_tls_valid() {
     let url =
         std::env::var("NATS_TLS_ADDRESS").unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: Some(TlsEnableableConfig {
-            enabled: Some(true),
-            options: TlsConfig {
-                ca_file: Some("tests/data/nats/rootCA.pem".into()),
-                ..Default::default()
-            },
-        }),
-        auth: None,
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.tls = Some(TlsEnableableConfig {
+        enabled: Some(true),
+        options: TlsConfig {
+            ca_file: Some("tests/data/nats/rootCA.pem".into()),
+            ..Default::default()
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -319,17 +269,7 @@ async fn nats_tls_invalid() {
     let url =
         std::env::var("NATS_TLS_ADDRESS").unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: None,
-        auth: None,
-        request: Default::default(),
-        jetstream: false,
-    };
+    let conf = generate_sink_config(&url, &subject);
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -346,25 +286,16 @@ async fn nats_tls_client_cert_valid() {
     let url = std::env::var("NATS_TLS_CLIENT_CERT_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: Some(TlsEnableableConfig {
-            enabled: Some(true),
-            options: TlsConfig {
-                ca_file: Some("tests/data/nats/rootCA.pem".into()),
-                crt_file: Some("tests/data/nats/nats-client.pem".into()),
-                key_file: Some("tests/data/nats/nats-client.key".into()),
-                ..Default::default()
-            },
-        }),
-        auth: None,
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.tls = Some(TlsEnableableConfig {
+        enabled: Some(true),
+        options: TlsConfig {
+            ca_file: Some("tests/data/nats/rootCA.pem".into()),
+            crt_file: Some("tests/data/nats/nats-client.pem".into()),
+            key_file: Some("tests/data/nats/nats-client.key".into()),
+            ..Default::default()
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -381,23 +312,14 @@ async fn nats_tls_client_cert_invalid() {
     let url = std::env::var("NATS_TLS_CLIENT_CERT_ADDRESS")
         .unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: Some(TlsEnableableConfig {
-            enabled: Some(true),
-            options: TlsConfig {
-                ca_file: Some("tests/data/nats/rootCA.pem".into()),
-                ..Default::default()
-            },
-        }),
-        auth: None,
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.tls = Some(TlsEnableableConfig {
+        enabled: Some(true),
+        options: TlsConfig {
+            ca_file: Some("tests/data/nats/rootCA.pem".into()),
+            ..Default::default()
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -414,27 +336,19 @@ async fn nats_tls_jwt_auth_valid() {
     let url =
         std::env::var("NATS_JWT_ADDRESS").unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: Some(TlsEnableableConfig {
-            enabled: Some(true),
-            options: TlsConfig {
-                ca_file: Some("tests/data/nats/rootCA.pem".into()),
-                ..Default::default()
-            },
-        }),
-        auth: Some(NatsAuthConfig::CredentialsFile {
-            credentials_file: NatsAuthCredentialsFile {
-                path: "tests/data/nats/nats.creds".into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.tls = Some(TlsEnableableConfig {
+        enabled: Some(true),
+        options: TlsConfig {
+            ca_file: Some("tests/data/nats/rootCA.pem".into()),
+            ..Default::default()
+        },
+    });
+    conf.auth = Some(NatsAuthConfig::CredentialsFile {
+        credentials_file: NatsAuthCredentialsFile {
+            path: "tests/data/nats/nats.creds".into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -451,27 +365,19 @@ async fn nats_tls_jwt_auth_invalid() {
     let url =
         std::env::var("NATS_JWT_ADDRESS").unwrap_or_else(|_| String::from("nats://localhost:4222"));
 
-    let conf = NatsSinkConfig {
-        acknowledgements: Default::default(),
-        encoding: TextSerializerConfig::default().into(),
-        connection_name: "".to_owned(),
-        subject: Template::try_from(subject.as_str()).unwrap(),
-        url,
-        tls: Some(TlsEnableableConfig {
-            enabled: Some(true),
-            options: TlsConfig {
-                ca_file: Some("tests/data/nats/rootCA.pem".into()),
-                ..Default::default()
-            },
-        }),
-        auth: Some(NatsAuthConfig::CredentialsFile {
-            credentials_file: NatsAuthCredentialsFile {
-                path: "tests/data/nats/nats-bad.creds".into(),
-            },
-        }),
-        request: Default::default(),
-        jetstream: false,
-    };
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.tls = Some(TlsEnableableConfig {
+        enabled: Some(true),
+        options: TlsConfig {
+            ca_file: Some("tests/data/nats/rootCA.pem".into()),
+            ..Default::default()
+        },
+    });
+    conf.auth = Some(NatsAuthConfig::CredentialsFile {
+        credentials_file: NatsAuthCredentialsFile {
+            path: "tests/data/nats/nats-bad.creds".into(),
+        },
+    });
 
     let r = publish_and_check(conf).await;
     assert!(
@@ -479,3 +385,139 @@ async fn nats_tls_jwt_auth_invalid() {
         "publish_and_check failed, expected NatsError::Connect, got: {r:?}"
     );
 }
+
+#[tokio::test]
+async fn nats_jetstream_valid() {
+    trace_init();
+    let stream_name = format!("EVENTS_{}", random_string(10));
+    let subject = format!("events.test-{}", random_string(10));
+
+    let url = std::env::var("NATS_JETSTREAM_ADDRESS")
+        .unwrap_or_else(|_| String::from("nats://localhost:4222"));
+
+    let client = async_nats::connect(&url)
+        .await
+        .expect("Failed to connect to NATS");
+
+    let js = async_nats::jetstream::new(client);
+
+    js.get_or_create_stream(async_nats::jetstream::stream::Config {
+        name: stream_name.clone(),
+        subjects: vec![subject.clone()],
+        storage: StorageType::Memory,
+        ..Default::default()
+    })
+    .await
+    .expect("Failed to create JetStream stream");
+
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.jetstream = JetStreamConfig {
+        enabled: true,
+        ..Default::default()
+    };
+
+    let r = publish_and_check(conf).await;
+    assert!(
+        r.is_ok(),
+        "publish_and_check failed, expected Ok(()), got: {r:?}"
+    );
+}
+
+#[derive(Debug, Deserialize)]
+struct JetstreamTestEvent {
+    id: String,
+    message: String,
+}
+
+// Check that message ids are dynamically assigned
+// and that they are correctly deduplicated at the stream level
+#[tokio::test]
+async fn nats_jetstream_message_id_valid() {
+    trace_init();
+    let stream_name = format!("EVENTS_MSG_ID_{}", random_string(10));
+    let subject = format!("events.test.msg.id-{}", random_string(10));
+
+    let url = std::env::var("NATS_JETSTREAM_ADDRESS")
+        .unwrap_or_else(|_| String::from("nats://localhost:4222"));
+
+    let client = async_nats::connect(&url).await.unwrap();
+    let js = async_nats::jetstream::new(client);
+    js.get_or_create_stream(async_nats::jetstream::stream::Config {
+        name: stream_name.clone(),
+        subjects: vec![subject.clone()],
+        storage: StorageType::Memory,
+        duplicate_window: std::time::Duration::from_secs(60),
+        ..Default::default()
+    })
+    .await
+    .unwrap();
+
+    let mut conf = generate_sink_config(&url, &subject);
+    conf.encoding = JsonSerializerConfig::default().into();
+
+    let header_config = NatsHeaderConfig {
+        message_id: Some(Template::try_from("{{ id }}").unwrap()),
+    };
+
+    conf.jetstream = JetStreamConfig {
+        enabled: true,
+        headers: Some(header_config),
+    };
+
+    let sink = NatsSink::new(conf.clone()).await.unwrap();
+    let sink = VectorSink::from_event_streamsink(sink);
+
+    let event_id = "123";
+
+    let event1 = LogEvent::from(value!({
+        "id": event_id,
+        "message": "first message",
+    }));
+
+    let event2 = LogEvent::from(value!({
+        "id": event_id,
+        "message": "second message",
+    }));
+
+    let event3 = LogEvent::from(value!({
+        "id": event_id,
+        "message": "third message",
+    }));
+
+    let event_array = EventArray::Logs(vec![event1, event2, event3]);
+    sink.run(futures::stream::iter(vec![event_array]).boxed())
+        .await
+        .unwrap();
+
+    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+
+    let stream = js.get_stream(&stream_name).await.unwrap();
+    let consumer = stream
+        .create_consumer(async_nats::jetstream::consumer::pull::Config {
+            durable_name: Some(format!("test-consumer-{}", random_string(5))),
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+
+    let mut processed_messages = Vec::new();
+
+    let messages = consumer.fetch().max_messages(2).messages().await;
+    let mut stream = messages.expect("Failed to get stream");
+
+    while let Some(Ok(msg)) = stream.next().await {
+        msg.ack().await.unwrap();
+        processed_messages.push(msg);
+    }
+
+    assert_eq!(
+        processed_messages.len(),
+        1,
+        "Expected only one message due to deduplication"
+    );
+
+    let msg = &processed_messages[0];
+    let received_event: JetstreamTestEvent = serde_json::from_slice(&msg.payload).unwrap();
+    assert_eq!(received_event.id, event_id);
+    assert_eq!(received_event.message, "first message");
+}
diff --git a/src/sinks/nats/request_builder.rs b/src/sinks/nats/request_builder.rs
index 08368e138d368..660810f9b7b53 100644
--- a/src/sinks/nats/request_builder.rs
+++ b/src/sinks/nats/request_builder.rs
@@ -1,12 +1,13 @@
 use std::io;
 
+use async_nats::HeaderMap;
 use bytes::{Bytes, BytesMut};
 use tokio_util::codec::Encoder as _;
 use vector_lib::config::telemetry;
 
 use crate::sinks::prelude::*;
 
-use super::sink::NatsEvent;
+use super::{config::NatsHeaderConfig, sink::NatsEvent};
 
 pub(super) struct NatsEncoder {
     pub(super) transformer: Transformer,
@@ -39,17 +40,20 @@ impl encoding::Encoder<Event> for NatsEncoder {
 
 pub(super) struct NatsMetadata {
     subject: String,
+    headers: HeaderMap,
     finalizers: EventFinalizers,
 }
 
 pub(super) struct NatsRequestBuilder {
     pub(super) encoder: NatsEncoder,
+    pub(super) headers: Option<NatsHeaderConfig>,
 }
 
 #[derive(Clone)]
 pub(super) struct NatsRequest {
     pub(super) bytes: Bytes,
     pub(super) subject: String,
+    pub(super) headers: HeaderMap,
     finalizers: EventFinalizers,
     pub(super) metadata: RequestMetadata,
 }
@@ -91,9 +95,15 @@ impl RequestBuilder<NatsEvent> for NatsRequestBuilder {
         mut input: NatsEvent,
     ) -> (Self::Metadata, RequestMetadataBuilder, Self::Events) {
         let builder = RequestMetadataBuilder::from_event(&input.event);
+        let headers = self
+            .headers
+            .as_ref()
+            .map(|config| config.build_headers(&input.event))
+            .unwrap_or_default();
 
         let metadata = NatsMetadata {
             subject: input.subject,
+            headers,
             finalizers: input.event.take_finalizers(),
         };
 
@@ -110,6 +120,7 @@ impl RequestBuilder<NatsEvent> for NatsRequestBuilder {
         NatsRequest {
             bytes: body,
             subject: nats_metadata.subject,
+            headers: nats_metadata.headers,
             finalizers: nats_metadata.finalizers,
             metadata,
         }
diff --git a/src/sinks/nats/service.rs b/src/sinks/nats/service.rs
index 1aeccced639f9..08b098aae46ff 100644
--- a/src/sinks/nats/service.rs
+++ b/src/sinks/nats/service.rs
@@ -48,7 +48,7 @@ impl Service<NatsRequest> for NatsService {
 
         Box::pin(async move {
             match publisher
-                .publish(req.subject, req.bytes)
+                .publish(req.subject, req.headers, req.bytes)
                 .map_err(async_nats::Error::from)
                 .await
             {
diff --git a/src/sinks/nats/sink.rs b/src/sinks/nats/sink.rs
index a5b94d4a9c759..bfec91a999ae7 100644
--- a/src/sinks/nats/sink.rs
+++ b/src/sinks/nats/sink.rs
@@ -5,7 +5,7 @@ use snafu::ResultExt;
 use crate::sinks::prelude::*;
 
 use super::{
-    config::{NatsPublisher, NatsSinkConfig, NatsTowerRequestConfigDefaults},
+    config::{NatsHeaderConfig, NatsPublisher, NatsSinkConfig, NatsTowerRequestConfigDefaults},
     request_builder::{NatsEncoder, NatsRequestBuilder},
     service::{NatsResponse, NatsService},
     EncodingSnafu, NatsError,
@@ -22,6 +22,7 @@ pub(super) struct NatsSink {
     encoder: Encoder<()>,
     publisher: Arc<NatsPublisher>,
     subject: Template,
+    headers: Option<NatsHeaderConfig>,
 }
 
 impl NatsSink {
@@ -48,6 +49,7 @@ impl NatsSink {
         let encoder = Encoder::<()>::new(serializer);
         let request = config.request;
         let subject = config.subject;
+        let headers = config.jetstream.headers;
 
         Ok(NatsSink {
             request,
@@ -55,6 +57,7 @@ impl NatsSink {
             encoder,
             publisher,
             subject,
+            headers,
         })
     }
 
@@ -66,6 +69,7 @@ impl NatsSink {
                 encoder: self.encoder.clone(),
                 transformer: self.transformer.clone(),
             },
+            headers: self.headers.clone(),
         };
 
         let service = ServiceBuilder::new()
diff --git a/tests/data/nats/nats-jetstream.conf b/tests/data/nats/nats-jetstream.conf
new file mode 100644
index 0000000000000..528e6ab71d92f
--- /dev/null
+++ b/tests/data/nats/nats-jetstream.conf
@@ -0,0 +1,3 @@
+jetstream {
+  store_dir: "/tmp/nats/jetstream"
+}
diff --git a/website/cue/reference/components/sinks/generated/nats.cue b/website/cue/reference/components/sinks/generated/nats.cue
index edb77b2b1bbe1..0260a118b89a4 100644
--- a/website/cue/reference/components/sinks/generated/nats.cue
+++ b/website/cue/reference/components/sinks/generated/nats.cue
@@ -484,7 +484,29 @@ generated: components: sinks: nats: configuration: {
 			[jetstream]: https://docs.nats.io/nats-concepts/jetstream
 			"""
 		required: false
-		type: bool: default: false
+		type: object: options: {
+			enabled: {
+				description: "Whether to enable Jetstream."
+				required:    false
+				type: bool: default: false
+			}
+			headers: {
+				description: "A map of NATS headers to be included in each message."
+				required:    false
+				type: object: options: message_id: {
+					description: """
+						A unique identifier for the message. Useful for deduplication.
+
+						Can be a template that references fields in the event, e.g., `{{ event_id }}`.
+						"""
+					required: false
+					type: string: {
+						examples: ["{{ event_id }}"]
+						syntax: "template"
+					}
+				}
+			}
+		}
 	}
 	request: {
 		description: """