diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
index faaaa38fc..f8c246638 100644
--- a/python/CHANGELOG.md
+++ b/python/CHANGELOG.md
@@ -7,6 +7,26 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ### What's New
 
+#### MCAP imports
+
+`client.data_import` now supports MCAP (`.mcap`) files with ROS 2 (`ros2msg`/`cdr`) topics.
+
+```python
+job = client.data_import.import_from_path("recording.mcap", asset=my_asset)
+```
+
+Importing without a config ingests every supported channel. Topics that cannot be decoded fail the import unless `McapParseErrorPolicy.IGNORE_ERROR` is set.
+
+`detect_config` reads the file's channels locally without decoding messages, returning an `McapImportConfig` whose `data` holds one entry per field. Edit it to select, rename, or retype channels before importing.
+
+```python
+config = client.data_import.detect_config("recording.mcap")
+config.complex_types_import_mode = McapComplexTypesImportMode.STRING
+```
+
+Variable-cardinality fields (dynamic and bounded arrays) are typed `BYTES`. As with Parquet, `complex_types_import_mode` on the config decides what each becomes: Arrow IPC bytes, a JSON string under `<name>.json`, both (the default), or neither.
+
+Reading a file locally needs the new `mcap` extra (`pip install sift-stack-py[mcap]`), so both `detect_config` and importing without a config require it.
 #### List and get data imports
 
 New in `client.data_import`: `list_` and `get` (plus `find`), and a `run.data_imports` property.
diff --git a/python/docs/guides/pytest_plugin/index.md b/python/docs/guides/pytest_plugin/index.md
index 215dc9048..6d33f1ff3 100644
--- a/python/docs/guides/pytest_plugin/index.md
+++ b/python/docs/guides/pytest_plugin/index.md
@@ -100,11 +100,11 @@ The plugin runs in one of three modes, picked at invocation.
 | **Offline** | `--sift-offline` | No; records to a log file for later replay | Environments without Sift access.                             |
 | **Disabled** | `--sift-disabled` | No | Local dev. Bounds still evaluate and return a real pass/fail. |
 
-Online mode pings Sift once at session start and aborts if Sift is unreachable or the credentials are invalid, 
-so a misconfigured job fails immediately instead of silently producing no report. 
-During the run, every create and update is appended to a JSONL log file. 
-A background worker uploads new entries to Sift incrementally. 
-If the connection drops mid-test, the test keeps running and the log keeps writing locally. 
+Online mode pings Sift once at session start and aborts if Sift is unreachable or the credentials are invalid,
+so a misconfigured job fails immediately instead of silently producing no report.
+During the run, every create and update is appended to a JSONL log file.
+A background worker uploads new entries to Sift incrementally.
+If the connection drops mid-test, the test keeps running and the log keeps writing locally.
 The remaining entries can be uploaded afterward by running import-test-result-log, which the plugin prints on exit. That command resumes into the report the interrupted run created rather than starting a second one.
 
 See [Running Modes](running_modes.md) for the log-file and replay pipeline,
diff --git a/python/examples/data_import/mcap/main.py b/python/examples/data_import/mcap/main.py
new file mode 100644
index 000000000..05b3932c5
--- /dev/null
+++ b/python/examples/data_import/mcap/main.py
@@ -0,0 +1,79 @@
+"""Import an MCAP (.mcap) file into Sift.
+
+MCAP files are self-describing, so the import needs no column mapping: every
+channel of every supported topic (ros2msg schemas with cdr messages) is
+imported. Channel names combine the topic and flattened field path; the
+bundled sample_data.mcap produces "/imu/data.angular_velocity.x",
+"/imu/data.linear_acceleration.x", "/battery.voltage", and their siblings.
+
+Swap sample_data.mcap for your own .mcap recording.
+"""
+
+import os
+
+from dotenv import load_dotenv
+from sift_client import SiftClient
+
+if __name__ == "__main__":
+    load_dotenv()
+
+    grpc_uri = os.getenv("SIFT_GRPC_URI")
+    assert grpc_uri, "expected 'SIFT_GRPC_URI' environment variable to be set"
+
+    rest_uri = os.getenv("SIFT_REST_URI")
+    assert rest_uri, "expected 'SIFT_REST_URI' environment variable to be set"
+
+    apikey = os.getenv("SIFT_API_KEY")
+    assert apikey, "expected 'SIFT_API_KEY' environment variable to be set"
+
+    asset_name = os.getenv("ASSET_NAME")
+    assert asset_name, "expected 'ASSET_NAME' environment variable to be set"
+
+    client = SiftClient(api_key=apikey, grpc_url=grpc_uri, rest_url=rest_uri)
+
+    # Auto-detect the config and import the file.
+    import_job = client.data_import.import_from_path(
+        "sample_data.mcap",
+        asset=asset_name,
+    )
+
+    import_job.wait_until_complete()
+
+    # If auto-detect doesn't quite match your file, inspect the config and patch
+    # it before importing. Common fixes: drop channels you don't need, rename or
+    # retype a channel, skip undecodable topics, set a start time for logs on a
+    # non-Unix epoch, or pick metadata records to import.
+    #
+    # from datetime import datetime, timezone
+    #
+    # from sift_client.sift_types.data_import import (
+    #     McapComplexTypesImportMode,
+    #     McapParseErrorPolicy,
+    # )
+    #
+    # config = client.data_import.detect_config("sample_data.mcap")
+    # print(config)  # inspect every detected channel
+    #
+    # # Example: import array fields only as JSON strings, instead of the
+    # # default of both JSON and Arrow IPC bytes
+    # config.complex_types_import_mode = McapComplexTypesImportMode.STRING
+    #
+    # # Example: import only the IMU topic
+    # config.data = [d for d in config.data if d.topic == "/imu/data"]
+    #
+    # # Example: skip undecodable topics and records instead of failing
+    # config.parse_error_policy = McapParseErrorPolicy.IGNORE_ERROR
+    #
+    # # Example: reinterpret log_time as elapsed nanoseconds from an explicit
+    # # start; only for recorders whose clock did not track Unix time
+    # config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    #
+    # # Example: import every key of a named metadata record as run metadata
+    # config.metadata_records = ["calibration"]
+    #
+    # import_job = client.data_import.import_from_path(
+    #     "sample_data.mcap",
+    #     asset=asset_name,
+    #     config=config,
+    # )
+    # import_job.wait_until_complete()
diff --git a/python/examples/data_import/mcap/requirements.txt b/python/examples/data_import/mcap/requirements.txt
new file mode 100644
index 000000000..a0739c91b
--- /dev/null
+++ b/python/examples/data_import/mcap/requirements.txt
@@ -0,0 +1,2 @@
+python-dotenv
+sift-stack-py[mcap]
diff --git a/python/examples/data_import/mcap/sample_data.mcap b/python/examples/data_import/mcap/sample_data.mcap
new file mode 100644
index 000000000..0e72f96e3
Binary files /dev/null and b/python/examples/data_import/mcap/sample_data.mcap differ
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py b/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py
index 4ce2d8bdb..a5a7c5f63 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py
@@ -20,6 +20,7 @@
     DataImport,
     Hdf5ImportConfig,
     ImportConfig,
+    McapImportConfig,
     ParquetFlatDatasetImportConfig,
     ParquetSingleChannelPerRowImportConfig,
     TdmsImportConfig,
