diff --git a/dataconnect/__init__.py b/dataconnect/__init__.py
index 3907ded..f9b6f02 100644
--- a/dataconnect/__init__.py
+++ b/dataconnect/__init__.py
@@ -3,21 +3,31 @@
 from __future__ import annotations
 
 from dataconnect.client import DataConnectClient
-from dataconnect.exceptions import AuthenticationError, ConnectionError, DataConnectError
-from dataconnect.models import (
-    Dataset,
-    DatasetVersion,
-    Study,
-    StudyEnvironment,
+from dataconnect.exceptions import (
+    AuthenticationError,
+    AuthorizationError,
+    ConnectionError,
+    DataConnectError,
+    NotFoundError,
+    QueryError,
+    ServerError,
+    ValidationError,
 )
+from dataconnect.models import Study, StudyEnvironment
 
 __all__ = [
+    # Client
     "DataConnectClient",
+    # Domain models
     "Study",
     "StudyEnvironment",
-    "Dataset",
-    "DatasetVersion",
+    # Exceptions — catch these in user application code
     "DataConnectError",
     "ConnectionError",
     "AuthenticationError",
+    "AuthorizationError",
+    "NotFoundError",
+    "QueryError",
+    "ServerError",
+    "ValidationError",
 ]
diff --git a/dataconnect/client.py b/dataconnect/client.py
index 9a6f270..84ce20d 100644
--- a/dataconnect/client.py
+++ b/dataconnect/client.py
@@ -1,26 +1,16 @@
-"""Public API for the DataConnect client library."""
+"""Public API for the DataConnect client library.
+
+``DataConnectClient`` is a thin façade over ``DataConnectService``.
+The ``connect()`` class method is the composition root — the only place in
+the SDK where concrete implementation types are wired together.
+"""
 
 from __future__ import annotations
 
-import json
 from types import TracebackType
-from typing import Any
-
-import pyarrow as pa
-
-from dataconnect import _encoding
-from dataconnect.auth import BearerTokenAuth
-from dataconnect.framework.pyarrow_transport import PyArrowFlightTransport
-from dataconnect.framework.transport import FlightTransport
-from dataconnect.models import Dataset, Study
 
-# Flight actions / commands
-_ACTION_LIST_STUDIES = "studies.list"
-_ACTION_LIST_DATASETS = "datasets.list"
-_ACTION_LIST_DATASET_VERSIONS = "dataset_versions.list"
-_ACTION_FETCH_TICKET = "data.fetch_ticket"
-_CMD_PUBLISH = "publish"
-_CMD_DRY_PUBLISH = "dry_publish"
+from dataconnect.models import Study
+from dataconnect.service import DataConnectService, DefaultDataConnectService
 
 _DEFAULT_HOST = "enodia-gateway.platform.imedidata.com"
 _DEFAULT_PORT = 443
@@ -29,9 +19,9 @@
 class DataConnectClient:
     """Client for interacting with DataConnect services."""
 
-    def __init__(self, transport: FlightTransport) -> None:
-        """Initialize the DataConnect client with a specified transport."""
-        self._transport = transport
+    def __init__(self, service: DataConnectService) -> None:
+        """Initialize the client with an injected service implementation."""
+        self._service = service
 
     @classmethod
     def connect(
@@ -41,37 +31,26 @@ def connect(
         use_tls: bool = True,
         token: str = "",
     ) -> DataConnectClient:
-        """Open connection to a Flight server."""
-        location = f"grpc+tls://{host}:{port}"
-        transport = PyArrowFlightTransport(
-            location=location,
-            credentials=BearerTokenAuth(token),
-        )
-        return cls(transport)
-
-    def studies(self) -> list[Study]:
+
+        # Import is deferred so pyarrow.flight is only loaded when this factory
+        # is called — callers injecting a custom transport are unaffected.
+        from dataconnect.transport.arrow_flight.transport import ArrowFlightTransport
+
+        transport = ArrowFlightTransport(host=host, port=port, use_tls=use_tls, token=token)
+
+        return cls(DefaultDataConnectService(transport))
+
+    # Public API
+
+    def get_studies(self) -> list[Study]:
         """List the studies the client is authorized to access."""
-        rows = self._action_json(_ACTION_LIST_STUDIES, None)
-        return [Study(**r) for r in rows]
-
-    def datasets(self, study_uuid: str) -> list[Dataset]:
-        """List the datasets available for a given study."""
-        body = {"study_uuid": study_uuid}
-        rows = self._action_json(_ACTION_LIST_DATASETS, {"study_uuid": body})
-        return [Dataset(**r) for r in rows]
-
-    def fetch_data(self, dataset_uuid: str) -> pa.Table:
-        """Fetch the data for a given dataset as a PyArrow Table."""
-        body = {"dataset_uuid": dataset_uuid}
-        results = self._transport.do_action(_ACTION_FETCH_TICKET, _encoding.dumps(body))
-        if not results:
-            raise RuntimeError("Server returned no data for the fetch_data action.")
-        return self._transport.do_get(results).read_all()
+        return self._service.get_studies()
 
     # Lifecycle
+
     def close(self) -> None:
-        """Close the underlying transport connection."""
-        self._transport.close()
+        """Close the underlying connection."""
+        self._service.close()
 
     def __enter__(self) -> DataConnectClient:
         return self
@@ -82,12 +61,4 @@ def __exit__(
         exc_value: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        self._transport.close()
-
-    # Helpers
-    def _action_json(self, action: str, body: dict[str, Any] | None) -> Any:
-        """Execute a Flight action and return the result as JSON."""
-        results = self._transport.do_action(action, _encoding.dumps(body or {}))
-        if not results:
-            return []
-        return json.loads(results.decode("utf-8"))
+        self._service.close()
diff --git a/dataconnect/exceptions.py b/dataconnect/exceptions.py
index b686ce7..299f016 100644
--- a/dataconnect/exceptions.py
+++ b/dataconnect/exceptions.py
@@ -1,4 +1,16 @@
-"""Public exceptions for DataConnect."""
+"""Public exceptions for DataConnect.
+
+Hierarchy
+---------
+DataConnectError
+├── ConnectionError        — unable to reach server
+├── AuthenticationError    — authentication failure from server
+├── AuthorizationError     — authorization failure from server
+├── NotFoundError          — requested resource does not exist
+├── QueryError             — server rejected query / stream read failure
+├── ServerError            — unexpected server-side error
+└── ValidationError        — server response was malformed or unexpected
+"""
 
 from __future__ import annotations
 
@@ -8,12 +20,32 @@ class DataConnectError(Exception):
 
 
 class ConnectionError(DataConnectError):
-    """Error connecting to the DataConnect server."""
+    """Unable to establish or maintain a connection to the server."""
 
 
 class AuthenticationError(DataConnectError):
-    """Error authenticating with the DataConnect server."""
+    """Authentication failure."""
+
+
+class AuthorizationError(DataConnectError):
+    """Authorization failure."""
+
+
+class NotFoundError(DataConnectError):
+    """The requested resource (study, dataset, etc.) was not found."""
 
 
 class QueryError(DataConnectError):
-    """Error executing a query."""
+    """The server rejected the query or a data-stream read failed."""
+
+
+class ServerError(DataConnectError):
+    """Unexpected server-side error."""
+
+    def __init__(self, message: str, status_code: int = 0) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class ValidationError(DataConnectError):
+    """Server returned data in an unexpected or invalid format."""
diff --git a/dataconnect/models.py b/dataconnect/models.py
index c003133..c429250 100644
--- a/dataconnect/models.py
+++ b/dataconnect/models.py
@@ -1,28 +1,17 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-
-
-@dataclass(frozen=True)
-class Study:
-    id: str
-    name: str
+from dataclasses import dataclass, field
+from uuid import UUID
 
 
 @dataclass(frozen=True)
 class StudyEnvironment:
-    """Environment variables for a study."""
-
-
-@dataclass(frozen=True)
-class Dataset:
-    id: str
-    study_id: str
+    uuid: UUID
     name: str
 
 
 @dataclass(frozen=True)
-class DatasetVersion:
-    id: str
-    dataset_id: str
+class Study:
+    uuid: UUID
     name: str
+    environments: list[StudyEnvironment] = field(default_factory=list)
diff --git a/dataconnect/service/__init__.py b/dataconnect/service/__init__.py
new file mode 100644
index 0000000..8f66da5
--- /dev/null
+++ b/dataconnect/service/__init__.py
@@ -0,0 +1,9 @@
+"""Service layer — public symbols re-exported for import convenience."""
+
+from dataconnect.service.base import DataConnectService
+from dataconnect.service.default import DefaultDataConnectService
+
+__all__ = [
+    "DataConnectService",
+    "DefaultDataConnectService",
+]
diff --git a/dataconnect/service/base.py b/dataconnect/service/base.py
new file mode 100644
index 0000000..78a26b0
--- /dev/null
+++ b/dataconnect/service/base.py
@@ -0,0 +1,17 @@
+"""Abstract service interface for DataConnect."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from dataconnect.models import Study
+
+
+class DataConnectService(ABC):
+    """Abstract service interface — defines all operations available to the client."""
+
+    @abstractmethod
+    def get_studies(self) -> list[Study]: ...
+
+    @abstractmethod
+    def close(self) -> None: ...
diff --git a/dataconnect/service/default.py b/dataconnect/service/default.py
new file mode 100644
index 0000000..2135b89
--- /dev/null
+++ b/dataconnect/service/default.py
@@ -0,0 +1,80 @@
+"""Default service implementation — domain logic, encoding, and error translation."""
+
+from __future__ import annotations
+
+from dataconnect.exceptions import (
+    AuthenticationError,
+    AuthorizationError,
+    ConnectionError,
+    DataConnectError,
+    NotFoundError,
+    QueryError,
+    ServerError,
+    ValidationError,
+)
+from dataconnect.models import Study
+from dataconnect.service.base import DataConnectService
+from dataconnect.service.mappers import resource_to_study
+from dataconnect.transport.base import Transport
+from dataconnect.transport.errors import (
+    TransportAuthenticationError,
+    TransportAuthorizationError,
+    TransportConnectionError,
+    TransportError,
+    TransportIOError,
+    TransportNotFoundError,
+    TransportStatusError,
+)
+from dataconnect.transport.models import ResourceQuery
+
+# Server action identifiers
+_ACTION_LIST_STUDIES = "studies.list"
+
+
+def _translate_error(ex: TransportError) -> DataConnectError:
+    """Map a ``TransportError`` to the appropriate public ``DataConnectError``."""
+
+    if isinstance(ex, TransportAuthenticationError):
+        return AuthenticationError(str(ex))
+    if isinstance(ex, TransportAuthorizationError):
+        return AuthorizationError(str(ex))
+    if isinstance(ex, TransportNotFoundError):
+        return NotFoundError(str(ex))
+    if isinstance(ex, TransportStatusError):
+        return ServerError(str(ex), status_code=ex.status_code)
+    if isinstance(ex, TransportConnectionError):
+        return ConnectionError(str(ex))
+    if isinstance(ex, TransportIOError):
+        return QueryError(str(ex))
+
+    return ServerError(str(ex))
+
+
+class DefaultDataConnectService(DataConnectService):
+    """Concrete service injected with an abstract ``Transport``."""
+
+    def __init__(self, transport: Transport) -> None:
+        self._transport = transport
+
+    # DataConnectService
+
+    def get_studies(self) -> list[Study]:
+
+        request = ResourceQuery(action=_ACTION_LIST_STUDIES)
+
+        try:
+            resources = self._transport.list_resources(request)
+        except TransportError as ex:
+            raise _translate_error(ex) from ex
+
+        try:
+            return [resource_to_study(r) for r in resources]
+        except (KeyError, TypeError, ValueError) as ex:
+            raise ValidationError(f"Unexpected studies response format: {ex}") from ex
+
+    def close(self) -> None:
+
+        try:
+            self._transport.close()
+        except TransportError as ex:
+            raise ConnectionError(str(ex)) from ex
diff --git a/dataconnect/service/mappers.py b/dataconnect/service/mappers.py
new file mode 100644
index 0000000..00eb9c2
--- /dev/null
+++ b/dataconnect/service/mappers.py
@@ -0,0 +1,26 @@
+"""Resource → domain model mappers.
+
+Each function takes a transport-layer ``ResourceInfo`` and returns a public
+domain model.  All wire-format knowledge (JSON encoding, field names, byte
+decoding) is isolated here.
+"""
+
+from __future__ import annotations
+
+import json
+from uuid import UUID
+
+from dataconnect.models import Study, StudyEnvironment
+from dataconnect.transport.models import ResourceInfo
+
+
+def resource_to_study(resource: ResourceInfo) -> Study:
+    """Parse a transport-layer ``ResourceInfo`` into a ``Study`` domain object."""
+
+    data = json.loads(resource.endpoints[0].ticket.decode("utf-8"))
+
+    return Study(
+        uuid=UUID(data["uuid"]),
+        name=data["name"],
+        environments=[StudyEnvironment(uuid=UUID(e["uuid"]), name=e["name"]) for e in data.get("environments", [])],
+    )
diff --git a/dataconnect/transport/__init__.py b/dataconnect/transport/__init__.py
new file mode 100644
index 0000000..3402521
--- /dev/null
+++ b/dataconnect/transport/__init__.py
@@ -0,0 +1,16 @@
+"""Transport layer — public exports.
+
+Transport errors (``transport/errors.py``) are intentionally NOT re-exported
+here. They are internal to the transport layer and must not be caught by
+user-facing code.
+"""
+
+from dataconnect.transport.base import Transport
+from dataconnect.transport.models import DataRef, ResourceInfo, ResourceQuery
+
+__all__ = [
+    "Transport",
+    "ResourceQuery",
+    "ResourceInfo",
+    "DataRef",
+]
diff --git a/dataconnect/transport/arrow_flight/__init__.py b/dataconnect/transport/arrow_flight/__init__.py
new file mode 100644
index 0000000..e666446
--- /dev/null
+++ b/dataconnect/transport/arrow_flight/__init__.py
@@ -0,0 +1,3 @@
+from dataconnect.transport.arrow_flight.transport import ArrowFlightTransport
+
+__all__ = ["ArrowFlightTransport"]
diff --git a/dataconnect/transport/arrow_flight/transport.py b/dataconnect/transport/arrow_flight/transport.py
new file mode 100644
index 0000000..8fae243
--- /dev/null
+++ b/dataconnect/transport/arrow_flight/transport.py
@@ -0,0 +1,103 @@
+"""Arrow Flight implementation of the Transport interface.
+
+This is the ONLY file in the SDK that imports ``pyarrow.flight``.
+All pyarrow Flight exceptions are caught here and translated into
+technology-agnostic ``TransportError`` subtypes before propagating up.
+"""
+
+from __future__ import annotations
+
+import json
+
+from pyarrow import flight
+
+from dataconnect.transport.base import Transport
+from dataconnect.transport.errors import (
+    TransportAuthenticationError,
+    TransportAuthorizationError,
+    TransportConnectionError,
+    TransportStatusError,
+)
+from dataconnect.transport.models import DataRef, ResourceInfo, ResourceQuery
+
+
+def _to_resource_info(info: flight.FlightInfo) -> ResourceInfo:
+    """Convert a pyarrow ``FlightInfo`` to a technology-agnostic ``ResourceInfo``."""
+
+    descriptor_bytes = info.descriptor.command if info.descriptor else b""
+    endpoints = [DataRef(ticket=e.ticket.ticket) for e in info.endpoints]
+    schema_bytes = info.schema.serialize().to_pybytes()
+
+    return ResourceInfo(
+        descriptor=descriptor_bytes,
+        endpoints=endpoints,
+        schema_bytes=schema_bytes,
+        total_records=info.total_records,
+    )
+
+
+# Maps service-layer action names to the flight_type value the Arrow Flight server expects.
+_ACTION_FLIGHT_TYPE: dict[str, str] = {
+    "studies.list": "STUDIES",
+}
+
+
+class ArrowFlightTransport(Transport):
+    """Flight transport implementation using pyarrow (default implementation)."""
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        use_tls: bool,
+        token: str = "",
+    ) -> None:
+        self._call_headers: list[tuple[bytes, bytes]] = []
+
+        scheme = "grpc+tls" if use_tls else "grpc"
+        location = f"{scheme}://{host}:{port}"
+
+        try:
+            tls_root_certs = None  # pending
+            self._client = flight.FlightClient(location, tls_root_certs=tls_root_certs)
+        except Exception as exc:
+            raise TransportConnectionError(f"Failed to connect to {location}: {exc}") from exc
+
+        if token:
+            self._call_headers.append((b"authorization", f"Bearer {token}".encode()))
+
+    def _options(self) -> flight.FlightCallOptions:
+        return flight.FlightCallOptions(headers=self._call_headers)
+
+    # Transport
+
+    def list_resources(self, request: ResourceQuery) -> list[ResourceInfo]:
+        """Translate the action name to Arrow Flight criteria and return resource records."""
+
+        flight_type = _ACTION_FLIGHT_TYPE.get(request.action)
+
+        if flight_type is None:
+            raise TransportConnectionError(f"Unknown action: {request.action!r}")
+
+        body = json.loads(request.body) if request.body else {}
+        criteria = json.dumps({**body, "flight_type": flight_type}, separators=(",", ":")).encode("utf-8")
+
+        try:
+            raw_flights = self._client.list_flights(criteria, self._options())
+            return [_to_resource_info(f) for f in raw_flights]
+
+        except flight.FlightUnauthenticatedError as ex:
+            raise TransportAuthenticationError(str(ex)) from ex
+        except flight.FlightUnauthorizedError as ex:
+            raise TransportAuthorizationError(str(ex)) from ex
+        except flight.FlightUnavailableError as ex:
+            raise TransportConnectionError(str(ex)) from ex
+        except flight.FlightInternalError as ex:
+            raise TransportStatusError(str(ex), status_code=13, grpc_status="INTERNAL") from ex
+        except flight.FlightError as ex:
+            raise TransportConnectionError(str(ex)) from ex
+        except Exception as ex:
+            raise TransportConnectionError(f"Unexpected error during list_resources: {ex}") from ex
+
+    def close(self) -> None:
+        self._client.close()
diff --git a/dataconnect/transport/base.py b/dataconnect/transport/base.py
new file mode 100644
index 0000000..df12c50
--- /dev/null
+++ b/dataconnect/transport/base.py
@@ -0,0 +1,28 @@
+"""Abstract Transport interface — technology-agnostic.
+
+No pyarrow imports here.
+
+Layer 3 deals only with transport-level DTOs defined in ``transport/models.py``.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from dataconnect.transport.models import ResourceInfo, ResourceQuery
+
+
+class Transport(ABC):
+    """Minimal abstract transport for DataConnect operations."""
+
+    @abstractmethod
+    def list_resources(self, request: ResourceQuery) -> list[ResourceInfo]:
+        """List available data resources matching the given query.
+
+        The transport does not interpret the action name or body — that is the
+        service layer's responsibility.
+        """
+
+    @abstractmethod
+    def close(self) -> None:
+        """Close the transport connection."""
diff --git a/dataconnect/transport/errors.py b/dataconnect/transport/errors.py
new file mode 100644
index 0000000..235c435
--- /dev/null
+++ b/dataconnect/transport/errors.py
@@ -0,0 +1,45 @@
+"""Internal transport-layer exceptions.
+
+These are NEVER re-exported publicly.
+
+The service layer catches them and translates them into
+public ``DataConnectError`` subtypes.
+"""
+
+from __future__ import annotations
+
+
+class TransportError(Exception):
+    """Base class for all transport-layer errors."""
+
+
+class TransportConnectionError(TransportError):
+    """Raised when a connection to the server cannot be established."""
+
+
+class TransportAuthenticationError(TransportError):
+    """Raised on authentication failures."""
+
+
+class TransportAuthorizationError(TransportError):
+    """Raised on authorization failures."""
+
+
+class TransportStatusError(TransportError):
+    """Raised when the server returns an explicit error status."""
+
+    def __init__(self, message: str, status_code: int, grpc_status: str = "") -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.grpc_status = grpc_status
+
+
+class TransportNotFoundError(TransportStatusError):
+    """Raised when the server returns not-found response."""
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message, status_code=5, grpc_status="NOT_FOUND")
+
+
+class TransportIOError(TransportError):
+    """Raised when reading from or writing to a data stream fails."""
diff --git a/dataconnect/transport/models.py b/dataconnect/transport/models.py
new file mode 100644
index 0000000..65ea576
--- /dev/null
+++ b/dataconnect/transport/models.py
@@ -0,0 +1,40 @@
+"""Transport-layer DTOs — technology-agnostic contract."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class ResourceQuery:
+    """An outbound request to query available resources."""
+
+    action: str
+    body: str = field(default="")
+
+    def append_body(self, extra: dict[str, Any]) -> ResourceQuery:
+        """Return a new ResourceQuery with extra fields merged into the JSON body."""
+
+        body_dict = json.loads(self.body) if self.body else {}
+        merged_body = {**body_dict, **extra}
+
+        return ResourceQuery(action=self.action, body=json.dumps(merged_body, separators=(",", ":")))
+
+
+@dataclass(frozen=True)
+class DataRef:
+    """An opaque server-side reference to a data stream."""
+
+    ticket: bytes
+
+
+@dataclass(frozen=True)
+class ResourceInfo:
+    """Technology-agnostic representation of a single resource."""
+
+    descriptor: bytes
+    endpoints: list[DataRef]
+    total_records: int
+    schema_bytes: bytes