diff --git a/packages/pynumaflow-lite/Cargo.toml b/packages/pynumaflow-lite/Cargo.toml
index 2d0d4dd6..3c0939ac 100644
--- a/packages/pynumaflow-lite/Cargo.toml
+++ b/packages/pynumaflow-lite/Cargo.toml
@@ -48,6 +48,10 @@ path = "tests/bin/session_reduce.rs"
 name = "test_accumulator"
 path = "tests/bin/accumulator.rs"
 
+[[bin]]
+name = "test_reducestream"
+path = "tests/bin/reducestream.rs"
+
 [[bin]]
 name = "test_sink"
 path = "tests/bin/sink.rs"
diff --git a/packages/pynumaflow-lite/pynumaflow_lite/__init__.py b/packages/pynumaflow-lite/pynumaflow_lite/__init__.py
index da65d338..f29648be 100644
--- a/packages/pynumaflow-lite/pynumaflow_lite/__init__.py
+++ b/packages/pynumaflow-lite/pynumaflow_lite/__init__.py
@@ -28,6 +28,11 @@
 except Exception:  # pragma: no cover
     session_reducer = None
 
+try:
+    reducestreamer = _import_module(__name__ + ".reducestreamer")
+except Exception:  # pragma: no cover
+    reducestreamer = None
+
 try:
     accumulator = _import_module(__name__ + ".accumulator")
 except Exception:  # pragma: no cover
@@ -43,12 +48,13 @@
 except Exception:  # pragma: no cover
     sourcer = None
 
-# Surface the Python Mapper, BatchMapper, MapStreamer, Reducer, SessionReducer, Accumulator, Sinker, and Sourcer classes under the extension submodules for convenient access
+# Surface the Python Mapper, BatchMapper, MapStreamer, Reducer, SessionReducer, ReduceStreamer, Accumulator, Sinker, and Sourcer classes under the extension submodules for convenient access
 from ._map_dtypes import Mapper
 from ._batchmapper_dtypes import BatchMapper
 from ._mapstream_dtypes import MapStreamer
 from ._reduce_dtypes import Reducer
 from ._session_reduce_dtypes import SessionReducer
+from ._reducestreamer_dtypes import ReduceStreamer
 from ._accumulator_dtypes import Accumulator
 from ._sink_dtypes import Sinker
 from ._source_dtypes import Sourcer
@@ -83,6 +89,12 @@
     except Exception:
         pass
 
+if reducestreamer is not None:
+    try:
+        setattr(reducestreamer, "ReduceStreamer", ReduceStreamer)
+    except Exception:
+        pass
+
 if accumulator is not None:
     try:
         setattr(accumulator, "Accumulator", Accumulator)
@@ -102,7 +114,7 @@
         pass
 
 # Public API
-__all__ = ["mapper", "batchmapper", "mapstreamer", "reducer", "session_reducer", "accumulator", "sinker", "sourcer"]
+__all__ = ["mapper", "batchmapper", "mapstreamer", "reducer", "session_reducer", "reducestreamer", "accumulator", "sinker", "sourcer"]
 
 __doc__ = pynumaflow_lite.__doc__
 if hasattr(pynumaflow_lite, "__all__"):