@@ -50,6 +51,8 @@ def _set_config_on_request(
         request.hdf5_config.CopyFrom(config._to_proto())
     elif isinstance(config, UlogImportConfig):
         request.ulog_config.CopyFrom(config._to_proto())
+    elif isinstance(config, McapImportConfig):
+        request.mcap_config.CopyFrom(config._to_proto())
     else:
         raise TypeError(f"Unsupported import config type: {type(config).__name__}")
 
diff --git a/python/lib/sift_client/_internal/util/mcap.py b/python/lib/sift_client/_internal/util/mcap.py
new file mode 100644
index 000000000..3957d5022
--- /dev/null
+++ b/python/lib/sift_client/_internal/util/mcap.py
@@ -0,0 +1,439 @@
+"""Detect channels in MCAP (``.mcap``) files.
+
+Reads the file's schema and channel records without decoding message
+payloads, then flattens each topic's ros2msg schema into leaf fields, one
+channel per leaf.
+"""
+
+from __future__ import annotations
+
+import warnings
+from collections import defaultdict
+from pathlib import Path
+from typing import NamedTuple
+
+from mcap import records as mcap_records
+from mcap.reader import make_reader
+from mcap.stream_reader import StreamReader
+from mcap_ros2 import _dynamic as ros2_dynamic
+
+from sift_client.sift_types.channel import ChannelDataType
+from sift_client.sift_types.data_import import McapDataColumn, McapImportConfig
+
+MCAP_MAGIC = b"\x89MCAP0\r\n"
+
+# Chunk compressions Sift can read. The reader skips a chunk it cannot
+# decompress without complaining, hiding every channel inside it, so anything
+# else has to be rejected explicitly.
+SUPPORTED_COMPRESSIONS = frozenset(("", "zstd", "lz4"))
+
+# ROS 2 scalar types to Sift channel types. Narrow integers widen to 32-bit
+# like the other import types. byte and char both map to UINT_32: ROS 2
+# defines them as unsigned 8-bit (octet and uint8).
+ROS2_TO_SIFT_TYPE: dict[str, ChannelDataType] = {
+    "bool": ChannelDataType.BOOL,
+    "int8": ChannelDataType.INT_32,
+    "int16": ChannelDataType.INT_32,
+    "int32": ChannelDataType.INT_32,
+    "int64": ChannelDataType.INT_64,
+    "uint8": ChannelDataType.UINT_32,
+    "uint16": ChannelDataType.UINT_32,
+    "uint32": ChannelDataType.UINT_32,
+    "uint64": ChannelDataType.UINT_64,
+    "byte": ChannelDataType.UINT_32,
+    "char": ChannelDataType.UINT_32,
+    "float32": ChannelDataType.FLOAT,
+    "float64": ChannelDataType.DOUBLE,
+    "string": ChannelDataType.STRING,
+}
+
+# builtin_interfaces Time and Duration import as one INT_64 nanosecond channel.
+TIME_MESSAGE_TYPES = frozenset(("builtin_interfaces/Time", "builtin_interfaces/Duration"))
+ROS2_TIME_TYPE = "__time__"
+
+# Guards malformed schemas with self-referential fixed nesting.
+MAX_FIELD_DEPTH = 32
+
+
+class UnsupportedTopicError(Exception):
+    """The topic's schema cannot be decoded."""
+
+
+class LeafField(NamedTuple):
+    """A leaf field of a topic's message type. Scalar leaves carry one value
+    per message; complex leaves are variable-cardinality.
+    """
+
+    field_path: str
+    kind: str  # "scalar" or "complex"
+    # The ROS 2 base type for scalar leaves (ROS2_TIME_TYPE for
+    # builtin_interfaces Time/Duration). None for complex leaves.
+    ros_type: str | None
+
+    def sift_type(self) -> ChannelDataType:
+        if self.ros_type == ROS2_TIME_TYPE:
+            return ChannelDataType.INT_64
+        return ROS2_TO_SIFT_TYPE[self.ros_type]  # type: ignore[index]
+
+
+class TopicInfo(NamedTuple):
+    """A supported topic and its importable leaves."""
+
+    topic: str
+    leaves: list[LeafField]
+
+
+def parse_schema_defs(schema: mcap_records.Schema):
+    """Parse a ros2msg concatenated schema into (root msgdef, msgdefs by
+    name)."""
+    msgdefs: dict = {
+        "builtin_interfaces/Time": ros2_dynamic.TimeDefinition,
+        "builtin_interfaces/Duration": ros2_dynamic.TimeDefinition,
+    }
+
+    def handle(cur_schema_name: str, short_name: str, msgdef) -> None:
+        msgdefs[cur_schema_name] = msgdef
+        msgdefs[short_name] = msgdef
+
+    try:
+        text = schema.data.decode("utf-8")
+        ros2_dynamic._for_each_msgdef(schema.name, text, handle)
+    except Exception as e:
+        raise UnsupportedTopicError(f"its schema failed to parse ({e})") from e
+
+    root = msgdefs.get(schema.name) or msgdefs.get(
+        "/".join((schema.name.split("/")[0], schema.name.split("/")[-1]))
+    )
+    if root is None:
+        raise UnsupportedTopicError("its schema does not define the root message")
+    return root, msgdefs
+
+
+def is_variable_array(ftype) -> bool:
+    """Unbounded and bounded ([<=N]) arrays decode with a dynamic length."""
+    return ftype.is_array and (ftype.array_size is None or ftype.is_upper_bound)
+
+
+def check_primitive_supported(type_name: str, label: str) -> None:
+    if type_name == "wstring":
+        raise UnsupportedTopicError(
+            f"field '{label}' uses wstring, which the decoder does not implement"
+        )
+    if type_name not in ROS2_TO_SIFT_TYPE:
+        raise UnsupportedTopicError(f"field '{label}' has unsupported type '{type_name}'")
+
+
+def resolve_or_raise(msgdefs: dict, ftype, label: str):
+    nested = msgdefs.get(f"{ftype.pkg_name}/{ftype.type}")
+    if nested is None:
+        raise UnsupportedTopicError(
+            f"field '{label}' has unknown type '{ftype.pkg_name}/{ftype.type}'"
+        )
+    return nested
+
+
+def check_ftype_decodable(
+    msgdefs: dict, ftype, label: str, visited: frozenset = frozenset(), depth: int = 0
+) -> None:
+    """Raise UnsupportedTopicError if the field type cannot be decoded.
+
+    Variable-cardinality fields do not expand into leaves, but every element
+    is still decoded on import, so a wstring or unknown type anywhere in the
+    subtree makes the whole topic unsupported.
+    """
+    if ftype.is_primitive_type():
+        check_primitive_supported(ftype.type, label)
+        return
+    if f"{ftype.pkg_name}/{ftype.type}" in TIME_MESSAGE_TYPES:
+        return
+    check_subtree_decodable(msgdefs, resolve_or_raise(msgdefs, ftype, label), visited, depth)
+
+
+def check_subtree_decodable(msgdefs: dict, msgdef, visited: frozenset, depth: int) -> None:
+    if depth > MAX_FIELD_DEPTH:
+        raise UnsupportedTopicError(f"its schema nests deeper than {MAX_FIELD_DEPTH} levels")
+    name = f"{msgdef.base_type.pkg_name}/{msgdef.msg_name}"
+    if name in visited:
+        return
+    visited = visited | {name}
+    for field in msgdef.fields:
+        check_ftype_decodable(msgdefs, field.type, field.name, visited, depth + 1)
+
+
+def expand_message_fields(root_msgdef, msgdefs: dict) -> list[LeafField]:
+    """Flatten the root message into importable leaves.
+
+    Scalar leaves append their dot-delimited field path. Fixed-size arrays
+    expand with bracketed indexes. Variable-cardinality fields become one
+    complex leaf. Named constants are not message fields and never appear.
+    """
+    leaves: list[LeafField] = []
+
+    def walk(prefix: str, msgdef, depth: int) -> None:
+        if depth > MAX_FIELD_DEPTH:
+            raise UnsupportedTopicError(f"its schema nests deeper than {MAX_FIELD_DEPTH} levels")
+        for field in msgdef.fields:
+            ftype = field.type
+            path = f"{prefix}{field.name}"
+            if is_variable_array(ftype):
+                # The leaf imports whole, but its elements are still
+                # decoded, so check the element type.
+                check_ftype_decodable(msgdefs, ftype, path)
+                leaves.append(LeafField(path, "complex", None))
+                continue
+
+            indexes = [f"[{i}]" for i in range(ftype.array_size)] if ftype.is_array else [""]
+            if ftype.is_primitive_type():
+                check_primitive_supported(ftype.type, path)
+                leaves.extend(
+                    LeafField(f"{path}{index}", "scalar", ftype.type) for index in indexes
+                )
+            elif f"{ftype.pkg_name}/{ftype.type}" in TIME_MESSAGE_TYPES:
+                leaves.extend(
+                    LeafField(f"{path}{index}", "scalar", ROS2_TIME_TYPE) for index in indexes
+                )
+            else:
+                nested = resolve_or_raise(msgdefs, ftype, path)
+                for index in indexes:
+                    walk(f"{path}{index}.", nested, depth + 1)
+
+    walk("", root_msgdef, 0)
+    return leaves
+
+
+def read_schemas_and_channels(
+    path: Path,
+) -> tuple[dict[int, mcap_records.Schema], list[mcap_records.Channel], list[str]]:
+    """Read schema and channel records without decoding message payloads.
+
+    A well-formed file repeats them in its summary section, a few kilobytes at
+    the end, and that is all we need. Otherwise they only exist in the data
+    section and the whole file has to be read: unchunked files keep them at the
+    top level, chunked files keep them inside the chunks.
+
+    A file that stops parsing partway keeps what was read, with a warning.
+    """
+    parse_warnings: list[str] = []
+    schemas: dict[int, mcap_records.Schema] = {}
+    channels: list[mcap_records.Channel] = []
+    seen_channel_ids: set[int] = set()
+    attachment_count = 0
+
+    def add_channel(channel: mcap_records.Channel) -> None:
+        if channel.id not in seen_channel_ids:
+            seen_channel_ids.add(channel.id)
+            channels.append(channel)
+
+    def scan(stream: StreamReader) -> None:
+        nonlocal attachment_count
+        records = iter(stream.records)
+        while True:
+            try:
+                record = next(records)
+            except StopIteration:
+                return
+            except Exception as e:
+                # Damaged files raise McapError, struct.error, decompression and
+                # decode errors. EndOfFile stringifies empty, so name the type.
+                detail = str(e) or type(e).__name__
+                message = (
+                    "stopped reading at an unparseable record; the detected "
+                    f"channels may be incomplete: {detail}"
+                )
+                # Both passes hit the same broken spot.
+                if message not in parse_warnings:
+                    parse_warnings.append(message)
+                return
+            if isinstance(record, mcap_records.Chunk):
+                if record.compression not in SUPPORTED_COMPRESSIONS:
+                    raise ValueError(
+                        f"unsupported chunk compression '{record.compression}'; "
+                        "supported compressions are none, zstd, and lz4"
+                    )
+            elif isinstance(record, mcap_records.Schema):
+                schemas[record.id] = record
+            elif isinstance(record, mcap_records.Channel):
+                add_channel(record)
+            elif isinstance(record, mcap_records.Attachment):
+                attachment_count += 1
+
+    with open(path, "rb") as file:
+        if file.read(len(MCAP_MAGIC)) != MCAP_MAGIC:
+            raise ValueError(f"'{path.name}' is not an MCAP file (bad magic bytes)")
+
+        file.seek(0)
+        try:
+            summary = make_reader(file).get_summary()
+        except Exception:
+            # A damaged footer or summary raises McapError, struct.error or
+            # UnicodeDecodeError; fall back to a linear scan either way.
+            summary = None
+
+        if summary is not None:
+            for chunk_index in summary.chunk_indexes:
+                if chunk_index.compression not in SUPPORTED_COMPRESSIONS:
+                    raise ValueError(
+                        f"unsupported chunk compression '{chunk_index.compression}'; "
+                        "supported compressions are none, zstd, and lz4"
+                    )
+            schemas.update(summary.schemas)
+            for channel in sorted(summary.channels.values(), key=lambda c: c.id):
+                add_channel(channel)
+            attachment_count = len(summary.attachment_indexes)
+
+        if not channels:
+            # Repeating the records in the summary is optional, so fall back to
+            # reading the data section. Anything counted above gets counted
+            # again there, so start over.
+            attachment_count = 0
+            if summary is None:
+                # Chunk records carry the only copy of the compression strings
+                # when there are no chunk indexes to read them from, and they
+                # are visible only while the chunks stay unopened.
+                file.seek(0)
+                scan(StreamReader(file, emit_chunks=True))
+            file.seek(0)
+            scan(StreamReader(file, emit_chunks=False))
+
+    if attachment_count:
+        parse_warnings.append(
+            f"the file has {attachment_count} attachment(s); attachments are not imported"
+        )
+
+    return schemas, channels, parse_warnings
+
+
+def detect_mcap_topics(
+    schemas: dict[int, mcap_records.Schema],
+    channels: list[mcap_records.Channel],
+    parse_warnings: list[str],
+) -> list[TopicInfo]:
+    """Derive the supported topics and their importable leaves.
+
+    Same-topic channels merge only when their schemas and message encodings
+    match. Distinct topics colliding case-insensitively keep the first.
+    Topics that cannot be decoded are skipped with a warning.
+    """
+    channels_by_topic: defaultdict[str, list[mcap_records.Channel]] = defaultdict(list)
+    for channel in channels:
+        channels_by_topic[channel.topic].append(channel)
+
+    # Sift channel names compare case-insensitively, so distinct topics
+    # colliding only by case conflict; the first wins.
+    kept_by_lower: dict[str, str] = {}
+    for topic in channels_by_topic:
+        first = kept_by_lower.setdefault(topic.lower(), topic)
+        if first != topic:
+            parse_warnings.append(
+                f"topic '{topic}' collides with topic '{first}' by case only; kept the "
+                "first. Set McapParseErrorPolicy.IGNORE_ERROR to import it and skip the "
+                "rest, otherwise the import fails"
+            )
+
+    topics: list[TopicInfo] = []
+    unsupported: dict[str, str] = {}
+    for topic, topic_channels in channels_by_topic.items():
+        if kept_by_lower[topic.lower()] != topic:
+            continue
+        # Same-topic channels merge only when they agree.
+        encodings = {c.message_encoding for c in topic_channels}
+        topic_schemas = [schemas.get(c.schema_id) for c in topic_channels]
+        signatures = {None if s is None else (s.name, s.encoding, s.data) for s in topic_schemas}
+        if len(encodings) > 1 or len(signatures) > 1:
+            unsupported[topic] = (
+                "it has multiple channels with mismatched schemas or message encodings"
+            )
+            continue
+        channel = topic_channels[0]
+        schema = topic_schemas[0]
+        if schema is None:
+            unsupported[topic] = "it has no schema"
+            continue
+        if channel.message_encoding != "cdr":
+            unsupported[topic] = (
+                f"its message encoding is '{channel.message_encoding}' (only cdr is supported)"
+            )
+            continue
+        if schema.encoding != "ros2msg":
+            unsupported[topic] = (
+                f"its schema encoding is '{schema.encoding}' (only ros2msg is supported)"
+            )
+            continue
+        try:
+            root, msgdefs = parse_schema_defs(schema)
+            leaves = expand_message_fields(root, msgdefs)
+        except UnsupportedTopicError as e:
+            unsupported[topic] = str(e)
+            continue
+        topics.append(TopicInfo(topic=topic, leaves=leaves))
+
+    if unsupported:
+        details = "; ".join(f"'{t}': {reason}" for t, reason in sorted(unsupported.items()))
+        parse_warnings.append(f"skipped unsupported topics: {details}")
+    return topics
+
+
+def detect_mcap_fields(topics: list[TopicInfo]) -> list[McapDataColumn]:
+    """Return one ``McapDataColumn`` per leaf field, named ``<topic>.<field_path>``.
+
+    Variable-cardinality fields are marked ``BYTES``; the config's
+    ``complex_types_import_mode`` (default ``BOTH``) decides what they import as.
+    """
+    channels: list[McapDataColumn] = []
+    # Sift channel names are unique per asset and compare case-insensitively.
+    # Values are the (name, topic, field path) that first claimed the key.
+    taken_names: dict[str, tuple[str, str, str]] = {}
+    for topic in topics:
+        for leaf in topic.leaves:
+            name = f"{topic.topic}.{leaf.field_path}"
+            # BYTES marks a complex field; complex_types_import_mode expands it.
+            data_type = ChannelDataType.BYTES if leaf.kind == "complex" else leaf.sift_type()
+            existing = taken_names.get(name.lower())
+            if existing is not None:
+                raise ValueError(
+                    f"two channels are both named '{name}': topic '{topic.topic}' field "
+                    f"'{leaf.field_path}' and topic '{existing[1]}' field '{existing[2]}'. "
+                    "Build an McapImportConfig by hand to give them distinct names."
+                )
+            taken_names[name.lower()] = (name, topic.topic, leaf.field_path)
+            channels.append(
+                McapDataColumn(
+                    topic=topic.topic,
+                    field_path=leaf.field_path,
+                    name=name,
+                    data_type=data_type,
+                )
+            )
+    return channels
+
+
+def detect_mcap_config(file_path: str | Path, asset_name: str = "") -> McapImportConfig:
+    """Detect an MCAP import config by enumerating the file's channels.
+
+    Channels come from the file's schema and channel records; message payloads
+    are not read, so a topic is listed even when it logged no messages. Topics
+    that cannot be decoded (non-cdr message encodings, non-ros2msg schemas,
+    undecodable schemas) are skipped with a warning. Importing such a file
+    fails unless ``parse_error_policy`` is ``McapParseErrorPolicy.IGNORE_ERROR``.
+
+    Args:
+        file_path: Path to the ``.mcap`` file.
+        asset_name: The asset name to set on the config.
+
+    Returns:
+        A config whose ``data`` is a flattened list of fields with default
+        Sift names and types. Edit or remove entries before importing.
+        Leaving ``data`` empty imports every field with the defaults.
+
+    Raises:
+        ValueError: If the file is not MCAP, uses an unsupported chunk
+            compression, or two channels share a name.
+    """
+    path = Path(file_path)
+    schemas, channels, parse_warnings = read_schemas_and_channels(path)
+    topics = detect_mcap_topics(schemas, channels, parse_warnings)
+    # Emitted before the names are built so a name clash does not discard
+    # what the scan found.
+    for message in parse_warnings:
+        warnings.warn(f"'{path.name}': {message}", stacklevel=2)
+    return McapImportConfig(asset_name=asset_name, data=detect_mcap_fields(topics))
diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py
new file mode 100644
index 000000000..ba2f0939c
--- /dev/null
+++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py
@@ -0,0 +1,34 @@
+"""Tests for the data imports low-level wrapper."""
+
+from __future__ import annotations
+
+import pytest
+from sift.data_imports.v2.data_imports_pb2 import CreateDataImportFromUploadRequest
+
+from sift_client._internal.low_level_wrappers.data_imports import _set_config_on_request
+from sift_client.sift_types.channel import ChannelDataType
+from sift_client.sift_types.data_import import McapDataColumn, McapImportConfig
+
+
+class TestSetConfigOnRequest:
+    def test_mcap_config_sets_mcap_field(self):
+        request = CreateDataImportFromUploadRequest()
+        config = McapImportConfig(
+            asset_name="my_asset",
+            data=[
+                McapDataColumn(
+                    topic="/imu", field_path="orientation.x", data_type=ChannelDataType.DOUBLE
+                )
+            ],
+        )
+
+        _set_config_on_request(request, config)
+
+        assert request.HasField("mcap_config")
+        assert request.mcap_config.asset_name == "my_asset"
+        assert request.mcap_config.data[0].ros2.field_path == "orientation.x"
+
+    def test_unknown_config_type_raises(self):
+        request = CreateDataImportFromUploadRequest()
+        with pytest.raises(TypeError, match="Unsupported import config type"):
+            _set_config_on_request(request, object())  # type: ignore[arg-type]
diff --git a/python/lib/sift_client/_tests/_internal/test_mcap.py b/python/lib/sift_client/_tests/_internal/test_mcap.py
new file mode 100644
index 000000000..be248831b
--- /dev/null
+++ b/python/lib/sift_client/_tests/_internal/test_mcap.py
@@ -0,0 +1,545 @@
+"""Tests for MCAP channel detection."""
+
+from __future__ import annotations
+
+import pytest
+from mcap.records import Channel, Schema
+from mcap.writer import Writer
+
+from sift_client._internal.util.mcap import (
+    UnsupportedTopicError,
+    detect_mcap_config,
+    detect_mcap_topics,
+    expand_message_fields,
+    parse_schema_defs,
+)
+from sift_client.sift_types.channel import ChannelDataType
+from sift_client.sift_types.data_import import McapComplexTypesImportMode
+
+
+def _schema(data: str, schema_id: int = 1, name: str = "test_msgs/msg/Test") -> Schema:
+    return Schema(id=schema_id, data=data.encode(), encoding="ros2msg", name=name)
+
+
+def _channel(topic: str, schema_id: int = 1, channel_id: int = 1, encoding: str = "cdr") -> Channel:
+    return Channel(
+        id=channel_id, topic=topic, message_encoding=encoding, metadata={}, schema_id=schema_id
+    )
+
+
+def _leaves(data: str, name: str = "test_msgs/msg/Test"):
+    root, msgdefs = parse_schema_defs(_schema(data, name=name))
+    return expand_message_fields(root, msgdefs)
+
+
+IMU_SCHEMA = """geometry_msgs/Vector3 gyro
+float64 temp
+================================================================================
+MSG: geometry_msgs/Vector3
+float64 x
+float64 y
+float64 z
+"""
+
+
+def _write_mcap(
+    path,
+    schemas_and_topics: list[tuple[str, str, str]],
+    messages: int = 0,
+    attachment: bool = False,
+    **writer_kwargs,
+) -> None:
+    """Write an MCAP file with one channel per (schema_name, schema_text, topic).
+
+    Each channel logs ``messages`` cdr messages, and ``writer_kwargs`` go to the
+    mcap ``Writer`` (chunking, schema and channel repeats, statistics).
+    """
+    with open(path, "wb") as f:
+        writer = Writer(f, **writer_kwargs)
+        writer.start()
+        for schema_name, schema_text, topic in schemas_and_topics:
+            schema_id = writer.register_schema(
+                name=schema_name, encoding="ros2msg", data=schema_text.encode()
+            )
+            channel_id = writer.register_channel(
+                topic=topic, message_encoding="cdr", schema_id=schema_id
+            )
+            for i in range(messages):
+                writer.add_message(
+                    channel_id=channel_id, log_time=i, publish_time=i, data=b"\x00" * 32
+                )
+        if attachment:
+            writer.add_attachment(
+                create_time=1, log_time=1, name="notes.txt", media_type="text/plain", data=b"hi"
+            )
+        writer.finish()
+
+
+class TestExpandMessageFields:
+    def test_scalars(self):
+        assert _leaves("float64 x\nuint32 seq\nstring status\n") == [
+            ("x", "scalar", "float64"),
+            ("seq", "scalar", "uint32"),
+            ("status", "scalar", "string"),
+        ]
+
+    def test_nested_message_uses_dotted_paths(self):
+        leaves = _leaves(IMU_SCHEMA)
+        assert [leaf.field_path for leaf in leaves] == ["gyro.x", "gyro.y", "gyro.z", "temp"]
+
+    def test_fixed_array_expands_per_element(self):
+        assert [leaf.field_path for leaf in _leaves("float32[3] accel\n")] == [
+            "accel[0]",
+            "accel[1]",
+            "accel[2]",
+        ]
+
+    @pytest.mark.parametrize(
+        "definition",
+        [
+            "int32[] samples\n",
+            "int32[<=4] samples\n",
+            "geometry_msgs/Vector3[] samples\n"
+            + "=" * 80
+            + "\nMSG: geometry_msgs/Vector3\nfloat64 x\n",
+        ],
+        ids=["unbounded", "bounded", "of_messages"],
+    )
+    def test_variable_array_is_one_complex_leaf(self, definition):
+        assert _leaves(definition) == [("samples", "complex", None)]
+
+    def test_fixed_array_of_messages_expands_per_element(self):
+        schema = (
+            "geometry_msgs/Vector3[2] corners\n"
+            + "=" * 80
+            + "\nMSG: geometry_msgs/Vector3\nfloat64 x\nfloat64 y\nfloat64 z\n"
+        )
+        assert [leaf.field_path for leaf in _leaves(schema)] == [
+            "corners[0].x",
+            "corners[0].y",
+            "corners[0].z",
+            "corners[1].x",
+            "corners[1].y",
+            "corners[1].z",
+        ]
+
+    def test_time_and_duration_collapse_to_int64(self):
+        leaves = _leaves("builtin_interfaces/Time stamp\nbuiltin_interfaces/Duration elapsed\n")
+        assert [leaf.field_path for leaf in leaves] == ["stamp", "elapsed"]
+        assert all(leaf.sift_type() == ChannelDataType.INT_64 for leaf in leaves)
+
+    def test_constants_are_not_fields(self):
+        assert _leaves("int32 STATUS_OK=0\nint32 status\n") == [
+            ("status", "scalar", "int32"),
+        ]
+
+    def test_maps_every_ros2_scalar_type(self):
+        # Narrow ints widen to 32-bit; byte and char are unsigned 8-bit in ROS 2.
+        definition = "\n".join(
+            f"{ros_type} f_{ros_type}"
+            for ros_type in (
+                "bool",
+                "int8",
+                "int16",
+                "int32",
+                "int64",
+                "uint8",
+                "uint16",
+                "uint32",
+                "uint64",
+                "byte",
+                "char",
+                "float32",
+                "float64",
+                "string",
+            )
+        )
+        assert [leaf.sift_type() for leaf in _leaves(definition)] == [
+            ChannelDataType.BOOL,
+            ChannelDataType.INT_32,
+            ChannelDataType.INT_32,
+            ChannelDataType.INT_32,
+            ChannelDataType.INT_64,
+            ChannelDataType.UINT_32,
+            ChannelDataType.UINT_32,
+            ChannelDataType.UINT_32,
+            ChannelDataType.UINT_64,
+            ChannelDataType.UINT_32,
+            ChannelDataType.UINT_32,
+            ChannelDataType.FLOAT,
+            ChannelDataType.DOUBLE,
+            ChannelDataType.STRING,
+        ]
+
+    @pytest.mark.parametrize(
+        "definition",
+        [
+            "wstring label\n",
+            "wstring[] labels\n",
+            "pkg/Bad[] items\n" + "=" * 80 + "\nMSG: pkg/Bad\nwstring label\n",
+        ],
+        ids=["scalar", "array", "nested_in_array"],
+    )
+    def test_wstring_raises_unsupported(self, definition):
+        # Complex leaves still decode every element, so an undecodable element
+        # type anywhere makes the whole topic unsupported.
+        with pytest.raises(UnsupportedTopicError, match="wstring"):
+            _leaves(definition)
+
+    def test_nesting_beyond_max_depth_raises(self):
+        # A chain of 40 nested message types exceeds MAX_FIELD_DEPTH (32).
+        parts = ["pkg/M0 child\n"]
+        parts.extend("=" * 80 + f"\nMSG: pkg/M{i}\npkg/M{i + 1} child\n" for i in range(40))
+        parts.append("=" * 80 + "\nMSG: pkg/M40\nfloat64 x\n")
+        with pytest.raises(UnsupportedTopicError, match="nests deeper"):
+            _leaves("".join(parts))
+
+    def test_unknown_nested_type_raises(self):
+        with pytest.raises(UnsupportedTopicError, match="unknown type"):
+            _leaves("other_msgs/Missing part\n")
+
+    def test_root_message_missing_raises(self):
+        with pytest.raises(UnsupportedTopicError, match="root message"):
+            parse_schema_defs(
+                Schema(
+                    id=1,
+                    data=b"MSG: other_msgs/Other\nfloat64 x\n",
+                    encoding="ros2msg",
+                    name="test_msgs/msg/Test",
+                )
+            )
+
+
+class TestDetectMcapTopics:
+    def test_supported_topic_yields_leaves(self):
+        warnings: list[str] = []
+        topics = detect_mcap_topics({1: _schema(IMU_SCHEMA)}, [_channel("/imu")], warnings)
+        assert [t.topic for t in topics] == ["/imu"]
+        assert [leaf.field_path for leaf in topics[0].leaves] == [
+            "gyro.x",
+            "gyro.y",
+            "gyro.z",
+            "temp",
+        ]
+        assert warnings == []
+
+    def test_non_cdr_encoding_skipped_with_warning(self):
+        warnings: list[str] = []
+        topics = detect_mcap_topics(
+            {1: _schema(IMU_SCHEMA)}, [_channel("/imu", encoding="json")], warnings
+        )
+        assert topics == []
+        assert any("only cdr is supported" in w for w in warnings)
+
+    def test_non_ros2msg_schema_skipped_with_warning(self):
+        schema = Schema(id=1, data=b"{}", encoding="jsonschema", name="Test")
+        warnings: list[str] = []
+        assert detect_mcap_topics({1: schema}, [_channel("/diag")], warnings) == []
+        assert any("only ros2msg is supported" in w for w in warnings)
+
+    def test_missing_schema_skipped_with_warning(self):
+        warnings: list[str] = []
+        assert detect_mcap_topics({}, [_channel("/imu")], warnings) == []
+        assert any("no schema" in w for w in warnings)
+
+    def test_case_colliding_topics_keep_first(self):
+        schemas = {1: _schema(IMU_SCHEMA)}
+        channels = [
+            _channel("/imu", channel_id=1),
+            _channel("/IMU", channel_id=2),
+        ]
+        warnings: list[str] = []
+        topics = detect_mcap_topics(schemas, channels, warnings)
+        assert [t.topic for t in topics] == ["/imu"]
+        assert any("by case only" in w for w in warnings)
+        # The importer rejects the collision unless the policy is IGNORE_ERROR,
+        # so the warning has to say so.
+        assert any("IGNORE_ERROR" in w for w in warnings)
+
+    def test_same_topic_channels_merge_when_identical(self):
+        schemas = {1: _schema(IMU_SCHEMA)}
+        channels = [_channel("/imu", channel_id=1), _channel("/imu", channel_id=2)]
+        warnings: list[str] = []
+        topics = detect_mcap_topics(schemas, channels, warnings)
+        assert [t.topic for t in topics] == ["/imu"]
+        assert warnings == []
+
+    def test_same_topic_mismatched_schemas_skipped(self):
+        schemas = {
+            1: _schema(IMU_SCHEMA, schema_id=1),
+            2: _schema("float64 other\n", schema_id=2),
+        }
+        channels = [
+            _channel("/imu", schema_id=1, channel_id=1),
+            _channel("/imu", schema_id=2, channel_id=2),
+        ]
+        warnings: list[str] = []
+        assert detect_mcap_topics(schemas, channels, warnings) == []
+        assert any("mismatched schemas" in w for w in warnings)
+
+
+class TestDetectMcapConfig:
+    def test_detects_channel_per_field(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu/data")])
+
+        config = detect_mcap_config(path, asset_name="robot")
+        assert config.asset_name == "robot"
+        assert len(config.data) == 4
+        channels = {(d.topic, d.field_path, d.name, d.data_type) for d in config.data}
+        assert channels == {
+            ("/imu/data", "gyro.x", "/imu/data.gyro.x", ChannelDataType.DOUBLE),
+            ("/imu/data", "gyro.y", "/imu/data.gyro.y", ChannelDataType.DOUBLE),
+            ("/imu/data", "gyro.z", "/imu/data.gyro.z", ChannelDataType.DOUBLE),
+            ("/imu/data", "temp", "/imu/data.temp", ChannelDataType.DOUBLE),
+        }
+
+    def test_complex_field_is_one_bytes_channel(self, tmp_path):
+        # The mode turns this single entry into the channels that get imported.
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("test_msgs/msg/Samples", "int32[] samples\n", "/samples")])
+
+        config = detect_mcap_config(path)
+        assert [(d.field_path, d.name, d.data_type) for d in config.data] == [
+            ("samples", "/samples.samples", ChannelDataType.BYTES),
+        ]
+
+    def test_unsupported_topic_warns_and_keeps_supported(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        with open(path, "wb") as f:
+            writer = Writer(f)
+            writer.start()
+            imu = writer.register_schema(
+                name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode()
+            )
+            writer.register_channel(topic="/imu", message_encoding="cdr", schema_id=imu)
+            diag = writer.register_schema(name="diag", encoding="jsonschema", data=b"{}")
+            writer.register_channel(topic="/diag", message_encoding="json", schema_id=diag)
+            writer.finish()
+
+        with pytest.warns(UserWarning, match="skipped unsupported topics"):
+            config = detect_mcap_config(path)
+        assert {d.topic for d in config.data} == {"/imu"}
+
+    def test_clean_file_does_not_warn(self, tmp_path, recwarn):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")])
+
+        detect_mcap_config(path)
+        assert not [w for w in recwarn.list if issubclass(w.category, UserWarning)]
+
+    def test_truncated_file_scans_linearly(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")])
+        # Cutting the trailing magic invalidates the footer, so detection
+        # falls back to a linear scan of the intact data section.
+        data = path.read_bytes()
+        path.write_bytes(data[:-8])
+
+        with pytest.warns(UserWarning, match="stopped reading"):
+            config = detect_mcap_config(path)
+        assert {d.topic for d in config.data} == {"/imu"}
+
+    def test_rejects_non_mcap_file(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        path.write_bytes(b"NOTMCAP!" + b"\x00" * 64)
+        with pytest.raises(ValueError, match="not an MCAP file"):
+            detect_mcap_config(path)
+
+    def test_records_only_in_data_section_are_detected(self, tmp_path):
+        # An unchunked file whose summary omits the schema and channel repeats
+        # is spec-legal; the records are only in the data section.
+        path = tmp_path / "log.mcap"
+        _write_mcap(
+            path,
+            [("sensors/msg/Imu", IMU_SCHEMA, "/imu")],
+            use_chunking=False,
+            repeat_channels=False,
+            repeat_schemas=False,
+        )
+
+        config = detect_mcap_config(path)
+        assert {d.topic for d in config.data} == {"/imu"}
+
+    def test_rejects_unsupported_chunk_compression(self, tmp_path):
+        # Channels inside a chunk we cannot decompress are invisible, so the
+        # file can never import; refuse it rather than listing nothing.
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], messages=1)
+        path.write_bytes(path.read_bytes().replace(b"zstd", b"lzma"))
+
+        with pytest.raises(ValueError, match="unsupported chunk compression"):
+            detect_mcap_config(path)
+
+    def test_duplicate_channel_names_raise(self, tmp_path):
+        # '/a' field 'b.c' and '/a.b' field 'c' both name a channel '/a.b.c'.
+        path = tmp_path / "log.mcap"
+        _write_mcap(
+            path,
+            [
+                ("pkg/msg/A", "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n", "/a"),
+                ("pkg/msg/C", "int32 c\n", "/a.b"),
+            ],
+        )
+
+        with pytest.raises(ValueError, match="both named '/a.b.c'") as excinfo:
+            detect_mcap_config(path)
+        assert "topic '/a' field 'b.c'" in str(excinfo.value)
+        assert "topic '/a.b' field 'c'" in str(excinfo.value)
+
+
+class TestComplexTypesImportMode:
+    """A variable-cardinality field is one entry in ``data``; the mode decides
+    which channels it becomes when the config is sent.
+    """
+
+    SCHEMA = "int32 count\nint32[] samples\n"
+
+    def _config(self, tmp_path, mode):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("test_msgs/msg/Samples", self.SCHEMA, "/samples")])
+        config = detect_mcap_config(path)
+        config.complex_types_import_mode = mode
+        return config
+
+    @pytest.mark.parametrize(
+        ("mode", "expected"),
+        [
+            (
+                McapComplexTypesImportMode.BOTH,
+                [
+                    ("/samples.count", ChannelDataType.INT_32),
+                    ("/samples.samples", ChannelDataType.BYTES),
+                    ("/samples.samples.json", ChannelDataType.STRING),
+                ],
+            ),
+            (
+                McapComplexTypesImportMode.BYTES,
+                [
+                    ("/samples.count", ChannelDataType.INT_32),
+                    ("/samples.samples", ChannelDataType.BYTES),
+                ],
+            ),
+            (
+                McapComplexTypesImportMode.STRING,
+                [
+                    ("/samples.count", ChannelDataType.INT_32),
+                    ("/samples.samples.json", ChannelDataType.STRING),
+                ],
+            ),
+            (
+                McapComplexTypesImportMode.IGNORE,
+                [("/samples.count", ChannelDataType.INT_32)],
+            ),
+        ],
+    )
+    def test_mode_decides_the_channels_sent(self, tmp_path, mode, expected):
+        proto = self._config(tmp_path, mode)._to_proto()
+        assert [
+            (d.channel_config.name, ChannelDataType(d.channel_config.data_type)) for d in proto.data
+        ] == expected
+
+    def test_both_channels_share_the_field_selector(self, tmp_path):
+        proto = self._config(tmp_path, McapComplexTypesImportMode.BOTH)._to_proto()
+        selectors = [(d.topic, d.ros2.field_path) for d in proto.data]
+        assert selectors[1] == selectors[2] == ("/samples", "samples")
+
+    def test_default_is_both(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("test_msgs/msg/Samples", self.SCHEMA, "/samples")])
+        assert detect_mcap_config(path).complex_types_import_mode is (
+            McapComplexTypesImportMode.BOTH
+        )
+
+    def test_ignoring_every_configured_channel_raises(self, tmp_path):
+        # An empty list would mean "import the whole file", so refuse instead.
+        config = self._config(tmp_path, McapComplexTypesImportMode.IGNORE)
+        config.data = [d for d in config.data if d.data_type == ChannelDataType.BYTES]
+        with pytest.raises(ValueError, match="nothing would be imported"):
+            config._to_proto()
+
+    def test_generated_json_name_clash_depends_on_the_mode(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        _write_mcap(
+            path,
+            [
+                ("pkg/msg/A", "int32[] b\n", "/a"),
+                ("pkg/msg/B", "int32 json\n", "/a.b"),
+            ],
+        )
+        config = detect_mcap_config(path)
+        assert {d.name for d in config.data} == {"/a.b", "/a.b.json"}
+
+        # BOTH generates a second '/a.b.json' from '/a.b'; BYTES does not.
+        with pytest.raises(ValueError, match="would both be imported as '/a.b.json'"):
+            config._to_proto()
+        config.complex_types_import_mode = McapComplexTypesImportMode.BYTES
+        assert {d.channel_config.name for d in config._to_proto().data} == {"/a.b", "/a.b.json"}
+
+
+class TestFileScanning:
+    """A well-formed file is read from its summary alone. The data section is
+    only read when the summary comes up short.
+    """
+
+    def test_well_formed_file_never_reads_the_data_section(self, tmp_path, monkeypatch):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], messages=200)
+
+        def fail(*args, **kwargs):
+            raise AssertionError("read the data section for a file with a usable summary")
+
+        monkeypatch.setattr("sift_client._internal.util.mcap.StreamReader", fail)
+        config = detect_mcap_config(path)
+        assert {d.topic for d in config.data} == {"/imu"}
+
+    def test_summary_without_repeats_falls_back_to_the_chunks(self, tmp_path):
+        # Statistics alone does not make the summary usable: without the
+        # channel repeats the channels exist only inside the chunks.
+        path = tmp_path / "log.mcap"
+        _write_mcap(
+            path,
+            [("sensors/msg/Imu", IMU_SCHEMA, "/imu")],
+            messages=10,
+            repeat_channels=False,
+            repeat_schemas=False,
+        )
+
+        config = detect_mcap_config(path)
+        assert {d.topic for d in config.data} == {"/imu"}
+
+    def test_unsupported_compression_rejected_without_a_summary(self, tmp_path):
+        # No summary means no chunk indexes, so the compression string is only
+        # readable from the chunk records themselves.
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], messages=1)
+        path.write_bytes(path.read_bytes().replace(b"zstd", b"lzma")[:-8])
+
+        with pytest.raises(ValueError, match="unsupported chunk compression"):
+            detect_mcap_config(path)
+
+    def test_attachments_warn(self, tmp_path):
+        path = tmp_path / "log.mcap"
+        _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], attachment=True)
+
+        with pytest.warns(UserWarning, match="1 attachment"):
+            config = detect_mcap_config(path)
+        assert {d.topic for d in config.data} == {"/imu"}
+
+    def test_warnings_survive_a_name_clash(self, tmp_path):
+        # The clash raises, but what the scan already found must still reach
+        # the caller instead of being discarded with the exception.
+        path = tmp_path / "log.mcap"
+        _write_mcap(
+            path,
+            [
+                ("pkg/msg/A", "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n", "/a"),
+                ("pkg/msg/C", "int32 c\n", "/a.b"),
+            ],
+            attachment=True,
+        )
+
+        with pytest.warns(UserWarning, match="1 attachment"), pytest.raises(
+            ValueError, match="both named"
+        ):
+            detect_mcap_config(path)
diff --git a/python/lib/sift_client/_tests/resources/test_data_imports.py b/python/lib/sift_client/_tests/resources/test_data_imports.py
index 893b504db..cd2e9b4f0 100644
--- a/python/lib/sift_client/_tests/resources/test_data_imports.py
+++ b/python/lib/sift_client/_tests/resources/test_data_imports.py
@@ -41,6 +41,10 @@
     DataTypeKey,
     Hdf5DataColumn,
     Hdf5ImportConfig,
