diff --git a/obd_ai/__init__.py b/obd_ai/__init__.py
new file mode 100644
index 00000000..80d1b8d4
--- /dev/null
+++ b/obd_ai/__init__.py
@@ -0,0 +1,16 @@
+"""Safe, high-level interfaces for LLM/MCP-style OBD access."""
+
+from .catalog import (
+    ApprovedCommand,
+    ApprovedCommandCatalog,
+    DEFAULT_APPROVED_COMMAND_CATALOG,
+)
+from .session import OBDAISession, OBDAISessionManager
+
+__all__ = [
+    "ApprovedCommand",
+    "ApprovedCommandCatalog",
+    "DEFAULT_APPROVED_COMMAND_CATALOG",
+    "OBDAISession",
+    "OBDAISessionManager",
+]
diff --git a/obd_ai/catalog.py b/obd_ai/catalog.py
new file mode 100644
index 00000000..b3ad7e72
--- /dev/null
+++ b/obd_ai/catalog.py
@@ -0,0 +1,183 @@
+"""Approved read-only command catalog for the OBD AI layer."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Iterable, Mapping, Tuple
+
+import obd
+from obd.OBDCommand import OBDCommand
+
+
+@dataclass(frozen=True)
+class ApprovedCommand:
+    """Metadata for a command that is safe to expose to the OBD AI layer."""
+
+    key: str
+    public_name: str
+    category: str
+    description: str
+    obd_command_name: str
+    obd_command: OBDCommand
+
+
+@dataclass(frozen=True)
+class _CommandDefinition:
+    key: str
+    public_name: str
+    category: str
+    description: str
+    obd_command_name: str
+
+
+class ApprovedCommandCatalog:
+    """Read-only catalog that maps public command keys to OBD commands."""
+
+    def __init__(self, definitions: Iterable[_CommandDefinition]):
+        mutable_map = {}
+        ordered = []
+
+        for definition in definitions:
+            if definition.key in mutable_map:
+                raise ValueError(f"Duplicate approved command key: {definition.key}")
+
+            try:
+                obd_command = getattr(obd.commands, definition.obd_command_name)
+            except AttributeError as exc:
+                raise ValueError(
+                    f"Unknown obd command mapping: {definition.obd_command_name}"
+                ) from exc
+
+            approved = ApprovedCommand(
+                key=definition.key,
+                public_name=definition.public_name,
+                category=definition.category,
+                description=definition.description,
+                obd_command_name=definition.obd_command_name,
+                obd_command=obd_command,
+            )
+
+            mutable_map[approved.key] = approved
+            ordered.append(approved)
+
+        self._commands: Mapping[str, ApprovedCommand] = MappingProxyType(mutable_map)
+        self._ordered_commands: Tuple[ApprovedCommand, ...] = tuple(ordered)
+
+    def __contains__(self, key: str) -> bool:
+        return key in self._commands
+
+    def __len__(self) -> int:
+        return len(self._ordered_commands)
+
+    def get(self, key: str) -> ApprovedCommand:
+        try:
+            return self._commands[key]
+        except KeyError as exc:
+            raise KeyError(f"Unapproved command key: {key}") from exc
+
+    def keys(self) -> Tuple[str, ...]:
+        return tuple(command.key for command in self._ordered_commands)
+
+    def list(self) -> Tuple[ApprovedCommand, ...]:
+        return self._ordered_commands
+
+    def as_mapping(self) -> Mapping[str, ApprovedCommand]:
+        return self._commands
+
+
+_MVP_READ_ONLY_COMMANDS = (
+    _CommandDefinition(
+        key="status_since_dtc_clear",
+        public_name="Status Since DTC Clear",
+        category="diagnostics",
+        description="Read current monitor status and MIL flags.",
+        obd_command_name="STATUS",
+    ),
+    _CommandDefinition(
+        key="engine_rpm",
+        public_name="Engine RPM",
+        category="powertrain",
+        description="Read current engine speed in revolutions per minute.",
+        obd_command_name="RPM",
+    ),
+    _CommandDefinition(
+        key="vehicle_speed",
+        public_name="Vehicle Speed",
+        category="powertrain",
+        description="Read current vehicle speed.",
+        obd_command_name="SPEED",
+    ),
+    _CommandDefinition(
+        key="engine_coolant_temperature",
+        public_name="Engine Coolant Temperature",
+        category="powertrain",
+        description="Read current coolant temperature.",
+        obd_command_name="COOLANT_TEMP",
+    ),
+    _CommandDefinition(
+        key="intake_air_temperature",
+        public_name="Intake Air Temperature",
+        category="powertrain",
+        description="Read current intake air temperature.",
+        obd_command_name="INTAKE_TEMP",
+    ),
+    _CommandDefinition(
+        key="calculated_engine_load",
+        public_name="Calculated Engine Load",
+        category="powertrain",
+        description="Read current calculated engine load percentage.",
+        obd_command_name="ENGINE_LOAD",
+    ),
+    _CommandDefinition(
+        key="throttle_position",
+        public_name="Throttle Position",
+        category="powertrain",
+        description="Read current throttle position percentage.",
+        obd_command_name="THROTTLE_POS",
+    ),
+    _CommandDefinition(
+        key="fuel_level",
+        public_name="Fuel Level",
+        category="fuel",
+        description="Read current fuel level percentage.",
+        obd_command_name="FUEL_LEVEL",
+    ),
+    _CommandDefinition(
+        key="control_module_voltage",
+        public_name="Control Module Voltage",
+        category="electrical",
+        description="Read current ECU/control module voltage.",
+        obd_command_name="CONTROL_MODULE_VOLTAGE",
+    ),
+    _CommandDefinition(
+        key="engine_run_time",
+        public_name="Engine Run Time",
+        category="diagnostics",
+        description="Read total engine run time since startup.",
+        obd_command_name="RUN_TIME",
+    ),
+    _CommandDefinition(
+        key="stored_trouble_codes",
+        public_name="Stored Trouble Codes",
+        category="diagnostics",
+        description="Read currently stored diagnostic trouble codes.",
+        obd_command_name="GET_DTC",
+    ),
+    _CommandDefinition(
+        key="pending_trouble_codes",
+        public_name="Pending Trouble Codes",
+        category="diagnostics",
+        description="Read pending trouble codes from current drive cycle.",
+        obd_command_name="GET_CURRENT_DTC",
+    ),
+    _CommandDefinition(
+        key="vehicle_identification_number",
+        public_name="Vehicle Identification Number",
+        category="vehicle_info",
+        description="Read the vehicle identification number (VIN).",
+        obd_command_name="VIN",
+    ),
+)
+
+DEFAULT_APPROVED_COMMAND_CATALOG = ApprovedCommandCatalog(_MVP_READ_ONLY_COMMANDS)
diff --git a/obd_ai/session.py b/obd_ai/session.py
new file mode 100644
index 00000000..b113e194
--- /dev/null
+++ b/obd_ai/session.py
@@ -0,0 +1,86 @@
+"""Session abstractions for the OBD AI layer."""
+
+from __future__ import annotations
+
+from typing import Callable, Optional
+
+import obd
+from obd.OBDResponse import OBDResponse
+from obd.utils import OBDStatus
+
+from .catalog import ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG
+
+OBDFactory = Callable[..., obd.OBD]
+
+
+class OBDAISession:
+    """Safe wrapper around an OBD connection constrained by an approved catalog."""
+
+    def __init__(self, connection: obd.OBD, catalog: ApprovedCommandCatalog):
+        self._connection = connection
+        self._catalog = catalog
+
+    def __enter__(self) -> "OBDAISession":
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        self.close()
+
+    def list_approved_commands(self):
+        return self._catalog.list()
+
+    def query(self, approved_key: str) -> OBDResponse:
+        approved = self._catalog.get(approved_key)
+        return self._connection.query(approved.obd_command)
+
+    def supports(self, approved_key: str) -> bool:
+        approved = self._catalog.get(approved_key)
+        return self._connection.supports(approved.obd_command)
+
+    def status(self) -> OBDStatus:
+        return self._connection.status()
+
+    def is_connected(self) -> bool:
+        return self._connection.is_connected()
+
+    def port_name(self) -> str:
+        return self._connection.port_name()
+
+    def close(self) -> None:
+        self._connection.close()
+
+
+class OBDAISessionManager:
+    """Factory for creating OBD AI sessions over `obd.OBD`."""
+
+    def __init__(
+        self,
+        catalog: ApprovedCommandCatalog = DEFAULT_APPROVED_COMMAND_CATALOG,
+        obd_factory: Optional[OBDFactory] = None,
+    ):
+        self._catalog = catalog
+        self._obd_factory = obd_factory or obd.OBD
+
+    def open_session(
+        self,
+        portstr=None,
+        baudrate=None,
+        protocol=None,
+        fast: bool = True,
+        timeout: float = 0.1,
+        check_voltage: bool = True,
+        start_low_power: bool = False,
+    ) -> OBDAISession:
+        connection = self._obd_factory(
+            portstr=portstr,
+            baudrate=baudrate,
+            protocol=protocol,
+            fast=fast,
+            timeout=timeout,
+            check_voltage=check_voltage,
+            start_low_power=start_low_power,
+        )
+        return OBDAISession(connection=connection, catalog=self._catalog)
+
+    def wrap_existing(self, connection: obd.OBD) -> OBDAISession:
+        return OBDAISession(connection=connection, catalog=self._catalog)
diff --git a/tests/test_obd_ai_catalog.py b/tests/test_obd_ai_catalog.py
new file mode 100644
index 00000000..5edddf63
--- /dev/null
+++ b/tests/test_obd_ai_catalog.py
@@ -0,0 +1,35 @@
+import pytest
+
+import obd
+from obd_ai.catalog import DEFAULT_APPROVED_COMMAND_CATALOG
+
+
+def test_default_catalog_maps_to_real_obd_commands():
+    catalog = DEFAULT_APPROVED_COMMAND_CATALOG
+
+    rpm = catalog.get("engine_rpm")
+    assert rpm.public_name == "Engine RPM"
+    assert rpm.category == "powertrain"
+    assert rpm.obd_command_name == "RPM"
+    assert rpm.obd_command is obd.commands.RPM
+
+
+def test_default_catalog_excludes_write_commands_for_mvp_safety():
+    catalog = DEFAULT_APPROVED_COMMAND_CATALOG
+
+    assert "stored_trouble_codes" in catalog
+    assert "pending_trouble_codes" in catalog
+    assert "clear_trouble_codes" not in catalog
+    assert "CLEAR_DTC" not in catalog
+
+
+def test_catalog_unknown_key_raises_key_error():
+    with pytest.raises(KeyError, match="Unapproved command key"):
+        DEFAULT_APPROVED_COMMAND_CATALOG.get("clear_trouble_codes")
+
+
+def test_catalog_mapping_is_read_only():
+    mapping = DEFAULT_APPROVED_COMMAND_CATALOG.as_mapping()
+
+    with pytest.raises(TypeError):
+        mapping["new"] = "value"
diff --git a/tests/test_obd_ai_session.py b/tests/test_obd_ai_session.py
new file mode 100644
index 00000000..712d91b4
--- /dev/null
+++ b/tests/test_obd_ai_session.py
@@ -0,0 +1,90 @@
+import pytest
+
+import obd
+from obd.utils import OBDStatus
+
+from obd_ai.session import OBDAISessionManager
+
+
+class FakeConnection:
+    def __init__(self):
+        self.query_calls = []
+        self.closed = False
+
+    def query(self, command, force=False):
+        self.query_calls.append((command, force))
+        return {"command": command.name, "force": force}
+
+    @staticmethod
+    def status():
+        return OBDStatus.CAR_CONNECTED
+
+    @staticmethod
+    def is_connected():
+        return True
+
+    @staticmethod
+    def port_name():
+        return "FAKEPORT"
+
+    @staticmethod
+    def supports(command):
+        return command is obd.commands.RPM
+
+    def close(self):
+        self.closed = True
+
+
+class FakeFactory:
+    def __init__(self):
+        self.last_kwargs = None
+        self.connection = FakeConnection()
+
+    def __call__(self, **kwargs):
+        self.last_kwargs = kwargs
+        return self.connection
+
+
+def test_manager_builds_session_with_expected_obd_init_args():
+    factory = FakeFactory()
+    manager = OBDAISessionManager(obd_factory=factory)
+
+    manager.open_session(portstr="/dev/ttyUSB9", fast=False, timeout=0.25)
+
+    assert factory.last_kwargs["portstr"] == "/dev/ttyUSB9"
+    assert factory.last_kwargs["fast"] is False
+    assert factory.last_kwargs["timeout"] == 0.25
+
+
+def test_session_query_only_uses_approved_catalog_commands():
+    factory = FakeFactory()
+    manager = OBDAISessionManager(obd_factory=factory)
+    session = manager.open_session()
+
+    response = session.query("engine_rpm")
+
+    assert response == {"command": "RPM", "force": False}
+    assert factory.connection.query_calls == [(obd.commands.RPM, False)]
+
+
+def test_session_query_rejects_unapproved_command_keys():
+    factory = FakeFactory()
+    manager = OBDAISessionManager(obd_factory=factory)
+    session = manager.open_session()
+
+    with pytest.raises(KeyError, match="Unapproved command key"):
+        session.query("clear_trouble_codes")
+
+
+def test_session_lifecycle_helpers_delegate_to_connection():
+    factory = FakeFactory()
+    manager = OBDAISessionManager(obd_factory=factory)
+    session = manager.open_session()
+
+    assert session.status() == OBDStatus.CAR_CONNECTED
+    assert session.is_connected() is True
+    assert session.port_name() == "FAKEPORT"
+    assert session.supports("engine_rpm") is True
+
+    session.close()
+    assert factory.connection.closed is True