diff --git a/packages/pynumaflow-lite/pynumaflow_lite/_reducestreamer_dtypes.py b/packages/pynumaflow-lite/pynumaflow_lite/_reducestreamer_dtypes.py
new file mode 100644
index 00000000..ef84aebd
--- /dev/null
+++ b/packages/pynumaflow-lite/pynumaflow_lite/_reducestreamer_dtypes.py
@@ -0,0 +1,36 @@
+from abc import ABCMeta, abstractmethod
+from pynumaflow_lite.reducestreamer import Datum, Message, Metadata
+from collections.abc import AsyncIterable, AsyncIterator
+
+
+class ReduceStreamer(metaclass=ABCMeta):
+    """
+    Interface for reduce streaming handlers. A new instance will be created per window.
+    
+    Unlike regular Reducer which returns all messages at once, ReduceStreamer
+    allows you to yield messages incrementally as an async iterator.
+    """
+
+    def __call__(self, *args, **kwargs):
+        return self.handler(*args, **kwargs)
+
+    @abstractmethod
+    async def handler(
+        self, 
+        keys: list[str], 
+        datums: AsyncIterable[Datum], 
+        md: Metadata
+    ) -> AsyncIterator[Message]:
+        """
+        Implement this handler; consume `datums` async iterable and yield Messages incrementally.
+        
+        Args:
+            keys: List of keys for this window
+            datums: An async iterator of Datum objects
+            md: Metadata containing window information
+            
+        Yields:
+            Message objects to be sent to the next vertex
+        """
+        pass
+
diff --git a/packages/pynumaflow-lite/pynumaflow_lite/reducestreamer.pyi b/packages/pynumaflow-lite/pynumaflow_lite/reducestreamer.pyi
new file mode 100644
index 00000000..4f134cc0
--- /dev/null
+++ b/packages/pynumaflow-lite/pynumaflow_lite/reducestreamer.pyi
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import datetime as _dt
+from typing import Optional, List, Dict, Awaitable, AsyncIterator
+
+# Re-export the Python ABC for user convenience and typing
+from ._reducestreamer_dtypes import ReduceStreamer as ReduceStreamer
+
+
+class Message:
+    keys: Optional[List[str]]
+    value: bytes
+    tags: Optional[List[str]]
+
+    def __init__(
+            self,
+            value: bytes,
+            keys: Optional[List[str]] = ...,
+            tags: Optional[List[str]] = ...,
+    ) -> None: ...
+
+    @staticmethod
+    def message_to_drop() -> Message: ...
+
+
+class Datum:
+    keys: List[str]
+    value: bytes
+    watermark: _dt.datetime
+    eventtime: _dt.datetime
+    headers: Dict[str, str]
+
+    def __repr__(self) -> str: ...
+
+    def __str__(self) -> str: ...
+
+
+class IntervalWindow:
+    start: _dt.datetime
+    end: _dt.datetime
+
+
+class Metadata:
+    interval_window: IntervalWindow
+
+
+class PyAsyncDatumStream:
+    """
+    Python-visible async iterator that yields Datum items from a Tokio mpsc channel.
+    """
+
+    def __init__(self) -> None: ...
+    def __aiter__(self) -> PyAsyncDatumStream: ...
+    def __anext__(self) -> Datum: ...
+
+
+class ReduceStreamAsyncServer:
+    def __init__(
+            self,
+            sock_file: str = ...,
+            info_file: str = ...,
+    ) -> None: ...
+
+    def start(self, py_creator: type, init_args: tuple | None = ...) -> Awaitable[None]: ...
+
+    def stop(self) -> None: ...
+
+
+__all__ = [
+    "Message",
+    "Datum",
+    "IntervalWindow",
+    "Metadata",
+    "PyAsyncDatumStream",
+    "ReduceStreamAsyncServer",
+    "ReduceStreamer",
+]
+
diff --git a/packages/pynumaflow-lite/src/lib.rs b/packages/pynumaflow-lite/src/lib.rs
index 1522ae8e..6b386d11 100644
--- a/packages/pynumaflow-lite/src/lib.rs
+++ b/packages/pynumaflow-lite/src/lib.rs
@@ -5,6 +5,7 @@ pub mod mapstream;
 pub mod pyiterables;
 pub mod pyrs;
 pub mod reduce;
+pub mod reducestream;
 pub mod session_reduce;
 pub mod sink;
 pub mod source;
@@ -46,6 +47,13 @@ fn session_reducer(_py: Python, m: &Bound<PyModule>) -> PyResult<()> {
     Ok(())
 }
 
