diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,22 @@
+# Changelog
+
+All notable changes to `kafka-effectful` are documented here.
+
+This package follows the [Haskell Package Versioning Policy](https://pvp.haskell.org/).
+
+## 0.1.0.0 — 2026-04-16
+
+Initial release.
+
+This is an experimental release. Breaking changes are expected in
+subsequent 0.x versions. Pin to an exact version in production until the
+API stabilizes at 1.0.
+
+- `KafkaProducer` effect with `produceMessage` and `flushProducer`
+  operations.
+- `KafkaConsumer` effect with polling, offset management, partition
+  management, and querying operations.
+- Resource-safe interpreters that acquire and release Kafka handles
+  via `bracket`.
+- Errors surfaced through `Effectful.Error.Static` as
+  `Error KafkaError`.
diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Nadeem Bitar
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/README.md b/README.md
new file mode 100644
--- /dev/null
+++ b/README.md
@@ -0,0 +1,100 @@
+# kafka-effectful
+
+Effectful effects and interpreters for [hw-kafka-client](https://hackage.haskell.org/package/hw-kafka-client), a Haskell binding to Apache Kafka via librdkafka.
+
+Provides typed, composable `KafkaProducer` and `KafkaConsumer` effects for the [effectful](https://hackage.haskell.org/package/effectful) ecosystem.
+
+> **Status: experimental.** This package is on its first release. The API may change in breaking ways in subsequent 0.x versions. Pin to an exact version in production until 1.0 is tagged.
+
+## Features
+
+- **KafkaProducer** -- send messages and flush the producer queue
+- **KafkaConsumer** -- poll messages (single or batch), manage offsets, assign/pause/resume/seek partitions, and query committed offsets, positions, assignments, and subscriptions
+- Resource-safe interpreters that acquire and release Kafka handles via `bracket`
+- Errors surfaced through `Effectful.Error.Static` (`Error KafkaError`)
+
+## Usage
+
+### Producer
+
+```haskell
+import Kafka.Effectful
+
+example :: (IOE :> es, Error KafkaError :> es) => Eff es ()
+example =
+  runKafkaProducer producerProps $ do
+    produceMessage record
+    flushProducer
+```
+
+### Consumer
+
+```haskell
+import Kafka.Effectful
+
+example :: (IOE :> es, Error KafkaError :> es) => Eff es ()
+example =
+  runKafkaConsumer consumerProps subscription loop
+  where
+    loop = do
+      mbMsg <- pollMessage (Timeout 1000)
+      case mbMsg of
+        Nothing  -> loop
+        Just msg -> do
+          commitOffsetMessage OffsetCommit msg
+          loop
+```
+
+`pollMessage` returns `Nothing` when the timeout elapses without a message
+arriving; non-timeout failures are thrown via the `Error KafkaError` effect.
+
+### Running it
+
+The effect handlers `runKafkaProducer` and `runKafkaConsumer` require `IOE` and
+`Error KafkaError` in the effect stack. A complete program wires them with
+`runEff` and `runError`:
+
+```haskell
+{-# LANGUAGE TypeApplications #-}
+
+import Effectful
+import Effectful.Error.Static (runError)
+import Kafka.Effectful
+
+main :: IO ()
+main = do
+  result <- runEff . runError @KafkaError $ runProgram
+  case result of
+    Left (_, err) -> putStrLn ("Kafka error: " <> show err)
+    Right ()      -> pure ()
+  where
+    runProgram =
+      runKafkaProducer producerProps $ do
+        produceMessage record
+        flushProducer
+```
+
+Replace `producerProps` and `record` with your own `ProducerProperties` and
+`ProducerRecord` values (see the `Kafka.Effectful.Producer` module for the
+available builders).
+
+## Module Structure
+
+| Module | Description |
+|--------|-------------|
+| `Kafka.Effectful` | Convenience re-export of both effects and common types |
+| `Kafka.Effectful.Producer` | Producer effect, interpreter, and types |
+| `Kafka.Effectful.Consumer` | Consumer effect, interpreter, and types |
+| `Kafka.Effectful.Producer.Effect` | `KafkaProducer` effect definition and operations |
+| `Kafka.Effectful.Producer.Interpreter` | `runKafkaProducer` interpreter |
+| `Kafka.Effectful.Consumer.Effect` | `KafkaConsumer` effect definition and operations |
+| `Kafka.Effectful.Consumer.Interpreter` | `runKafkaConsumer` interpreter |
+
+## Requirements
+
+- GHC >= 9.12
+- librdkafka (system dependency)
+
+## License
+
+MIT
diff --git a/kafka-effectful.cabal b/kafka-effectful.cabal
new file mode 100644
--- /dev/null
+++ b/kafka-effectful.cabal
@@ -0,0 +1,63 @@
+cabal-version:   3.4
+name:            kafka-effectful
+version:         0.1.0.0
+synopsis:        Effectful effects for hw-kafka-client
+description:
+  Effectful effects and interpreters for hw-kafka-client, a Haskell
+  binding to Apache Kafka via librdkafka. Provides typed, composable
+  KafkaProducer and KafkaConsumer effects.
+
+license:         MIT
+license-file:    LICENSE
+author:          Nadeem Bitar
+maintainer:      Nadeem Bitar
+category:        Network, Messaging
+build-type:      Simple
+extra-doc-files:
+  CHANGELOG.md
+  README.md
+
+source-repository head
+  type:     git
+  location: https://github.com/shinzui/kafka-effectful.git
+
+common warnings
+  ghc-options:
+    -Wall -Wcompat -Widentities -Wincomplete-uni-patterns
+    -Wincomplete-record-updates -Wredundant-constraints
+    -fhide-source-paths -Wmissing-export-lists -Wpartial-fields
+    -Wmissing-deriving-strategies
+
+library
+  import:             warnings
+  exposed-modules:
+    Kafka.Effectful
+    Kafka.Effectful.Consumer
+    Kafka.Effectful.Consumer.Effect
+    Kafka.Effectful.Consumer.Interpreter
+    Kafka.Effectful.Producer
+    Kafka.Effectful.Producer.Effect
+    Kafka.Effectful.Producer.Interpreter
+
+  build-depends:
+    , base             >=4.21 && <5
+    , bytestring       >=0.11 && <0.13
+    , containers       >=0.6  && <0.8
+    , effectful-core   >=2.5  && <2.7
+    , hw-kafka-client  >=5.3  && <6
+    , text             >=2.0  && <2.2
+
+  hs-source-dirs:     src
+  default-language:   GHC2024
+  default-extensions:
+    DataKinds
+    DuplicateRecordFields
+    GADTs
+    GeneralisedNewtypeDeriving
+    ImportQualifiedPost
+    LambdaCase
+    NoFieldSelectors
+    OverloadedRecordDot
+    OverloadedStrings
+    TypeFamilies
+    TypeOperators
diff --git a/src/Kafka/Effectful.hs b/src/Kafka/Effectful.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful.hs
@@ -0,0 +1,141 @@
+module Kafka.Effectful (
+    -- * Producer Effect
+    KafkaProducer,
+    runKafkaProducer,
+    produceMessage,
+    flushProducer,
+
+    -- * Consumer Effect
+    KafkaConsumer,
+    runKafkaConsumer,
+
+    -- ** Polling
+    pollMessage,
+    pollMessageBatch,
+
+    -- ** Offset Management
+    commitOffsetMessage,
+    commitAllOffsets,
+    commitPartitionsOffsets,
+    storeOffsets,
+    storeOffsetMessage,
+
+    -- ** Partition Management
+    assign,
+    pausePartitions,
+    resumePartitions,
+    seekPartitions,
+
+    -- ** Querying
+    committed,
+    position,
+    assignment,
+    subscription,
+
+    -- * Producer Types
+    ProducerRecord (..),
+    ProducePartition (..),
+    DeliveryReport (..),
+    ImmediateError (..),
+    ProducerProperties (..),
+
+    -- * Consumer Types
+    ConsumerRecord (..),
+    ConsumerProperties (..),
+    CallbackPollMode (..),
+    Subscription (..),
+    Offset (..),
+    OffsetReset (..),
+    OffsetCommit (..),
+    TopicPartition (..),
+    SubscribedPartitions (..),
+    ConsumerGroupId (..),
+    PartitionOffset (..),
+    Timestamp (..),
+    RebalanceEvent (..),
+
+    -- * Subscription Builders
+    topics,
+    offsetReset,
+    extraSubscriptionProps,
+
+    -- * Common Types
+    KafkaError (..),
+    TopicName (..),
+    BrokerAddress (..),
+    Timeout (..),
+    BatchSize (..),
+    PartitionId (..),
+    Millis (..),
+    ClientId (..),
+    KafkaLogLevel (..),
+    KafkaDebug (..),
+    KafkaCompressionCodec (..),
+    Headers,
+    headersFromList,
+    headersToList,
+)
+where
+
+import Kafka.Effectful.Consumer (
+    CallbackPollMode (..),
+    ConsumerGroupId (..),
+    ConsumerProperties (..),
+    ConsumerRecord (..),
+    KafkaConsumer,
+    Offset (..),
+    OffsetCommit (..),
+    OffsetReset (..),
+    PartitionOffset (..),
+    RebalanceEvent (..),
+    SubscribedPartitions (..),
+    Subscription (..),
+    Timestamp (..),
+    TopicPartition (..),
+    assign,
+    assignment,
+    commitAllOffsets,
+    commitOffsetMessage,
+    commitPartitionsOffsets,
+    committed,
+    extraSubscriptionProps,
+    offsetReset,
+    pausePartitions,
+    pollMessage,
+    pollMessageBatch,
+    position,
+    resumePartitions,
+    runKafkaConsumer,
+    seekPartitions,
+    storeOffsetMessage,
+    storeOffsets,
+    subscription,
+    topics,
+ )
+import Kafka.Effectful.Producer (
+    DeliveryReport (..),
+    ImmediateError (..),
+    KafkaProducer,
+    ProducePartition (..),
+    ProducerProperties (..),
+    ProducerRecord (..),
+    flushProducer,
+    produceMessage,
+    runKafkaProducer,
+ )
+import Kafka.Types (
+    BatchSize (..),
+    BrokerAddress (..),
+    ClientId (..),
+    Headers,
+    KafkaCompressionCodec (..),
+    KafkaDebug (..),
+    KafkaError (..),
+    KafkaLogLevel (..),
+    Millis (..),
+    PartitionId (..),
+    Timeout (..),
+    TopicName (..),
+    headersFromList,
+    headersToList,
+ )
diff --git a/src/Kafka/Effectful/Consumer.hs b/src/Kafka/Effectful/Consumer.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful/Consumer.hs
@@ -0,0 +1,101 @@
+module Kafka.Effectful.Consumer (
+    -- * Effect
+    KafkaConsumer,
+
+    -- * Interpreter
+    runKafkaConsumer,
+
+    -- * Polling
+    pollMessage,
+    pollMessageBatch,
+
+    -- * Offset Management
+    commitOffsetMessage,
+    commitAllOffsets,
+    commitPartitionsOffsets,
+    storeOffsets,
+    storeOffsetMessage,
+
+    -- * Partition Management
+    assign,
+    pausePartitions,
+    resumePartitions,
+    seekPartitions,
+
+    -- * Querying
+    committed,
+    position,
+    assignment,
+    subscription,
+
+    -- * Consumer Types
+    ConsumerRecord (..),
+    Offset (..),
+    OffsetReset (..),
+    OffsetCommit (..),
+    TopicPartition (..),
+    SubscribedPartitions (..),
+    ConsumerGroupId (..),
+    PartitionOffset (..),
+    Timestamp (..),
+    RebalanceEvent (..),
+
+    -- * Configuration
+    ConsumerProperties (..),
+    CallbackPollMode (..),
+    K.brokersList,
+    K.autoCommit,
+    K.noAutoCommit,
+    K.noAutoOffsetStore,
+    K.groupId,
+    K.clientId,
+    K.setCallback,
+    K.logLevel,
+    K.compression,
+    K.suppressDisconnectLogs,
+    K.statisticsInterval,
+    K.extraProps,
+    K.extraProp,
+    K.debugOptions,
+    K.queuedMaxMessagesKBytes,
+    K.callbackPollMode,
+
+    -- * Subscription
+    Subscription (..),
+    topics,
+    offsetReset,
+    extraSubscriptionProps,
+
+    -- * Callbacks
+    K.rebalanceCallback,
+    K.offsetCommitCallback,
+    K.errorCallback,
+    K.logCallback,
+    K.statsCallback,
+    K.Callback,
+
+    -- * Common Types
+    KafkaError (..),
+    TopicName (..),
+    BrokerAddress (..),
+    Timeout (..),
+    BatchSize (..),
+    PartitionId (..),
+    Millis (..),
+    ClientId (..),
+    KafkaLogLevel (..),
+    KafkaDebug (..),
+    KafkaCompressionCodec (..),
+    Headers,
+    headersFromList,
+    headersToList,
+)
+where
+
+import Kafka.Consumer.ConsumerProperties (CallbackPollMode (..), ConsumerProperties (..))
+import Kafka.Consumer.ConsumerProperties qualified as K
+import Kafka.Consumer.Subscription (Subscription (..), extraSubscriptionProps, offsetReset, topics)
+import Kafka.Consumer.Types (ConsumerGroupId (..), ConsumerRecord (..), Offset (..), OffsetCommit (..), OffsetReset (..), PartitionOffset (..), RebalanceEvent (..), SubscribedPartitions (..), Timestamp (..), TopicPartition (..))
+import Kafka.Effectful.Consumer.Effect (KafkaConsumer, assign, assignment, commitAllOffsets, commitOffsetMessage, commitPartitionsOffsets, committed, pausePartitions, pollMessage, pollMessageBatch, position, resumePartitions, seekPartitions, storeOffsetMessage, storeOffsets, subscription)
+import Kafka.Effectful.Consumer.Interpreter (runKafkaConsumer)
+import Kafka.Types (BatchSize (..), BrokerAddress (..), ClientId (..), Headers, KafkaCompressionCodec (..), KafkaDebug (..), KafkaError (..), KafkaLogLevel (..), Millis (..), PartitionId (..), Timeout (..), TopicName (..), headersFromList, headersToList)
diff --git a/src/Kafka/Effectful/Consumer/Effect.hs b/src/Kafka/Effectful/Consumer/Effect.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful/Consumer/Effect.hs
@@ -0,0 +1,188 @@
+module Kafka.Effectful.Consumer.Effect (
+    -- * Effect
+    KafkaConsumer (..),
+
+    -- * Polling
+    pollMessage,
+    pollMessageBatch,
+
+    -- * Offset Management
+    commitOffsetMessage,
+    commitAllOffsets,
+    commitPartitionsOffsets,
+    storeOffsets,
+    storeOffsetMessage,
+
+    -- * Partition Management
+    assign,
+    pausePartitions,
+    resumePartitions,
+    seekPartitions,
+
+    -- * Querying
+    committed,
+    position,
+    assignment,
+    subscription,
+)
+where
+
+import Data.ByteString (ByteString)
+import Data.Map.Strict (Map)
+import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>))
+import Effectful.Dispatch.Dynamic (send)
+import Kafka.Consumer.Types (
+    ConsumerRecord,
+    OffsetCommit,
+    SubscribedPartitions,
+    TopicPartition,
+ )
+import Kafka.Types (
+    BatchSize,
+    KafkaError,
+    PartitionId,
+    Timeout,
+    TopicName,
+ )
+
+-- | Effect for Kafka consumer operations.
+data KafkaConsumer :: Effect where
+    PollMessage ::
+        Timeout ->
+        KafkaConsumer m (Maybe (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))
+    PollMessageBatch ::
+        Timeout ->
+        BatchSize ->
+        KafkaConsumer m [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]
+    CommitOffsetMessage ::
+        OffsetCommit ->
+        ConsumerRecord k v ->
+        KafkaConsumer m ()
+    CommitAllOffsets ::
+        OffsetCommit ->
+        KafkaConsumer m ()
+    CommitPartitionsOffsets ::
+        OffsetCommit ->
+        [TopicPartition] ->
+        KafkaConsumer m ()
+    StoreOffsets ::
+        [TopicPartition] ->
+        KafkaConsumer m ()
+    StoreOffsetMessage ::
+        ConsumerRecord k v ->
+        KafkaConsumer m ()
+    Assign ::
+        [TopicPartition] ->
+        KafkaConsumer m ()
+    PausePartitions ::
+        [(TopicName, PartitionId)] ->
+        KafkaConsumer m ()
+    ResumePartitions ::
+        [(TopicName, PartitionId)] ->
+        KafkaConsumer m ()
+    SeekPartitions ::
+        [TopicPartition] ->
+        Timeout ->
+        KafkaConsumer m ()
+    Committed ::
+        Timeout ->
+        [(TopicName, PartitionId)] ->
+        KafkaConsumer m [TopicPartition]
+    Position ::
+        [(TopicName, PartitionId)] ->
+        KafkaConsumer m [TopicPartition]
+    Assignment ::
+        KafkaConsumer m (Map TopicName [PartitionId])
+    Subscription ::
+        KafkaConsumer m [(TopicName, SubscribedPartitions)]
+
+type instance DispatchOf KafkaConsumer = 'Dynamic
+
+-- Polling
+
+{- | Poll for a single message.
+
+Returns 'Nothing' when the timeout elapses without a message arriving.
+Throws 'KafkaError' via the 'Error' effect for any non-timeout failure
+(for example, a broker transport error or an assignment revocation).
+-}
+pollMessage ::
+    (KafkaConsumer :> es) =>
+    Timeout ->
+    Eff es (Maybe (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))
+pollMessage = send . PollMessage
+
+-- | Poll for a batch of messages. Per-message errors are preserved in the 'Either'.
+pollMessageBatch ::
+    (KafkaConsumer :> es) =>
+    Timeout ->
+    BatchSize ->
+    Eff es [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]
+pollMessageBatch t b = send $ PollMessageBatch t b
+
+-- Offset Management
+
+-- | Commit the offset of a specific message. Throws 'KafkaError' on failure.
+commitOffsetMessage ::
+    (KafkaConsumer :> es) => OffsetCommit -> ConsumerRecord k v -> Eff es ()
+commitOffsetMessage oc cr = send $ CommitOffsetMessage oc cr
+
+-- | Commit offsets for all currently assigned partitions. Throws 'KafkaError' on failure.
+commitAllOffsets :: (KafkaConsumer :> es) => OffsetCommit -> Eff es ()
+commitAllOffsets = send . CommitAllOffsets
+
+-- | Commit offsets for specific partitions. Throws 'KafkaError' on failure.
+commitPartitionsOffsets ::
+    (KafkaConsumer :> es) => OffsetCommit -> [TopicPartition] -> Eff es ()
+commitPartitionsOffsets oc tps = send $ CommitPartitionsOffsets oc tps
+
+-- | Store offsets locally without committing to the broker. Throws 'KafkaError' on failure.
+storeOffsets :: (KafkaConsumer :> es) => [TopicPartition] -> Eff es ()
+storeOffsets = send . StoreOffsets
+
+-- | Store a message's offset locally without committing. Throws 'KafkaError' on failure.
+storeOffsetMessage :: (KafkaConsumer :> es) => ConsumerRecord k v -> Eff es ()
+storeOffsetMessage = send . StoreOffsetMessage
+
+-- Partition Management
+
+-- | Manually assign partitions to the consumer. Throws 'KafkaError' on failure.
+assign :: (KafkaConsumer :> es) => [TopicPartition] -> Eff es ()
+assign = send . Assign
+
+-- | Pause consuming from the specified partitions. Throws 'KafkaError' on failure.
+pausePartitions :: (KafkaConsumer :> es) => [(TopicName, PartitionId)] -> Eff es ()
+pausePartitions = send . PausePartitions
+
+-- | Resume consuming from the specified partitions. Throws 'KafkaError' on failure.
+resumePartitions :: (KafkaConsumer :> es) => [(TopicName, PartitionId)] -> Eff es ()
+resumePartitions = send . ResumePartitions
+
+-- | Seek to specific offsets for partitions. Throws 'KafkaError' on failure.
+seekPartitions :: (KafkaConsumer :> es) => [TopicPartition] -> Timeout -> Eff es ()
+seekPartitions tps t = send $ SeekPartitions tps t
+
+-- Querying
+
+-- | Get committed offsets for the specified partitions. Throws 'KafkaError' on failure.
+committed ::
+    (KafkaConsumer :> es) =>
+    Timeout ->
+    [(TopicName, PartitionId)] ->
+    Eff es [TopicPartition]
+committed t ps = send $ Committed t ps
+
+-- | Get the current position (last consumed offset + 1). Throws 'KafkaError' on failure.
+position ::
+    (KafkaConsumer :> es) =>
+    [(TopicName, PartitionId)] ->
+    Eff es [TopicPartition]
+position = send . Position
+
+-- | Get the current partition assignment.
+assignment :: (KafkaConsumer :> es) => Eff es (Map TopicName [PartitionId])
+assignment = send Assignment
+
+-- | Get the current topic subscription.
+subscription :: (KafkaConsumer :> es) => Eff es [(TopicName, SubscribedPartitions)]
+subscription = send Subscription
diff --git a/src/Kafka/Effectful/Consumer/Interpreter.hs b/src/Kafka/Effectful/Consumer/Interpreter.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful/Consumer/Interpreter.hs
@@ -0,0 +1,105 @@
+-- The 'EffectHandler' type synonym in effectful-core expands to a
+-- constraint that GHC's redundant-constraint check flags on the
+-- handler's signature, even though the constraint is required for
+-- 'interpret' to type-check. Suppress the warning at the file level.
+{-# OPTIONS_GHC -Wno-redundant-constraints #-}
+
+module Kafka.Effectful.Consumer.Interpreter (
+    -- * Interpreter
+    runKafkaConsumer,
+)
+where
+
+import Control.Monad (void)
+import Data.Foldable (for_)
+import Effectful (Eff, IOE, (:>))
+import Effectful qualified
+import Effectful.Dispatch.Dynamic (EffectHandler, interpret)
+import Effectful.Error.Static (Error, throwError)
+import Effectful.Exception (ExitCase (..))
+import Effectful.Exception qualified as Exception
+import Kafka.Consumer (RdKafkaRespErrT (..))
+import Kafka.Consumer qualified as K
+import Kafka.Consumer.ConsumerProperties (ConsumerProperties)
+import Kafka.Consumer.Subscription (Subscription)
+import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..))
+import Kafka.Types (KafkaError (..))
+
+{- | Run the 'KafkaConsumer' effect.
+
+Acquires a consumer handle from the given properties and subscription,
+and releases it when the effect scope ends. Errors are thrown via the
+'Error' effect.
+-}
+runKafkaConsumer ::
+    (IOE :> es, Error KafkaError :> es) =>
+    ConsumerProperties ->
+    Subscription ->
+    Eff (KafkaConsumer : es) a ->
+    Eff es a
+runKafkaConsumer props sub action =
+    fst
+        <$> Exception.generalBracket
+            acquire
+            release
+            (\consumer -> interpret (handleConsumer consumer) action)
+  where
+    acquire = do
+        result <- Effectful.liftIO $ K.newConsumer props sub
+        case result of
+            Left err -> throwError err
+            Right consumer -> pure consumer
+
+    release consumer = \case
+        ExitCaseSuccess _ -> do
+            mbErr <- Effectful.liftIO $ K.closeConsumer consumer
+            for_ mbErr throwError
+        ExitCaseException _ ->
+            Effectful.liftIO . void $ K.closeConsumer consumer
+        ExitCaseAbort ->
+            Effectful.liftIO . void $ K.closeConsumer consumer
+
+handleConsumer ::
+    (IOE :> es, Error KafkaError :> es) =>
+    K.KafkaConsumer ->
+    EffectHandler KafkaConsumer es
+handleConsumer consumer _env = \case
+    PollMessage timeout -> do
+        result <- Effectful.liftIO $ K.pollMessage consumer timeout
+        case result of
+            Left (KafkaResponseError RdKafkaRespErrTimedOut) -> pure Nothing
+            Left err -> throwError err
+            Right msg -> pure (Just msg)
+    PollMessageBatch timeout batchSize ->
+        Effectful.liftIO $ K.pollMessageBatch consumer timeout batchSize
+    CommitOffsetMessage oc cr -> throwOnJust $ K.commitOffsetMessage oc consumer cr
+    CommitAllOffsets oc -> throwOnJust $ K.commitAllOffsets oc consumer
+    CommitPartitionsOffsets oc tps -> throwOnJust $ K.commitPartitionsOffsets oc consumer tps
+    StoreOffsets tps -> throwOnJust $ K.storeOffsets consumer tps
+    StoreOffsetMessage cr -> throwOnJust $ K.storeOffsetMessage consumer cr
+    Assign tps -> throwOnJust $ K.assign consumer tps
+    PausePartitions parts ->
+        throwOnKafkaErr (K.pausePartitions consumer parts)
+    ResumePartitions parts ->
+        throwOnKafkaErr (K.resumePartitions consumer parts)
+    SeekPartitions tps timeout -> throwOnJust $ K.seekPartitions consumer tps timeout
+    Committed timeout parts -> throwOnLeft $ K.committed consumer timeout parts
+    Position parts -> throwOnLeft $ K.position consumer parts
+    Assignment -> throwOnLeft $ K.assignment consumer
+    Subscription -> throwOnLeft $ K.subscription consumer
+  where
+    throwOnJust action' = do
+        mbErr <- Effectful.liftIO action'
+        for_ mbErr throwError
+
+    throwOnLeft action' = do
+        result <- Effectful.liftIO action'
+        case result of
+            Left err -> throwError err
+            Right a -> pure a
+
+    throwOnKafkaErr action' = do
+        err <- Effectful.liftIO action'
+        case err of
+            KafkaResponseError RdKafkaRespErrNoError -> pure ()
+            _ -> throwError err
diff --git a/src/Kafka/Effectful/Producer.hs b/src/Kafka/Effectful/Producer.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful/Producer.hs
@@ -0,0 +1,63 @@
+module Kafka.Effectful.Producer (
+    -- * Effect
+    KafkaProducer,
+
+    -- * Interpreter
+    runKafkaProducer,
+
+    -- * Operations
+    produceMessage,
+    flushProducer,
+
+    -- * Types
+    ProducerRecord (..),
+    ProducePartition (..),
+    DeliveryReport (..),
+    ImmediateError (..),
+
+    -- * Configuration
+    ProducerProperties (..),
+    K.brokersList,
+    K.setCallback,
+    K.logLevel,
+    K.compression,
+    K.topicCompression,
+    K.sendTimeout,
+    K.statisticsInterval,
+    K.extraProps,
+    K.extraProp,
+    K.suppressDisconnectLogs,
+    K.extraTopicProps,
+    K.debugOptions,
+
+    -- * Callbacks
+    K.deliveryCallback,
+    K.errorCallback,
+    K.logCallback,
+    K.statsCallback,
+    K.Callback,
+
+    -- * Common Types
+    KafkaError (..),
+    TopicName (..),
+    BrokerAddress (..),
+    Timeout (..),
+    KafkaLogLevel (..),
+    KafkaDebug (..),
+    KafkaCompressionCodec (..),
+    Headers,
+    headersFromList,
+    headersToList,
+    Offset (..),
+)
+where
+
+import Kafka.Effectful.Producer.Effect (KafkaProducer, flushProducer, produceMessage)
+import Kafka.Effectful.Producer.Interpreter (runKafkaProducer)
+import Kafka.Producer.ProducerProperties (ProducerProperties (..))
+import Kafka.Producer.ProducerProperties qualified as K
+import Kafka.Producer.Types (DeliveryReport (..), ImmediateError (..), ProducePartition (..), ProducerRecord (..))
+import Kafka.Types (BrokerAddress (..), Headers, KafkaCompressionCodec (..), KafkaDebug (..), KafkaError (..), KafkaLogLevel (..), Timeout (..), TopicName (..), headersFromList, headersToList)
+
+-- Offset is in Consumer.Types
+import Kafka.Consumer.Types (Offset (..))
diff --git a/src/Kafka/Effectful/Producer/Effect.hs b/src/Kafka/Effectful/Producer/Effect.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful/Producer/Effect.hs
@@ -0,0 +1,30 @@
+module Kafka.Effectful.Producer.Effect (
+    -- * Effect
+    KafkaProducer (..),
+
+    -- * Operations
+    produceMessage,
+    flushProducer,
+)
+where
+
+import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>))
+import Effectful.Dispatch.Dynamic (send)
+import Kafka.Producer.Types (ProducerRecord)
+
+-- | Effect for Kafka producer operations.
+data KafkaProducer :: Effect where
+    ProduceMessage :: ProducerRecord -> KafkaProducer m ()
+    FlushProducer :: KafkaProducer m ()
+
+type instance DispatchOf KafkaProducer = 'Dynamic
+
+{- | Send a single message to Kafka.
+Throws 'KafkaError' via the 'Error' effect on failure.
+-}
+produceMessage :: (KafkaProducer :> es) => ProducerRecord -> Eff es ()
+produceMessage = send . ProduceMessage
+
+-- | Flush the producer's outbound queue, blocking until all messages are sent.
+flushProducer :: (KafkaProducer :> es) => Eff es ()
+flushProducer = send FlushProducer
diff --git a/src/Kafka/Effectful/Producer/Interpreter.hs b/src/Kafka/Effectful/Producer/Interpreter.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Effectful/Producer/Interpreter.hs
@@ -0,0 +1,55 @@
+-- The 'EffectHandler' type synonym in effectful-core expands to a
+-- constraint that GHC's redundant-constraint check flags on the
+-- handler's signature, even though the constraint is required for
+-- 'interpret' to type-check. Suppress the warning at the file level.
+{-# OPTIONS_GHC -Wno-redundant-constraints #-}
+
+module Kafka.Effectful.Producer.Interpreter (
+    -- * Interpreter
+    runKafkaProducer,
+)
+where
+
+import Data.Foldable (for_)
+import Effectful (Eff, IOE, (:>))
+import Effectful qualified
+import Effectful.Dispatch.Dynamic (EffectHandler, interpret)
+import Effectful.Error.Static (Error, throwError)
+import Effectful.Exception qualified as Exception
+import Kafka.Effectful.Producer.Effect (KafkaProducer (..))
+import Kafka.Producer qualified as K
+import Kafka.Producer.ProducerProperties (ProducerProperties)
+import Kafka.Types (KafkaError)
+
+{- | Run the 'KafkaProducer' effect.
+
+Acquires a producer handle from the given properties and releases it
+when the effect scope ends. Errors are thrown via the 'Error' effect.
+-}
+runKafkaProducer ::
+    (IOE :> es, Error KafkaError :> es) =>
+    ProducerProperties ->
+    Eff (KafkaProducer : es) a ->
+    Eff es a
+runKafkaProducer props action =
+    Exception.bracket
+        acquire
+        (Effectful.liftIO . K.closeProducer)
+        (\producer -> interpret (handleProducer producer) action)
+  where
+    acquire = do
+        result <- Effectful.liftIO $ K.newProducer props
+        case result of
+            Left err -> throwError err
+            Right producer -> pure producer
+
+handleProducer ::
+    (IOE :> es, Error KafkaError :> es) =>
+    K.KafkaProducer ->
+    EffectHandler KafkaProducer es
+handleProducer producer _env = \case
+    ProduceMessage record -> do
+        mbErr <- Effectful.liftIO $ K.produceMessage producer record
+        for_ mbErr throwError
+    FlushProducer ->
+        Effectful.liftIO $ K.flushProducer producer