diff --git a/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerClient.groovy b/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerClient.groovy
index c5b57be256..0b56c19393 100644
--- a/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerClient.groovy
+++ b/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerClient.groovy
@@ -363,10 +363,12 @@ class TowerClient implements TraceObserver {
     }
 
     String getAccessToken() {
-        // access token
-        def token = session.config.navigate('tower.accessToken')
-        if( !token )
-            token = env.get('TOWER_ACCESS_TOKEN')
+        // when 'TOWER_WORKFLOW_ID' is provided in the env, it's a tower made launch
+        // therefore the access token should only be taken from the env
+        // otherwise check into the config file and fallback in the env
+        def token = env.get('TOWER_WORKFLOW_ID')
+                ? env.get('TOWER_ACCESS_TOKEN')
+                : session.config.navigate('tower.accessToken', env.get('TOWER_ACCESS_TOKEN'))
         if( !token )
             throw new AbortOperationException("Missing Nextflow Tower access token -- Make sure there's a variable TOWER_ACCESS_TOKEN in your environment")
         return token
diff --git a/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerFactory.groovy b/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerFactory.groovy
index 24919f31b3..a4124ad218 100644
--- a/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerFactory.groovy
+++ b/plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerFactory.groovy
@@ -38,7 +38,7 @@ class TowerFactory implements TraceObserverFactory {
     @Override
     Collection<TraceObserver> create(Session session) {
         final config = session.config
-        Boolean isEnabled = config.navigate('tower.enabled') as Boolean
+        Boolean isEnabled = config.navigate('tower.enabled') as Boolean || env.get('TOWER_WORKFLOW_ID')
         String endpoint = config.navigate('tower.endpoint') as String
         Duration requestInterval = config.navigate('tower.requestInterval') as Duration
         Duration aliveInterval = config.navigate('tower.aliveInterval') as Duration
@@ -58,7 +58,12 @@ class TowerFactory implements TraceObserverFactory {
         tower.maxRetries = config.navigate('tower.maxRetries', 5) as int
         tower.backOffBase = config.navigate('tower.backOffBase', SimpleHttpClient.DEFAULT_BACK_OFF_BASE) as int
         tower.backOffDelay = config.navigate('tower.backOffDelay', SimpleHttpClient.DEFAULT_BACK_OFF_DELAY  ) as int
-        tower.workspaceId = config.navigate('tower.workspaceId', env.get('TOWER_WORKSPACE_ID'))
+        // when 'TOWER_WORKFLOW_ID' is provided in the env, it's a tower made launch
+        // therefore the workspace should only be taken from the env
+        // otherwise check into the config file and fallback in the env
+        tower.workspaceId = env.get('TOWER_WORKFLOW_ID')
+                ? env.get('TOWER_WORKSPACE_ID')
+                : config.navigate('tower.workspaceId', env.get('TOWER_WORKSPACE_ID'))
         final result = new ArrayList(1)
         result.add(tower)
         // register auth provider
diff --git a/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerClientTest.groovy b/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerClientTest.groovy
index 9d85aa3a49..1fec077b27 100644
--- a/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerClientTest.groovy
+++ b/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerClientTest.groovy
@@ -146,19 +146,28 @@ class TowerClientTest extends Specification {
 
     def 'should get access token' () {
         given:
-        def ENV = [TOWER_ACCESS_TOKEN: 'xyz']
         def session = Mock(Session)
 
         when:
-        def observer = new TowerClient(session: session)
+        def observer = new TowerClient(session: session, env: [TOWER_ACCESS_TOKEN: 'xyz'])
         def result = observer.getAccessToken()
         then:
         session.getConfig() >> [tower:[accessToken: 'abc'] ]
         and:
+        // the token in the config overrides the one in the env
         result == 'abc'
 
         when:
-        observer = new TowerClient(session: session, env: ENV)
+        observer = new TowerClient(session: session, env: [TOWER_ACCESS_TOKEN: 'xyz', TOWER_WORKFLOW_ID: '111222333'])
+        result = observer.getAccessToken()
+        then:
+        session.getConfig() >> [tower:[accessToken: 'abc'] ]
+        and:
+        // the token from the env is taken because is a tower launch aka TOWER_WORKFLOW_ID is set
+        result == 'xyz'
+
+        when:
+        observer = new TowerClient(session: session, env: [TOWER_ACCESS_TOKEN: 'xyz'])
         result = observer.getAccessToken()
         then:
         session.getConfig() >> [:]
@@ -172,7 +181,6 @@ class TowerClientTest extends Specification {
         session.getConfig() >> [:]
         then:
         thrown(AbortOperationException)
-
     }
 
     def 'should post task records' () {
diff --git a/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerFactoryTest.groovy b/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerFactoryTest.groovy
index 5a95840420..8faed8cacd 100644
--- a/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerFactoryTest.groovy
+++ b/plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerFactoryTest.groovy
@@ -61,22 +61,65 @@ class TowerFactoryTest extends Specification {
     def 'should create with with workspace id'() {
         given:
         def session = Mock(Session)
-        def factory = new TowerFactory(env: [TOWER_WORKSPACE_ID: '100'])
 
+        //
+        // the workspace id is taken from the env
+        //
         when:
+        def factory = new TowerFactory(env: [TOWER_WORKSPACE_ID: '100'])
         def client = (TowerClient) factory.create(session)[0]
         then:
         session.getConfig() >> [tower: [enabled: true, accessToken: 'xyz']]
         and:
         client.getWorkspaceId() == '100'
 
+        //
+        // the workspace id is taken from the config
+        //
+        when:
+        factory = new TowerFactory(env: [:])
+        client = (TowerClient) factory.create(session)[0]
+        then:
+        session.getConfig() >> [tower: [enabled: true, workspaceId: '200', accessToken: 'xyz']]
+        and:
+        client.getWorkspaceId() == '200'
 
+        //
+        // the workspace id is set both in the config and the env
+        // the config has the priority
+        //
         when:
+        factory = new TowerFactory(env: [TOWER_WORKSPACE_ID: '100'])
         client = (TowerClient) factory.create(session)[0]
         then:
         session.getConfig() >> [tower: [enabled: true, workspaceId: '200', accessToken: 'xyz']]
         and:
         client.getWorkspaceId() == '200'
+
+        //
+        // when TOWER_WORKFLOW_ID is set is a tower launch
+        // then the workspace id is only taken from the env
+        //
+        when:
+        factory = new TowerFactory(env: [TOWER_WORKSPACE_ID: '100', TOWER_WORKFLOW_ID: '111222333', TOWER_ACCESS_TOKEN: 'xyz'])
+        client = (TowerClient) factory.create(session)[0]
+        then:
+        session.getConfig() >> [tower: [enabled: true, workspaceId: '200', accessToken: 'xyz']]
+        and:
+        client.getWorkspaceId() == '100'
+
+
+        //
+        // when enabled is false but `TOWER_WORKFLOW_ID` is provided
+        // then the client should be created
+        //
+        when:
+        factory = new TowerFactory(env: [TOWER_WORKSPACE_ID: '100', TOWER_WORKFLOW_ID: '111222333', TOWER_ACCESS_TOKEN: 'xyz'])
+        client = (TowerClient) factory.create(session)[0]
+        then:
+        session.getConfig() >> [tower: [enabled: false]]
+        and:
+        client.getWorkspaceId() == '100'
     }
 
     @Unroll
@@ -103,7 +146,6 @@ class TowerFactoryTest extends Specification {
         'https://tower.nf/this/that'    | true          | 'Bearer xyz123'
         'HTTPS://TOWER.NF/THIS/THAT'    | true          | 'Bearer xyz123'
 
-
     }
 
 }