+    McapComplexTypesImportMode,
+    McapDataColumn,
+    McapImportConfig,
+    McapParseErrorPolicy,
     ParquetDataColumn,
     ParquetFlatDatasetImportConfig,
     ParquetSingleChannelConfig,
@@ -472,6 +476,177 @@ def test_getitem_not_found(self):
             self._config()["nonexistent"]
 
 
+class TestMcapConfig:
+    def _config(self):
+        return McapImportConfig(
+            asset_name="my_asset",
+            run_name="run1",
+            data=[
+                McapDataColumn(
+                    topic="/imu/data",
+                    field_path="orientation.x",
+                    data_type=ChannelDataType.DOUBLE,
+                ),
+                McapDataColumn(
+                    topic="/battery",
+                    field_path="voltage",
+                    name="battery_voltage",
+                    data_type=ChannelDataType.FLOAT,
+                    units="V",
+                    description="pack voltage",
+                ),
+            ],
+            metadata_records=["calibration"],
+            parse_error_policy=McapParseErrorPolicy.IGNORE_ERROR,
+            complex_types_import_mode=McapComplexTypesImportMode.STRING,
+        )
+
+    def test_to_proto(self):
+        proto = self._config()._to_proto()
+        assert proto.asset_name == "my_asset"
+        assert proto.run_name == "run1"
+        assert len(proto.data) == 2
+        assert proto.data[0].topic == "/imu/data"
+        assert proto.data[0].ros2.field_path == "orientation.x"
+        assert proto.data[0].channel_config.name == "/imu/data.orientation.x"
+        assert proto.data[1].topic == "/battery"
+        assert proto.data[1].channel_config.name == "battery_voltage"
+        assert proto.data[1].channel_config.units == "V"
+        assert list(proto.metadata_records) == ["calibration"]
+
+    def test_to_proto_defaults(self):
+        """An empty config imports all channels; the default policy fails on
+        error and imports complex fields as both bytes and JSON strings.
+        """
+        from sift.data_imports.v2.data_imports_pb2 import (
+            MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH,
+            MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR,
+        )
+
+        proto = McapImportConfig(asset_name="a")._to_proto()
+        assert len(proto.data) == 0
+        assert proto.run_id == ""
+        assert proto.parse_error_policy == MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR
+        assert proto.complex_types_import_mode == MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH
+        assert not proto.HasField("relative_start_time")
+
+    def test_relative_start_time_round_trips(self):
+        config = McapImportConfig(
+            asset_name="a",
+            relative_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        )
+        proto = config._to_proto()
+        assert proto.HasField("relative_start_time")
+        restored = McapImportConfig._from_proto(proto)
+        assert restored.relative_start_time == config.relative_start_time
+
+    def test_from_proto_round_trip(self):
+        config = self._config()
+        restored = McapImportConfig._from_proto(config._to_proto())
+        assert restored.asset_name == config.asset_name
+        assert restored.run_name == config.run_name
+        assert restored.metadata_records == config.metadata_records
+        assert restored.parse_error_policy == McapParseErrorPolicy.IGNORE_ERROR
+        assert restored.complex_types_import_mode == McapComplexTypesImportMode.STRING
+        assert len(restored.data) == 2
+        assert restored.data[0].topic == "/imu/data"
+        assert restored.data[0].field_path == "orientation.x"
+        assert restored.data[0].name == "/imu/data.orientation.x"
+        assert restored.data[1].name == "battery_voltage"
+        assert restored.data[1].data_type == ChannelDataType.FLOAT
+        assert restored.data[1].units == "V"
+        assert restored.data[1].description == "pack voltage"
+
+    def test_from_proto_unspecified_enums_fall_back_to_defaults(self):
+        """UNSPECIFIED proto values mean FAIL_ON_ERROR and BOTH on the server."""
+        from sift.data_imports.v2.data_imports_pb2 import McapConfig as McapConfigProto
+
+        restored = McapImportConfig._from_proto(McapConfigProto(asset_name="a"))
+        assert restored.parse_error_policy == McapParseErrorPolicy.FAIL_ON_ERROR
+        assert restored.complex_types_import_mode == McapComplexTypesImportMode.BOTH
+
+    def test_run_id_takes_precedence(self):
+        proto = McapImportConfig(asset_name="a", run_name="ignored", run_id="run_123")._to_proto()
+        assert proto.run_id == "run_123"
+
+    def test_name_defaults_to_channel(self):
+        col = McapDataColumn(
+            topic="/imu/data", field_path="orientation.x", data_type=ChannelDataType.DOUBLE
+        )
+        assert col.default_channel_name == "/imu/data.orientation.x"
+        assert col.name == "/imu/data.orientation.x"
+
+    def test_explicit_name_overrides_channel(self):
+        col = McapDataColumn(
+            topic="/battery",
+            field_path="voltage",
+            name="battery_voltage",
+            data_type=ChannelDataType.FLOAT,
+        )
+        assert col.name == "battery_voltage"
+
+    def test_getitem(self):
+        col = self._config()["battery_voltage"]
+        assert col.field_path == "voltage"
+
+    def test_getitem_not_found(self):
+        with pytest.raises(KeyError, match="nonexistent"):
+            self._config()["nonexistent"]
+
+
+class TestImportFromPathClearsDetectedChannels:
+    """Auto-detected ULog and MCAP configs import with an empty channel list
+    so the server imports every channel instead of strictly filtering on a
+    list that client detection may have misread.
+    """
+
+    async def _import(self, tmp_path, filename, detected):
+        path = tmp_path / filename
+        path.write_bytes(b"")
+
+        api = DataImportAPIAsync(MagicMock())
+        api.detect_config = AsyncMock(return_value=detected)
+        api._low_level_client = MagicMock()
+        api._low_level_client.create_from_upload = AsyncMock(return_value=("import_1", "url"))
+        job = MagicMock()
+        api.client.async_.jobs.get = AsyncMock(return_value=job)
+
+        assert await api.import_from_path(path, asset="my_asset", show_progress=False) is job
+
+    @pytest.mark.asyncio
+    async def test_mcap_data_cleared(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(
+            "sift_client.resources.data_imports.upload_file", lambda *a, **k: {"jobId": "j1"}
+        )
+        detected = McapImportConfig(
+            asset_name="",
+            data=[McapDataColumn(topic="/imu", field_path="x", data_type=ChannelDataType.DOUBLE)],
+        )
+
+        await self._import(tmp_path, "log.mcap", detected)
+
+        assert detected.data == []
+        assert detected.asset_name == "my_asset"
+
+    @pytest.mark.asyncio
+    async def test_ulog_data_cleared(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(
+            "sift_client.resources.data_imports.upload_file", lambda *a, **k: {"jobId": "j1"}
+        )
+        detected = UlogImportConfig(
+            asset_name="",
+            data=[
+                UlogDataColumn(
+                    message_name="sensor_accel", field_name="x", data_type=ChannelDataType.FLOAT
+                )
+            ],
+        )
+
+        await self._import(tmp_path, "log.ulg", detected)
+
+        assert detected.data == []
+
+
 class TestCsvToProto:
     def test_to_proto(self, csv_config):
         proto = csv_config._to_proto()
@@ -565,6 +740,9 @@ def test_known_extension_uses_map(self):
     def test_ulog_extension_uses_map(self):
         assert _resolve_data_type_key(".ulg", None) == DataTypeKey.ULOG
 
+    def test_mcap_extension_uses_map(self):
+        assert _resolve_data_type_key(".mcap", None) == DataTypeKey.MCAP
+
     def test_explicit_data_type_overrides_extension(self):
         result = _resolve_data_type_key(".csv", DataTypeKey.TDMS)
         assert result == DataTypeKey.TDMS
diff --git a/python/lib/sift_client/resources/data_imports.py b/python/lib/sift_client/resources/data_imports.py
index 794d03b5a..2f1612579 100644
--- a/python/lib/sift_client/resources/data_imports.py
+++ b/python/lib/sift_client/resources/data_imports.py
@@ -17,6 +17,7 @@
     DataTypeKey,
     Hdf5ImportConfig,
     ImportConfig,
+    McapImportConfig,
     ParquetFlatDatasetImportConfig,
     ParquetSingleChannelPerRowImportConfig,
     ParquetTimeColumn,
@@ -70,7 +71,7 @@ async def import_from_path(
         completion before proceeding.
 
         When ``config`` is omitted the file format is auto-detected via
-        ``detect_config`` (CSV, Parquet, HDF5, TDMS, and ULog).
+        ``detect_config`` (CSV, Parquet, HDF5, TDMS, ULog, and MCAP).
         When ``asset`` is provided it overrides the config value;
         otherwise the config's ``asset_name`` is used.
         If neither ``run`` nor ``run_name`` is provided (and none is
@@ -109,16 +110,17 @@ async def import_from_path(
             config: Import configuration describing the file format and column
                 mapping. When provided, ``data_type`` is ignored. If omitted,
                 the config is auto-detected via ``detect_config`` (for ULog
-                the detected channel list is dropped so every channel in the
-                file is imported). You can call ``detect_config`` yourself to
-                inspect and modify the config before passing it here.
+                and MCAP the detected channel list is dropped so every channel
+                in the file is imported). You can call ``detect_config``
+                yourself to inspect and modify the config before passing it
+                here.
             data_type: Explicit data type key. Required for formats with
                 multiple supported layouts (Parquet, HDF5) where the file
                 extension alone is ambiguous. Only used when ``config`` is
                 not provided.
             time_format: Time format override for CSV, Parquet, HDF5, and TDMS.
-                Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the
-                detected format if available, otherwise
+                Ignored for ULog and MCAP. When omitted, CSV, Parquet, and
+                HDF5 use the detected format if available, otherwise
                 ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its
                 detected/default time handling. Only used when ``config`` is
                 not provided.
@@ -147,11 +149,8 @@ async def import_from_path(
                 data_type=data_type,
                 time_format=time_format,
             )
-            if isinstance(config, UlogImportConfig):
-                # An empty channel list imports every channel. Keeping the
-                # detected list adds nothing and can fail the import when
-                # detection misreads a damaged file and lists channels the
-                # file does not contain.
+            if isinstance(config, (UlogImportConfig, McapImportConfig)):
+                # An empty channel list imports every channel
                 config.data = []
 
         if asset is not None:
@@ -328,9 +327,10 @@ async def detect_config(
         Returns the detected configuration, inferring the file format from the
         extension when ``data_type`` is not provided. CSV and Parquet are
         detected by sending a sample of the file to the server's DetectConfig
-        endpoint; TDMS, HDF5, and ULog are detected locally on the client.
+        endpoint; TDMS, HDF5, ULog, and MCAP are detected locally on the
+        client.
 
-        CSV, Parquet, HDF5, TDMS, and ULog files are supported for
+        CSV, Parquet, HDF5, TDMS, ULog, and MCAP files are supported for
         auto-detection.
 
         For CSV files, the server scans the first two rows for an optional
@@ -357,6 +357,15 @@ async def detect_config(
         to exactly those channels; the import fails if a listed channel is
         not in the file. Clear ``data`` to import every channel.
 
+        For MCAP files, ``data`` lists one channel per flattened field of each
+        supported topic, without decoding messages. A variable-cardinality
+        field is one entry; ``complex_types_import_mode`` on the config decides
+        whether it imports as Arrow IPC bytes, a JSON string under
+        ``<name>.json``, both, or neither. The same non-empty ``data``
+        semantics as ULog apply. Topics that cannot be decoded are skipped with
+        a warning; importing such a file fails unless
+        ``McapParseErrorPolicy.IGNORE_ERROR`` is set.
+
         For file types with multiple supported layouts (Parquet, HDF5),
         ``data_type`` must be specified explicitly.
 
@@ -366,8 +375,8 @@ async def detect_config(
                 multiple supported layouts (Parquet, HDF5) where the file
                 extension alone is ambiguous.
             time_format: Time format override for CSV, Parquet, HDF5, and TDMS.
-                Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the
-                detected format if available, otherwise
+                Ignored for ULog and MCAP. When omitted, CSV, Parquet, and
+                HDF5 use the detected format if available, otherwise
                 ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its
                 detected/default time handling.
 
@@ -389,7 +398,7 @@ async def detect_config(
         if time_format is not None:
             _apply_time_format(config, time_format)
         elif (
-            not isinstance(config, (TdmsImportConfig, UlogImportConfig))
+            not isinstance(config, (TdmsImportConfig, UlogImportConfig, McapImportConfig))
             and _get_time_format(config) is None
         ):
             _apply_time_format(config, TimeFormat.ABSOLUTE_UNIX_NANOSECONDS)
@@ -431,6 +440,15 @@ async def _detect_config_for_type(
                     "Install it via `pip install sift-stack-py[ulog]`."
                 ) from e
             return await run_sync_function(lambda: detect_ulog_config(path))
+        if data_type_key == DataTypeKey.MCAP:
+            try:
+                from sift_client._internal.util.mcap import detect_mcap_config
+            except ImportError as e:
+                raise RuntimeError(
+                    "mcap and mcap-ros2-support are required for MCAP import. "
+                    "Install them via `pip install sift-stack-py[mcap]`."
+                ) from e
+            return await run_sync_function(lambda: detect_mcap_config(path))
 
         is_parquet = data_type_key in (
             DataTypeKey.PARQUET_FLATDATASET,
@@ -468,7 +486,7 @@ def _read_sample() -> bytes:
 
         raise ValueError(
             f"No supported configuration detected for '{path.name}'. "
-            "Only CSV, Parquet, HDF5, TDMS, and ULog are supported by auto-detection."
+            "Only CSV, Parquet, HDF5, TDMS, ULog, and MCAP are supported by auto-detection."
         )
 
 
@@ -476,7 +494,8 @@ def _apply_time_format(config: ImportConfig, time_format: TimeFormat) -> None:
     """Set the time format on a detected config, dispatching by config type.
 
     CSV and Parquet store the format under ``time_column.format``. TDMS and
-    HDF5 store it on ``time_format``. ULog has no configurable time format.
+    HDF5 store it on ``time_format``. ULog and MCAP have no configurable
+    time format.
     """
     if isinstance(
         config,
diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi
index a030f4206..6d4f360dc 100644
--- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi
+++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi
@@ -717,9 +717,10 @@ class DataImportAPI:
         Returns the detected configuration, inferring the file format from the
         extension when ``data_type`` is not provided. CSV and Parquet are
         detected by sending a sample of the file to the server's DetectConfig
-        endpoint; TDMS, HDF5, and ULog are detected locally on the client.
+        endpoint; TDMS, HDF5, ULog, and MCAP are detected locally on the
+        client.
 
-        CSV, Parquet, HDF5, TDMS, and ULog files are supported for
+        CSV, Parquet, HDF5, TDMS, ULog, and MCAP files are supported for
         auto-detection.
 
         For CSV files, the server scans the first two rows for an optional
@@ -746,6 +747,15 @@ class DataImportAPI:
         to exactly those channels; the import fails if a listed channel is
         not in the file. Clear ``data`` to import every channel.
 
+        For MCAP files, ``data`` lists one channel per flattened field of each
+        supported topic, without decoding messages. A variable-cardinality
+        field is one entry; ``complex_types_import_mode`` on the config decides
+        whether it imports as Arrow IPC bytes, a JSON string under
+        ``<name>.json``, both, or neither. The same non-empty ``data``
+        semantics as ULog apply. Topics that cannot be decoded are skipped with
+        a warning; importing such a file fails unless
+        ``McapParseErrorPolicy.IGNORE_ERROR`` is set.
+
         For file types with multiple supported layouts (Parquet, HDF5),
         ``data_type`` must be specified explicitly.
 
@@ -755,8 +765,8 @@ class DataImportAPI:
                 multiple supported layouts (Parquet, HDF5) where the file
                 extension alone is ambiguous.
             time_format: Time format override for CSV, Parquet, HDF5, and TDMS.
-                Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the
-                detected format if available, otherwise
+                Ignored for ULog and MCAP. When omitted, CSV, Parquet, and
+                HDF5 use the detected format if available, otherwise
                 ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its
                 detected/default time handling.
 
@@ -840,7 +850,7 @@ class DataImportAPI:
         completion before proceeding.
 
         When ``config`` is omitted the file format is auto-detected via
-        ``detect_config`` (CSV, Parquet, HDF5, TDMS, and ULog).
+        ``detect_config`` (CSV, Parquet, HDF5, TDMS, ULog, and MCAP).
         When ``asset`` is provided it overrides the config value;
         otherwise the config's ``asset_name`` is used.
         If neither ``run`` nor ``run_name`` is provided (and none is
@@ -879,16 +889,17 @@ class DataImportAPI:
             config: Import configuration describing the file format and column
                 mapping. When provided, ``data_type`` is ignored. If omitted,
                 the config is auto-detected via ``detect_config`` (for ULog
-                the detected channel list is dropped so every channel in the
-                file is imported). You can call ``detect_config`` yourself to
-                inspect and modify the config before passing it here.
+                and MCAP the detected channel list is dropped so every channel
+                in the file is imported). You can call ``detect_config``
+                yourself to inspect and modify the config before passing it
+                here.
             data_type: Explicit data type key. Required for formats with
                 multiple supported layouts (Parquet, HDF5) where the file
                 extension alone is ambiguous. Only used when ``config`` is
                 not provided.
             time_format: Time format override for CSV, Parquet, HDF5, and TDMS.
-                Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the
-                detected format if available, otherwise
+                Ignored for ULog and MCAP. When omitted, CSV, Parquet, and
+                HDF5 use the detected format if available, otherwise
                 ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its
                 detected/default time handling. Only used when ``config`` is
                 not provided.
diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py
index 1966109df..570cc38a4 100644
--- a/python/lib/sift_client/sift_types/data_import.py
+++ b/python/lib/sift_client/sift_types/data_import.py
@@ -11,10 +11,17 @@
 from sift.data_imports.v2.data_imports_pb2 import (
     DATA_TYPE_KEY_CSV,
     DATA_TYPE_KEY_HDF5,
+    DATA_TYPE_KEY_MCAP,
     DATA_TYPE_KEY_PARQUET_FLATDATASET,
     DATA_TYPE_KEY_PARQUET_SINGLE_CHANNEL_PER_ROW,
     DATA_TYPE_KEY_TDMS,
     DATA_TYPE_KEY_ULOG,
+    MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH,
+    MCAP_COMPLEX_TYPES_IMPORT_MODE_BYTES,
+    MCAP_COMPLEX_TYPES_IMPORT_MODE_IGNORE,
+    MCAP_COMPLEX_TYPES_IMPORT_MODE_STRING,
+    MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR,
+    MCAP_PARSE_ERROR_POLICY_IGNORE_ERROR,
     PARQUET_COMPLEX_TYPES_IMPORT_MODE_BOTH,
     PARQUET_COMPLEX_TYPES_IMPORT_MODE_BYTES,
     PARQUET_COMPLEX_TYPES_IMPORT_MODE_IGNORE,
@@ -34,6 +41,9 @@
 from sift.data_imports.v2.data_imports_pb2 import DataImportStatus as DataImportStatusProto
 from sift.data_imports.v2.data_imports_pb2 import Hdf5Config as Hdf5ConfigProto
 from sift.data_imports.v2.data_imports_pb2 import Hdf5DataConfig as Hdf5DataConfigProto
+from sift.data_imports.v2.data_imports_pb2 import McapConfig as McapConfigProto
+from sift.data_imports.v2.data_imports_pb2 import McapDataConfig as McapDataConfigProto
+from sift.data_imports.v2.data_imports_pb2 import McapRos2Selector as McapRos2SelectorProto
 from sift.data_imports.v2.data_imports_pb2 import ParquetConfig as ParquetConfigProto
 from sift.data_imports.v2.data_imports_pb2 import ParquetDataColumn as ParquetDataColumnProto
 from sift.data_imports.v2.data_imports_pb2 import (
@@ -91,6 +101,7 @@ class DataTypeKey(Enum):
     HDF5_TWO_D = "hdf5_two_d"
     HDF5_COMPOUND = "hdf5_compound"
     ULOG = "ulog"
+    MCAP = "mcap"
 
 
 DATA_TYPE_KEY_TO_PROTO = {
@@ -102,6 +113,7 @@ class DataTypeKey(Enum):
     DataTypeKey.HDF5_TWO_D: DATA_TYPE_KEY_HDF5,
     DataTypeKey.HDF5_COMPOUND: DATA_TYPE_KEY_HDF5,
     DataTypeKey.ULOG: DATA_TYPE_KEY_ULOG,
+    DataTypeKey.MCAP: DATA_TYPE_KEY_MCAP,
 }
 
 
@@ -109,6 +121,7 @@ class DataTypeKey(Enum):
     ".csv": DataTypeKey.CSV,
     ".tdms": DataTypeKey.TDMS,
     ".ulg": DataTypeKey.ULOG,
+    ".mcap": DataTypeKey.MCAP,
 }
 
 
@@ -1014,6 +1027,216 @@ def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig:
         )
 
 
+class McapParseErrorPolicy(Enum):
+    """Controls how MCAP import handles recoverable parse errors.
+
+    Recoverable errors include truncated or undecodable records and
+    unsupported topics. The policy applies when the file is imported, not
+    during ``detect_config``.
+    """
+
+    FAIL_ON_ERROR = MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR
+    """Fail the import on any recoverable parse error."""
+
+    IGNORE_ERROR = MCAP_PARSE_ERROR_POLICY_IGNORE_ERROR
+    """Import what decoded. Skipped topics and records surface as warnings."""
+
+
+# Suffix given to the JSON channel of a variable-cardinality field.
+MCAP_JSON_CHANNEL_SUFFIX = ".json"
+
+
+class McapComplexTypesImportMode(Enum):
+    """Controls how variable-cardinality MCAP fields (dynamic and bounded
+    arrays) are imported.
+
+    Under ``BOTH``, each such field imports as two channels: Arrow IPC bytes
+    under the field's base name and a JSON string under ``<base name>.json``.
+    """
+
+    IGNORE = MCAP_COMPLEX_TYPES_IMPORT_MODE_IGNORE
+    BOTH = MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH
+    STRING = MCAP_COMPLEX_TYPES_IMPORT_MODE_STRING
+    BYTES = MCAP_COMPLEX_TYPES_IMPORT_MODE_BYTES
+
+
+class McapDataColumn(DataColumnBase):
+    """A single MCAP channel selection.
+
+    Channels are selected by topic and flattened field path, as returned by
+    ``detect_config``. A variable-cardinality field is selected whole by its
+    base path and imports per its ``data_type``: ``BYTES`` (Arrow IPC) or
+    ``STRING`` (JSON), which ``complex_types_import_mode`` must allow.
+
+    Attributes:
+        topic: The topic the channel comes from (e.g. ``"/imu/data"``).
+        field_path: The dot-delimited field path within the decoded message
+            (e.g. ``"orientation.x"``, ``"orientation_covariance[0]"``).
+        name: Sift channel name to create. Defaults to ``default_channel_name``,
+            e.g. ``"/imu/data.orientation.x"``.
+    """
+
+    topic: str
+    field_path: str
+    name: str = ""
+
+    @property
+    def default_channel_name(self) -> str:
+        """The default Sift channel name for this selection,
+        ``<topic>.<field_path>`` (e.g. ``"/imu/data.orientation.x"``).
+        """
+        return f"{self.topic}.{self.field_path}"
+
+    @model_validator(mode="after")
+    def _apply_default_name(self) -> McapDataColumn:
+        if not self.name:
+            self.name = self.default_channel_name
+        return self
+
+
+class McapImportConfig(ImportConfigBase):
+    """Configuration for importing an MCAP (``.mcap``) file.
+
+    MCAP files describe their own channels. Leave ``data`` empty to import
+    every detected channel, or call ``detect_config`` and edit the returned
+    ``data`` list to skip, rename, retype, or annotate channels before
+    importing.
+
+    Attributes:
+        data: Channel selections. If empty, imports all detected channels with
+            default names and data types. If non-empty, imports only these
+            channels.
+        relative_start_time: Log-start UTC, only for logs on a non-Unix epoch.
+            When set, ``log_time`` is reinterpreted as elapsed nanoseconds from
+            this start.
+        metadata_records: Metadata records to import as run metadata. Every key
+            of each named record is stored as ``<record_name>.<key>``. Empty
+            imports none.
+        parse_error_policy: How to handle recoverable parse errors. Defaults to
+            failing the import.
+        complex_types_import_mode: How to import variable-cardinality fields.
+            Defaults to importing them as both Arrow IPC bytes and JSON strings.
+            ``data`` lists one entry per field; the mode decides which channels
+            that entry becomes, so it can be changed on a detected config.
+    """
+
+    data: list[McapDataColumn] = []
+    relative_start_time: datetime | None = None
+    metadata_records: list[str] = []
+    parse_error_policy: McapParseErrorPolicy = McapParseErrorPolicy.FAIL_ON_ERROR
+    complex_types_import_mode: McapComplexTypesImportMode = McapComplexTypesImportMode.BOTH
+
+    def __getitem__(self, name: str) -> McapDataColumn:
+        """Look up a configured MCAP channel by Sift channel name.
+
+        Example::
+
+            config["/imu/data.orientation.x"].data_type = ChannelDataType.FLOAT
+        """
+        for dc in self.data:
+            if dc.name == name:
+                return dc
+        raise KeyError(f"No data column named '{name}'")
+
+    def _to_proto(self) -> McapConfigProto:
+        proto = McapConfigProto(
+            asset_name=self.asset_name,
+            run_name=self.run_name or "",
+            run_id=self.run_id or "",
+            metadata_records=self.metadata_records,
+            parse_error_policy=self.parse_error_policy.value,
+            complex_types_import_mode=self.complex_types_import_mode.value,
+        )
+        if self.relative_start_time is not None:
+            proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time))
+
+        mode = self.complex_types_import_mode
+        # Channel names are unique per asset and compare case-insensitively.
+        taken_names: dict[str, str] = {}
+
+        def add(dc: McapDataColumn, name: str, data_type: ChannelDataType) -> None:
+            source = taken_names.get(name.lower())
+            if source is not None:
+                raise ValueError(
+                    f"channels '{source}' and '{dc.name}' would both be imported as "
+                    f"'{name}'. Rename or remove one before importing."
+                )
+            taken_names[name.lower()] = dc.name
+            proto.data.append(
+                McapDataConfigProto(
+                    topic=dc.topic,
+                    ros2=McapRos2SelectorProto(field_path=dc.field_path),
+                    channel_config=ChannelConfigProto(
+                        name=name,
+                        data_type=data_type.value,
+                        units=dc.units,
+                        description=dc.description,
+                    ),
+                )
+            )
+
+        for dc in self.data:
+            # Only variable-cardinality fields can be BYTES, and the mode
+            # decides which channels they become.
+            if dc.data_type != ChannelDataType.BYTES:
+                add(dc, dc.name, dc.data_type)
+                continue
+            if mode is McapComplexTypesImportMode.IGNORE:
+                continue
+            if mode in (McapComplexTypesImportMode.BYTES, McapComplexTypesImportMode.BOTH):
+                add(dc, dc.name, ChannelDataType.BYTES)
+            if mode in (McapComplexTypesImportMode.STRING, McapComplexTypesImportMode.BOTH):
+                add(dc, dc.name + MCAP_JSON_CHANNEL_SUFFIX, ChannelDataType.STRING)
+
+        if self.data and not proto.data:
+            # An empty list means "import everything", which is not what
+            # selecting channels and then dropping them all should do.
+            raise ValueError(
+                "complex_types_import_mode is IGNORE and every configured channel is "
+                "variable-cardinality, so nothing would be imported. Choose another mode "
+                "or clear 'data' to import the whole file."
+            )
+        return proto
+
+    @classmethod
+    def _from_proto(cls, proto: McapConfigProto) -> McapImportConfig:
+        """Create from a proto McapConfig (e.g. from a GetDataImport response)."""
+        relative_start_time = None
+        if proto.HasField("relative_start_time"):
+            from datetime import timezone
+
+            relative_start_time = proto.relative_start_time.ToDatetime(tzinfo=timezone.utc)
+
+        parse_error_policy = McapParseErrorPolicy.FAIL_ON_ERROR
+        if proto.parse_error_policy == MCAP_PARSE_ERROR_POLICY_IGNORE_ERROR:
+            parse_error_policy = McapParseErrorPolicy.IGNORE_ERROR
+
+        mode = proto.complex_types_import_mode
+        data = [
+            McapDataColumn(
+                topic=d.topic,
+                field_path=d.ros2.field_path,
+                name=d.channel_config.name,
+                data_type=ChannelDataType(d.channel_config.data_type),
+                units=d.channel_config.units,
+                description=d.channel_config.description,
+            )
+            for d in proto.data
+        ]
+        return cls(
+            asset_name=proto.asset_name,
+            run_name=proto.run_name or None,
+            run_id=proto.run_id or None,
+            data=data,
+            relative_start_time=relative_start_time,
+            metadata_records=list(proto.metadata_records),
+            parse_error_policy=parse_error_policy,
+            complex_types_import_mode=McapComplexTypesImportMode(mode)
+            if mode
+            else McapComplexTypesImportMode.BOTH,
+        )
+
+
 ImportConfig = Union[
     CsvImportConfig,
     ParquetFlatDatasetImportConfig,
@@ -1021,6 +1244,7 @@ def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig:
     TdmsImportConfig,
     Hdf5ImportConfig,
     UlogImportConfig,
+    McapImportConfig,
 ]
 
 
diff --git a/python/pyproject.toml b/python/pyproject.toml
index 4aa459cd1..81dc5a893 100644
--- a/python/pyproject.toml
+++ b/python/pyproject.toml
@@ -63,6 +63,8 @@ import-test-result-log = "sift_client.scripts.import_test_result_log:main"
 all = [
     'cffi~=1.14',
     'h5py~=3.11',
+    'mcap-ros2-support~=0.5.7',
+    'mcap~=1.4',
     'npTDMS~=1.9',
     'polars~=1.8',
     'pyOpenSSL<24.0.0',
@@ -98,6 +100,8 @@ dev-all = [
     'cffi~=1.14',
     'grpcio-testing~=1.13',
     'h5py~=3.11',
+    'mcap-ros2-support~=0.5.7',
+    'mcap~=1.4',
     'mypy==1.10.0',
     'npTDMS~=1.9',
     'pdoc==14.5.0',
@@ -149,6 +153,8 @@ docs-build = [
     "griffe-pydantic==1.3.1 ; python_version >= '3.10'",
     'grpcio-testing~=1.13',
     'h5py~=3.11',
+    'mcap-ros2-support~=0.5.7',
+    'mcap~=1.4',
     "mike==2.1.3 ; python_version >= '3.10'",
     "mkdocs-api-autonav==0.4.0 ; python_version >= '3.10'",
     "mkdocs-include-markdown-plugin==7.1.6 ; python_version >= '3.10'",
@@ -179,6 +185,8 @@ docs-build = [
 ]
 file-imports = [
     'h5py~=3.11',
+    'mcap-ros2-support~=0.5.7',
+    'mcap~=1.4',
     'npTDMS~=1.9',
     'polars~=1.8',
     'pyulog~=1.2.2',
@@ -188,6 +196,10 @@ hdf5 = [
     'h5py~=3.11',
     'polars~=1.8',
 ]
+mcap = [
+    'mcap-ros2-support~=0.5.7',
+    'mcap~=1.4',
+]
 openssl = [
     'cffi~=1.14',
     'pyOpenSSL<24.0.0',
@@ -251,6 +263,7 @@ rosbags = ["rosbags~=0.0 ; python_full_version >= '3.8.2'"]
 sift-stream = ["sift-stream-bindings==0.5.0"]
 hdf5 = ["h5py~=3.11", "polars~=1.8"]  # polars is only used by sift_py; remove once sift_py is fully deprecated
 ulog = ["pyulog~=1.2.2"]
+mcap = ["mcap~=1.4", "mcap-ros2-support~=0.5.7"]
 data-review = ["pyarrow>=17.0.0"]
 
 [tool.sift.extras.combine]
@@ -261,7 +274,7 @@ dev = ["development"]
 sift-stream-bindings = ["sift-stream"]
 
 # combinations
-file-imports = ["tdms", "rosbags", "hdf5", "ulog"]
+file-imports = ["tdms", "rosbags", "hdf5", "ulog", "mcap"]
 all = ["openssl", "sift-stream", "file-imports", "data-review"]
 
 dev-all = ["development", "all", "build"]
@@ -394,6 +407,13 @@ module = "alive_progress"
 follow_imports = "skip"
 ignore_errors = true
 
+# mcap's writer does `from .__init__ import __version__`, which makes mypy
+# discover mcap/__init__.py under two module names and abort. Only the MCAP
+# detection tests import the writer.
+[[tool.mypy.overrides]]
+module = "mcap.writer"
+follow_imports = "skip"
+
 [tool.setuptools.packages.find]
 where = ["lib"]
 exclude = ["sift_client._tests", "sift_client._tests.*"]
diff --git a/python/uv.lock b/python/uv.lock
index 2a91f3bb5..b8a7b1bdc 100644
--- a/python/uv.lock
+++ b/python/uv.lock
@@ -1480,6 +1480,7 @@ version = "4.3.3"
 source = { registry = "https://pypi.org/simple" }
 resolution-markers = [
     "python_full_version >= '3.8.2' and python_full_version < '3.9'",
+    "python_full_version < '3.8.2'",
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/a4/31/ec1259ca8ad11568abaf090a7da719616ca96b60d097ccc5799cd0ff599c/lz4-4.3.3.tar.gz", hash = "sha256:01fe674ef2889dbb9899d8a67361e0c4a2c833af5aeb37dd505727cf5d2a131e", size = 171509, upload-time = "2024-01-01T23:03:13.535Z" }
 wheels = [
@@ -1795,6 +1796,33 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/41/09/5b161152e2d90f7b87f781c2e1267494aef9c32498df793f73ad0a0a494a/matplotlib_inline-0.2.2-py3-none-any.whl", hash = "sha256:3c821cf1c209f59fb2d2d64abbf5b23b67bcb2210d663f9918dd851c6da1fcf6", size = 9534, upload-time = "2026-05-08T17:33:32.055Z" },
 ]
 
+[[package]]
+name = "mcap"
+version = "1.4.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "lz4", version = "4.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
+    { name = "lz4", version = "4.4.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" },
+    { name = "zstandard", version = "0.23.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
+    { name = "zstandard", version = "0.25.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/f8/d7/0f17e59733a71bd4d5b38afc8484531c2cd7648b08c79863ee95fc42b002/mcap-1.4.0.tar.gz", hash = "sha256:0528e2f86a61bfec73779e0628e6cf27af83d01d89e20b27d5ec9f0b556a63ac", size = 22155, upload-time = "2026-06-18T21:50:07.965Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/48/14/7e0b2a74b67e16e5f40ab78cc3e5aa4e7bdd55e0aec963d573edc07a15cd/mcap-1.4.0-py3-none-any.whl", hash = "sha256:0b48b1cc951b8d5aabd2599e60d410bae4f1be1819094f54117b7cbf6b3ee2e9", size = 20826, upload-time = "2026-06-18T21:50:06.704Z" },
+]
+
+[[package]]
+name = "mcap-ros2-support"
+version = "0.5.7"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "mcap" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/84/0d/f6da01d8e4b73861dba17fd2ebd9a78d2fd4fa582ad348cf18d16a79a7e1/mcap_ros2_support-0.5.7.tar.gz", hash = "sha256:8ddb67e452a6e2963664e29bc8868e61d7011b72feb2b62832ed444e27ca3ab8", size = 23295, upload-time = "2025-12-24T21:25:37.614Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/92/c5/8e8636099c180031436a011908a22216c25e6cb7790ba748212e203317c4/mcap_ros2_support-0.5.7-py3-none-any.whl", hash = "sha256:349f0e0f7af8ebeb516003e801ea5c5de5b4b65359ce4a17a8d98b25a9cea260", size = 22086, upload-time = "2025-12-24T21:25:36.425Z" },
+]
+
 [[package]]
 name = "mdit-py-plugins"
 version = "0.6.1"
@@ -4486,6 +4514,8 @@ all = [
     { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
     { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
     { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+    { name = "mcap" },
+    { name = "mcap-ros2-support" },
     { name = "nptdms" },
     { name = "polars", version = "1.8.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
     { name = "polars", version = "1.36.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
@@ -4532,6 +4562,8 @@ dev-all = [
     { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
     { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
     { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+    { name = "mcap" },
+    { name = "mcap-ros2-support" },
     { name = "mypy" },
     { name = "nptdms" },
     { name = "pdoc" },
@@ -4592,6 +4624,8 @@ docs-build = [
     { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
     { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
     { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+    { name = "mcap" },
+    { name = "mcap-ros2-support" },
     { name = "mike", marker = "python_full_version >= '3.10'" },
     { name = "mkdocs", marker = "python_full_version >= '3.10'" },
     { name = "mkdocs-api-autonav", marker = "python_full_version >= '3.10'" },
@@ -4629,6 +4663,8 @@ file-imports = [
     { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
     { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
     { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+    { name = "mcap" },
+    { name = "mcap-ros2-support" },
     { name = "nptdms" },
     { name = "polars", version = "1.8.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" },
     { name = "polars", version = "1.36.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
@@ -4645,6 +4681,10 @@ hdf5 = [
     { name = "polars", version = "1.36.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" },
     { name = "polars", version = "1.40.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
 ]
+mcap = [
+    { name = "mcap" },
+    { name = "mcap-ros2-support" },
+]
 openssl = [
     { name = "cffi" },
     { name = "pyopenssl" },
@@ -4693,6 +4733,16 @@ requires-dist = [
     { name = "h5py", marker = "extra == 'docs-build'", specifier = "~=3.11" },
     { name = "h5py", marker = "extra == 'file-imports'", specifier = "~=3.11" },
     { name = "h5py", marker = "extra == 'hdf5'", specifier = "~=3.11" },
+    { name = "mcap", marker = "extra == 'all'", specifier = "~=1.4" },
+    { name = "mcap", marker = "extra == 'dev-all'", specifier = "~=1.4" },
+    { name = "mcap", marker = "extra == 'docs-build'", specifier = "~=1.4" },
+    { name = "mcap", marker = "extra == 'file-imports'", specifier = "~=1.4" },
+    { name = "mcap", marker = "extra == 'mcap'", specifier = "~=1.4" },
+    { name = "mcap-ros2-support", marker = "extra == 'all'", specifier = "~=0.5.7" },
+    { name = "mcap-ros2-support", marker = "extra == 'dev-all'", specifier = "~=0.5.7" },
+    { name = "mcap-ros2-support", marker = "extra == 'docs-build'", specifier = "~=0.5.7" },
+    { name = "mcap-ros2-support", marker = "extra == 'file-imports'", specifier = "~=0.5.7" },
+    { name = "mcap-ros2-support", marker = "extra == 'mcap'", specifier = "~=0.5.7" },
     { name = "mike", marker = "python_full_version >= '3.10' and extra == 'docs'", specifier = "==2.1.3" },
     { name = "mike", marker = "python_full_version >= '3.10' and extra == 'docs-build'", specifier = "==2.1.3" },
     { name = "mkdocs", marker = "python_full_version >= '3.10' and extra == 'docs'", specifier = "==1.6.1" },
@@ -4807,7 +4857,7 @@ requires-dist = [
     { name = "types-requests", specifier = "~=2.25" },
     { name = "typing-extensions", specifier = "~=4.6" },
 ]
-provides-extras = ["all", "build", "data-review", "dev", "dev-all", "development", "docs", "docs-build", "file-imports", "hdf5", "openssl", "rosbags", "sift-stream", "sift-stream-bindings", "tdms", "ulog"]
+provides-extras = ["all", "build", "data-review", "dev", "dev-all", "development", "docs", "docs-build", "file-imports", "hdf5", "mcap", "openssl", "rosbags", "sift-stream", "sift-stream-bindings", "tdms", "ulog"]
 
 [[package]]
 name = "sift-stream-bindings"
@@ -5433,9 +5483,10 @@ version = "0.23.0"
 source = { registry = "https://pypi.org/simple" }
 resolution-markers = [
     "python_full_version >= '3.8.2' and python_full_version < '3.9'",
+    "python_full_version < '3.8.2'",
 ]
 dependencies = [
-    { name = "cffi", marker = "python_full_version >= '3.8.2' and python_full_version < '3.9' and platform_python_implementation == 'PyPy'" },
+    { name = "cffi", marker = "python_full_version < '3.9' and platform_python_implementation == 'PyPy'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/ed/f6/2ac0287b442160a89d726b17a9184a4c615bb5237db763791a7fd16d9df1/zstandard-0.23.0.tar.gz", hash = "sha256:b2d8c62d08e7255f68f7a740bae85b3c9b8e5466baa9cbf7f57f1cde0ac6bc09", size = 681701, upload-time = "2024-07-15T00:18:06.141Z" }
 wheels = [