diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1cb7448..c10cbc7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.12.2] - 2024-03-26
+
+### Fixed
+
+- Use file for batch url computation
+
 ## [1.12.1] - 2024-03-22
 
 ### Fixed
diff --git a/pyproject.toml b/pyproject.toml
index 332381e..a473d93 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "sekoia-automation-sdk"
 
-version = "1.12.1"
+version = "1.12.2"
 description = "SDK to create Sekoia.io playbook modules"
 license = "MIT"
 readme = "README.md"
diff --git a/sekoia_automation/aio/connector.py b/sekoia_automation/aio/connector.py
index c7dba8d..8be9cb7 100644
--- a/sekoia_automation/aio/connector.py
+++ b/sekoia_automation/aio/connector.py
@@ -1,6 +1,5 @@
 """Contains connector with async version."""
 
-import os
 from abc import ABC
 from asyncio import AbstractEventLoop, get_event_loop
 from collections.abc import AsyncGenerator
@@ -154,8 +153,10 @@ async def push_data_to_intakes(
             list[str]:
         """
         self._last_events_time = datetime.utcnow()
-        intake_host = os.getenv("INTAKE_URL", self.configuration.intake_server)
-        batch_api = urljoin(intake_host, "batch")
+        if intake_server := self.configuration.intake_server:
+            batch_api = urljoin(intake_server, "batch")
+        else:
+            batch_api = urljoin(self.intake_url, "batch")
 
         result_ids = []
 
diff --git a/sekoia_automation/connector/__init__.py b/sekoia_automation/connector/__init__.py
index 69f4267..9a6a882 100644
--- a/sekoia_automation/connector/__init__.py
+++ b/sekoia_automation/connector/__init__.py
@@ -1,4 +1,3 @@
-import os
 import uuid
 from abc import ABC
 from collections.abc import Generator, Sequence
@@ -36,7 +35,7 @@
 
 
 class DefaultConnectorConfiguration(BaseModel):
-    intake_server: str = "https://intake.sekoia.io"
+    intake_server: str | None = None
     intake_key: str
 
 
@@ -171,8 +170,10 @@ def push_events_to_intakes(
         # Reset the consecutive error count
         self._error_count = 0
         self._last_events_time = datetime.utcnow()
-        intake_host = os.getenv("INTAKE_URL", self.configuration.intake_server)
-        batch_api = urljoin(intake_host, "batch")
+        if intake_server := self.configuration.intake_server:
+            batch_api = urljoin(intake_server, "batch")
+        else:
+            batch_api = urljoin(self.intake_url, "batch")
 
         # Dict to collect event_ids for the API
         collect_ids: dict[int, list] = {}
diff --git a/sekoia_automation/module.py b/sekoia_automation/module.py
index 3c3eeda..6ff4f6b 100644
--- a/sekoia_automation/module.py
+++ b/sekoia_automation/module.py
@@ -322,6 +322,7 @@ class ModuleItem(ABC):
     CALLBACK_URL_FILE_NAME = "url_callback"
     SECRETS_URL_FILE_NAME = "url_secrets"
     LOGS_URL_FILE_NAME = "url_logs"
+    INTAKE_URL_FILE_NAME = "intake_url"
 
     name: str | None = None
     description: str | None = None
@@ -419,6 +420,10 @@ def secrets_url(self) -> str:
         except FileNotFoundError:
             return self.callback_url.replace("/callback", "/secrets")
 
+    @cached_property
+    def intake_url(self) -> str:
+        return self.module.load_config(self.INTAKE_URL_FILE_NAME)
+
     @property
     def _headers(self) -> dict[str, str]:
         return {"Authorization": f"Bearer {self.token}"}
diff --git a/tests/aio/test_connector.py b/tests/aio/test_connector.py
index 14c9e2d..967c313 100644
--- a/tests/aio/test_connector.py
+++ b/tests/aio/test_connector.py
@@ -1,6 +1,7 @@
 """Test async connector."""
 
-from unittest.mock import Mock, patch
+import json
+from unittest.mock import Mock, mock_open, patch
 from urllib.parse import urljoin
 
 import pytest
diff --git a/tests/connectors/test_connector.py b/tests/connectors/test_connector.py
index 5ed868e..684af5f 100644
--- a/tests/connectors/test_connector.py
+++ b/tests/connectors/test_connector.py
@@ -1,4 +1,3 @@
-import os
 from unittest.mock import Mock, PropertyMock, patch
 
 import pytest
@@ -14,6 +13,12 @@
 EVENTS = ["foo", "bar"]
 
 
+@pytest.fixture(autouse=True)
+def configure_intake_url(config_storage):
+    with (config_storage / "intake_url").open("w") as f:
+        f.write("https://intake.sekoia.io")
+
+
 class DummyConnector(Connector):
     def run(self):
         raise NotImplementedError
@@ -136,25 +141,42 @@ def test_push_event_to_intake_with_chunks(test_connector, mocked_trigger_logs):
     assert result == ["001", "002", "003", "004"]
 
 
-def test_push_event_to_intake_custom_url(test_connector, mocked_trigger_logs):
+def test_push_event_to_intake_custom_url(
+    test_connector, mocked_trigger_logs, config_storage
+):
+    assert test_connector.configuration.intake_server is None
+
     url = "https://fra2.app.sekoia.io/v1/intake-http/batch"
     batch_mock = mocked_trigger_logs.post(
         url, json={"event_ids": ["001"]}, additional_matcher=match_events("foo")
     )
     # With trailing slash
-    with patch.dict(
-        os.environ, {"INTAKE_URL": "https://fra2.app.sekoia.io/v1/intake-http/"}
-    ):
-        test_connector.push_events_to_intakes(["foo"])
-        assert batch_mock.call_count == 1
+    with (config_storage / "intake_url").open("w") as f:
+        f.write("https://fra2.app.sekoia.io/v1/intake-http/")
+    test_connector.push_events_to_intakes(["foo"])
+    assert batch_mock.call_count == 1
 
     # Without trailing slash
+    with (config_storage / "intake_url").open("w") as f:
+        f.write("https://fra2.app.sekoia.io/v1/intake-http")
     mocked_trigger_logs.reset_mock()
-    with patch.dict(
-        os.environ, {"INTAKE_URL": "https://fra2.app.sekoia.io/v1/intake-http"}
-    ):
-        test_connector.push_events_to_intakes(["foo"])
-        assert batch_mock.call_count == 1
+    test_connector.push_events_to_intakes(["foo"])
+    assert batch_mock.call_count == 1
+
+
+def test_push_event_to_intake_custom_url_configuration(
+    test_connector, mocked_trigger_logs
+):
+    url = "https://fra2.app.sekoia.io/v1/intake-http/batch"
+    batch_mock = mocked_trigger_logs.post(
+        url, json={"event_ids": ["001"]}, additional_matcher=match_events("foo")
+    )
+
+    test_connector.configuration.intake_server = (
+        "https://fra2.app.sekoia.io/v1/intake-http"
+    )
+    test_connector.push_events_to_intakes(["foo"])
+    assert batch_mock.call_count == 1
 
 
 def test_push_event_to_intake_with_chunks_executor_stopped(
diff --git a/tests/expectations/sample_module/connector_sample_connector.json b/tests/expectations/sample_module/connector_sample_connector.json
index a15b2e5..822d4c9 100644
--- a/tests/expectations/sample_module/connector_sample_connector.json
+++ b/tests/expectations/sample_module/connector_sample_connector.json
@@ -9,7 +9,6 @@
     "properties": {
       "intake_server": {
         "title": "Intake Server",
-        "default": "https://intake.sekoia.io",
         "type": "string"
       },
       "intake_key": {
@@ -22,4 +21,4 @@
     ]
   },
   "results": {}
-}
\ No newline at end of file
+}
diff --git a/tests/test_trigger.py b/tests/test_trigger.py
index 00c1319..9c47576 100644
--- a/tests/test_trigger.py
+++ b/tests/test_trigger.py
@@ -61,6 +61,13 @@ def test_secrets_url():
     mock.assert_called_with(trigger.SECRETS_URL_FILE_NAME)
 
 
+def test_intake_url():
+    trigger = DummyTrigger()
+    with patch.object(Module, "load_config", return_value="intake") as mock:
+        assert trigger.intake_url == "intake"
+    mock.assert_called_with(trigger.INTAKE_URL_FILE_NAME)
+
+
 def test_logs_url():
     trigger = DummyTrigger()
     with patch.object(Module, "load_config", return_value="logs") as mock: