Merge pull request #60 from datagouv/feat/specific-user-agent-for-mcp

feat: mcp sends a specific user agent to datagouv services

Author: bolinocroustibat

helpers/crawler_api_client.py

@@ -5,6 +5,7 @@
 import httpx
 
 from helpers import env_config
+from helpers.user_agent import USER_AGENT
 
 logger = logging.getLogger("datagouv_mcp")
 
@@ -19,7 +20,7 @@ async def _get_session(
 ) -> tuple[httpx.AsyncClient, bool]:
     if session is not None:
         return session, False
-    new_session = httpx.AsyncClient()
+    new_session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     return new_session, True
 
 

helpers/datagouv_api_client.py

@@ -6,6 +6,7 @@
 import yaml
 
 from helpers import env_config
+from helpers.user_agent import USER_AGENT
 
 logger = logging.getLogger("datagouv_mcp")
 
@@ -29,7 +30,7 @@ async def get_resource_details(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         base_url: str = env_config.get_base_url("datagouv_api")
@@ -45,7 +46,7 @@ async def get_resource_metadata(
 ) -> dict[str, Any]:
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         data = await get_resource_details(resource_id, session=session)
@@ -69,7 +70,7 @@ async def get_dataset_details(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         base_url: str = env_config.get_base_url("datagouv_api")
@@ -85,7 +86,7 @@ async def get_dataset_metadata(
 ) -> dict[str, Any]:
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         data = await get_dataset_details(dataset_id, session=session)
@@ -105,7 +106,7 @@ async def get_resource_and_dataset_metadata(
 ) -> dict[str, Any]:
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     try:
         res: dict[str, Any] = await get_resource_metadata(resource_id, session=session)
         ds: dict[str, Any] = {}
@@ -129,7 +130,7 @@ async def get_resources_for_dataset(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     try:
         ds = await get_dataset_metadata(dataset_id, session=session)
         base_url: str = env_config.get_base_url("datagouv_api")
@@ -164,7 +165,7 @@ async def fetch_openapi_spec(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         logger.debug("Fetching OpenAPI spec from %s", url)
@@ -196,7 +197,7 @@ async def get_dataservice_details(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         base_url: str = env_config.get_base_url("datagouv_api")
@@ -226,7 +227,7 @@ async def search_dataservices(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         base_url: str = env_config.get_base_url("datagouv_api")
@@ -295,7 +296,7 @@ async def search_datasets(
     """
     own = session is None
     if own:
-        session = httpx.AsyncClient()
+        session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     assert session is not None
     try:
         base_url: str = env_config.get_base_url("datagouv_api")

helpers/metrics_api_client.py

@@ -4,6 +4,7 @@
 import httpx
 
 from helpers import env_config
+from helpers.user_agent import USER_AGENT
 
 logger = logging.getLogger("datagouv_mcp")
 
@@ -13,7 +14,7 @@ async def _get_session(
 ) -> tuple[httpx.AsyncClient, bool]:
     if session is not None:
         return session, False
-    new_session = httpx.AsyncClient()
+    new_session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     return new_session, True
 
 

helpers/tabular_api_client.py

@@ -4,6 +4,7 @@
 import httpx
 
 from helpers import env_config
+from helpers.user_agent import USER_AGENT
 
 logger = logging.getLogger("datagouv_mcp")
 
@@ -17,7 +18,7 @@ async def _get_session(
 ) -> tuple[httpx.AsyncClient, bool]:
     if session is not None:
         return session, False
-    new_session = httpx.AsyncClient()
+    new_session = httpx.AsyncClient(headers={"User-Agent": USER_AGENT})
     return new_session, True
 
 

helpers/user_agent.py

@@ -0,0 +1,5 @@
+"""User-Agent sent to data.gouv.fr services for identification and support."""
+
+from importlib.metadata import version
+
+USER_AGENT = f"datagouv-mcp/{version('datagouv-mcp')}"

tests/test_datagouv_api.py

@@ -1,10 +1,12 @@
 """Tests for the datagouv_api_client helper."""
 
 import os
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 from helpers import datagouv_api_client
+from helpers.user_agent import USER_AGENT
 
 
 @pytest.fixture
@@ -34,6 +36,28 @@ async def test_get_dataset_metadata(self, known_dataset_id):
         assert "title" in metadata
         assert metadata["title"] is not None
 
+    async def test_get_dataset_metadata_sends_user_agent(self, known_dataset_id):
+        """Test that get_dataset_metadata creates a client with User-Agent header."""
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "id": known_dataset_id,
+            "title": "Test Dataset",
+        }
+        mock_response.raise_for_status = MagicMock()
+        mock_client.get = AsyncMock(return_value=mock_response)
+        mock_client.aclose = AsyncMock(return_value=None)
+
+        with patch(
+            "helpers.datagouv_api_client.httpx.AsyncClient",
+            return_value=mock_client,
+        ) as mock_async_client:
+            await datagouv_api_client.get_dataset_metadata(
+                known_dataset_id, session=None
+            )
+
+        mock_async_client.assert_called_once_with(headers={"User-Agent": USER_AGENT})
+
     async def test_get_resource_metadata(self, known_resource_id):
         """Test fetching resource metadata."""
         metadata = await datagouv_api_client.get_resource_metadata(known_resource_id)