diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -1,5 +1,9 @@
 # ChangeLog for conduit
 
+## 1.3.2
+
+* Add `mapInputM` [#435](https://github.com/snoyberg/conduit/pull/435)
+
 ## 1.3.1.2
 
 * More eagerly emit groups in `chunksOf` [#427](https://github.com/snoyberg/conduit/pull/427)
diff --git a/conduit.cabal b/conduit.cabal
--- a/conduit.cabal
+++ b/conduit.cabal
@@ -1,5 +1,5 @@
 Name:                conduit
-Version:             1.3.1.2
+Version:             1.3.2
 Synopsis:            Streaming data processing library.
 description:
     `conduit` is a solution to the streaming data problem, allowing for production,
diff --git a/src/Data/Conduit.hs b/src/Data/Conduit.hs
--- a/src/Data/Conduit.hs
+++ b/src/Data/Conduit.hs
@@ -57,6 +57,7 @@
     , mapOutput
     , mapOutputMaybe
     , mapInput
+    , mapInputM
     , mergeSource
     , passthroughSink
     , sourceToList
diff --git a/src/Data/Conduit/Internal.hs b/src/Data/Conduit/Internal.hs
--- a/src/Data/Conduit/Internal.hs
+++ b/src/Data/Conduit/Internal.hs
@@ -11,8 +11,9 @@
 
 import           Data.Conduit.Internal.Conduit hiding (await,
                                                 awaitForever, bracketP,
-                                                leftover, mapInput, mapOutput,
-                                                mapOutputMaybe, transPipe,
+                                                leftover, mapInput, mapInputM,
+                                                mapOutput, mapOutputMaybe,
+                                                transPipe,
                                                 yield, yieldM)
 import           Data.Conduit.Internal.Pipe
 import           Data.Conduit.Internal.Fusion
diff --git a/src/Data/Conduit/Internal/Conduit.hs b/src/Data/Conduit/Internal/Conduit.hs
--- a/src/Data/Conduit/Internal/Conduit.hs
+++ b/src/Data/Conduit/Internal/Conduit.hs
@@ -72,6 +72,7 @@
     , Data.Conduit.Internal.Conduit.mapOutput
     , Data.Conduit.Internal.Conduit.mapOutputMaybe
     , Data.Conduit.Internal.Conduit.mapInput
+    , Data.Conduit.Internal.Conduit.mapInputM
     , zipSinks
     , zipSources
     , zipSourcesApp
@@ -987,6 +988,22 @@
     go (Done r) = rest r
     go (PipeM mp) = PipeM $ liftM go mp
     go (Leftover p i) = maybe id (flip Leftover) (f' i) (go p)
+    in go (c0 Done)
+
+-- | Apply a monadic action to all the input values of a @ConduitT@.
+--
+-- Since 1.3.2
+mapInputM :: Monad m
+          => (i1 -> m i2) -- ^ map initial input to new input
+          -> (i2 -> m (Maybe i1)) -- ^ map new leftovers to initial leftovers
+          -> ConduitT i2 o m r
+          -> ConduitT i1 o m r
+mapInputM f f' (ConduitT c0) = ConduitT $ \rest -> let
+    go (HaveOutput p o) = HaveOutput (go p) o
+    go (NeedInput p c)  = NeedInput (\i -> PipeM $ go . p <$> f i) (go . c)
+    go (Done r)         = rest r
+    go (PipeM mp)       = PipeM $ fmap go mp
+    go (Leftover p i)   = PipeM $ (\x -> maybe id (flip Leftover) x (go p)) <$> f' i
     in go (c0 Done)
 
 -- | The connect-and-resume operator. This does not close the @Source@, but