diff --git a/api/controllers/console/datasets/external.py b/api/controllers/console/datasets/external.py
index b7af30f451a..34f6aecfcbd 100644
--- a/api/controllers/console/datasets/external.py
+++ b/api/controllers/console/datasets/external.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
@@ -37,6 +39,7 @@ from models import Account
 from models.dataset import ExternalKnowledgeApis
 from services.dataset_service import DatasetService
 from services.enterprise import rbac_service as enterprise_rbac_service
+from services.entities.external_knowledge_entities.external_knowledge_entities import ExternalDatasetCreatePayload
 from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
 from services.knowledge_service import BedrockRetrievalSetting, ExternalDatasetTestService
@@ -47,14 +50,6 @@ class ExternalKnowledgeApiPayload(BaseModel):
     settings: dict[str, Any]
 
 
-class ExternalDatasetCreatePayload(BaseModel):
-    external_knowledge_api_id: str
-    external_knowledge_id: str
-    name: str = Field(..., min_length=1, max_length=100)
-    description: str | None = Field(None, max_length=400)
-    external_retrieval_model: dict[str, Any] | None = None
-
-
 class ExternalHitTestingPayload(BaseModel):
     query: str
     external_retrieval_model: dict[str, Any] | None = None
@@ -62,7 +57,7 @@ class ExternalHitTestingPayload(BaseModel):
 
 
 class BedrockRetrievalPayload(BaseModel):
-    retrieval_setting: "BedrockRetrievalSetting"
+    retrieval_setting: BedrockRetrievalSetting
     query: str
     knowledge_id: str
 
@@ -362,7 +357,6 @@ class ExternalDatasetCreateApi(Resource):
     def post(self, session: Session, current_tenant_id: str, current_user: Account):
         # The role of the current user in the ta table must be admin, owner, or editor
         payload = ExternalDatasetCreatePayload.model_validate(console_ns.payload or {})
-        args = payload.model_dump(exclude_none=True)
 
         # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
         if not current_user.is_dataset_editor:
@@ -372,7 +366,7 @@ class ExternalDatasetCreateApi(Resource):
             dataset = ExternalDatasetService.create_external_dataset(
                 tenant_id=current_tenant_id,
                 user_id=current_user.id,
-                args=args,
+                args=payload,
                 session=session,
             )
         except services.errors.dataset.DatasetNameDuplicateError:
diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md
index 232ec62e765..545df2a320e 100644
--- a/api/openapi/markdown/console-openapi.md
+++ b/api/openapi/markdown/console-openapi.md
@@ -18081,6 +18081,13 @@ Built-in tool icons are URL strings; API-based tool icons are provider-defined p
 
 #### ExternalDatasetCreatePayload
 
+Validated fields required to create an external dataset binding.
+
+The console controller owns HTTP concerns, but the service also needs this
+contract when creating the tenant-scoped dataset and external knowledge
+binding. Keep it outside controllers so service imports do not depend on
+Flask blueprint initialization.
+
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | description | string |  | No |
diff --git a/api/services/entities/external_knowledge_entities/external_knowledge_entities.py b/api/services/entities/external_knowledge_entities/external_knowledge_entities.py
index 110dbe5a5e7..be4bd467731 100644
--- a/api/services/entities/external_knowledge_entities/external_knowledge_entities.py
+++ b/api/services/entities/external_knowledge_entities/external_knowledge_entities.py
@@ -1,6 +1,6 @@
 from typing import Any, Literal, Union
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class AuthorizationConfig(BaseModel):
@@ -24,3 +24,19 @@ class ExternalKnowledgeApiSetting(BaseModel):
     request_method: str
     headers: dict[str, Any] | None = None
     params: dict[str, Any] | None = None
