feat: use httpx async in project resolver modules

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: DevonFulcher

src/dbt_mcp/oauth/fastapi_app.py

@@ -132,22 +132,22 @@ def shutdown_server() -> dict[str, bool]:
         return {"ok": True}
 
     @app.get("/projects")
-    def projects() -> list[DbtPlatformProject]:
+    async def projects() -> list[DbtPlatformProject]:
         if app.state.decoded_access_token is None:
             raise RuntimeError("Access token missing; OAuth flow not completed")
         access_token = app.state.decoded_access_token.access_token_response.access_token
         headers = {
             "Accept": "application/json",
             "Authorization": f"Bearer {access_token}",
         }
-        accounts = get_all_accounts(
+        accounts = await get_all_accounts(
             dbt_platform_url=dbt_platform_url,
             headers=headers,
         )
         projects: list[DbtPlatformProject] = []
         for account in [a for a in accounts if a.state == 1 and not a.locked]:
             projects.extend(
-                get_all_projects_for_account(
+                await get_all_projects_for_account(
                     dbt_platform_url=dbt_platform_url,
                     account=account,
                     headers=headers,
@@ -161,7 +161,7 @@ def get_dbt_platform_context() -> DbtPlatformContext:
         return dbt_platform_context_manager.read_context() or DbtPlatformContext()
 
     @app.post("/environments")
-    def get_deployment_environments(
+    async def get_deployment_environments(
         request: GetEnvironmentsRequest,
     ) -> list[DbtPlatformEnvironmentResponse]:
         """Get all deployment environments for a project, excluding development environments."""
@@ -173,7 +173,7 @@ def get_deployment_environments(
             "Accept": "application/json",
             "Authorization": f"Bearer {access_token}",
         }
-        environments = _get_all_environments_for_project(
+        environments = await _get_all_environments_for_project(
             dbt_platform_url=dbt_platform_url,
             account_id=request.account_id,
             project_id=request.project_id,
@@ -188,7 +188,7 @@ def get_deployment_environments(
         ]
 
     @app.post("/selected_project")
-    def set_selected_project(
+    async def set_selected_project(
         selected_project_request: SelectedProjectRequest,
     ) -> DbtPlatformContext:
         logger.info("Selected project received")
@@ -199,7 +199,7 @@ def set_selected_project(
             "Accept": "application/json",
             "Authorization": f"Bearer {access_token}",
         }
-        accounts = get_all_accounts(
+        accounts = await get_all_accounts(
             dbt_platform_url=dbt_platform_url,
             headers=headers,
         )
@@ -208,7 +208,7 @@ def set_selected_project(
         )
         if account is None:
             raise ValueError(f"Account {selected_project_request.account_id} not found")
-        environments = _get_all_environments_for_project(
+        environments = await _get_all_environments_for_project(
             dbt_platform_url=dbt_platform_url,
             account_id=selected_project_request.account_id,
             project_id=selected_project_request.project_id,

src/dbt_mcp/project/environment_resolver.py

@@ -1,6 +1,6 @@
 import logging
 
-import requests
+import httpx
 
 from dbt_mcp.oauth.dbt_platform import (
     DbtPlatformEnvironment,
@@ -10,7 +10,7 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_all_environments_for_project(
+async def _get_all_environments_for_project(
     *,
     dbt_platform_url: str,
     account_id: int,
@@ -21,19 +21,20 @@ def _get_all_environments_for_project(
     """Fetch all environments for a project using offset/page_size pagination."""
     offset = 0
     environments: list[DbtPlatformEnvironmentResponse] = []
-    while True:
-        environments_response = requests.get(
-            f"{dbt_platform_url}/api/v3/accounts/{account_id}/projects/{project_id}/environments/?state=1&offset={offset}&limit={page_size}",
-            headers=headers,
-        )
-        environments_response.raise_for_status()
-        page = environments_response.json()["data"]
-        environments.extend(
-            DbtPlatformEnvironmentResponse(**environment) for environment in page
-        )
-        if len(page) < page_size:
-            break
-        offset += page_size
+    async with httpx.AsyncClient() as client:
+        while True:
+            response = await client.get(
+                f"{dbt_platform_url}/api/v3/accounts/{account_id}/projects/{project_id}/environments/?state=1&offset={offset}&limit={page_size}",
+                headers=headers,
+            )
+            response.raise_for_status()
+            page = response.json()["data"]
+            environments.extend(
+                DbtPlatformEnvironmentResponse(**environment) for environment in page
+            )
+            if len(page) < page_size:
+                break
+            offset += page_size
     return environments
 
 
@@ -90,7 +91,7 @@ def resolve_environments(
     return prod_environment, dev_environment
 
 
-def get_environments_for_project(
+async def get_environments_for_project(
     *,
     dbt_platform_url: str,
     account_id: int,
@@ -102,7 +103,7 @@ def get_environments_for_project(
 
     Returns a tuple of (prod_environment, dev_environment).
     """
-    environments = _get_all_environments_for_project(
+    environments = await _get_all_environments_for_project(
         dbt_platform_url=dbt_platform_url,
         account_id=account_id,
         project_id=project_id,

src/dbt_mcp/project/project_resolver.py

@@ -1,6 +1,6 @@
 import logging
 
-import requests
+import httpx
 
 from dbt_mcp.oauth.dbt_platform import (
     DbtPlatformAccount,
@@ -10,22 +10,22 @@
 logger = logging.getLogger(__name__)
 
 
-def get_all_accounts(
+async def get_all_accounts(
     *,
     dbt_platform_url: str,
     headers: dict[str, str],
 ) -> list[DbtPlatformAccount]:
-    accounts_response = requests.get(
-        url=f"{dbt_platform_url}/api/v3/accounts/",
-        headers=headers,
-    )
-    accounts_response.raise_for_status()
-    return [
-        DbtPlatformAccount(**account) for account in accounts_response.json()["data"]
-    ]
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            url=f"{dbt_platform_url}/api/v3/accounts/",
+            headers=headers,
+        )
+        response.raise_for_status()
+        data = response.json()
+    return [DbtPlatformAccount(**account) for account in data["data"]]
 
 
-def get_all_projects_for_account(
+async def get_all_projects_for_account(
     *,
     dbt_platform_url: str,
     account: DbtPlatformAccount,
@@ -35,17 +35,19 @@ def get_all_projects_for_account(
     """Fetch all projects for an account using offset/page_size pagination."""
     offset = 0
     projects: list[DbtPlatformProject] = []
-    while True:
-        projects_response = requests.get(
-            f"{dbt_platform_url}/api/v3/accounts/{account.id}/projects/?state=1&offset={offset}&limit={page_size}",
-            headers=headers,
-        )
-        projects_response.raise_for_status()
-        page = projects_response.json()["data"]
-        projects.extend(
-            DbtPlatformProject(**project, account_name=account.name) for project in page
-        )
-        if len(page) < page_size:
-            break
-        offset += page_size
+    async with httpx.AsyncClient() as client:
+        while True:
+            response = await client.get(
+                f"{dbt_platform_url}/api/v3/accounts/{account.id}/projects/?state=1&offset={offset}&limit={page_size}",
+                headers=headers,
+            )
+            response.raise_for_status()
+            page = response.json()["data"]
+            projects.extend(
+                DbtPlatformProject(**project, account_name=account.name)
+                for project in page
+            )
+            if len(page) < page_size:
+                break
+            offset += page_size
     return projects

tests/unit/oauth/test_fastapi_app_pagination.py

@@ -1,4 +1,4 @@
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -24,35 +24,49 @@ def account():
     )
 
 
-@patch("dbt_mcp.project.project_resolver.requests.get")
-def test_get_all_projects_for_account_paginates(mock_get: Mock, base_headers, account):
+def create_mock_response(data: dict) -> MagicMock:
+    resp = MagicMock()
+    resp.json.return_value = data
+    resp.raise_for_status.return_value = None
+    return resp
+
+
+def create_mock_httpx_client(responses: list) -> AsyncMock:
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(side_effect=responses)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=None)
+    return mock_client
+
+
+async def test_get_all_projects_for_account_paginates(base_headers, account):
     # Two pages: first full page (limit=2), second partial page (1 item) -> stop
-    first_page_resp = Mock()
-    first_page_resp.json.return_value = {
-        "data": [
-            {"id": 101, "name": "Proj A", "account_id": account.id},
-            {"id": 102, "name": "Proj B", "account_id": account.id},
-        ]
-    }
-    first_page_resp.raise_for_status.return_value = None
-
-    second_page_resp = Mock()
-    second_page_resp.json.return_value = {
-        "data": [
-            {"id": 103, "name": "Proj C", "account_id": account.id},
-        ]
-    }
-    second_page_resp.raise_for_status.return_value = None
-
-    mock_get.side_effect = [first_page_resp, second_page_resp]
-
-    result = get_all_projects_for_account(
-        dbt_platform_url="https://cloud.getdbt.com",
-        account=account,
-        headers=base_headers,
-        page_size=2,
+    first_page_resp = create_mock_response(
+        {
+            "data": [
+                {"id": 101, "name": "Proj A", "account_id": account.id},
+                {"id": 102, "name": "Proj B", "account_id": account.id},
+            ]
+        }
+    )
+    second_page_resp = create_mock_response(
+        {
+            "data": [
+                {"id": 103, "name": "Proj C", "account_id": account.id},
+            ]
+        }
     )
 
+    mock_client = create_mock_httpx_client([first_page_resp, second_page_resp])
+
+    with patch("httpx.AsyncClient", return_value=mock_client):
+        result = await get_all_projects_for_account(
+            dbt_platform_url="https://cloud.getdbt.com",
+            account=account,
+            headers=base_headers,
+            page_size=2,
+        )
+
     # Should aggregate 3 projects and include account_name field
     assert len(result) == 3
     assert {p.id for p in result} == {101, 102, 103}
@@ -63,52 +77,45 @@ def test_get_all_projects_for_account_paginates(mock_get: Mock, base_headers, ac
         "https://cloud.getdbt.com/api/v3/accounts/1/projects/?state=1&offset=0&limit=2",
         "https://cloud.getdbt.com/api/v3/accounts/1/projects/?state=1&offset=2&limit=2",
     ]
-    actual_urls = [
-        call.kwargs["url"] if "url" in call.kwargs else call.args[0]
-        for call in mock_get.call_args_list
-    ]
+    actual_urls = [call.args[0] for call in mock_client.get.call_args_list]
     assert actual_urls == expected_urls
 
 
-@patch("dbt_mcp.project.environment_resolver.requests.get")
-def test_get_all_environments_for_project_paginates(mock_get: Mock, base_headers):
+async def test_get_all_environments_for_project_paginates(base_headers):
     # Two pages: first full page (limit=2), second partial (1 item)
-    first_page_resp = Mock()
-    first_page_resp.json.return_value = {
-        "data": [
-            {"id": 201, "name": "Dev", "deployment_type": "development"},
-            {"id": 202, "name": "Prod", "deployment_type": "production"},
-        ]
-    }
-    first_page_resp.raise_for_status.return_value = None
-
-    second_page_resp = Mock()
-    second_page_resp.json.return_value = {
-        "data": [
-            {"id": 203, "name": "Staging", "deployment_type": "development"},
-        ]
-    }
-    second_page_resp.raise_for_status.return_value = None
-
-    mock_get.side_effect = [first_page_resp, second_page_resp]
-
-    result = _get_all_environments_for_project(
-        dbt_platform_url="https://cloud.getdbt.com",
-        account_id=1,
-        project_id=9,
-        headers=base_headers,
-        page_size=2,
+    first_page_resp = create_mock_response(
+        {
+            "data": [
+                {"id": 201, "name": "Dev", "deployment_type": "development"},
+                {"id": 202, "name": "Prod", "deployment_type": "production"},
+            ]
+        }
+    )
+    second_page_resp = create_mock_response(
+        {
+            "data": [
+                {"id": 203, "name": "Staging", "deployment_type": "development"},
+            ]
+        }
     )
 
+    mock_client = create_mock_httpx_client([first_page_resp, second_page_resp])
+
+    with patch("httpx.AsyncClient", return_value=mock_client):
+        result = await _get_all_environments_for_project(
+            dbt_platform_url="https://cloud.getdbt.com",
+            account_id=1,
+            project_id=9,
+            headers=base_headers,
+            page_size=2,
+        )
+
     assert len(result) == 3
     assert {e.id for e in result} == {201, 202, 203}
 
     expected_urls = [
         "https://cloud.getdbt.com/api/v3/accounts/1/projects/9/environments/?state=1&offset=0&limit=2",
         "https://cloud.getdbt.com/api/v3/accounts/1/projects/9/environments/?state=1&offset=2&limit=2",
     ]
-    actual_urls = [
-        call.kwargs["url"] if "url" in call.kwargs else call.args[0]
-        for call in mock_get.call_args_list
-    ]
+    actual_urls = [call.args[0] for call in mock_client.get.call_args_list]
     assert actual_urls == expected_urls