diff --git a/Cargo.lock b/Cargo.lock
index 2e0ebfeba..653d90009 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3029,6 +3029,7 @@ dependencies = [
  "eyre",
  "reth",
  "reth-chainspec",
+ "reth-ethereum-payload-builder",
  "reth-evm",
  "reth-evm-ethereum",
  "reth-node-api",
@@ -7205,7 +7206,6 @@ dependencies = [
  "reth-node-types",
  "reth-optimism-chainspec",
  "reth-payload-builder",
- "reth-payload-builder-primitives",
  "reth-payload-primitives",
  "reth-provider",
  "reth-prune",
@@ -7300,7 +7300,6 @@ dependencies = [
  "reth-metrics",
  "reth-network-p2p",
  "reth-payload-builder",
- "reth-payload-builder-primitives",
  "reth-payload-primitives",
  "reth-primitives-traits",
  "reth-provider",
@@ -7696,6 +7695,7 @@ dependencies = [
  "reth-consensus",
  "reth-db",
  "reth-db-common",
+ "reth-ethereum-payload-builder",
  "reth-evm",
  "reth-execution-types",
  "reth-exex",
@@ -8011,6 +8011,7 @@ version = "1.1.5"
 dependencies = [
  "alloy-rpc-types-engine",
  "eyre",
+ "reth-basic-payload-builder",
  "reth-consensus",
  "reth-db-api",
  "reth-engine-primitives",
@@ -8018,6 +8019,7 @@ dependencies = [
  "reth-network-api",
  "reth-node-core",
  "reth-node-types",
+ "reth-payload-builder",
  "reth-payload-builder-primitives",
  "reth-payload-primitives",
  "reth-provider",
@@ -8039,6 +8041,7 @@ dependencies = [
  "futures",
  "jsonrpsee",
  "rayon",
+ "reth-basic-payload-builder",
  "reth-chain-state",
  "reth-chainspec",
  "reth-cli-util",
@@ -8155,7 +8158,6 @@ dependencies = [
  "eyre",
  "futures",
  "rand 0.8.5",
- "reth-basic-payload-builder",
  "reth-chainspec",
  "reth-consensus",
  "reth-db",
@@ -8171,7 +8173,6 @@ dependencies = [
  "reth-node-api",
  "reth-node-builder",
  "reth-node-core",
- "reth-payload-builder",
  "reth-payload-primitives",
  "reth-provider",
  "reth-revm",
@@ -8588,7 +8589,6 @@ dependencies = [
  "alloy-consensus",
  "alloy-primitives",
  "alloy-rpc-types",
- "async-trait",
  "futures-util",
  "metrics",
  "reth-chain-state",
@@ -8608,8 +8608,6 @@ dependencies = [
 name = "reth-payload-builder-primitives"
 version = "1.1.5"
 dependencies = [
- "alloy-rpc-types-engine",
- "async-trait",
  "pin-project",
  "reth-payload-primitives",
  "tokio",
diff --git a/crates/e2e-test-utils/src/node.rs b/crates/e2e-test-utils/src/node.rs
index 63aaf1bd1..0e02dc57b 100644
--- a/crates/e2e-test-utils/src/node.rs
+++ b/crates/e2e-test-utils/src/node.rs
@@ -60,11 +60,13 @@ where
         node: FullNode<Node, AddOns>,
         attributes_generator: impl Fn(u64) -> Engine::PayloadBuilderAttributes + 'static,
     ) -> eyre::Result<Self> {
-        let builder = node.payload_builder.clone();
-
         Ok(Self {
             inner: node.clone(),
-            payload: PayloadTestContext::new(builder, attributes_generator).await?,
+            payload: PayloadTestContext::new(
+                node.payload_builder_handle.clone(),
+                attributes_generator,
+            )
+            .await?,
             network: NetworkTestContext::new(node.network.clone()),
             engine_api: EngineApiTestContext {
                 chain_spec: node.chain_spec(),
diff --git a/crates/e2e-test-utils/src/payload.rs b/crates/e2e-test-utils/src/payload.rs
index 858e311ca..f98e85f17 100644
--- a/crates/e2e-test-utils/src/payload.rs
+++ b/crates/e2e-test-utils/src/payload.rs
@@ -1,7 +1,7 @@
 use futures_util::StreamExt;
 use reth_node_api::BlockBody;
 use reth_payload_builder::{PayloadBuilderHandle, PayloadId};
-use reth_payload_builder_primitives::{Events, PayloadBuilder};
+use reth_payload_builder_primitives::Events;
 use reth_payload_primitives::{BuiltPayload, PayloadBuilderAttributes, PayloadTypes};
 use tokio_stream::wrappers::BroadcastStream;
 
diff --git a/crates/engine/local/Cargo.toml b/crates/engine/local/Cargo.toml
index bd4910cf7..a6a6ae371 100644
--- a/crates/engine/local/Cargo.toml
+++ b/crates/engine/local/Cargo.toml
@@ -19,7 +19,6 @@ reth-node-types.workspace = true
 reth-evm.workspace = true
 reth-ethereum-engine-primitives.workspace = true
 reth-payload-builder.workspace = true
-reth-payload-builder-primitives.workspace = true
 reth-payload-primitives.workspace = true
 reth-provider.workspace = true
 reth-prune.workspace = true
diff --git a/crates/engine/local/src/miner.rs b/crates/engine/local/src/miner.rs
index cee570e4e..484a1eeae 100644
--- a/crates/engine/local/src/miner.rs
+++ b/crates/engine/local/src/miner.rs
@@ -7,7 +7,6 @@ use eyre::OptionExt;
 use futures_util::{stream::Fuse, StreamExt};
 use reth_engine_primitives::{BeaconEngineMessage, EngineTypes};
 use reth_payload_builder::PayloadBuilderHandle;
-use reth_payload_builder_primitives::PayloadBuilder;
 use reth_payload_primitives::{
     BuiltPayload, EngineApiMessageVersion, PayloadAttributesBuilder, PayloadKind, PayloadTypes,
 };
diff --git a/crates/engine/tree/Cargo.toml b/crates/engine/tree/Cargo.toml
index 9b9085d82..eb5ac3db1 100644
--- a/crates/engine/tree/Cargo.toml
+++ b/crates/engine/tree/Cargo.toml
@@ -20,7 +20,6 @@ reth-engine-primitives.workspace = true
 reth-errors.workspace = true
 reth-evm.workspace = true
 reth-network-p2p.workspace = true
-reth-payload-builder-primitives.workspace = true
 reth-payload-builder.workspace = true
 reth-payload-primitives.workspace = true
 reth-primitives-traits.workspace = true
diff --git a/crates/engine/tree/src/tree/mod.rs b/crates/engine/tree/src/tree/mod.rs
index 288134194..7ba9a1a2c 100644
--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -40,7 +40,6 @@ use reth_evm::{
     ConfigureEvm, Evm, TransactionEnv,
 };
 use reth_payload_builder::PayloadBuilderHandle;
-use reth_payload_builder_primitives::PayloadBuilder;
 use reth_payload_primitives::{EngineApiMessageVersion, PayloadBuilderAttributes};
 use reth_primitives_traits::{
     Block, GotExpected, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader,
diff --git a/crates/ethereum/node/Cargo.toml b/crates/ethereum/node/Cargo.toml
index f8d3542e4..609a5a122 100644
--- a/crates/ethereum/node/Cargo.toml
+++ b/crates/ethereum/node/Cargo.toml
@@ -12,9 +12,7 @@ workspace = true
 
 [dependencies]
 # reth
-reth-payload-builder.workspace = true
 reth-ethereum-engine-primitives.workspace = true
-reth-basic-payload-builder.workspace = true
 reth-ethereum-payload-builder.workspace = true
 reth-ethereum-consensus.workspace = true
 reth-ethereum-primitives.workspace = true
@@ -75,7 +73,6 @@ test-utils = [
     "reth-chainspec/test-utils",
     "reth-consensus/test-utils",
     "reth-network/test-utils",
-    "reth-payload-builder/test-utils",
     "reth-ethereum-primitives/test-utils",
     "reth-revm/test-utils",
     "reth-db/test-utils",
diff --git a/crates/ethereum/node/src/payload.rs b/crates/ethereum/node/src/payload.rs
index 8c6d3e3e6..ad156bf52 100644
--- a/crates/ethereum/node/src/payload.rs
+++ b/crates/ethereum/node/src/payload.rs
@@ -1,6 +1,5 @@
 //! Payload component configuration for the Ethereum node.
 
-use reth_basic_payload_builder::{BasicPayloadJobGenerator, BasicPayloadJobGeneratorConfig};
 use reth_chainspec::ChainSpec;
 use reth_ethereum_engine_primitives::{
     EthBuiltPayload, EthPayloadAttributes, EthPayloadBuilderAttributes,
@@ -13,8 +12,6 @@ use reth_node_api::{FullNodeTypes, NodeTypesWithEngine, PrimitivesTy, TxTy};
 use reth_node_builder::{
     components::PayloadServiceBuilder, BuilderContext, PayloadBuilderConfig, PayloadTypes,
 };
-use reth_payload_builder::{PayloadBuilderHandle, PayloadBuilderService};
-use reth_provider::CanonStateSubscriptions;
 use reth_transaction_pool::{PoolTransaction, TransactionPool};
 
 /// A basic ethereum payload service.
@@ -23,13 +20,16 @@ use reth_transaction_pool::{PoolTransaction, TransactionPool};
 pub struct EthereumPayloadBuilder;
 
 impl EthereumPayloadBuilder {
-    /// A helper method initializing [`PayloadBuilderService`] with the given EVM config.
-    pub fn spawn<Types, Node, Evm, Pool>(
-        self,
+    /// A helper method initializing [`reth_ethereum_payload_builder::EthereumPayloadBuilder`] with
+    /// the given EVM config.
+    pub fn build<Types, Node, Evm, Pool>(
+        &self,
         evm_config: Evm,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<PayloadBuilderHandle<Types::Engine>>
+    ) -> eyre::Result<
+        reth_ethereum_payload_builder::EthereumPayloadBuilder<Pool, Node::Provider, Evm>,
+    >
     where
         Types: NodeTypesWithEngine<ChainSpec = ChainSpec, Primitives = EthPrimitives>,
         Node: FullNodeTypes<Types = Types>,
@@ -44,30 +44,12 @@ impl EthereumPayloadBuilder {
         >,
     {
         let conf = ctx.payload_builder_config();
-        let payload_builder = reth_ethereum_payload_builder::EthereumPayloadBuilder::new(
+        Ok(reth_ethereum_payload_builder::EthereumPayloadBuilder::new(
             ctx.provider().clone(),
             pool,
             evm_config,
             EthereumBuilderConfig::new(conf.extra_data_bytes()).with_gas_limit(conf.gas_limit()),
-        );
-
-        let payload_job_config = BasicPayloadJobGeneratorConfig::default()
-            .interval(conf.interval())
-            .deadline(conf.deadline())
-            .max_payload_tasks(conf.max_payload_tasks());
-
-        let payload_generator = BasicPayloadJobGenerator::with_builder(
-            ctx.provider().clone(),
-            ctx.task_executor().clone(),
-            payload_job_config,
-            payload_builder,
-        );
-        let (payload_service, payload_builder) =
-            PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
-
-        ctx.task_executor().spawn_critical("payload builder service", Box::pin(payload_service));
-
-        Ok(payload_builder)
+        ))
     }
 }
 
@@ -84,11 +66,14 @@ where
         PayloadBuilderAttributes = EthPayloadBuilderAttributes,
     >,
 {
-    async fn spawn_payload_service(
-        self,
+    type PayloadBuilder =
+        reth_ethereum_payload_builder::EthereumPayloadBuilder<Pool, Node::Provider, EthEvmConfig>;
+
+    async fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<PayloadBuilderHandle<Types::Engine>> {
-        self.spawn(EthEvmConfig::new(ctx.chain_spec()), ctx, pool)
+    ) -> eyre::Result<Self::PayloadBuilder> {
+        self.build(EthEvmConfig::new(ctx.chain_spec()), ctx, pool)
     }
 }
diff --git a/crates/ethereum/node/tests/e2e/utils.rs b/crates/ethereum/node/tests/e2e/utils.rs
index 362002c03..d33dc3650 100644
--- a/crates/ethereum/node/tests/e2e/utils.rs
+++ b/crates/ethereum/node/tests/e2e/utils.rs
@@ -11,10 +11,10 @@ use alloy_rpc_types_eth::TransactionRequest;
 use alloy_signer::SignerSync;
 use rand::{seq::SliceRandom, Rng};
 use reth_e2e_test_utils::{wallet::Wallet, NodeHelperType, TmpDB};
+use reth_ethereum_engine_primitives::EthPayloadBuilderAttributes;
 use reth_ethereum_primitives::TxType;
 use reth_node_api::NodeTypesWithDBAdapter;
 use reth_node_ethereum::EthereumNode;
-use reth_payload_builder::EthPayloadBuilderAttributes;
 use reth_provider::FullProvider;
 use revm::primitives::{AccessListItem, Authorization};
 
diff --git a/crates/exex/test-utils/Cargo.toml b/crates/exex/test-utils/Cargo.toml
index 1bb82e97f..ec8f9e397 100644
--- a/crates/exex/test-utils/Cargo.toml
+++ b/crates/exex/test-utils/Cargo.toml
@@ -20,18 +20,19 @@ reth-db-common.workspace = true
 reth-evm = { workspace = true, features = ["test-utils"] }
 reth-execution-types.workspace = true
 reth-exex.workspace = true
+reth-payload-builder.workspace = true
 reth-network.workspace = true
 reth-node-api.workspace = true
 reth-node-core.workspace = true
 reth-node-builder = { workspace = true, features = ["test-utils"] }
 reth-node-ethereum.workspace = true
-reth-payload-builder.workspace = true
 reth-primitives.workspace = true
 reth-primitives-traits.workspace = true
 reth-provider = { workspace = true, features = ["test-utils"] }
 reth-tasks.workspace = true
 reth-transaction-pool = { workspace = true, features = ["test-utils"] }
 reth-trie-db.workspace = true
+reth-ethereum-payload-builder.workspace = true
 
 ## alloy
 alloy-eips.workspace = true
diff --git a/crates/exex/test-utils/src/lib.rs b/crates/exex/test-utils/src/lib.rs
index b29353d15..0ffdecf0a 100644
--- a/crates/exex/test-utils/src/lib.rs
+++ b/crates/exex/test-utils/src/lib.rs
@@ -24,6 +24,7 @@ use reth_db::{
     DatabaseEnv,
 };
 use reth_db_common::init::init_genesis;
+use reth_ethereum_payload_builder::EthereumBuilderConfig;
 use reth_evm::test_utils::MockExecutorProvider;
 use reth_execution_types::Chain;
 use reth_exex::{ExExContext, ExExEvent, ExExNotification, ExExNotifications, Wal};
@@ -287,7 +288,14 @@ pub async fn test_exex_context_with_chain_spec(
     let task_executor = tasks.executor();
     tasks.executor().spawn(network_manager);
 
-    let (_, payload_builder) = NoopPayloadBuilderService::<EthEngineTypes>::new();
+    let payload_builder = reth_ethereum_payload_builder::EthereumPayloadBuilder::new(
+        provider.clone(),
+        transaction_pool.clone(),
+        evm_config.clone(),
+        EthereumBuilderConfig::new(Default::default()),
+    );
+
+    let (_, payload_builder_handle) = NoopPayloadBuilderService::<EthEngineTypes>::new();
 
     let components = NodeAdapter::<FullNodeTypesAdapter<_, _, _>, _> {
         components: Components {
@@ -297,6 +305,7 @@ pub async fn test_exex_context_with_chain_spec(
             consensus,
             network,
             payload_builder,
+            payload_builder_handle,
         },
         task_executor,
         provider,
diff --git a/crates/node/api/Cargo.toml b/crates/node/api/Cargo.toml
index fa323108d..afa33a273 100644
--- a/crates/node/api/Cargo.toml
+++ b/crates/node/api/Cargo.toml
@@ -12,12 +12,14 @@ workspace = true
 
 [dependencies]
 # reth
+reth-basic-payload-builder.workspace = true
 reth-db-api.workspace = true
 reth-consensus.workspace = true
 reth-evm.workspace = true
 reth-provider.workspace = true
 reth-engine-primitives.workspace = true
 reth-transaction-pool.workspace = true
+reth-payload-builder.workspace = true
 reth-payload-builder-primitives.workspace = true
 reth-payload-primitives.workspace = true
 reth-tasks.workspace = true
diff --git a/crates/node/api/src/node.rs b/crates/node/api/src/node.rs
index 088729dd8..3e2122321 100644
--- a/crates/node/api/src/node.rs
+++ b/crates/node/api/src/node.rs
@@ -1,6 +1,8 @@
 //! Traits for configuring a node.
 
+use crate::PayloadTypes;
 use alloy_rpc_types_engine::JwtSecret;
+use reth_basic_payload_builder::PayloadBuilder;
 use reth_consensus::{ConsensusError, FullConsensus};
 use reth_db_api::{database_metrics::DatabaseMetrics, Database};
 use reth_engine_primitives::BeaconConsensusEngineHandle;
@@ -8,7 +10,7 @@ use reth_evm::{execute::BlockExecutorProvider, ConfigureEvmFor};
 use reth_network_api::FullNetwork;
 use reth_node_core::node_config::NodeConfig;
 use reth_node_types::{NodeTypes, NodeTypesWithDBAdapter, NodeTypesWithEngine, TxTy};
-use reth_payload_builder_primitives::PayloadBuilder;
+use reth_payload_builder::PayloadBuilderHandle;
 use reth_provider::FullProvider;
 use reth_tasks::TaskExecutor;
 use reth_transaction_pool::{PoolTransaction, TransactionPool};
@@ -42,6 +44,23 @@ where
     type Provider = Provider;
 }
 
+/// Helper trait to bound [`PayloadBuilder`] to the node's engine types.
+pub trait PayloadBuilderFor<N: NodeTypesWithEngine>:
+    PayloadBuilder<
+    Attributes = <N::Engine as PayloadTypes>::PayloadBuilderAttributes,
+    BuiltPayload = <N::Engine as PayloadTypes>::BuiltPayload,
+>
+{
+}
+
+impl<T, N: NodeTypesWithEngine> PayloadBuilderFor<N> for T where
+    T: PayloadBuilder<
+        Attributes = <N::Engine as PayloadTypes>::PayloadBuilderAttributes,
+        BuiltPayload = <N::Engine as PayloadTypes>::BuiltPayload,
+    >
+{
+}
+
 /// Encapsulates all types and components of the node.
 pub trait FullNodeComponents: FullNodeTypes + Clone + 'static {
     /// The transaction pool of the node.
@@ -63,8 +82,7 @@ pub trait FullNodeComponents: FullNodeTypes + Clone + 'static {
     type Network: FullNetwork;
 
     /// Builds new blocks.
-    type PayloadBuilder: PayloadBuilder<PayloadType = <Self::Types as NodeTypesWithEngine>::Engine>
-        + Clone;
+    type PayloadBuilder: PayloadBuilderFor<Self::Types>;
 
     /// Returns the transaction pool of the node.
     fn pool(&self) -> &Self::Pool;
@@ -81,9 +99,15 @@ pub trait FullNodeComponents: FullNodeTypes + Clone + 'static {
     /// Returns the handle to the network
     fn network(&self) -> &Self::Network;
 
-    /// Returns the handle to the payload builder service.
+    /// Returns the configured payload builder.
     fn payload_builder(&self) -> &Self::PayloadBuilder;
 
+    /// Returns the handle to the payload builder service handling payload building requests from
+    /// the engine.
+    fn payload_builder_handle(
+        &self,
+    ) -> &PayloadBuilderHandle<<Self::Types as NodeTypesWithEngine>::Engine>;
+
     /// Returns the provider of the node.
     fn provider(&self) -> &Self::Provider;
 
diff --git a/crates/node/builder/Cargo.toml b/crates/node/builder/Cargo.toml
index b455f6af4..38f3f0a3a 100644
--- a/crates/node/builder/Cargo.toml
+++ b/crates/node/builder/Cargo.toml
@@ -54,6 +54,7 @@ reth-tasks.workspace = true
 reth-tokio-util.workspace = true
 reth-tracing.workspace = true
 reth-transaction-pool.workspace = true
+reth-basic-payload-builder.workspace = true
 
 ## ethereum
 alloy-consensus.workspace = true
diff --git a/crates/node/builder/src/builder/states.rs b/crates/node/builder/src/builder/states.rs
index 058700348..a07caccbd 100644
--- a/crates/node/builder/src/builder/states.rs
+++ b/crates/node/builder/src/builder/states.rs
@@ -119,6 +119,14 @@ impl<T: FullNodeTypes, C: NodeComponents<T>> FullNodeComponents for NodeAdapter<
         self.components.payload_builder()
     }
 
+    fn payload_builder_handle(
+        &self,
+    ) -> &reth_payload_builder::PayloadBuilderHandle<
+        <Self::Types as reth_node_api::NodeTypesWithEngine>::Engine,
+    > {
+        self.components.payload_builder_handle()
+    }
+
     fn provider(&self) -> &Self::Provider {
         &self.provider
     }
diff --git a/crates/node/builder/src/components/builder.rs b/crates/node/builder/src/components/builder.rs
index a9576d8c1..8b8b35dd4 100644
--- a/crates/node/builder/src/components/builder.rs
+++ b/crates/node/builder/src/components/builder.rs
@@ -10,11 +10,12 @@ use crate::{
 use reth_consensus::{ConsensusError, FullConsensus};
 use reth_evm::{execute::BlockExecutorProvider, ConfigureEvmFor};
 use reth_network::NetworkPrimitives;
-use reth_node_api::{BlockTy, BodyTy, HeaderTy, NodeTypesWithEngine, PrimitivesTy, TxTy};
-use reth_payload_builder::PayloadBuilderHandle;
+use reth_node_api::{BlockTy, BodyTy, HeaderTy, PrimitivesTy, TxTy};
 use reth_transaction_pool::{PoolTransaction, TransactionPool};
 use std::{future::Future, marker::PhantomData};
 
+use super::PayloadBuilderFor;
+
 /// A generic, general purpose and customizable [`NodeComponentsBuilder`] implementation.
 ///
 /// This type is stateful and captures the configuration of the node's components.
@@ -324,6 +325,7 @@ where
         ExecB::EVM,
         ExecB::Executor,
         ConsB::Consensus,
+        PayloadB::PayloadBuilder,
     >;
 
     async fn build_components(
@@ -332,7 +334,7 @@ where
     ) -> eyre::Result<Self::Components> {
         let Self {
             pool_builder,
-            payload_builder,
+            payload_builder: payload_builder_builder,
             network_builder,
             executor_builder: evm_builder,
             consensus_builder,
@@ -342,7 +344,10 @@ where
         let (evm_config, executor) = evm_builder.build_evm(context).await?;
         let pool = pool_builder.build_pool(context).await?;
         let network = network_builder.build_network(context, pool.clone()).await?;
-        let payload_builder = payload_builder.spawn_payload_service(context, pool.clone()).await?;
+        let payload_builder =
+            payload_builder_builder.build_payload_builder(context, pool.clone()).await?;
+        let payload_builder_handle =
+            payload_builder_builder.spawn_payload_builder_service(context, payload_builder.clone());
         let consensus = consensus_builder.build_consensus(context).await?;
 
         Ok(Components {
@@ -350,6 +355,7 @@ where
             evm_config,
             network,
             payload_builder,
+            payload_builder_handle,
             executor,
             consensus,
         })
@@ -380,10 +386,7 @@ impl Default for ComponentsBuilder<(), (), (), (), (), ()> {
 /// A type that's responsible for building the components of the node.
 pub trait NodeComponentsBuilder<Node: FullNodeTypes>: Send {
     /// The components for the node with the given types
-    type Components: NodeComponents<
-        Node,
-        PayloadBuilder = PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>,
-    >;
+    type Components: NodeComponents<Node>;
 
     /// Consumes the type and returns the created components.
     fn build_components(
@@ -392,7 +395,7 @@ pub trait NodeComponentsBuilder<Node: FullNodeTypes>: Send {
     ) -> impl Future<Output = eyre::Result<Self::Components>> + Send;
 }
 
-impl<Node, N, F, Fut, Pool, EVM, Executor, Cons> NodeComponentsBuilder<Node> for F
+impl<Node, N, F, Fut, Pool, EVM, Executor, Cons, Payload> NodeComponentsBuilder<Node> for F
 where
     N: NetworkPrimitives<
         BlockHeader = HeaderTy<Node::Types>,
@@ -401,7 +404,8 @@ where
     >,
     Node: FullNodeTypes,
     F: FnOnce(&BuilderContext<Node>) -> Fut + Send,
-    Fut: Future<Output = eyre::Result<Components<Node, N, Pool, EVM, Executor, Cons>>> + Send,
+    Fut: Future<Output = eyre::Result<Components<Node, N, Pool, EVM, Executor, Cons, Payload>>>
+        + Send,
     Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
         + Unpin
         + 'static,
@@ -409,8 +413,9 @@ where
     Executor: BlockExecutorProvider<Primitives = PrimitivesTy<Node::Types>>,
     Cons:
         FullConsensus<PrimitivesTy<Node::Types>, Error = ConsensusError> + Clone + Unpin + 'static,
+    Payload: PayloadBuilderFor<Node::Types> + Unpin + 'static,
 {
-    type Components = Components<Node, N, Pool, EVM, Executor, Cons>;
+    type Components = Components<Node, N, Pool, EVM, Executor, Cons, Payload>;
 
     fn build_components(
         self,
diff --git a/crates/node/builder/src/components/mod.rs b/crates/node/builder/src/components/mod.rs
index 538427289..e2ebdc058 100644
--- a/crates/node/builder/src/components/mod.rs
+++ b/crates/node/builder/src/components/mod.rs
@@ -21,6 +21,7 @@ pub use network::*;
 pub use payload::*;
 pub use pool::*;
 use reth_network_p2p::BlockClient;
+use reth_payload_builder::PayloadBuilderHandle;
 
 use crate::{ConfigureEvm, FullNodeTypes};
 use reth_consensus::{ConsensusError, FullConsensus};
@@ -28,9 +29,9 @@ use reth_evm::{execute::BlockExecutorProvider, ConfigureEvmFor};
 use reth_network::{NetworkHandle, NetworkPrimitives};
 use reth_network_api::FullNetwork;
 use reth_node_api::{
-    BlockTy, BodyTy, HeaderTy, NodeTypes, NodeTypesWithEngine, PayloadBuilder, PrimitivesTy, TxTy,
+    BlockTy, BodyTy, HeaderTy, NodeTypes, NodeTypesWithEngine, PayloadBuilderFor, PrimitivesTy,
+    TxTy,
 };
-use reth_payload_builder::PayloadBuilderHandle;
 use reth_transaction_pool::{PoolTransaction, TransactionPool};
 
 /// An abstraction over the components of a node, consisting of:
@@ -58,8 +59,7 @@ pub trait NodeComponents<T: FullNodeTypes>: Clone + Unpin + Send + Sync + 'stati
     type Network: FullNetwork<Client: BlockClient<Block = BlockTy<T::Types>>>;
 
     /// Builds new blocks.
-    type PayloadBuilder: PayloadBuilder<PayloadType = <T::Types as NodeTypesWithEngine>::Engine>
-        + Clone;
+    type PayloadBuilder: PayloadBuilderFor<T::Types> + Clone + Unpin + 'static;
 
     /// Returns the transaction pool of the node.
     fn pool(&self) -> &Self::Pool;
@@ -76,15 +76,29 @@ pub trait NodeComponents<T: FullNodeTypes>: Clone + Unpin + Send + Sync + 'stati
     /// Returns the handle to the network
     fn network(&self) -> &Self::Network;
 
-    /// Returns the handle to the payload builder service.
+    /// Returns the payload builder that knows how to build blocks.
     fn payload_builder(&self) -> &Self::PayloadBuilder;
+
+    /// Returns the handle to the payload builder service handling payload building requests from
+    /// the engine.
+    fn payload_builder_handle(
+        &self,
+    ) -> &PayloadBuilderHandle<<T::Types as NodeTypesWithEngine>::Engine>;
 }
 
 /// All the components of the node.
 ///
 /// This provides access to all the components of the node.
 #[derive(Debug)]
-pub struct Components<Node: FullNodeTypes, N: NetworkPrimitives, Pool, EVM, Executor, Consensus> {
+pub struct Components<
+    Node: FullNodeTypes,
+    N: NetworkPrimitives,
+    Pool,
+    EVM,
+    Executor,
+    Consensus,
+    Payload,
+> {
     /// The transaction pool of the node.
     pub transaction_pool: Pool,
     /// The node's EVM configuration, defining settings for the Ethereum Virtual Machine.
@@ -95,19 +109,21 @@ pub struct Components<Node: FullNodeTypes, N: NetworkPrimitives, Pool, EVM, Exec
     pub consensus: Consensus,
     /// The network implementation of the node.
     pub network: NetworkHandle<N>,
+    /// The payload builder.
+    pub payload_builder: Payload,
     /// The handle to the payload builder service.
-    pub payload_builder: PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>,
+    pub payload_builder_handle: PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>,
 }
 
-impl<Node, Pool, EVM, Executor, Cons, N> NodeComponents<Node>
-    for Components<Node, N, Pool, EVM, Executor, Cons>
+impl<Node, Pool, EVM, Executor, Cons, N, Payload> NodeComponents<Node>
+    for Components<Node, N, Pool, EVM, Executor, Cons, Payload>
 where
+    Node: FullNodeTypes,
     N: NetworkPrimitives<
         BlockHeader = HeaderTy<Node::Types>,
         BlockBody = BodyTy<Node::Types>,
         Block = BlockTy<Node::Types>,
     >,
-    Node: FullNodeTypes,
     Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
         + Unpin
         + 'static,
@@ -115,13 +131,14 @@ where
     Executor: BlockExecutorProvider<Primitives = PrimitivesTy<Node::Types>>,
     Cons:
         FullConsensus<PrimitivesTy<Node::Types>, Error = ConsensusError> + Clone + Unpin + 'static,
+    Payload: PayloadBuilderFor<Node::Types> + Clone + Unpin + 'static,
 {
     type Pool = Pool;
     type Evm = EVM;
     type Executor = Executor;
     type Consensus = Cons;
     type Network = NetworkHandle<N>;
-    type PayloadBuilder = PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>;
+    type PayloadBuilder = Payload;
 
     fn pool(&self) -> &Self::Pool {
         &self.transaction_pool
@@ -146,16 +163,24 @@ where
     fn payload_builder(&self) -> &Self::PayloadBuilder {
         &self.payload_builder
     }
+
+    fn payload_builder_handle(
+        &self,
+    ) -> &PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine> {
+        &self.payload_builder_handle
+    }
 }
 
-impl<Node, N, Pool, EVM, Executor, Cons> Clone for Components<Node, N, Pool, EVM, Executor, Cons>
+impl<Node, N, Pool, EVM, Executor, Cons, Payload> Clone
+    for Components<Node, N, Pool, EVM, Executor, Cons, Payload>
 where
     N: NetworkPrimitives,
     Node: FullNodeTypes,
     Pool: TransactionPool,
-    EVM: ConfigureEvm<Header = HeaderTy<Node::Types>, Transaction = TxTy<Node::Types>>,
+    EVM: ConfigureEvm,
     Executor: BlockExecutorProvider,
     Cons: Clone,
+    Payload: Clone,
 {
     fn clone(&self) -> Self {
         Self {
@@ -165,6 +190,7 @@ where
             consensus: self.consensus.clone(),
             network: self.network.clone(),
             payload_builder: self.payload_builder.clone(),
+            payload_builder_handle: self.payload_builder_handle.clone(),
         }
     }
 }
diff --git a/crates/node/builder/src/components/payload.rs b/crates/node/builder/src/components/payload.rs
index 0efad9ba5..8ad1f5ec3 100644
--- a/crates/node/builder/src/components/payload.rs
+++ b/crates/node/builder/src/components/payload.rs
@@ -1,45 +1,73 @@
 //! Payload service component for the node builder.
 
+use crate::{BuilderContext, FullNodeTypes, NodeTypesWithEngine};
+use reth_basic_payload_builder::{BasicPayloadJobGenerator, BasicPayloadJobGeneratorConfig};
+use reth_chain_state::CanonStateSubscriptions;
+use reth_node_api::PayloadBuilderFor;
+use reth_payload_builder::{PayloadBuilderHandle, PayloadBuilderService};
+use reth_transaction_pool::TransactionPool;
 use std::future::Future;
 
-use reth_node_api::NodeTypesWithEngine;
-use reth_payload_builder::PayloadBuilderHandle;
-use reth_transaction_pool::TransactionPool;
-
-use crate::{BuilderContext, FullNodeTypes};
-
 /// A type that knows how to spawn the payload service.
-pub trait PayloadServiceBuilder<Node: FullNodeTypes, Pool: TransactionPool>: Send {
+pub trait PayloadServiceBuilder<Node: FullNodeTypes, Pool: TransactionPool>: Send + Sized {
+    /// Payload builder implementation.
+    type PayloadBuilder: PayloadBuilderFor<Node::Types> + Unpin + 'static;
+
     /// Spawns the payload service and returns the handle to it.
     ///
     /// The [`BuilderContext`] is provided to allow access to the node's configuration.
-    fn spawn_payload_service(
-        self,
+    fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> impl Future<
-        Output = eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>>,
-    > + Send;
+    ) -> impl Future<Output = eyre::Result<Self::PayloadBuilder>> + Send;
+
+    /// Spawns the [`PayloadBuilderService`] and returns the handle to it for use by the engine.
+    ///
+    /// We provide default implementation via [`BasicPayloadJobGenerator`] but it can be overridden
+    /// for custom job orchestration logic,
+    fn spawn_payload_builder_service(
+        self,
+        ctx: &BuilderContext<Node>,
+        payload_builder: Self::PayloadBuilder,
+    ) -> PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine> {
+        let conf = ctx.config().builder.clone();
+
+        let payload_job_config = BasicPayloadJobGeneratorConfig::default()
+            .interval(conf.interval)
+            .deadline(conf.deadline)
+            .max_payload_tasks(conf.max_payload_tasks);
+
+        let payload_generator = BasicPayloadJobGenerator::with_builder(
+            ctx.provider().clone(),
+            ctx.task_executor().clone(),
+            payload_job_config,
+            payload_builder,
+        );
+        let (payload_service, payload_service_handle) =
+            PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
+
+        ctx.task_executor().spawn_critical("payload builder service", Box::pin(payload_service));
+
+        payload_service_handle
+    }
 }
 
-impl<Node, F, Fut, Pool> PayloadServiceBuilder<Node, Pool> for F
+impl<Node, F, Fut, Pool, Builder> PayloadServiceBuilder<Node, Pool> for F
 where
     Node: FullNodeTypes,
     Pool: TransactionPool,
     F: Fn(&BuilderContext<Node>, Pool) -> Fut + Send,
-    Fut: Future<
-            Output = eyre::Result<
-                PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>,
-            >,
-        > + Send,
+    Fut: Future<Output = eyre::Result<Builder>> + Send,
+    Builder: PayloadBuilderFor<Node::Types> + Unpin + 'static,
 {
-    fn spawn_payload_service(
-        self,
+    type PayloadBuilder = Builder;
+
+    fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> impl Future<
-        Output = eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>>,
-    > + Send {
+    ) -> impl Future<Output = eyre::Result<Self::PayloadBuilder>> + Send {
         self(ctx, pool)
     }
 }
diff --git a/crates/node/builder/src/launch/engine.rs b/crates/node/builder/src/launch/engine.rs
index 7ae03dc3a..445734a4e 100644
--- a/crates/node/builder/src/launch/engine.rs
+++ b/crates/node/builder/src/launch/engine.rs
@@ -17,7 +17,7 @@ use reth_network::{NetworkSyncUpdater, SyncState};
 use reth_network_api::BlockDownloaderProvider;
 use reth_node_api::{
     BeaconConsensusEngineHandle, BuiltPayload, FullNodeTypes, NodeTypesWithDBAdapter,
-    NodeTypesWithEngine, PayloadAttributesBuilder, PayloadBuilder, PayloadTypes,
+    NodeTypesWithEngine, PayloadAttributesBuilder, PayloadTypes,
 };
 use reth_node_core::{
     dirs::{ChainPath, DataDirPath},
@@ -218,7 +218,7 @@ where
                 ctx.provider_factory().clone(),
                 ctx.blockchain_db().clone(),
                 pruner,
-                ctx.components().payload_builder().clone(),
+                ctx.components().payload_builder_handle().clone(),
                 engine_payload_validator,
                 engine_tree_config,
                 ctx.invalid_block_hook()?,
@@ -243,7 +243,7 @@ where
                 ctx.provider_factory().clone(),
                 ctx.blockchain_db().clone(),
                 pruner,
-                ctx.components().payload_builder().clone(),
+                ctx.components().payload_builder_handle().clone(),
                 engine_payload_validator,
                 engine_tree_config,
                 ctx.invalid_block_hook()?,
@@ -317,7 +317,7 @@ where
         let network_handle = ctx.components().network().clone();
         let mut built_payloads = ctx
             .components()
-            .payload_builder()
+            .payload_builder_handle()
             .subscribe()
             .await
             .map_err(|e| eyre::eyre!("Failed to subscribe to payload builder events: {:?}", e))?
@@ -399,6 +399,7 @@ where
             network: ctx.components().network().clone(),
             provider: ctx.node_adapter().provider.clone(),
             payload_builder: ctx.components().payload_builder().clone(),
+            payload_builder_handle: ctx.components().payload_builder_handle().clone(),
             task_executor: ctx.task_executor().clone(),
             config: ctx.node_config().clone(),
             data_dir: ctx.data_dir().clone(),
diff --git a/crates/node/builder/src/node.rs b/crates/node/builder/src/node.rs
index ce7d12fee..d2e6ae8ae 100644
--- a/crates/node/builder/src/node.rs
+++ b/crates/node/builder/src/node.rs
@@ -1,5 +1,6 @@
 // re-export the node api types
 pub use reth_node_api::{FullNodeTypes, NodeTypes, NodeTypesWithEngine};
+use reth_payload_builder::PayloadBuilderHandle;
 
 use std::{
     marker::PhantomData,
@@ -12,7 +13,6 @@ use reth_node_core::{
     dirs::{ChainPath, DataDirPath},
     node_config::NodeConfig,
 };
-use reth_payload_builder::PayloadBuilderHandle;
 use reth_provider::ChainSpecProvider;
 use reth_rpc_api::EngineApiClient;
 use reth_rpc_builder::{auth::AuthServerHandle, RpcServerHandle};
@@ -117,8 +117,10 @@ pub struct FullNode<Node: FullNodeComponents, AddOns: NodeAddOns<Node>> {
     pub network: Node::Network,
     /// Provider to interact with the node's database
     pub provider: Node::Provider,
+    /// Node's configured payload builder.
+    pub payload_builder: Node::PayloadBuilder,
     /// Handle to the node's payload builder service.
-    pub payload_builder: PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>,
+    pub payload_builder_handle: PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>,
     /// Task executor for the node.
     pub task_executor: TaskExecutor,
     /// The initial node config.
@@ -138,6 +140,7 @@ impl<Node: FullNodeComponents, AddOns: NodeAddOns<Node>> Clone for FullNode<Node
             network: self.network.clone(),
             provider: self.provider.clone(),
             payload_builder: self.payload_builder.clone(),
+            payload_builder_handle: self.payload_builder_handle.clone(),
             task_executor: self.task_executor.clone(),
             config: self.config.clone(),
             data_dir: self.data_dir.clone(),
diff --git a/crates/node/builder/src/rpc.rs b/crates/node/builder/src/rpc.rs
index 7006a7d2c..ef597c911 100644
--- a/crates/node/builder/src/rpc.rs
+++ b/crates/node/builder/src/rpc.rs
@@ -437,7 +437,7 @@ where
             node.provider().clone(),
             config.chain.clone(),
             beacon_engine_handle,
-            PayloadStore::new(node.payload_builder().clone()),
+            PayloadStore::new(node.payload_builder_handle().clone()),
             node.pool().clone(),
             Box::new(node.task_executor().clone()),
             client,
diff --git a/crates/optimism/node/src/node.rs b/crates/optimism/node/src/node.rs
index 4725bc701..68c5bdc32 100644
--- a/crates/optimism/node/src/node.rs
+++ b/crates/optimism/node/src/node.rs
@@ -7,7 +7,6 @@ use crate::{
     OpEngineTypes,
 };
 use op_alloy_consensus::OpPooledTransaction;
-use reth_basic_payload_builder::{BasicPayloadJobGenerator, BasicPayloadJobGeneratorConfig};
 use reth_chainspec::{EthChainSpec, Hardforks};
 use reth_evm::{
     execute::BasicBlockExecutorProvider, ConfigureEvm, ConfigureEvmEnv, ConfigureEvmFor,
@@ -23,7 +22,7 @@ use reth_node_builder::{
     },
     node::{FullNodeTypes, NodeTypes, NodeTypesWithEngine},
     rpc::{EngineValidatorAddOn, EngineValidatorBuilder, RethRpcAddOns, RpcAddOns, RpcHandle},
-    BuilderContext, Node, NodeAdapter, NodeComponentsBuilder, PayloadBuilderConfig,
+    BuilderContext, Node, NodeAdapter, NodeComponentsBuilder,
 };
 use reth_optimism_chainspec::OpChainSpec;
 use reth_optimism_consensus::OpBeaconConsensus;
@@ -39,7 +38,6 @@ use reth_optimism_rpc::{
     witness::{DebugExecutionWitnessApiServer, OpDebugWitnessApi},
     OpEthApi, OpEthApiError, SequencerClient,
 };
-use reth_payload_builder::{PayloadBuilderHandle, PayloadBuilderService};
 use reth_provider::{CanonStateSubscriptions, EthStorage};
 use reth_rpc_eth_types::error::FromEvmError;
 use reth_rpc_server_types::RethRpcModule;
@@ -485,13 +483,23 @@ impl<Txs> OpPayloadBuilder<Txs> {
         OpPayloadBuilder { compute_pending_block, best_transactions, da_config }
     }
 
-    /// A helper method to initialize [`PayloadBuilderService`] with the given EVM config.
-    pub fn spawn<Node, Evm, Pool>(
-        self,
+    /// A helper method to initialize [`reth_optimism_payload_builder::OpPayloadBuilder`] with the
+    /// given EVM config.
+    #[expect(clippy::type_complexity)]
+    pub fn build<Node, Evm, Pool>(
+        &self,
         evm_config: Evm,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<PayloadBuilderHandle<OpEngineTypes>>
+    ) -> eyre::Result<
+        reth_optimism_payload_builder::OpPayloadBuilder<
+            Pool,
+            Node::Provider,
+            Evm,
+            PrimitivesTy<Node::Types>,
+            Txs,
+        >,
+    >
     where
         Node: FullNodeTypes<
             Types: NodeTypesWithEngine<
@@ -511,28 +519,10 @@ impl<Txs> OpPayloadBuilder<Txs> {
             ctx.provider().clone(),
             evm_config,
             BasicOpReceiptBuilder::default(),
-            OpBuilderConfig { da_config: self.da_config },
+            OpBuilderConfig { da_config: self.da_config.clone() },
         )
-        .with_transactions(self.best_transactions)
+        .with_transactions(self.best_transactions.clone())
         .set_compute_pending_block(self.compute_pending_block);
-        let conf = ctx.payload_builder_config();
-
-        let payload_job_config = BasicPayloadJobGeneratorConfig::default()
-            .interval(conf.interval())
-            .deadline(conf.deadline())
-            .max_payload_tasks(conf.max_payload_tasks());
-
-        let payload_generator = BasicPayloadJobGenerator::with_builder(
-            ctx.provider().clone(),
-            ctx.task_executor().clone(),
-            payload_job_config,
-            payload_builder,
-        );
-        let (payload_service, payload_builder) =
-            PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
-
-        ctx.task_executor().spawn_critical("payload builder service", Box::pin(payload_service));
-
         Ok(payload_builder)
     }
 }
@@ -551,12 +541,20 @@ where
         + 'static,
     Txs: OpPayloadTransactions<Pool::Transaction>,
 {
-    async fn spawn_payload_service(
-        self,
+    type PayloadBuilder = reth_optimism_payload_builder::OpPayloadBuilder<
+        Pool,
+        Node::Provider,
+        OpEvmConfig,
+        PrimitivesTy<Node::Types>,
+        Txs,
+    >;
+
+    async fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<PayloadBuilderHandle<OpEngineTypes>> {
-        self.spawn(OpEvmConfig::new(ctx.chain_spec()), ctx, pool)
+    ) -> eyre::Result<Self::PayloadBuilder> {
+        self.build(OpEvmConfig::new(ctx.chain_spec()), ctx, pool)
     }
 }
 
diff --git a/crates/payload/builder-primitives/Cargo.toml b/crates/payload/builder-primitives/Cargo.toml
index 6d89ea89d..70ff2d588 100644
--- a/crates/payload/builder-primitives/Cargo.toml
+++ b/crates/payload/builder-primitives/Cargo.toml
@@ -15,11 +15,7 @@ workspace = true
 # reth
 reth-payload-primitives.workspace = true
 
-# alloy
-alloy-rpc-types-engine = { workspace = true, features = ["serde"] }
-
 # async
-async-trait.workspace = true
 pin-project.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tokio-stream.workspace = true
diff --git a/crates/payload/builder-primitives/src/lib.rs b/crates/payload/builder-primitives/src/lib.rs
index af7ad736d..d181531ca 100644
--- a/crates/payload/builder-primitives/src/lib.rs
+++ b/crates/payload/builder-primitives/src/lib.rs
@@ -11,8 +11,4 @@
 mod events;
 pub use crate::events::{Events, PayloadEvents};
 
-/// Contains the payload builder trait to abstract over payload attributes.
-mod traits;
-pub use traits::{PayloadBuilder, PayloadStoreExt};
-
 pub use reth_payload_primitives::PayloadBuilderError;
diff --git a/crates/payload/builder-primitives/src/traits.rs b/crates/payload/builder-primitives/src/traits.rs
deleted file mode 100644
index b5e8910b6..000000000
--- a/crates/payload/builder-primitives/src/traits.rs
+++ /dev/null
@@ -1,111 +0,0 @@
-use crate::{PayloadBuilderError, PayloadEvents};
-use alloy_rpc_types_engine::PayloadId;
-use reth_payload_primitives::{PayloadKind, PayloadTypes};
-use std::fmt::Debug;
-use tokio::sync::oneshot;
-
-/// A helper trait for internal usage to retrieve and resolve payloads.
-#[async_trait::async_trait]
-pub trait PayloadStoreExt<T: PayloadTypes>: Debug + Send + Sync + Unpin {
-    /// Resolves the payload job and returns the best payload that has been built so far.
-    async fn resolve_kind(
-        &self,
-        id: PayloadId,
-        kind: PayloadKind,
-    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>>;
-
-    /// Resolves the payload job as fast and possible and returns the best payload that has been
-    /// built so far.
-    async fn resolve(&self, id: PayloadId) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
-        self.resolve_kind(id, PayloadKind::Earliest).await
-    }
-
-    /// Returns the best payload for the given identifier.
-    async fn best_payload(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>>;
-
-    /// Returns the payload attributes associated with the given identifier.
-    async fn payload_attributes(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<T::PayloadBuilderAttributes, PayloadBuilderError>>;
-}
-
-#[async_trait::async_trait]
-impl<T: PayloadTypes, P> PayloadStoreExt<T> for P
-where
-    P: PayloadBuilder<PayloadType = T>,
-{
-    async fn resolve_kind(
-        &self,
-        id: PayloadId,
-        kind: PayloadKind,
-    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
-        Some(PayloadBuilder::resolve_kind(self, id, kind).await?.map_err(Into::into))
-    }
-
-    async fn best_payload(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
-        Some(PayloadBuilder::best_payload(self, id).await?.map_err(Into::into))
-    }
-
-    async fn payload_attributes(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<T::PayloadBuilderAttributes, PayloadBuilderError>> {
-        Some(PayloadBuilder::payload_attributes(self, id).await?.map_err(Into::into))
-    }
-}
-
-/// A type that can request, subscribe to and resolve payloads.
-#[async_trait::async_trait]
-pub trait PayloadBuilder: Debug + Send + Sync + Unpin {
-    /// The Payload type for the builder.
-    type PayloadType: PayloadTypes;
-    /// The error type returned by the builder.
-    type Error: Into<PayloadBuilderError>;
-
-    /// Sends a message to the service to start building a new payload for the given payload.
-    ///
-    /// Returns a receiver that will receive the payload id.
-    fn send_new_payload(
-        &self,
-        attr: <Self::PayloadType as PayloadTypes>::PayloadBuilderAttributes,
-    ) -> oneshot::Receiver<Result<PayloadId, Self::Error>>;
-
-    /// Returns the best payload for the given identifier.
-    async fn best_payload(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<<Self::PayloadType as PayloadTypes>::BuiltPayload, Self::Error>>;
-
-    /// Resolves the payload job and returns the best payload that has been built so far.
-    async fn resolve_kind(
-        &self,
-        id: PayloadId,
-        kind: PayloadKind,
-    ) -> Option<Result<<Self::PayloadType as PayloadTypes>::BuiltPayload, Self::Error>>;
-
-    /// Resolves the payload job as fast and possible and returns the best payload that has been
-    /// built so far.
-    async fn resolve(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<<Self::PayloadType as PayloadTypes>::BuiltPayload, Self::Error>> {
-        self.resolve_kind(id, PayloadKind::Earliest).await
-    }
-
-    /// Sends a message to the service to subscribe to payload events.
-    /// Returns a receiver that will receive them.
-    async fn subscribe(&self) -> Result<PayloadEvents<Self::PayloadType>, Self::Error>;
-
-    /// Returns the payload attributes associated with the given identifier.
-    async fn payload_attributes(
-        &self,
-        id: PayloadId,
-    ) -> Option<Result<<Self::PayloadType as PayloadTypes>::PayloadBuilderAttributes, Self::Error>>;
-}
diff --git a/crates/payload/builder/Cargo.toml b/crates/payload/builder/Cargo.toml
index 33bed09f6..de04105df 100644
--- a/crates/payload/builder/Cargo.toml
+++ b/crates/payload/builder/Cargo.toml
@@ -26,7 +26,6 @@ alloy-primitives = { workspace = true, optional = true }
 alloy-rpc-types = { workspace = true, features = ["engine"] }
 
 # async
-async-trait.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tokio-stream.workspace = true
 futures-util.workspace = true
diff --git a/crates/payload/builder/src/service.rs b/crates/payload/builder/src/service.rs
index de51a791c..fc37a3830 100644
--- a/crates/payload/builder/src/service.rs
+++ b/crates/payload/builder/src/service.rs
@@ -11,9 +11,7 @@ use alloy_consensus::BlockHeader;
 use alloy_rpc_types::engine::PayloadId;
 use futures_util::{future::FutureExt, Stream, StreamExt};
 use reth_chain_state::CanonStateNotification;
-use reth_payload_builder_primitives::{
-    Events, PayloadBuilder, PayloadBuilderError, PayloadEvents, PayloadStoreExt,
-};
+use reth_payload_builder_primitives::{Events, PayloadBuilderError, PayloadEvents};
 use reth_payload_primitives::{BuiltPayload, PayloadBuilderAttributes, PayloadKind, PayloadTypes};
 use reth_primitives_traits::NodePrimitives;
 use std::{
@@ -38,7 +36,7 @@ type PayloadFuture<P> = Pin<Box<dyn Future<Output = Result<P, PayloadBuilderErro
 /// API).
 #[derive(Debug)]
 pub struct PayloadStore<T: PayloadTypes> {
-    inner: Arc<dyn PayloadStoreExt<T>>,
+    inner: Arc<PayloadBuilderHandle<T>>,
 }
 
 impl<T> PayloadStore<T>
@@ -91,10 +89,7 @@ where
     T: PayloadTypes,
 {
     /// Create a new instance
-    pub fn new<P>(inner: P) -> Self
-    where
-        P: PayloadStoreExt<T> + 'static,
-    {
+    pub fn new(inner: PayloadBuilderHandle<T>) -> Self {
         Self { inner: Arc::new(inner) }
     }
 }
@@ -117,36 +112,40 @@ pub struct PayloadBuilderHandle<T: PayloadTypes> {
     to_service: mpsc::UnboundedSender<PayloadServiceCommand<T>>,
 }
 
-// === impl PayloadBuilderHandle ===
+impl<T: PayloadTypes> PayloadBuilderHandle<T> {
+    /// Creates a new payload builder handle for the given channel.
+    ///
+    /// Note: this is only used internally by the [`PayloadBuilderService`] to manage the payload
+    /// building flow See [`PayloadBuilderService::poll`] for implementation details.
+    pub const fn new(to_service: mpsc::UnboundedSender<PayloadServiceCommand<T>>) -> Self {
+        Self { to_service }
+    }
 
-#[async_trait::async_trait]
-impl<T> PayloadBuilder for PayloadBuilderHandle<T>
-where
-    T: PayloadTypes,
-{
-    type PayloadType = T;
-    type Error = PayloadBuilderError;
-
-    fn send_new_payload(
+    /// Sends a message to the service to start building a new payload for the given payload.
+    ///
+    /// Returns a receiver that will receive the payload id.
+    pub fn send_new_payload(
         &self,
-        attr: <Self::PayloadType as PayloadTypes>::PayloadBuilderAttributes,
-    ) -> Receiver<Result<PayloadId, Self::Error>> {
+        attr: T::PayloadBuilderAttributes,
+    ) -> Receiver<Result<PayloadId, PayloadBuilderError>> {
         let (tx, rx) = oneshot::channel();
         let _ = self.to_service.send(PayloadServiceCommand::BuildNewPayload(attr, tx));
         rx
     }
 
+    /// Returns the best payload for the given identifier.
     /// Note: this does not resolve the job if it's still in progress.
-    async fn best_payload(
+    pub async fn best_payload(
         &self,
         id: PayloadId,
-    ) -> Option<Result<<Self::PayloadType as PayloadTypes>::BuiltPayload, Self::Error>> {
+    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
         let (tx, rx) = oneshot::channel();
         self.to_service.send(PayloadServiceCommand::BestPayload(id, tx)).ok()?;
         rx.await.ok()?
     }
 
-    async fn resolve_kind(
+    /// Resolves the payload job and returns the best payload that has been built so far.
+    pub async fn resolve_kind(
         &self,
         id: PayloadId,
         kind: PayloadKind,
@@ -159,7 +158,9 @@ where
         }
     }
 
-    async fn subscribe(&self) -> Result<PayloadEvents<Self::PayloadType>, Self::Error> {
+    /// Sends a message to the service to subscribe to payload events.
+    /// Returns a receiver that will receive them.
+    pub async fn subscribe(&self) -> Result<PayloadEvents<T>, PayloadBuilderError> {
         let (tx, rx) = oneshot::channel();
         let _ = self.to_service.send(PayloadServiceCommand::Subscribe(tx));
         Ok(PayloadEvents { receiver: rx.await? })
@@ -168,7 +169,7 @@ where
     /// Returns the payload attributes associated with the given identifier.
     ///
     /// Note: this returns the attributes of the payload and does not resolve the job.
-    async fn payload_attributes(
+    pub async fn payload_attributes(
         &self,
         id: PayloadId,
     ) -> Option<Result<T::PayloadBuilderAttributes, PayloadBuilderError>> {
@@ -178,19 +179,6 @@ where
     }
 }
 
-impl<T> PayloadBuilderHandle<T>
-where
-    T: PayloadTypes,
-{
-    /// Creates a new payload builder handle for the given channel.
-    ///
-    /// Note: this is only used internally by the [`PayloadBuilderService`] to manage the payload
-    /// building flow See [`PayloadBuilderService::poll`] for implementation details.
-    pub const fn new(to_service: mpsc::UnboundedSender<PayloadServiceCommand<T>>) -> Self {
-        Self { to_service }
-    }
-}
-
 impl<T> Clone for PayloadBuilderHandle<T>
 where
     T: PayloadTypes,
diff --git a/examples/custom-engine-types/src/main.rs b/examples/custom-engine-types/src/main.rs
index 7cb1ec4d0..ca89b82d4 100644
--- a/examples/custom-engine-types/src/main.rs
+++ b/examples/custom-engine-types/src/main.rs
@@ -34,12 +34,11 @@ use reth::{
         node::{NodeTypes, NodeTypesWithEngine},
         rpc::{EngineValidatorBuilder, RpcAddOns},
         BuilderContext, FullNodeTypes, Node, NodeAdapter, NodeBuilder, NodeComponentsBuilder,
-        PayloadBuilderConfig,
     },
     network::NetworkHandle,
     payload::ExecutionPayloadValidator,
     primitives::{Block, EthPrimitives, SealedBlock, TransactionSigned},
-    providers::{CanonStateSubscriptions, EthStorage, StateProviderFactory},
+    providers::{EthStorage, StateProviderFactory},
     rpc::{
         eth::EthApi,
         types::engine::{ExecutionPayload, PayloadError},
@@ -48,10 +47,7 @@ use reth::{
     transaction_pool::{PoolTransaction, TransactionPool},
     version::default_extra_data_bytes,
 };
-use reth_basic_payload_builder::{
-    BasicPayloadJobGenerator, BasicPayloadJobGeneratorConfig, BuildArguments, BuildOutcome,
-    PayloadBuilder, PayloadConfig,
-};
+use reth_basic_payload_builder::{BuildArguments, BuildOutcome, PayloadBuilder, PayloadConfig};
 use reth_chainspec::{Chain, ChainSpec, ChainSpecProvider};
 use reth_engine_local::payload::UnsupportedLocalAttributes;
 use reth_ethereum_payload_builder::EthereumBuilderConfig;
@@ -68,10 +64,7 @@ use reth_node_ethereum::{
     },
     EthEvmConfig,
 };
-use reth_payload_builder::{
-    EthBuiltPayload, EthPayloadBuilderAttributes, PayloadBuilderError, PayloadBuilderHandle,
-    PayloadBuilderService,
-};
+use reth_payload_builder::{EthBuiltPayload, EthPayloadBuilderAttributes, PayloadBuilderError};
 use reth_tracing::{RethTracer, Tracer};
 use reth_trie_db::MerklePatriciaTrie;
 use serde::{Deserialize, Serialize};
@@ -367,11 +360,13 @@ where
         + Unpin
         + 'static,
 {
-    async fn spawn_payload_service(
-        self,
+    type PayloadBuilder = CustomPayloadBuilder<Pool, Node::Provider>;
+
+    async fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>> {
+    ) -> eyre::Result<Self::PayloadBuilder> {
         let payload_builder = CustomPayloadBuilder {
             inner: reth_ethereum_payload_builder::EthereumPayloadBuilder::new(
                 ctx.provider().clone(),
@@ -380,24 +375,6 @@ where
                 EthereumBuilderConfig::new(default_extra_data_bytes()),
             ),
         };
-        let conf = ctx.payload_builder_config();
-
-        let payload_job_config = BasicPayloadJobGeneratorConfig::default()
-            .interval(conf.interval())
-            .deadline(conf.deadline())
-            .max_payload_tasks(conf.max_payload_tasks());
-
-        let payload_generator = BasicPayloadJobGenerator::with_builder(
-            ctx.provider().clone(),
-            ctx.task_executor().clone(),
-            payload_job_config,
-            payload_builder,
-        );
-        let (payload_service, payload_builder) =
-            PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
-
-        ctx.task_executor().spawn_critical("payload builder service", Box::pin(payload_service));
-
         Ok(payload_builder)
     }
 }
diff --git a/examples/custom-evm/Cargo.toml b/examples/custom-evm/Cargo.toml
index c9ce0c467..b236647f1 100644
--- a/examples/custom-evm/Cargo.toml
+++ b/examples/custom-evm/Cargo.toml
@@ -12,6 +12,7 @@ reth-evm-ethereum.workspace = true
 reth-node-api.workspace = true
 reth-node-core.workspace = true
 reth-primitives.workspace = true
+reth-ethereum-payload-builder.workspace = true
 reth-node-ethereum = { workspace = true, features = ["test-utils"] }
 reth-tracing.workspace = true
 reth-evm.workspace = true
diff --git a/examples/custom-evm/src/main.rs b/examples/custom-evm/src/main.rs
index f84708210..06c92960b 100644
--- a/examples/custom-evm/src/main.rs
+++ b/examples/custom-evm/src/main.rs
@@ -204,12 +204,15 @@ where
         PayloadBuilderAttributes = EthPayloadBuilderAttributes,
     >,
 {
-    async fn spawn_payload_service(
-        self,
+    type PayloadBuilder =
+        reth_ethereum_payload_builder::EthereumPayloadBuilder<Pool, Node::Provider, MyEvmConfig>;
+
+    async fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<reth::payload::PayloadBuilderHandle<Types::Engine>> {
-        self.inner.spawn(MyEvmConfig::new(ctx.chain_spec()), ctx, pool)
+    ) -> eyre::Result<Self::PayloadBuilder> {
+        self.inner.build(MyEvmConfig::new(ctx.chain_spec()), ctx, pool)
     }
 }
 #[tokio::main]
diff --git a/examples/custom-payload-builder/src/main.rs b/examples/custom-payload-builder/src/main.rs
index 1ea7f340c..f22611265 100644
--- a/examples/custom-payload-builder/src/main.rs
+++ b/examples/custom-payload-builder/src/main.rs
@@ -11,20 +11,15 @@
 
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 
-use generator::EmptyBlockPayloadJobGenerator;
 use reth::{
     builder::{components::PayloadServiceBuilder, node::FullNodeTypes, BuilderContext},
     cli::{config::PayloadBuilderConfig, Cli},
-    payload::PayloadBuilderHandle,
-    providers::CanonStateSubscriptions,
     transaction_pool::{PoolTransaction, TransactionPool},
 };
-use reth_basic_payload_builder::BasicPayloadJobGeneratorConfig;
 use reth_chainspec::ChainSpec;
-use reth_ethereum_payload_builder::EthereumBuilderConfig;
+use reth_ethereum_payload_builder::{EthereumBuilderConfig, EthereumPayloadBuilder};
 use reth_node_api::NodeTypesWithEngine;
 use reth_node_ethereum::{node::EthereumAddOns, EthEngineTypes, EthEvmConfig, EthereumNode};
-use reth_payload_builder::PayloadBuilderService;
 use reth_primitives::{EthPrimitives, TransactionSigned};
 
 pub mod generator;
@@ -47,38 +42,22 @@ where
         + Unpin
         + 'static,
 {
-    async fn spawn_payload_service(
-        self,
+    type PayloadBuilder = EthereumPayloadBuilder<Pool, Node::Provider, EthEvmConfig>;
+
+    async fn build_payload_builder(
+        &self,
         ctx: &BuilderContext<Node>,
         pool: Pool,
-    ) -> eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypesWithEngine>::Engine>> {
+    ) -> eyre::Result<Self::PayloadBuilder> {
         tracing::info!("Spawning a custom payload builder");
         let conf = ctx.payload_builder_config();
 
-        let payload_job_config = BasicPayloadJobGeneratorConfig::default()
-            .interval(conf.interval())
-            .deadline(conf.deadline())
-            .max_payload_tasks(conf.max_payload_tasks());
-
-        let payload_generator = EmptyBlockPayloadJobGenerator::with_builder(
+        Ok(reth_ethereum_payload_builder::EthereumPayloadBuilder::new(
             ctx.provider().clone(),
-            ctx.task_executor().clone(),
-            payload_job_config,
-            reth_ethereum_payload_builder::EthereumPayloadBuilder::new(
-                ctx.provider().clone(),
-                pool,
-                EthEvmConfig::new(ctx.chain_spec()),
-                EthereumBuilderConfig::new(conf.extra_data_bytes()),
-            ),
-        );
-
-        let (payload_service, payload_builder) =
-            PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
-
-        ctx.task_executor()
-            .spawn_critical("custom payload builder service", Box::pin(payload_service));
-
-        Ok(payload_builder)
+            pool,
+            EthEvmConfig::new(ctx.chain_spec()),
+            EthereumBuilderConfig::new(conf.extra_data_bytes()),
+        ))
     }
 }