+
+
+class ExternalDatasetCreatePayload(BaseModel):
+    """Validated fields required to create an external dataset binding.
+
+    The console controller owns HTTP concerns, but the service also needs this
+    contract when creating the tenant-scoped dataset and external knowledge
+    binding. Keep it outside controllers so service imports do not depend on
+    Flask blueprint initialization.
+    """
+
+    external_knowledge_api_id: str
+    external_knowledge_id: str
+    name: str = Field(..., min_length=1, max_length=100)
+    description: str | None = Field(None, max_length=400)
+    external_retrieval_model: dict[str, object] | None = Field(default=None)
diff --git a/api/services/external_knowledge_service.py b/api/services/external_knowledge_service.py
index 4b17e48662f..03fe274fe89 100644
--- a/api/services/external_knowledge_service.py
+++ b/api/services/external_knowledge_service.py
@@ -22,6 +22,7 @@ from models.dataset import (
 from services.enterprise import rbac_service as enterprise_rbac_service
 from services.entities.external_knowledge_entities.external_knowledge_entities import (
     Authorization,
+    ExternalDatasetCreatePayload,
     ExternalKnowledgeApiSetting,
 )
 from services.errors.dataset import DatasetNameDuplicateError
@@ -287,16 +288,17 @@ class ExternalDatasetService:
         return ExternalKnowledgeApiSetting.model_validate(settings)
 
     @staticmethod
-    def create_external_dataset(tenant_id: str, user_id: str, args: dict[str, Any], *, session: Session) -> Dataset:
+    def create_external_dataset(
+        tenant_id: str, user_id: str, args: ExternalDatasetCreatePayload, *, session: Session
+    ) -> Dataset:
+        """Create a tenant-scoped external dataset and binding in the caller's transaction."""
         # check if dataset name already exists
-        if session.scalar(
-            select(Dataset).where(Dataset.name == args.get("name"), Dataset.tenant_id == tenant_id).limit(1)
-        ):
-            raise DatasetNameDuplicateError(f"Dataset with name {args.get('name')} already exists.")
+        if session.scalar(select(Dataset).where(Dataset.name == args.name, Dataset.tenant_id == tenant_id).limit(1)):
+            raise DatasetNameDuplicateError(f"Dataset with name {args.name} already exists.")
         external_knowledge_api = session.scalar(
             select(ExternalKnowledgeApis)
             .where(
-                ExternalKnowledgeApis.id == args.get("external_knowledge_api_id"),
+                ExternalKnowledgeApis.id == args.external_knowledge_api_id,
                 ExternalKnowledgeApis.tenant_id == tenant_id,
             )
             .limit(1)
@@ -307,26 +309,22 @@ class ExternalDatasetService:
 
         dataset = Dataset(
             tenant_id=tenant_id,
-            name=args.get("name"),
-            description=args.get("description", ""),
+            name=args.name,
+            description=args.description or "",
             provider="external",
-            retrieval_model=args.get("external_retrieval_model"),
+            retrieval_model=args.external_retrieval_model,
             created_by=user_id,
             maintainer=user_id,
         )
 
         session.add(dataset)
         session.flush()
-        if args.get("external_knowledge_id") is None:
-            raise ValueError("external_knowledge_id is required")
-        if args.get("external_knowledge_api_id") is None:
-            raise ValueError("external_knowledge_api_id is required")
 
         external_knowledge_binding = ExternalKnowledgeBindings(
             tenant_id=tenant_id,
             dataset_id=dataset.id,
-            external_knowledge_api_id=args.get("external_knowledge_api_id") or "",
-            external_knowledge_id=args.get("external_knowledge_id") or "",
+            external_knowledge_api_id=args.external_knowledge_api_id or "",
+            external_knowledge_id=args.external_knowledge_id or "",
             created_by=user_id,
         )
         session.add(external_knowledge_binding)
diff --git a/api/tests/unit_tests/controllers/console/datasets/test_external.py b/api/tests/unit_tests/controllers/console/datasets/test_external.py
index 5addfd66e7f..edcc2451131 100644
--- a/api/tests/unit_tests/controllers/console/datasets/test_external.py
+++ b/api/tests/unit_tests/controllers/console/datasets/test_external.py
@@ -20,6 +20,7 @@ from controllers.console.datasets.external import (
 )
 from models.account import Account, TenantAccountRole
 from services.dataset_service import DatasetService
+from services.entities.external_knowledge_entities.external_knowledge_entities import ExternalDatasetCreatePayload
 from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
 from services.knowledge_service import ExternalDatasetTestService
@@ -411,7 +412,7 @@ class TestExternalDatasetCreateApi:
         create_external_dataset.assert_called_once_with(
             tenant_id="tenant-1",
             user_id="user-1",
-            args=payload,
+            args=ExternalDatasetCreatePayload.model_validate(payload),
             session=session,
         )
         dataset_response_source.assert_called_once_with(dataset, session=session)
diff --git a/api/tests/unit_tests/controllers/console/datasets/test_external_dataset_payload.py b/api/tests/unit_tests/controllers/console/datasets/test_external_dataset_payload.py
index 2ea1fcf5441..49c89b1adfc 100644
--- a/api/tests/unit_tests/controllers/console/datasets/test_external_dataset_payload.py
+++ b/api/tests/unit_tests/controllers/console/datasets/test_external_dataset_payload.py
@@ -11,7 +11,7 @@ full Flask/RESTX request stack.
 import pytest
 from pydantic import ValidationError
 
-from controllers.console.datasets.external import ExternalDatasetCreatePayload
+from services.entities.external_knowledge_entities.external_knowledge_entities import ExternalDatasetCreatePayload
 
 
 def test_external_dataset_create_payload_allows_name_length_100() -> None:
diff --git a/api/tests/unit_tests/services/test_external_dataset_service.py b/api/tests/unit_tests/services/test_external_dataset_service.py
index 629fc07ec7b..689af7300e6 100644
--- a/api/tests/unit_tests/services/test_external_dataset_service.py
+++ b/api/tests/unit_tests/services/test_external_dataset_service.py
@@ -1,8 +1,8 @@
-"""
-Comprehensive unit tests for ExternalDatasetService.
+"""Tests for external knowledge API, binding, dataset, and retrieval operations.
 
-This test suite provides extensive coverage of external knowledge API and dataset operations.
-Target: 1500+ lines of comprehensive test coverage.
+Database-facing cases persist the minimum required ``TypeBase`` tables through
+the shared in-memory SQLite fixture, including cross-tenant rows. HTTP calls,
+pagination, and clocks remain mocked at their genuine I/O boundaries.
 """
 
 import json
@@ -12,13 +12,15 @@ from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from constants import HIDDEN_VALUE
 from models.dataset import Dataset, ExternalKnowledgeApis, ExternalKnowledgeBindings
-from services.enterprise.rbac_service import RBACResourceType
 from services.entities.external_knowledge_entities.external_knowledge_entities import (
     Authorization,
     AuthorizationConfig,
+    ExternalDatasetCreatePayload,
     ExternalKnowledgeApiSetting,
 )
 from services.errors.dataset import DatasetNameDuplicateError
@@ -27,7 +29,7 @@ from services.external_knowledge_service import ExternalDatasetService
 
 
 class ExternalDatasetServiceTestDataFactory:
-    """Factory for creating test data and mock objects."""
+    """Build non-session value objects used by tests outside persistence paths."""
 
     @staticmethod
     def create_external_knowledge_api_mock(
@@ -136,6 +138,100 @@ def factory():
     return ExternalDatasetServiceTestDataFactory()
 
 
+def _make_external_knowledge_api(
+    *,
+    api_id: str = "api-123",
+    tenant_id: str = "tenant-123",
+    name: str = "Test API",
+    description: str = "Test description",
+    settings: dict[str, Any] | list[dict[str, Any]] | None = None,
+    created_by: str = "user-123",
+    updated_by: str = "user-123",
+) -> ExternalKnowledgeApis:
+    """Build a real ExternalKnowledgeApis row for SQLite-backed service tests."""
+    if settings is None:
+        settings = {"endpoint": "https://api.example.com", "api_key": "test-key-123"}
+    api = ExternalKnowledgeApis(
+        tenant_id=tenant_id,
+        created_by=created_by,
+        updated_by=updated_by,
+        name=name,
+        description=description,
+        settings=json.dumps(settings, ensure_ascii=False),
+    )
+    api.id = api_id
+    return api
+
+
+def _make_dataset(
+    *,
+    dataset_id: str = "dataset-123",
+    tenant_id: str = "tenant-123",
+    name: str = "Test Dataset",
+    provider: str = "external",
+    description: str = "",
+    retrieval_model: dict[str, Any] | None = None,
+    created_by: str = "user-123",
+) -> Dataset:
+    """Build a real Dataset row with the fields required by ExternalDatasetService."""
+    dataset = Dataset(
+        id=dataset_id,
+        tenant_id=tenant_id,
+        name=name,
+        description=description,
+        provider=provider,
+        retrieval_model=retrieval_model or {},
+        created_by=created_by,
+        maintainer=created_by,
+    )
+    return dataset
+
+
+def _make_external_knowledge_binding(
+    *,
+    binding_id: str = "binding-123",
+    tenant_id: str = "tenant-123",
+    dataset_id: str = "dataset-123",
+    external_knowledge_api_id: str = "api-123",
+    external_knowledge_id: str = "knowledge-123",
+    created_by: str = "user-123",
+) -> ExternalKnowledgeBindings:
+    """Build a real ExternalKnowledgeBindings row for tenant-scoped lookup tests."""
+    binding = ExternalKnowledgeBindings(
+        tenant_id=tenant_id,
+        dataset_id=dataset_id,
+        external_knowledge_api_id=external_knowledge_api_id,
+        external_knowledge_id=external_knowledge_id,
+        created_by=created_by,
+    )
+    binding.id = binding_id
+    return binding
+
+
+def _add_and_commit(session: Session, *objects: object) -> None:
+    """Persist rows so service methods exercise real SQLAlchemy queries."""
+    session.add_all(objects)
+    session.commit()
+
+
+def _seed_external_retrieval_dependencies(
+    session: Session,
+    *,
+    tenant_id: str = "tenant-123",
+    dataset_id: str = "dataset-123",
+    api_id: str = "api-123",
+) -> tuple[ExternalKnowledgeBindings, ExternalKnowledgeApis]:
+    """Seed the binding and API template required by fetch_external_knowledge_retrieval."""
+    binding = _make_external_knowledge_binding(
+        tenant_id=tenant_id,
+        dataset_id=dataset_id,
+        external_knowledge_api_id=api_id,
+    )
+    api = _make_external_knowledge_api(api_id=api_id, tenant_id=tenant_id)
+    _add_and_commit(session, binding, api)
+    return binding, api
+
+
 class TestExternalDatasetServiceGetAPIs:
     """Test get_external_knowledge_apis operations - comprehensive coverage."""
 
@@ -430,13 +526,13 @@ class TestExternalDatasetServiceValidateAPIList:
         ExternalDatasetService.validate_api_list(api_settings)
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeApis,)], indirect=True)
 class TestExternalDatasetServiceCreateAPI:
     """Test create_external_knowledge_api operations."""
 
-    @patch("services.external_knowledge_service.db")
     @patch("services.external_knowledge_service.ExternalDatasetService.check_endpoint_and_api_key")
     def test_create_external_knowledge_api_success_full(
-        self, mock_check, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_check, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test successful creation with all fields."""
         # Arrange
@@ -449,7 +545,7 @@ class TestExternalDatasetServiceCreateAPI:
         }
 
         # Act
-        result = ExternalDatasetService.create_external_knowledge_api(tenant_id, user_id, args, session=mock_db.session)
+        result = ExternalDatasetService.create_external_knowledge_api(tenant_id, user_id, args, session=sqlite_session)
 
         # Assert
         assert result.name == "Test API"
@@ -458,14 +554,12 @@ class TestExternalDatasetServiceCreateAPI:
         assert result.created_by == user_id
         assert result.updated_by == user_id
         mock_check.assert_called_once_with(args["settings"])
-        mock_db.session.add.assert_called_once()
-        mock_db.session.flush.assert_called_once()
-        mock_db.session.commit.assert_not_called()
+        persisted_api = sqlite_session.get(ExternalKnowledgeApis, result.id)
+        assert persisted_api is result
 
-    @patch("services.external_knowledge_service.db")
     @patch("services.external_knowledge_service.ExternalDatasetService.check_endpoint_and_api_key")
     def test_create_external_knowledge_api_minimal_fields(
-        self, mock_check, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_check, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test creation with minimal required fields."""
         # Arrange
@@ -476,16 +570,16 @@ class TestExternalDatasetServiceCreateAPI:
 
         # Act
         result = ExternalDatasetService.create_external_knowledge_api(
-            "tenant-123", "user-123", args, session=mock_db.session
+            "tenant-123", "user-123", args, session=sqlite_session
         )
 
         # Assert
         assert result.name == "Minimal API"
         assert result.description == ""
+        assert sqlite_session.get(ExternalKnowledgeApis, result.id) is result
 
-    @patch("services.external_knowledge_service.db")
     def test_create_external_knowledge_api_missing_settings(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test creation fails when settings are missing."""
         # Arrange
@@ -493,26 +587,22 @@ class TestExternalDatasetServiceCreateAPI:
 
         # Act & Assert
         with pytest.raises(ValueError, match="settings is required"):
-            ExternalDatasetService.create_external_knowledge_api(
-                "tenant-123", "user-123", args, session=mock_db.session
-            )
+            ExternalDatasetService.create_external_knowledge_api("tenant-123", "user-123", args, session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
-    def test_create_external_knowledge_api_none_settings(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_create_external_knowledge_api_none_settings(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test creation fails when settings are explicitly None."""
         # Arrange
         args = {"name": "Test API", "settings": None}
 
         # Act & Assert
         with pytest.raises(ValueError, match="settings is required"):
-            ExternalDatasetService.create_external_knowledge_api(
-                "tenant-123", "user-123", args, session=mock_db.session
-            )
+            ExternalDatasetService.create_external_knowledge_api("tenant-123", "user-123", args, session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
     @patch("services.external_knowledge_service.ExternalDatasetService.check_endpoint_and_api_key")
     def test_create_external_knowledge_api_settings_json_serialization(
-        self, mock_check, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_check, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test that settings are properly JSON serialized."""
         # Arrange
@@ -525,7 +615,7 @@ class TestExternalDatasetServiceCreateAPI:
 
         # Act
         result = ExternalDatasetService.create_external_knowledge_api(
-            "tenant-123", "user-123", args, session=mock_db.session
+            "tenant-123", "user-123", args, session=sqlite_session
         )
 
         # Assert
@@ -533,10 +623,9 @@ class TestExternalDatasetServiceCreateAPI:
         parsed_settings = json.loads(result.settings)
         assert parsed_settings == settings
 
-    @patch("services.external_knowledge_service.db")
     @patch("services.external_knowledge_service.ExternalDatasetService.check_endpoint_and_api_key")
     def test_create_external_knowledge_api_unicode_handling(
-        self, mock_check, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_check, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test proper handling of Unicode characters in name and description."""
         # Arrange
@@ -548,17 +637,16 @@ class TestExternalDatasetServiceCreateAPI:
 
         # Act
         result = ExternalDatasetService.create_external_knowledge_api(
-            "tenant-123", "user-123", args, session=mock_db.session
+            "tenant-123", "user-123", args, session=sqlite_session
         )
 
         # Assert
         assert result.name == "测试API"
         assert result.description == "テストの説明"
 
-    @patch("services.external_knowledge_service.db")
     @patch("services.external_knowledge_service.ExternalDatasetService.check_endpoint_and_api_key")
     def test_create_external_knowledge_api_long_description(
-        self, mock_check, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_check, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test creation with very long description."""
         # Arrange
@@ -571,7 +659,7 @@ class TestExternalDatasetServiceCreateAPI:
 
         # Act
         result = ExternalDatasetService.create_external_knowledge_api(
-            "tenant-123", "user-123", args, session=mock_db.session
+            "tenant-123", "user-123", args, session=sqlite_session
         )
 
         # Assert
@@ -894,43 +982,42 @@ class TestExternalDatasetServiceCheckEndpoint:
         assert call_kwargs["headers"]["Authorization"] == "Bearer test-key-123"
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeApis,)], indirect=True)
 class TestExternalDatasetServiceGetAPI:
     """Test get_external_knowledge_api operations."""
 
-    @patch("services.external_knowledge_service.db")
-    def test_get_external_knowledge_api_success(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_get_external_knowledge_api_success(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test successful retrieval of external knowledge API."""
         # Arrange
         api_id = "api-123"
-        expected_api = factory.create_external_knowledge_api_mock(api_id=api_id)
-
-        mock_db.session.scalar.return_value = expected_api
+        expected_api = _make_external_knowledge_api(api_id=api_id)
+        _add_and_commit(sqlite_session, expected_api)
 
         # Act
         tenant_id = "tenant-123"
-        result = ExternalDatasetService.get_external_knowledge_api(api_id, tenant_id, session=mock_db.session)
+        result = ExternalDatasetService.get_external_knowledge_api(api_id, tenant_id, session=sqlite_session)
 
         # Assert
         assert result.id == api_id
 
-    @patch("services.external_knowledge_service.db")
-    def test_get_external_knowledge_api_not_found(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_get_external_knowledge_api_not_found(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test error when API is not found."""
-        # Arrange
-        mock_db.session.scalar.return_value = None
-
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
-            ExternalDatasetService.get_external_knowledge_api("nonexistent-id", "tenant-123", session=mock_db.session)
+            ExternalDatasetService.get_external_knowledge_api("nonexistent-id", "tenant-123", session=sqlite_session)
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeApis,)], indirect=True)
 class TestExternalDatasetServiceUpdateAPI:
     """Test update_external_knowledge_api operations."""
 
     @patch("services.external_knowledge_service.naive_utc_now")
-    @patch("services.external_knowledge_service.db")
     def test_update_external_knowledge_api_success_all_fields(
-        self, mock_db, mock_now, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_now, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test successful update with all fields."""
         # Arrange
@@ -940,7 +1027,8 @@ class TestExternalDatasetServiceUpdateAPI:
         current_time = datetime(2024, 1, 2, 12, 0)
         mock_now.return_value = current_time
 
-        existing_api = factory.create_external_knowledge_api_mock(api_id=api_id, tenant_id=tenant_id)
+        existing_api = _make_external_knowledge_api(api_id=api_id, tenant_id=tenant_id)
+        _add_and_commit(sqlite_session, existing_api)
 
         args = {
             "name": "Updated API",
@@ -948,11 +1036,9 @@ class TestExternalDatasetServiceUpdateAPI:
             "settings": {"endpoint": "https://new.example.com", "api_key": "new-key"},
         }
 
-        mock_db.session.scalar.return_value = existing_api
-
         # Act
         result = ExternalDatasetService.update_external_knowledge_api(
-            tenant_id, user_id, api_id, args, session=mock_db.session
+            tenant_id, user_id, api_id, args, session=sqlite_session
         )
 
         # Assert
@@ -960,193 +1046,207 @@ class TestExternalDatasetServiceUpdateAPI:
         assert result.description == "Updated description"
         assert result.updated_by == user_id
         assert result.updated_at == current_time
-        mock_db.session.flush.assert_called_once()
-        mock_db.session.commit.assert_not_called()
+        assert sqlite_session.get(ExternalKnowledgeApis, api_id) is result
 
-    @patch("services.external_knowledge_service.db")
     def test_update_external_knowledge_api_preserve_hidden_api_key(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test that hidden API key is preserved from existing settings."""
         # Arrange
         api_id = "api-123"
         tenant_id = "tenant-123"
 
-        existing_api = factory.create_external_knowledge_api_mock(
+        existing_api = _make_external_knowledge_api(
             api_id=api_id,
             tenant_id=tenant_id,
             settings={"endpoint": "https://api.example.com", "api_key": "original-secret-key"},
         )
+        _add_and_commit(sqlite_session, existing_api)
 
         args = {
             "name": "Updated API",
             "settings": {"endpoint": "https://api.example.com", "api_key": HIDDEN_VALUE},
         }
 
-        mock_db.session.scalar.return_value = existing_api
-
         # Act
         result = ExternalDatasetService.update_external_knowledge_api(
-            tenant_id, "user-123", api_id, args, session=mock_db.session
+            tenant_id, "user-123", api_id, args, session=sqlite_session
         )
 
         # Assert
         settings = json.loads(result.settings)
         assert settings["api_key"] == "original-secret-key"
 
-    @patch("services.external_knowledge_service.db")
-    def test_update_external_knowledge_api_not_found(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_update_external_knowledge_api_not_found(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test error when API is not found."""
         # Arrange
-        mock_db.session.scalar.return_value = None
-
         args = {"name": "Updated API"}
 
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
             ExternalDatasetService.update_external_knowledge_api(
-                "tenant-123", "user-123", "api-123", args, session=mock_db.session
+                "tenant-123", "user-123", "api-123", args, session=sqlite_session
             )
 
-    @patch("services.external_knowledge_service.db")
     def test_update_external_knowledge_api_tenant_mismatch(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when tenant ID doesn't match."""
         # Arrange
-        mock_db.session.scalar.return_value = None
-
+        _add_and_commit(sqlite_session, _make_external_knowledge_api(api_id="api-123", tenant_id="tenant-123"))
         args = {"name": "Updated API"}
 
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
             ExternalDatasetService.update_external_knowledge_api(
-                "wrong-tenant", "user-123", "api-123", args, session=mock_db.session
+                "wrong-tenant", "user-123", "api-123", args, session=sqlite_session
             )
 
-    @patch("services.external_knowledge_service.db")
-    def test_update_external_knowledge_api_name_only(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_update_external_knowledge_api_name_only(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test updating only the name field."""
         # Arrange
-        existing_api = factory.create_external_knowledge_api_mock(
+        existing_api = _make_external_knowledge_api(
             description="Original description",
             settings={"endpoint": "https://api.example.com", "api_key": "key"},
         )
+        _add_and_commit(sqlite_session, existing_api)
 
         args = {"name": "New Name Only"}
 
-        mock_db.session.scalar.return_value = existing_api
-
         # Act
         result = ExternalDatasetService.update_external_knowledge_api(
-            "tenant-123", "user-123", "api-123", args, session=mock_db.session
+            "tenant-123", "user-123", "api-123", args, session=sqlite_session
         )
 
         # Assert
         assert result.name == "New Name Only"
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeApis,)], indirect=True)
 class TestExternalDatasetServiceDeleteAPI:
     """Test delete_external_knowledge_api operations."""
 
-    @patch("services.external_knowledge_service.db")
-    def test_delete_external_knowledge_api_success(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_delete_external_knowledge_api_success(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test successful deletion of external knowledge API."""
         # Arrange
         api_id = "api-123"
         tenant_id = "tenant-123"
 
-        existing_api = factory.create_external_knowledge_api_mock(api_id=api_id, tenant_id=tenant_id)
-
-        mock_db.session.scalar.return_value = existing_api
+        existing_api = _make_external_knowledge_api(api_id=api_id, tenant_id=tenant_id)
+        _add_and_commit(sqlite_session, existing_api)
 
         # Act
-        ExternalDatasetService.delete_external_knowledge_api(tenant_id, api_id, session=mock_db.session)
+        ExternalDatasetService.delete_external_knowledge_api(tenant_id, api_id, session=sqlite_session)
 
         # Assert
-        mock_db.session.delete.assert_called_once_with(existing_api)
-        mock_db.session.flush.assert_called_once()
-        mock_db.session.commit.assert_not_called()
+        assert sqlite_session.get(ExternalKnowledgeApis, api_id) is None
 
-    @patch("services.external_knowledge_service.db")
-    def test_delete_external_knowledge_api_not_found(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_delete_external_knowledge_api_not_found(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test error when API is not found."""
-        # Arrange
-        mock_db.session.scalar.return_value = None
-
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
-            ExternalDatasetService.delete_external_knowledge_api("tenant-123", "api-123", session=mock_db.session)
+            ExternalDatasetService.delete_external_knowledge_api("tenant-123", "api-123", session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
     def test_delete_external_knowledge_api_tenant_mismatch(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when tenant ID doesn't match."""
         # Arrange
-        mock_db.session.scalar.return_value = None
+        _add_and_commit(sqlite_session, _make_external_knowledge_api(api_id="api-123", tenant_id="tenant-123"))
 
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
-            ExternalDatasetService.delete_external_knowledge_api("wrong-tenant", "api-123", session=mock_db.session)
+            ExternalDatasetService.delete_external_knowledge_api("wrong-tenant", "api-123", session=sqlite_session)
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeBindings,)], indirect=True)
 class TestExternalDatasetServiceAPIUseCheck:
     """Test external_knowledge_api_use_check operations."""
 
-    @patch("services.external_knowledge_service.db")
     def test_external_knowledge_api_use_check_in_use_single(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test API use check when API has one binding."""
         # Arrange
         api_id = "api-123"
         tenant_id = "tenant-123"
 
-        mock_db.session.scalar.return_value = 1
+        _add_and_commit(
+            sqlite_session,
+            _make_external_knowledge_binding(external_knowledge_api_id=api_id, tenant_id=tenant_id),
+            _make_external_knowledge_binding(
+                binding_id="binding-other",
+                external_knowledge_api_id=api_id,
+                tenant_id="other-tenant",
+            ),
+        )
 
         # Act
         in_use, count = ExternalDatasetService.external_knowledge_api_use_check(
-            api_id, tenant_id, session=mock_db.session
+            api_id, tenant_id, session=sqlite_session
         )
 
         # Assert
         assert in_use is True
         assert count == 1
-        assert "tenant_id" in str(mock_db.session.scalar.call_args.args[0])
 
-    @patch("services.external_knowledge_service.db")
     def test_external_knowledge_api_use_check_in_use_multiple(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test API use check with multiple bindings."""
         # Arrange
         api_id = "api-123"
         tenant_id = "tenant-123"
 
-        mock_db.session.scalar.return_value = 10
+        _add_and_commit(
+            sqlite_session,
+            *[
+                _make_external_knowledge_binding(
+                    binding_id=f"binding-{index}",
+                    external_knowledge_api_id=api_id,
+                    tenant_id=tenant_id,
+                    dataset_id=f"dataset-{index}",
+                )
+                for index in range(10)
+            ],
+        )
 
         # Act
         in_use, count = ExternalDatasetService.external_knowledge_api_use_check(
-            api_id, tenant_id, session=mock_db.session
+            api_id, tenant_id, session=sqlite_session
         )
 
         # Assert
         assert in_use is True
         assert count == 10
 
-    @patch("services.external_knowledge_service.db")
-    def test_external_knowledge_api_use_check_not_in_use(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_external_knowledge_api_use_check_not_in_use(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test API use check when API is not in use."""
         # Arrange
         api_id = "api-123"
         tenant_id = "tenant-123"
 
-        mock_db.session.scalar.return_value = 0
+        _add_and_commit(
+            sqlite_session,
+            _make_external_knowledge_binding(
+                external_knowledge_api_id=api_id,
+                tenant_id="other-tenant",
+            ),
+        )
 
         # Act
         in_use, count = ExternalDatasetService.external_knowledge_api_use_check(
-            api_id, tenant_id, session=mock_db.session
+            api_id, tenant_id, session=sqlite_session
         )
 
         # Assert
@@ -1154,48 +1254,47 @@ class TestExternalDatasetServiceAPIUseCheck:
         assert count == 0
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeBindings,)], indirect=True)
 class TestExternalDatasetServiceGetBinding:
     """Test get_external_knowledge_binding_with_dataset_id operations."""
 
-    @patch("services.external_knowledge_service.db")
-    def test_get_external_knowledge_binding_success(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_get_external_knowledge_binding_success(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test successful retrieval of external knowledge binding."""
         # Arrange
         tenant_id = "tenant-123"
         dataset_id = "dataset-123"
 
-        expected_binding = factory.create_external_knowledge_binding_mock(tenant_id=tenant_id, dataset_id=dataset_id)
-
-        mock_db.session.scalar.return_value = expected_binding
+        expected_binding = _make_external_knowledge_binding(tenant_id=tenant_id, dataset_id=dataset_id)
+        _add_and_commit(sqlite_session, expected_binding)
 
         # Act
         result = ExternalDatasetService.get_external_knowledge_binding_with_dataset_id(
-            tenant_id, dataset_id, session=mock_db.session
+            tenant_id, dataset_id, session=sqlite_session
         )
 
         # Assert
         assert result.dataset_id == dataset_id
         assert result.tenant_id == tenant_id
 
-    @patch("services.external_knowledge_service.db")
-    def test_get_external_knowledge_binding_not_found(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_get_external_knowledge_binding_not_found(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test error when binding is not found."""
-        # Arrange
-        mock_db.session.scalar.return_value = None
-
         # Act & Assert
         with pytest.raises(ValueError, match="external knowledge binding not found"):
             ExternalDatasetService.get_external_knowledge_binding_with_dataset_id(
-                "tenant-123", "dataset-123", session=mock_db.session
+                "tenant-123", "dataset-123", session=sqlite_session
             )
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeApis,)], indirect=True)
 class TestExternalDatasetServiceDocumentValidate:
     """Test document_create_args_validate operations."""
 
-    @patch("services.external_knowledge_service.db")
     def test_document_create_args_validate_success_all_params(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test successful validation with all required parameters."""
         # Arrange
@@ -1210,20 +1309,18 @@ class TestExternalDatasetServiceDocumentValidate:
             ]
         }
 
-        api = factory.create_external_knowledge_api_mock(api_id=api_id, settings=[settings])
-
-        mock_db.session.scalar.return_value = api
+        api = _make_external_knowledge_api(api_id=api_id, tenant_id=tenant_id, settings=[settings])
+        _add_and_commit(sqlite_session, api)
 
         process_parameter = {"param1": "value1", "param2": "value2"}
 
         # Act & Assert - should not raise
         ExternalDatasetService.document_create_args_validate(
-            tenant_id, api_id, process_parameter, session=mock_db.session
+            tenant_id, api_id, process_parameter, session=sqlite_session
         )
 
-    @patch("services.external_knowledge_service.db")
     def test_document_create_args_validate_missing_required_param(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test validation fails when required parameter is missing."""
         # Arrange
@@ -1232,45 +1329,39 @@ class TestExternalDatasetServiceDocumentValidate:
 
         settings = {"document_process_setting": [{"name": "required_param", "required": True}]}
 
-        api = factory.create_external_knowledge_api_mock(api_id=api_id, settings=[settings])
-
-        mock_db.session.scalar.return_value = api
+        api = _make_external_knowledge_api(api_id=api_id, tenant_id=tenant_id, settings=[settings])
+        _add_and_commit(sqlite_session, api)
 
         process_parameter = {}
 
         # Act & Assert
         with pytest.raises(ValueError, match="required_param is required"):
             ExternalDatasetService.document_create_args_validate(
-                tenant_id, api_id, process_parameter, session=mock_db.session
+                tenant_id, api_id, process_parameter, session=sqlite_session
             )
 
-    @patch("services.external_knowledge_service.db")
-    def test_document_create_args_validate_api_not_found(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_document_create_args_validate_api_not_found(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test validation fails when API is not found."""
-        # Arrange
-        mock_db.session.scalar.return_value = None
-
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
-            ExternalDatasetService.document_create_args_validate("tenant-123", "api-123", {}, session=mock_db.session)
+            ExternalDatasetService.document_create_args_validate("tenant-123", "api-123", {}, session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
     def test_document_create_args_validate_no_custom_parameters(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test validation succeeds when no custom parameters defined."""
         # Arrange
         settings = {}
-        api = factory.create_external_knowledge_api_mock(settings=[settings])
-
-        mock_db.session.scalar.return_value = api
+        api = _make_external_knowledge_api(settings=[settings])
+        _add_and_commit(sqlite_session, api)
 
         # Act & Assert - should not raise
-        ExternalDatasetService.document_create_args_validate("tenant-123", "api-123", {}, session=mock_db.session)
+        ExternalDatasetService.document_create_args_validate("tenant-123", "api-123", {}, session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
     def test_document_create_args_validate_optional_params_not_required(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test that optional parameters don't cause validation failure."""
         # Arrange
@@ -1281,15 +1372,14 @@ class TestExternalDatasetServiceDocumentValidate:
             ]
         }
 
-        api = factory.create_external_knowledge_api_mock(settings=[settings])
-
-        mock_db.session.scalar.return_value = api
+        api = _make_external_knowledge_api(settings=[settings])
+        _add_and_commit(sqlite_session, api)
 
         process_parameter = {"required_param": "value"}
 
         # Act & Assert - should not raise
         ExternalDatasetService.document_create_args_validate(
-            "tenant-123", "api-123", process_parameter, session=mock_db.session
+            "tenant-123", "api-123", process_parameter, session=sqlite_session
         )
 
 
@@ -1577,142 +1667,120 @@ class TestExternalDatasetServiceGetSettings:
         assert result.params["key1"] == "value1"
 
 
+@pytest.mark.parametrize(
+    "sqlite_session",
+    [(Dataset, ExternalKnowledgeApis, ExternalKnowledgeBindings)],
+    indirect=True,
+)
 class TestExternalDatasetServiceCreateDataset:
     """Test create_external_dataset operations."""
 
-    @patch("services.external_knowledge_service.db")
-    def test_create_external_dataset_success_full(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_create_external_dataset_success_full(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test successful creation of external dataset with all fields."""
         # Arrange
         tenant_id = "tenant-123"
         user_id = "user-123"
-        args = {
-            "name": "Test External Dataset",
-            "description": "Comprehensive test description",
-            "external_knowledge_api_id": "api-123",
-            "external_knowledge_id": "knowledge-123",
-            "external_retrieval_model": {"top_k": 5, "score_threshold": 0.7},
-        }
+        args = ExternalDatasetCreatePayload.model_validate(
+            {
+                "name": "Test External Dataset",
+                "description": "Comprehensive test description",
+                "external_knowledge_api_id": "api-123",
+                "external_knowledge_id": "knowledge-123",
+                "external_retrieval_model": {"top_k": 5, "score_threshold": 0.7},
+            }
+        )
 
-        api = factory.create_external_knowledge_api_mock(api_id="api-123")
-
-        mock_db.session.scalar.side_effect = [None, api]
+        api = _make_external_knowledge_api(api_id="api-123", tenant_id=tenant_id)
+        _add_and_commit(sqlite_session, api)
 
         # Act
-        result = ExternalDatasetService.create_external_dataset(tenant_id, user_id, args, session=mock_db.session)
+        result = ExternalDatasetService.create_external_dataset(tenant_id, user_id, args, session=sqlite_session)
 
         # Assert
         assert result.name == "Test External Dataset"
         assert result.description == "Comprehensive test description"
         assert result.provider == "external"
         assert result.created_by == user_id
-        mock_db.session.add.assert_called()
-        mock_db.session.flush.assert_called_once()
-        mock_db.session.commit.assert_called_once()
-
-    @patch("services.external_knowledge_service.enterprise_rbac_service.try_sync_creator_access_policy_member_bindings")
-    @patch("services.external_knowledge_service.db")
-    def test_create_external_dataset_syncs_creator_access_policy_binding(
-        self, mock_db, mock_sync, factory: ExternalDatasetServiceTestDataFactory
-    ):
-        """The creator must be bound to the new dataset's access policy, as in create_empty_dataset."""
-
-        # Arrange
-        def assign_dataset_id(instance):
-            # The real INSERT populates the primary key; a mocked session never flushes.
-            if isinstance(instance, Dataset):
-                instance.id = "dataset-705"
-
-        args = {
-            "name": "Bound External Dataset",
-            "external_knowledge_api_id": "api-703",
-            "external_knowledge_id": "knowledge-704",
-        }
-        mock_db.session.scalar.side_effect = [None, factory.create_external_knowledge_api_mock(api_id="api-703")]
-        mock_db.session.add.side_effect = assign_dataset_id
-
-        # Act
-        dataset = ExternalDatasetService.create_external_dataset(
-            "tenant-701", "user-702", args, session=mock_db.session
+        binding = sqlite_session.scalar(
+            select(ExternalKnowledgeBindings).where(
+                ExternalKnowledgeBindings.dataset_id == result.id,
+                ExternalKnowledgeBindings.tenant_id == tenant_id,
+            )
         )
+        assert binding is not None
+        assert binding.external_knowledge_api_id == "api-123"
+        assert binding.external_knowledge_id == "knowledge-123"
 
-        # Assert
-        assert dataset.id == "dataset-705"
-        mock_sync.assert_called_once_with(
-            "tenant-701",
-            "user-702",
-            RBACResourceType.DATASET,
-            "dataset-705",
-        )
-
-    @patch("services.external_knowledge_service.db")
     def test_create_external_dataset_duplicate_name_error(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when dataset name already exists."""
         # Arrange
-        existing_dataset = factory.create_dataset_mock(name="Duplicate Dataset")
+        existing_dataset = _make_dataset(name="Duplicate Dataset")
+        _add_and_commit(sqlite_session, existing_dataset)
 
-        mock_db.session.scalar.return_value = existing_dataset
-
-        args = {"name": "Duplicate Dataset"}
+        args = ExternalDatasetCreatePayload.model_validate(
+            {
+                "name": "Duplicate Dataset",
+                "external_knowledge_api_id": "api-123",
+                "external_knowledge_id": "knowledge-123",
+            }
+        )
 
         # Act & Assert
         with pytest.raises(DatasetNameDuplicateError):
-            ExternalDatasetService.create_external_dataset("tenant-123", "user-123", args, session=mock_db.session)
+            ExternalDatasetService.create_external_dataset("tenant-123", "user-123", args, session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
-    def test_create_external_dataset_api_not_found_error(self, mock_db, factory: ExternalDatasetServiceTestDataFactory):
+    def test_create_external_dataset_api_not_found_error(
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
+    ):
         """Test error when external knowledge API is not found."""
-        # Arrange
-        mock_db.session.scalar.side_effect = [None, None]
-
-        args = {"name": "Test Dataset", "external_knowledge_api_id": "nonexistent-api"}
+        args = ExternalDatasetCreatePayload.model_validate(
+            {
+                "name": "Test Dataset",
+                "external_knowledge_api_id": "nonexistent-api",
+                "external_knowledge_id": "knowledge-123",
+            }
+        )
 
         # Act & Assert
         with pytest.raises(ValueError, match="api template not found"):
-            ExternalDatasetService.create_external_dataset("tenant-123", "user-123", args, session=mock_db.session)
+            ExternalDatasetService.create_external_dataset("tenant-123", "user-123", args, session=sqlite_session)
 
-    @patch("services.external_knowledge_service.db")
     def test_create_external_dataset_missing_knowledge_id_error(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when external_knowledge_id is missing."""
         # Arrange
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [None, api]
-
-        args = {"name": "Test Dataset", "external_knowledge_api_id": "api-123"}
+        api = _make_external_knowledge_api()
+        _add_and_commit(sqlite_session, api)
 
         # Act & Assert
-        with pytest.raises(ValueError, match="external_knowledge_id is required"):
-            ExternalDatasetService.create_external_dataset("tenant-123", "user-123", args, session=mock_db.session)
+        with pytest.raises(ValueError, match="external_knowledge_id"):
+            ExternalDatasetCreatePayload.model_validate(
+                {"name": "Test Dataset", "external_knowledge_api_id": "api-123"}
+            )
 
-    @patch("services.external_knowledge_service.db")
     def test_create_external_dataset_missing_api_id_error(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when external_knowledge_api_id is missing."""
-        # Arrange
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [None, api]
-
-        args = {"name": "Test Dataset", "external_knowledge_id": "knowledge-123"}
-
         # Act & Assert
-        with pytest.raises(ValueError, match="external_knowledge_api_id is required"):
-            ExternalDatasetService.create_external_dataset("tenant-123", "user-123", args, session=mock_db.session)
+        with pytest.raises(ValueError, match="external_knowledge_api_id"):
+            ExternalDatasetCreatePayload.model_validate(
+                {"name": "Test Dataset", "external_knowledge_id": "knowledge-123"}
+            )
 
 
+@pytest.mark.parametrize("sqlite_session", [(ExternalKnowledgeApis, ExternalKnowledgeBindings)], indirect=True)
 class TestExternalDatasetServiceFetchRetrieval:
     """Test fetch_external_knowledge_retrieval operations."""
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_success_with_results(
-        self, mock_db, mock_process, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_process, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test successful external knowledge retrieval with results."""
         # Arrange
@@ -1720,12 +1788,7 @@ class TestExternalDatasetServiceFetchRetrieval:
         dataset_id = "dataset-123"
         query = "test query for retrieval"
 
-        binding = factory.create_external_knowledge_binding_mock(
-            dataset_id=dataset_id, external_knowledge_api_id="api-123"
-        )
-        api = factory.create_external_knowledge_api_mock(api_id="api-123")
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session, tenant_id=tenant_id, dataset_id=dataset_id)
 
         mock_response = MagicMock()
         mock_response.status_code = 200
@@ -1741,11 +1804,7 @@ class TestExternalDatasetServiceFetchRetrieval:
 
         # Act
         result = ExternalDatasetService.fetch_external_knowledge_retrieval(
-            tenant_id,
-            dataset_id,
-            query,
-            external_retrieval_parameters,
-            session=mock_db.session,
+            tenant_id, dataset_id, query, external_retrieval_parameters, session=sqlite_session
         )
 
         # Assert
@@ -1753,46 +1812,38 @@ class TestExternalDatasetServiceFetchRetrieval:
         assert result[0]["content"] == "result 1"
         assert result[1]["score"] == 0.8
 
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_binding_not_found_error(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when external knowledge binding is not found."""
-        # Arrange
-        mock_db.session.scalar.return_value = None
-
         # Act & Assert
         with pytest.raises(ExternalKnowledgeRetrievalError, match="external knowledge binding not found"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123", "dataset-123", "query", {}, session=mock_db.session
+                "tenant-123", "dataset-123", "query", {}, session=sqlite_session
             )
 
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_cross_tenant_api_template_error(
-        self, mock_db, factory: ExternalDatasetServiceTestDataFactory
+        self, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test error when a binding points to an API template outside the dataset tenant."""
         # Arrange
-        binding = factory.create_external_knowledge_binding_mock()
-        mock_db.session.scalar.side_effect = [binding, None]
+        binding = _make_external_knowledge_binding(tenant_id="tenant-123", external_knowledge_api_id="api-123")
+        cross_tenant_api = _make_external_knowledge_api(api_id="api-123", tenant_id="other-tenant")
+        _add_and_commit(sqlite_session, binding, cross_tenant_api)
 
         # Act & Assert
         with pytest.raises(ExternalKnowledgeRetrievalError, match="external api template not found"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123", "dataset-123", "query", {}, session=mock_db.session
+                "tenant-123", "dataset-123", "query", {}, session=sqlite_session
             )
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_empty_results(
-        self, mock_db, mock_process, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_process, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test retrieval with empty results."""
         # Arrange
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 200
@@ -1801,27 +1852,19 @@ class TestExternalDatasetServiceFetchRetrieval:
 
         # Act
         result = ExternalDatasetService.fetch_external_knowledge_retrieval(
-            "tenant-123",
-            "dataset-123",
-            "query",
-            {"top_k": 5},
-            session=mock_db.session,
+            "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
         )
 
         # Assert
         assert len(result) == 0
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_with_score_threshold(
-        self, mock_db, mock_process, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_process, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test retrieval with score threshold enabled."""
         # Arrange
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 200
@@ -1840,7 +1883,7 @@ class TestExternalDatasetServiceFetchRetrieval:
             "dataset-123",
             "query",
             external_retrieval_parameters,
-            session=mock_db.session,
+            session=sqlite_session,
         )
 
         # Assert
@@ -1850,16 +1893,12 @@ class TestExternalDatasetServiceFetchRetrieval:
         assert call_args.params["retrieval_setting"]["score_threshold"] == 0.75
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_non_200_status_raises_exception(
-        self, mock_db, mock_process, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_process, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test that non-200 status code raises Exception with response text."""
         # Arrange
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 500
@@ -1869,11 +1908,7 @@ class TestExternalDatasetServiceFetchRetrieval:
         # Act & Assert
         with pytest.raises(ExternalKnowledgeRetrievalError, match="Internal Server Error: Database connection failed"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123",
-                "dataset-123",
-                "query",
-                {"top_k": 5},
-                session=mock_db.session,
+                "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
             )
 
     @pytest.mark.parametrize(
@@ -1890,21 +1925,20 @@ class TestExternalDatasetServiceFetchRetrieval:
         ],
     )
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_various_error_status_codes(
-        self, mock_db, mock_process, factory: ExternalDatasetServiceTestDataFactory, status_code, error_message
+        self,
+        mock_process,
+        factory: ExternalDatasetServiceTestDataFactory,
+        sqlite_session: Session,
+        status_code,
+        error_message,
     ):
         """Test that various error status codes raise exceptions with response text."""
         # Arrange
         tenant_id = "tenant-123"
         dataset_id = "dataset-123"
 
-        binding = factory.create_external_knowledge_binding_mock(
-            dataset_id=dataset_id, external_knowledge_api_id="api-123"
-        )
-        api = factory.create_external_knowledge_api_mock(api_id="api-123")
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session, tenant_id=tenant_id, dataset_id=dataset_id)
 
         mock_response = MagicMock()
         mock_response.status_code = status_code
@@ -1914,20 +1948,16 @@ class TestExternalDatasetServiceFetchRetrieval:
         # Act & Assert
         with pytest.raises(ExternalKnowledgeRetrievalError, match=re.escape(error_message)):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                tenant_id, dataset_id, "query", {"top_k": 5}, session=mock_db.session
+                tenant_id, dataset_id, "query", {"top_k": 5}, session=sqlite_session
             )
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
     def test_fetch_external_knowledge_retrieval_empty_response_text(
-        self, mock_db, mock_process, factory: ExternalDatasetServiceTestDataFactory
+        self, mock_process, factory: ExternalDatasetServiceTestDataFactory, sqlite_session: Session
     ):
         """Test exception with empty response text."""
         # Arrange
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 503
@@ -1937,21 +1967,15 @@ class TestExternalDatasetServiceFetchRetrieval:
         # Act & Assert
         with pytest.raises(ExternalKnowledgeRetrievalError):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123",
-                "dataset-123",
-                "query",
-                {"top_k": 5},
-                session=mock_db.session,
+                "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
             )
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
-    def test_fetch_external_knowledge_retrieval_invalid_json_response(self, mock_db, mock_process, factory):
+    def test_fetch_external_knowledge_retrieval_invalid_json_response(
+        self, mock_process, factory, sqlite_session: Session
+    ):
         """Test malformed JSON success responses are normalized to external retrieval errors."""
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 200
@@ -1960,21 +1984,15 @@ class TestExternalDatasetServiceFetchRetrieval:
 
         with pytest.raises(ExternalKnowledgeRetrievalError, match="invalid external knowledge response"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123",
-                "dataset-123",
-                "query",
-                {"top_k": 5},
-                session=mock_db.session,
+                "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
             )
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
-    def test_fetch_external_knowledge_retrieval_invalid_success_payload_shape(self, mock_db, mock_process, factory):
+    def test_fetch_external_knowledge_retrieval_invalid_success_payload_shape(
+        self, mock_process, factory, sqlite_session: Session
+    ):
         """Test malformed success payload shapes are normalized to external retrieval errors."""
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 200
@@ -1983,21 +2001,15 @@ class TestExternalDatasetServiceFetchRetrieval:
 
         with pytest.raises(ExternalKnowledgeRetrievalError, match="invalid external knowledge response"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123",
-                "dataset-123",
-                "query",
-                {"top_k": 5},
-                session=mock_db.session,
+                "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
             )
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
-    def test_fetch_external_knowledge_retrieval_invalid_records_shape(self, mock_db, mock_process, factory):
+    def test_fetch_external_knowledge_retrieval_invalid_records_shape(
+        self, mock_process, factory, sqlite_session: Session
+    ):
         """Test non-list records payloads are normalized to external retrieval errors."""
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
 
         mock_response = MagicMock()
         mock_response.status_code = 200
@@ -2006,28 +2018,18 @@ class TestExternalDatasetServiceFetchRetrieval:
 
         with pytest.raises(ExternalKnowledgeRetrievalError, match="invalid external knowledge response"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123",
-                "dataset-123",
-                "query",
-                {"top_k": 5},
-                session=mock_db.session,
+                "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
             )
 
     @patch("services.external_knowledge_service.ExternalDatasetService.process_external_api")
-    @patch("services.external_knowledge_service.db")
-    def test_fetch_external_knowledge_retrieval_wraps_transport_errors(self, mock_db, mock_process, factory):
+    def test_fetch_external_knowledge_retrieval_wraps_transport_errors(
+        self, mock_process, factory, sqlite_session: Session
+    ):
         """Test transport/runtime failures are normalized to external retrieval errors."""
-        binding = factory.create_external_knowledge_binding_mock()
-        api = factory.create_external_knowledge_api_mock()
-
-        mock_db.session.scalar.side_effect = [binding, api]
+        _seed_external_retrieval_dependencies(sqlite_session)
         mock_process.side_effect = RuntimeError("connection reset by peer")
 
         with pytest.raises(ExternalKnowledgeRetrievalError, match="connection reset by peer"):
             ExternalDatasetService.fetch_external_knowledge_retrieval(
-                "tenant-123",
-                "dataset-123",
-                "query",
-                {"top_k": 5},
-                session=mock_db.session,
+                "tenant-123", "dataset-123", "query", {"top_k": 5}, session=sqlite_session
             )
diff --git a/packages/contracts/generated/api/console/datasets/zod.gen.ts b/packages/contracts/generated/api/console/datasets/zod.gen.ts
index b809480eb51..21d6c55ac3c 100644
--- a/packages/contracts/generated/api/console/datasets/zod.gen.ts
+++ b/packages/contracts/generated/api/console/datasets/zod.gen.ts
@@ -44,6 +44,13 @@ export const zBatchImportPayload = z.object({
 
 /**
  * ExternalDatasetCreatePayload
+ *
+ * Validated fields required to create an external dataset binding.
+ *
+ * The console controller owns HTTP concerns, but the service also needs this
+ * contract when creating the tenant-scoped dataset and external knowledge
+ * binding. Keep it outside controllers so service imports do not depend on
+ * Flask blueprint initialization.
  */
 export const zExternalDatasetCreatePayload = z.object({
   description: z.string().max(400).nullish(),