+/// Submodule: pynumaflow_lite.reducestreamer
+#[pymodule]
+fn reducestreamer(_py: Python, m: &Bound<PyModule>) -> PyResult<()> {
+    crate::reducestream::populate_py_module(m)?;
+    Ok(())
+}
+
 /// Submodule: pynumaflow_lite.accumulator
 #[pymodule]
 fn accumulator(_py: Python, m: &Bound<PyModule>) -> PyResult<()> {
@@ -76,6 +84,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
     m.add_wrapped(pyo3::wrap_pymodule!(mapstreamer))?;
     m.add_wrapped(pyo3::wrap_pymodule!(reducer))?;
     m.add_wrapped(pyo3::wrap_pymodule!(session_reducer))?;
+    m.add_wrapped(pyo3::wrap_pymodule!(reducestreamer))?;
     m.add_wrapped(pyo3::wrap_pymodule!(accumulator))?;
     m.add_wrapped(pyo3::wrap_pymodule!(sinker))?;
     m.add_wrapped(pyo3::wrap_pymodule!(sourcer))?;
@@ -125,6 +134,15 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
         .getattr("modules")?
         .set_item(fullname, &sub)?;
 
+    // Ensure it's importable as `pynumaflow_lite.reducestreamer` as well
+    let binding = m.getattr("reducestreamer")?;
+    let sub = binding.cast::<PyModule>()?;
+    let fullname = "pynumaflow_lite.reducestreamer";
+    sub.setattr("__name__", fullname)?;
+    py.import("sys")?
+        .getattr("modules")?
+        .set_item(fullname, &sub)?;
+
     // Ensure it's importable as `pynumaflow_lite.accumulator` as well
     let binding = m.getattr("accumulator")?;
     let sub = binding.cast::<PyModule>()?;
diff --git a/packages/pynumaflow-lite/src/reducestream/mod.rs b/packages/pynumaflow-lite/src/reducestream/mod.rs
new file mode 100644
index 00000000..be9c7816
--- /dev/null
+++ b/packages/pynumaflow-lite/src/reducestream/mod.rs
@@ -0,0 +1,74 @@
+use std::sync::Mutex;
+
+pub mod server;
+
+use pyo3::prelude::*;
+
+// Re-export types from reduce module that are shared
+pub use crate::reduce::{Datum, IntervalWindow, Message, Metadata, PyAsyncDatumStream};
+
+/// Async Reduce Stream Server that can be started from Python code, taking a class (creator).
+#[pyclass(module = "pynumaflow_lite.reducestreamer")]
+pub struct ReduceStreamAsyncServer {
+    sock_file: String,
+    info_file: String,
+    shutdown_tx: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
+}
+
+#[pymethods]
+impl ReduceStreamAsyncServer {
+    #[new]
+    #[pyo3(signature = (sock_file="/var/run/numaflow/reducestream.sock".to_string(), info_file="/var/run/numaflow/reducestreamer-server-info".to_string()))]
+    fn new(sock_file: String, info_file: String) -> Self {
+        Self {
+            sock_file,
+            info_file,
+            shutdown_tx: Mutex::new(None),
+        }
+    }
+
+    /// Start the server with the given Python class (creator).
+    /// Only class-based reducers are supported (not function-based).
+    /// init_args is an optional tuple of positional arguments to pass to the class constructor.
+    #[pyo3(signature = (py_creator, init_args=None))]
+    pub fn start<'a>(
+        &self,
+        py: Python<'a>,
+        py_creator: Py<PyAny>,
+        init_args: Option<Py<PyAny>>,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let sock_file = self.sock_file.clone();
+        let info_file = self.info_file.clone();
+        let (tx, rx) = tokio::sync::oneshot::channel::<()>();
+        {
+            let mut guard = self.shutdown_tx.lock().unwrap();
+            *guard = Some(tx);
+        }
+
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            crate::reducestream::server::start(py_creator, init_args, sock_file, info_file, rx)
+                .await
+                .expect("server failed to start");
+            Ok(())
+        })
+    }
+
+    /// Trigger server shutdown from Python (idempotent).
+    pub fn stop(&self) -> PyResult<()> {
+        if let Some(tx) = self.shutdown_tx.lock().unwrap().take() {
+            let _ = tx.send(());
+        }
+        Ok(())
+    }
+}
+
+/// Helper to populate a PyModule with reduce stream types/functions.
+pub(crate) fn populate_py_module(m: &Bound<PyModule>) -> PyResult<()> {
+    m.add_class::<Message>()?;
+    m.add_class::<Datum>()?;
+    m.add_class::<IntervalWindow>()?;
+    m.add_class::<Metadata>()?;
+    m.add_class::<ReduceStreamAsyncServer>()?;
+    m.add_class::<PyAsyncDatumStream>()?;
+    Ok(())
+}
diff --git a/packages/pynumaflow-lite/src/reducestream/server.rs b/packages/pynumaflow-lite/src/reducestream/server.rs
new file mode 100644
index 00000000..9e406179
--- /dev/null
+++ b/packages/pynumaflow-lite/src/reducestream/server.rs
@@ -0,0 +1,171 @@
+use crate::pyiterables::PyAsyncIterStream;
+use crate::reducestream::{
+    Datum as PyDatum, IntervalWindow as PyIntervalWindow, Message as PyMessage,
+    Metadata as PyMetadata, PyAsyncDatumStream,
+};
+use numaflow::reducestream;
+use numaflow::shared::ServerExtras;
+use pyo3::prelude::*;
+use std::sync::Arc;
+use tokio_stream::StreamExt;
+
+pub(crate) struct PyReduceStreamerCreator {
+    /// handle to Python event loop
+    pub(crate) event_loop: Arc<Py<PyAny>>,
+    /// Python class to instantiate per window
+    pub(crate) py_creator: Arc<Py<PyAny>>,
+    /// optional tuple of positional args
+    pub(crate) init_args: Option<Arc<Py<PyAny>>>,
+}
+
+pub(crate) struct PyReduceStreamerRunner {
+    /// handle to Python event loop
+    pub(crate) event_loop: Arc<Py<PyAny>>,
+    /// Instance of class per window
+    pub(crate) py_instance: Arc<Py<PyAny>>,
+}
+
+impl reducestream::ReduceStreamerCreator for PyReduceStreamerCreator {
+    type R = PyReduceStreamerRunner;
+
+    fn create(&self) -> Self::R {
+        // Instantiate the Python class synchronously under the GIL.
+        let inst = Python::attach(|py| {
+            let class = self.py_creator.as_ref();
+            match &self.init_args {
+                Some(args) => {
+                    let bound = args.as_ref().bind(py);
+                    let py_tuple = bound.cast()?;
+                    class.call1(py, py_tuple)
+                }
+                None => class.call0(py),
+            }
+        })
+        .expect("failed to instantiate Python reduce streamer class");
+
+        PyReduceStreamerRunner {
+            event_loop: self.event_loop.clone(),
+            py_instance: Arc::new(inst),
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl reducestream::ReduceStreamer for PyReduceStreamerRunner {
+    async fn reducestream(
+        &self,
+        keys: Vec<String>,
+        mut input: tokio::sync::mpsc::Receiver<reducestream::ReduceStreamRequest>,
+        output: tokio::sync::mpsc::Sender<reducestream::Message>,
+        md: &reducestream::Metadata,
+    ) {
+        // Create a channel to stream Datum into Python as an async iterator
+        let (tx, rx) = tokio::sync::mpsc::channel::<PyDatum>(64);
+
+        // Spawn a task forwarding incoming datums to the Python-facing channel
+        let forwarder = tokio::spawn(async move {
+            while let Some(req) = input.recv().await {
+                let datum = PyDatum::from(req);
+                if tx.send(datum).await.is_err() {
+                    break;
+                }
+            }
+            // When input ends, dropping tx closes the channel and that is when
+            // the Python async iterable will stop.
+        });
+
+        // Call the Python coroutine:
+        // obj.handler(keys, datums: AsyncIterable[Datum], md: Metadata) -> AsyncIterator[Message]
+        let agen_obj = Python::attach(|py| {
+            let obj = self.py_instance.clone();
+            let stream = PyAsyncDatumStream::new_with(rx);
+
+            // Build metadata object
+            let interval =
+                PyIntervalWindow::new(md.interval_window.start_time, md.interval_window.end_time);
+            let md_py = PyMetadata::new(interval);
+
+            // Call handler method
+            let agen = obj
+                .call_method1(py, "handler", (keys, stream, md_py))
+                .expect("python handler method raised before returning async iterable");
+
+            // Keep as Py<PyAny>
+            agen.clone_ref(py).extract(py).unwrap_or(agen)
+        });
+
+        // Wrap the Python AsyncIterable in a Rust Stream that yields incrementally
+        let mut stream = PyAsyncIterStream::<PyMessage>::new(agen_obj, self.event_loop.clone())
+            .expect("failed to construct PyAsyncIterStream");
+
+        // Forward each yielded message immediately to the output sender
+        while let Some(item) = stream.next().await {
+            match item {
+                Ok(py_msg) => {
+                    let out: reducestream::Message = py_msg.into();
+                    if output.send(out).await.is_err() {
+                        break;
+                    }
+                }
+                Err(e) => {
+                    // Non-stop errors are surfaced per-item; log and stop this stream.
+                    eprintln!("Python async iteration error in reduce streamer: {:?}", e);
+                    break;
+                }
+            }
+        }
+
+        // Ensure forwarder completes
+        let _ = forwarder.await;
+    }
+}
+
+/// Start the reduce stream server by spinning up a dedicated Python asyncio loop and wiring shutdown.
+pub(super) async fn start(
+    py_creator: Py<PyAny>,
+    init_args: Option<Py<PyAny>>,
+    sock_file: String,
+    info_file: String,
+    shutdown_rx: tokio::sync::oneshot::Receiver<()>,
+) -> Result<(), pyo3::PyErr> {
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    let py_asyncio_loop_handle = tokio::task::spawn_blocking(move || crate::pyrs::run_asyncio(tx));
+    let event_loop = rx.await.unwrap();
+
+    let (sig_handle, combined_rx) = crate::pyrs::setup_sig_handler(shutdown_rx);
+
+    let creator = PyReduceStreamerCreator {
+        event_loop: event_loop.clone(),
+        py_creator: Arc::new(py_creator),
+        init_args: init_args.map(Arc::new),
+    };
+
+    let server = reducestream::Server::new(creator)
+        .with_socket_file(sock_file)
+        .with_server_info_file(info_file);
+
+    let result = server
+        .start_with_shutdown(combined_rx)
+        .await
+        .map_err(|e| pyo3::PyErr::new::<pyo3::exceptions::PyException, _>(e.to_string()));
+
+    // Ensure the event loop is stopped even if shutdown came from elsewhere.
+    Python::attach(|py| {
+        if let Ok(stop_cb) = event_loop.getattr(py, "stop") {
+            let _ = event_loop.call_method1(py, "call_soon_threadsafe", (stop_cb,));
+        }
+    });
+
+    println!("Numaflow Core (reduce stream) has shutdown...");
+
+    // Wait for the blocking asyncio thread to finish.
+    let _ = py_asyncio_loop_handle.await;
+
+    // if not finished, abort it
+    if !sig_handle.is_finished() {
+        println!("Aborting signal handler");
+        let _ = sig_handle.abort();
+    }
+
+    result
+}
diff --git a/packages/pynumaflow-lite/tests/bin/reducestream.rs b/packages/pynumaflow-lite/tests/bin/reducestream.rs
new file mode 100644
index 00000000..b8d606db
--- /dev/null
+++ b/packages/pynumaflow-lite/tests/bin/reducestream.rs
@@ -0,0 +1,153 @@
+use std::env;
+use std::path::PathBuf;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use tokio::net::UnixStream;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+use tonic::{Request, transport::Uri};
+use tower::service_fn;
+
+fn now_ts() -> prost_types::Timestamp {
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+    prost_types::Timestamp {
+        seconds: now.as_secs() as i64,
+        nanos: now.subsec_nanos() as i32,
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Default socket path with env/arg override
+    let sock_path = env::args()
+        .nth(1)
+        .or_else(|| env::var("NUMAFLOW_REDUCESTREAM_SOCK").ok())
+        .unwrap_or_else(|| "/tmp/var/run/numaflow/reducestream.sock".to_string());
+
+    // Connect over UDS
+    let channel = tonic::transport::Endpoint::try_from("http://[::]:50051")?
+        .connect_with_connector(service_fn(move |_: Uri| {
+            let sock = PathBuf::from(sock_path.clone());
+            async move {
+                Ok::<_, std::io::Error>(hyper_util::rt::TokioIo::new(
+                    UnixStream::connect(sock).await?,
+                ))
+            }
+        }))
+        .await?;
+
+    // Reduce client from generated proto (reducestream uses the same proto as reduce)
+    let mut client = numaflow::proto::reduce::reduce_client::ReduceClient::new(channel);
+
+    let (tx, rx) = mpsc::channel(16);
+
+    // Send 3 ReduceRequests to test streaming behavior
+    // The Python handler should yield 3 messages (one per datum)
+    {
+        use numaflow::proto::reduce as reduce_proto;
+
+        // First request with OPEN operation
+        let rr1 = reduce_proto::ReduceRequest {
+            payload: Some(reduce_proto::reduce_request::Payload {
+                keys: vec!["test-key".into()],
+                value: b"1".to_vec(),
+                watermark: Some(now_ts()),
+                event_time: Some(now_ts()),
+                headers: Default::default(),
+            }),
+            operation: Some(reduce_proto::reduce_request::WindowOperation {
+                event: 0, // OPEN
+                windows: vec![reduce_proto::Window {
+                    start: Some(now_ts()),
+                    end: Some(now_ts()),
+                    slot: "slot-0".to_string(),
+                }],
+            }),
+        };
+        tx.send(rr1).await.unwrap();
+
+        // Second request with APPEND operation
+        let rr2 = reduce_proto::ReduceRequest {
+            payload: Some(reduce_proto::reduce_request::Payload {
+                keys: vec!["test-key".into()],
+                value: b"2".to_vec(),
+                watermark: Some(now_ts()),
+                event_time: Some(now_ts()),
+                headers: Default::default(),
+            }),
+            operation: Some(reduce_proto::reduce_request::WindowOperation {
+                event: 4, // APPEND
+                windows: vec![reduce_proto::Window {
+                    start: Some(now_ts()),
+                    end: Some(now_ts()),
+                    slot: "slot-0".to_string(),
+                }],
+            }),
+        };
+        tx.send(rr2).await.unwrap();
+
+        // Third request with APPEND operation
+        let rr3 = reduce_proto::ReduceRequest {
+            payload: Some(reduce_proto::reduce_request::Payload {
+                keys: vec!["test-key".into()],
+                value: b"3".to_vec(),
+                watermark: Some(now_ts()),
+                event_time: Some(now_ts()),
+                headers: Default::default(),
+            }),
+            operation: Some(reduce_proto::reduce_request::WindowOperation {
+                event: 4, // APPEND
+                windows: vec![reduce_proto::Window {
+                    start: Some(now_ts()),
+                    end: Some(now_ts()),
+                    slot: "slot-0".to_string(),
+                }],
+            }),
+        };
+        tx.send(rr3).await.unwrap();
+    }
+
+    // Drop sender to signal end-of-stream (COB)
+    drop(tx);
+
+    // Start the RPC using a Request-wrapped stream
+    let request = Request::new(ReceiverStream::new(rx));
+    let mut resp = client.reduce_fn(request).await?.into_inner();
+
+    // Read responses - we should get 3 messages (one per datum) due to streaming
+    let mut message_count = 0;
+    let mut found_eof = false;
+
+    loop {
+        if let Some(r) = resp.message().await? {
+            if let Some(res) = r.result {
+                assert!(!res.value.is_empty());
+                message_count += 1;
+                println!(
+                    "Received message {}: {:?}",
+                    message_count,
+                    String::from_utf8_lossy(&res.value)
+                );
+            }
+            if r.eof {
+                found_eof = true;
+                println!("Received EOF");
+                break;
+            }
+        } else {
+            break;
+        }
+    }
+
+    // Verify we got 3 messages (streaming behavior)
+    assert_eq!(
+        message_count, 3,
+        "Expected 3 streamed messages, got {}",
+        message_count
+    );
+    assert!(found_eof, "Should have received EOF");
+
+    println!("Test passed! Received {} streamed messages", message_count);
+
+    Ok(())
+}
diff --git a/packages/pynumaflow-lite/tests/examples/reducestream_counter.py b/packages/pynumaflow-lite/tests/examples/reducestream_counter.py
new file mode 100644
index 00000000..fcebeb0e
--- /dev/null
+++ b/packages/pynumaflow-lite/tests/examples/reducestream_counter.py
@@ -0,0 +1,78 @@
+import asyncio
+import signal
+from collections.abc import AsyncIterable, AsyncIterator
+
+from pynumaflow_lite import reducestreamer
+
+
+class ReduceStreamCounter(reducestreamer.ReduceStreamer):
+    """
+    A reduce streaming counter that emits intermediate results.
+    
+    This test implementation counts datums and yields a message for every
+    datum received, demonstrating the streaming capability.
+    """
+    
+    def __init__(self, initial: int = 0) -> None:
+        self.counter = initial
+
+    async def handler(
+        self,
+        keys: list[str],
+        datums: AsyncIterable[reducestreamer.Datum],
+        md: reducestreamer.Metadata,
+    ) -> AsyncIterator[reducestreamer.Message]:
+        """
+        Process datums and yield messages incrementally.
+        
+        For testing purposes, we yield a message for each datum received.
+        """
+        iw = md.interval_window
+        self.counter = 0
+        
+        async for _ in datums:
+            self.counter += 1
+            # Yield a message for each datum (streaming behavior)
+            msg = (
+                f"counter:{self.counter} "
+                f"interval_window_start:{iw.start} "
+                f"interval_window_end:{iw.end}"
+            ).encode()
+            yield reducestreamer.Message(msg, keys=keys)
+
+
+# Optional: ensure default signal handlers are in place so asyncio.run can handle them cleanly.
+signal.signal(signal.SIGINT, signal.default_int_handler)
+try:
+    signal.signal(signal.SIGTERM, signal.SIG_DFL)
+except AttributeError:
+    pass
+
+
+async def start(creator: type, init_args: tuple):
+    """Start the reduce stream server."""
+    sock_file = "/tmp/var/run/numaflow/reducestream.sock"
+    server_info_file = "/tmp/var/run/numaflow/reducestreamer-server-info"
+    server = reducestreamer.ReduceStreamAsyncServer(sock_file, server_info_file)
+
+    loop = asyncio.get_running_loop()
+    try:
+        loop.add_signal_handler(signal.SIGINT, lambda: server.stop())
+        loop.add_signal_handler(signal.SIGTERM, lambda: server.stop())
+    except (NotImplementedError, RuntimeError):
+        pass
+
+    try:
+        await server.start(creator, init_args)
+        print("Shutting down gracefully...")
+    except asyncio.CancelledError:
+        try:
+            server.stop()
+        except Exception:
+            pass
+        return
+
+
+if __name__ == "__main__":
+    asyncio.run(start(ReduceStreamCounter, (0,)))
+
diff --git a/packages/pynumaflow-lite/tests/test_reducestream.py b/packages/pynumaflow-lite/tests/test_reducestream.py
new file mode 100644
index 00000000..b42cfa08
--- /dev/null
+++ b/packages/pynumaflow-lite/tests/test_reducestream.py
@@ -0,0 +1,23 @@
+from pathlib import Path
+
+import pytest
+
+from _test_utils import run_python_server_with_rust_client
+
+SOCK_PATH = Path("/tmp/var/run/numaflow/reducestream.sock")
+SERVER_INFO = Path("/tmp/var/run/numaflow/reducestreamer-server-info")
+
+SCRIPTS = [
+    "reducestream_counter.py",
+]
+
+
+@pytest.mark.parametrize("script", SCRIPTS)
+def test_python_reducestream_server_and_rust_client(script: str, tmp_path: Path):
+    run_python_server_with_rust_client(
+        script=script,
+        sock_path=SOCK_PATH,
+        server_info_path=SERVER_INFO,
+        rust_bin_name="test_reducestream",
+    )
+