diff --git a/flowbio/cli/_accession_sheet.py b/flowbio/cli/_accession_sheet.py
new file mode 100644
index 0000000..cace45a
--- /dev/null
+++ b/flowbio/cli/_accession_sheet.py
@@ -0,0 +1,168 @@
+"""CSV accession-sheet parsing for ``samples import``.
+
+An accession sheet is a CSV with one row per accession to import: required
+``accession`` and ``sample_type`` columns, plus optional ``name``/
+``organism`` and per-accession metadata columns. This mirrors ``_sheet.py``'s
+reads-based sample sheet, but the reserved columns differ — there is nothing
+to upload (no ``reads1``/``reads2``) and the import API has no project field.
+
+Domain rules (accession format, duplicates, sample type, organism, metadata)
+are all checked server-side when the sheet is submitted — duplicating that
+locally would just be a second, driftable copy of the same rules.
+``accession`` and ``sample_type`` are different: they are the two columns
+every row must have to mean anything at all, so a missing one is rejected
+here rather than silently skipped or shipped as an empty string the server
+would just reject anyway. There is deliberately no other way to supply a
+sample type for ``samples import`` — the sheet is the single source of it.
+A row with every cell blank (e.g. a trailing comma-only line some spreadsheet
+exports append below the data) is skipped rather than treated as a row
+missing values, since there is nothing there to be missing.
+"""
+from __future__ import annotations
+
+import csv
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import Path
+
+from flowbio.cli._exit_codes import CliUsageError
+from flowbio.cli._files import existing_file
+from flowbio.v2.samples import SampleImportSpec, SampleTypeId
+
+RESERVED_COLUMNS = ("accession", "name", "organism", "sample_type")
+
+
+@dataclass(frozen=True)
+class AccessionSheetRow:
+    """One data row of an accession sheet."""
+
+    row_number: int
+    accession: str
+    name: str | None
+    organism: str | None
+    sample_type: SampleTypeId
+    metadata: dict[str, str]
+
+    def __post_init__(self) -> None:
+        if not self.accession:
+            raise ValueError("accession must not be empty")
+        if not self.sample_type:
+            raise ValueError("sample_type must not be empty")
+
+    def to_spec(self) -> SampleImportSpec:
+        """Build the :class:`~flowbio.v2.samples.SampleImportSpec` for this row."""
+        return SampleImportSpec(
+            accession=self.accession,
+            sample_type=self.sample_type,
+            name=self.name,
+            organism_id=self.organism,
+            metadata=self.metadata or None,
+        )
+
+
+@dataclass(frozen=True)
+class AccessionSheet:
+    """A parsed accession sheet."""
+
+    path: Path
+    rows: list[AccessionSheetRow]
+
+
+def parse_accession_sheet(path: Path) -> AccessionSheet:
+    """Parse a CSV accession sheet into an :class:`AccessionSheet`.
+
+    :param path: The accession-sheet file. Must be a ``.csv`` — an ``.xlsx`` or
+        ``.tsv`` is a usage error directing the user to export to CSV.
+    :returns: The parsed sheet, with wholly-blank rows skipped, empty cells
+        dropped, and surrounding whitespace trimmed (including in header
+        names). Values are otherwise passed through unchanged, including
+        ``accession``, sent to the server as-entered.
+    :raises CliUsageError: If the file is not a readable ``.csv``, has no
+        rows, or has a row with no accession or no sample_type.
+    """
+    if path.suffix.lower() != ".csv":
+        raise CliUsageError(
+            f"Accession sheet must be a .csv file: {path}. "
+            f"Export your spreadsheet to CSV first.",
+        )
+    existing_file(path)
+    rows: list[AccessionSheetRow] = []
+    missing_accession: list[int] = []
+    missing_sample_type: list[int] = []
+    # utf-8-sig transparently strips a leading BOM, which spreadsheet tools
+    # (notably Excel's "CSV UTF-8" export) prepend — otherwise the first header
+    # parses as "accession" and every row reports a missing accession.
+    with path.open(newline="", encoding="utf-8-sig") as handle:
+        reader = csv.DictReader(handle)
+        # A hand-authored header (unlike upload-batch's template-generated
+        # one) routinely has a stray space after a comma; reassigning
+        # fieldnames makes every row dict keyed by the trimmed name too.
+        headers = [header.strip() for header in reader.fieldnames or []]
+        reader.fieldnames = headers
+        metadata_columns = [
+            header for header in headers if header not in RESERVED_COLUMNS
+        ]
+        # Row 1 is the first row after the header, matching _sheet.py's
+        # convention (and upload-batch's documented "1-based row number").
+        for row_number, record in enumerate(reader, start=1):
+            if _is_blank_row(record, headers):
+                continue
+            accession = _cell(record, "accession")
+            sample_type = _cell(record, "sample_type")
+            if accession is None:
+                missing_accession.append(row_number)
+            if sample_type is None:
+                missing_sample_type.append(row_number)
+            if accession is None or sample_type is None:
+                continue
+            rows.append(_build_row(record, row_number, metadata_columns, accession, sample_type))
+    if not rows and not missing_accession and not missing_sample_type:
+        raise CliUsageError(f"Accession sheet has no rows: {path}.")
+    if missing_accession or missing_sample_type:
+        clauses = [
+            clause for clause in (
+                _missing_value_clause("accession", missing_accession),
+                _missing_value_clause("sample_type", missing_sample_type),
+            ) if clause is not None
+        ]
+        raise CliUsageError(f"Accession sheet {'; '.join(clauses)}: {path}.")
+    return AccessionSheet(path=path, rows=rows)
+
+
+def _is_blank_row(record: dict[str, str], headers: Sequence[str]) -> bool:
+    return not any((record.get(header) or "").strip() for header in headers)
+
+
+def _missing_value_clause(column: str, missing: list[int]) -> str | None:
+    if not missing:
+        return None
+    numbers = ", ".join(str(number) for number in missing)
+    verb = "has" if len(missing) == 1 else "have"
+    return f"data row(s) {numbers} {verb} no {column}"
+
+
+def _cell(record: dict[str, str], column: str) -> str | None:
+    value = (record.get(column) or "").strip()
+    return value or None
+
+
+def _build_row(
+    record: dict[str, str],
+    row_number: int,
+    metadata_columns: list[str],
+    accession: str,
+    sample_type: str,
+) -> AccessionSheetRow:
+    metadata = {
+        column: value
+        for column in metadata_columns
+        if (value := _cell(record, column)) is not None
+    }
+    return AccessionSheetRow(
+        row_number=row_number,
+        accession=accession,
+        name=_cell(record, "name"),
+        organism=_cell(record, "organism"),
+        sample_type=SampleTypeId(sample_type),
+        metadata=metadata,
+    )
diff --git a/flowbio/cli/_samples.py b/flowbio/cli/_samples.py
index abd1896..9633221 100644
--- a/flowbio/cli/_samples.py
+++ b/flowbio/cli/_samples.py
@@ -11,9 +11,11 @@
 import argparse
 import json
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 from typing import Literal
 
+from flowbio.cli._accession_sheet import parse_accession_sheet
 from flowbio.cli._exit_codes import CliUsageError, ExitCode
 from flowbio.cli._files import existing_file
 from flowbio.cli._output import Output, format_issue
@@ -26,7 +28,12 @@
 from flowbio.cli._types import JsonValue
 from flowbio.v2.client import Client
 from flowbio.v2.exceptions import FlowApiError
-from flowbio.v2.samples import MetadataAttribute, SampleTypeId
+from flowbio.v2.samples import (
+    MetadataAttribute,
+    SampleImportJob,
+    SampleImportJobId,
+    SampleTypeId,
+)
 
 
 def register(
@@ -76,6 +83,22 @@ def register(
             "valid rows sequentially, reporting each row's outcome."
         ),
     ))
+    _configure_import(verbs.add_parser(
+        "import",
+        parents=[global_parent],
+        help="Import samples from public-repository accessions.",
+        description=(
+            "Kick off a batch import job from a CSV accession sheet and report "
+            "its id. Does not wait for the job to finish — check its progress "
+            "with 'samples import-status'."
+        ),
+    ))
+    _configure_import_status(verbs.add_parser(
+        "import-status",
+        parents=[global_parent],
+        help="Check the status of an import job.",
+        description="Fetch and report the current state of a 'samples import' job.",
+    ))
 
 
 def _configure_upload(upload: argparse.ArgumentParser) -> None:
@@ -233,6 +256,38 @@ def _configure_upload_batch(upload_batch: argparse.ArgumentParser) -> None:
     )
 
 
+def _configure_import(import_parser: argparse.ArgumentParser) -> None:
+    import_parser.set_defaults(command_parser=import_parser, handler=_import_command)
+    import_parser.add_argument(
+        "--sheet",
+        required=True,
+        metavar="PATH",
+        type=Path,
+        help=(
+            "CSV accession sheet (required accession/sample_type columns, "
+            "optional name/organism, plus metadata columns)."
+        ),
+    )
+
+
+def _job_id(value: str) -> SampleImportJobId:
+    try:
+        return SampleImportJobId(int(value))
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"job id must be an integer, got {value!r}") from None
+
+
+def _configure_import_status(import_status: argparse.ArgumentParser) -> None:
+    import_status.set_defaults(command_parser=import_status, handler=_import_status_command)
+    import_status.add_argument(
+        "--job-id",
+        required=True,
+        metavar="ID",
+        type=_job_id,
+        help="Import job id, as reported by 'samples import'.",
+    )
+
+
 def _upload_command(args: argparse.Namespace, client: Client, output: Output) -> ExitCode:
     """Upload a single sample and report its identifier.
 
@@ -558,6 +613,75 @@ def _invalid_line(row: SheetRow, reasons: list[str]) -> str:
     return f"Row {row.row_number} ({row.name}): {'; '.join(reasons)}"
 
 
+def _import_command(
+    args: argparse.Namespace, client: Client, output: Output,
+) -> ExitCode:
+    """Kick off a batch import job from an accession sheet and report its id.
+
+    Every row is submitted as-is: the accession format, sample type,
+    organism, and metadata rules are all validated server-side, so a
+    malformed sheet surfaces as a normal :class:`FlowApiError` rather than a
+    local pre-flight rejection. This command does not wait for the job to
+    finish — poll it yourself with ``samples import-status``.
+
+    :param args: Parsed command-line arguments.
+    :param client: The authenticated Flow client.
+    :param output: The result/error renderer.
+    :returns: :attr:`ExitCode.SUCCESS` once the job has been kicked off.
+    :raises CliUsageError: If the sheet is not a readable ``.csv``, has no
+        rows, or has a row with no accession or no sample_type.
+    """
+    sheet = parse_accession_sheet(args.sheet)
+    specs = [row.to_spec() for row in sheet.rows]
+    job = client.samples.import_samples(specs)
+    output.emit_result(
+        f"Started import job {job.id} for {len(specs)} accession(s) "
+        f"(status: {job.status}). Check progress with "
+        f"'flowbio samples import-status --job-id {job.id}'.",
+        job.model_dump(mode="json"),
+    )
+    return ExitCode.SUCCESS
+
+
+def _import_status_command(
+    args: argparse.Namespace, client: Client, output: Output,
+) -> ExitCode:
+    """Fetch and report the current state of an import job.
+
+    :param args: Parsed command-line arguments.
+    :param client: The authenticated Flow client.
+    :param output: The result/error renderer.
+    :returns: :attr:`ExitCode.SUCCESS` once the job's state has been fetched,
+        unless the job itself is ``"FAILED"``, in which case
+        :attr:`ExitCode.RUNTIME` — so a caller polling this command can
+        branch on its exit code alone, without parsing ``--json`` output.
+    """
+    job = client.samples.get_import(args.job_id)
+    if job.status == "FAILED":
+        output.emit_advisory(f"Job {job.id} failed: {job.error or 'no error message returned'}")
+    output.emit_result(_job_summary(job), job.model_dump(mode="json"))
+    return ExitCode.RUNTIME if job.status == "FAILED" else ExitCode.SUCCESS
+
+
+def _job_summary(job: SampleImportJob) -> str:
+    if job.status == "COMPLETED":
+        ids = ", ".join(str(sample_id) for sample_id in job.sample_ids) or "none"
+        suffix = _timestamp_suffix("finished", job.finished)
+        return f"Job {job.id}: COMPLETED{suffix}. Sample ids: {ids}."
+    if job.status == "FAILED":
+        # A human sees the error as a separate advisory on stderr, so it
+        # isn't repeated here.
+        return f"Job {job.id}: FAILED{_timestamp_suffix('finished', job.finished)}."
+    label = "started" if job.started else "created"
+    return f"Job {job.id}: {job.status}{_timestamp_suffix(label, job.started or job.created)}."
+
+
+def _timestamp_suffix(label: str, timestamp: datetime | None) -> str:
+    if timestamp is None:
+        return ""
+    return f" ({label} {timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')})"
+
+
 def _merge_metadata(
     pairs: list[str] | None, json_text: str | None,
 ) -> dict[str, str]:
diff --git a/flowbio/v2/__init__.py b/flowbio/v2/__init__.py
index 1b239ee..d0d5f1f 100644
--- a/flowbio/v2/__init__.py
+++ b/flowbio/v2/__init__.py
@@ -45,6 +45,10 @@
     Organism,
     Project,
     Sample,
+    SampleImportJob,
+    SampleImportJobId,
+    SampleImportSpec,
+    SampleImportStatus,
     SampleType,
     SampleTypeId,
 )
@@ -59,6 +63,10 @@
     "Organism",
     "Project",
     "Sample",
+    "SampleImportJob",
+    "SampleImportJobId",
+    "SampleImportSpec",
+    "SampleImportStatus",
     "SampleType",
     "SampleTypeId",
     "TokenCredentials",
diff --git a/flowbio/v2/samples.py b/flowbio/v2/samples.py
index fd888ed..26b48ea 100644
--- a/flowbio/v2/samples.py
+++ b/flowbio/v2/samples.py
@@ -27,10 +27,12 @@
 from __future__ import annotations
 
 from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING, NewType
+from typing import TYPE_CHECKING, Literal, NewType
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 
 from flowbio.v2._pagination import PageIterator
 from flowbio.v2.exceptions import (
@@ -139,6 +141,94 @@ class MultiplexedUpload(BaseModel, frozen=True):
     )
 
 
+SampleImportJobId = NewType("SampleImportJobId", int)
+"""The identifier of a sample-import job, as returned by
+:meth:`SampleResource.import_samples`."""
+
+
+SampleImportStatus = Literal["RUNNING", "COMPLETED", "FAILED"]
+"""The lifecycle state of a :class:`SampleImportJob`."""
+
+
+@dataclass(frozen=True)
+class SampleImportSpec:
+    """One accession to import, with its per-accession identity and metadata.
+
+    Example::
+
+        specs = [
+            SampleImportSpec(accession="ERR1160845", sample_type="RNA-Seq"),
+            SampleImportSpec(
+                accession="ERR10677146",
+                sample_type="RNA-Seq",
+                organism_id="Hs",
+                metadata={"strandedness": "reverse"},
+            ),
+        ]
+
+    :param accession: The public-repository run or experiment accession
+        (e.g. ``"ERR1160845"``), validated server-side.
+    :param sample_type: The sample type identifier (e.g. ``"RNA-Seq"``),
+        validated server-side. See :meth:`SampleResource.get_types`.
+    :param name: Optional sample name. Defaults to ``accession`` server-side
+        when omitted.
+    :param organism_id: Optional organism id (e.g. ``"Hs"``) to associate
+        with the sample, sent as ``organism``.
+    :param metadata: Optional metadata key-value pairs. See
+        :ref:`metadata-attributes` for details on required attributes.
+    """
+
+    accession: str
+    sample_type: SampleTypeId
+    name: str | None = None
+    organism_id: str | None = None
+    metadata: dict[str, str] | None = None
+
+
+class SampleImportJob(BaseModel, frozen=True):
+    """A batch job that imports one or more accessions into samples.
+
+    All accessions submitted in one :meth:`SampleResource.import_samples` call
+    share a single job: ``status`` and ``error`` describe the whole batch, and
+    ``accessions``/``sample_ids`` correspond positionally once ``status`` is
+    ``"COMPLETED"``.
+    """
+
+    id: SampleImportJobId = Field(description="Unique identifier for this import job.")
+    status: SampleImportStatus = Field(description="The job's current lifecycle state.")
+    created: datetime | None = Field(default=None, description="When the job was created.")
+    started: datetime | None = Field(default=None, description="When the job started running, if it has.")
+    finished: datetime | None = Field(
+        default=None, description="When the job finished (completed or failed), if it has.",
+    )
+    accessions: list[str] = Field(
+        default_factory=list, description="The accessions submitted with this job, in submission order.",
+    )
+    sample_ids: list[int] = Field(
+        default_factory=list,
+        description="The created samples' ids, corresponding to ``accessions`` once the job has completed.",
+    )
+    execution_id: int | None = Field(
+        default=None, description="The pipeline execution backing this job, if one was created.",
+    )
+    error: str | None = Field(
+        default=None, description='The failure reason, set only when status is "FAILED".',
+    )
+
+    @field_validator("created", "started", "finished", mode="after")
+    @classmethod
+    def _normalize_to_utc(cls, value: datetime | None) -> datetime | None:
+        if value is None:
+            return None
+        # A naive value (no tzinfo) is treated as already UTC, matching what
+        # the server always means by these timestamps, rather than left
+        # ambiguous for every consumer (CLI, --json, library) to decide on
+        # its own.
+        if value.tzinfo is None:
+            return value.replace(tzinfo=timezone.utc)
+        return value.astimezone(timezone.utc)
+
+
 class SampleResource:
     """Provides access to sample-related API endpoints.
 
@@ -399,6 +489,53 @@ def get_metadata_attributes(self) -> list[MetadataAttribute]:
         """
         return [self._create_metadata_attribute(item) for item in (self._transport.get("/samples/metadata"))]
 
+    def import_samples(self, imports: Sequence[SampleImportSpec]) -> SampleImportJob:
+        """Kick off a batch import of samples from public-repository accessions.
+
+        Every accession is submitted together and tracked as a single job — poll
+        it with :meth:`get_import` until its status leaves ``"RUNNING"``.
+
+        Requires authentication.
+
+        Example::
+
+            job = client.samples.import_samples([
+                SampleImportSpec(accession="ERR1160845", sample_type="RNA-Seq"),
+            ])
+            print(f"Import job: {job.id}")
+
+        :param imports: The accessions to import, one :class:`SampleImportSpec` each.
+        :raises FlowApiError: If any entry is invalid, e.g. an unsupported
+            accession format, unknown sample type, or missing required metadata.
+        """
+        payload = {"imports": [self._import_spec_fields(spec) for spec in imports]}
+        return SampleImportJob(**self._transport.post("/v2/sample-imports", json=payload))
+
+    def get_import(self, job_id: SampleImportJobId) -> SampleImportJob:
+        """Fetch the current state of an import job.
+
+        Example::
+
+            job = client.samples.get_import(job.id)
+            if job.status == "COMPLETED":
+                print(f"Imported samples: {job.sample_ids}")
+
+        :param job_id: The job id returned by :meth:`import_samples`.
+        :raises NotFoundError: If no import job with that id exists.
+        """
+        return SampleImportJob(**self._transport.get(f"/v2/sample-imports/{job_id}"))
+
+    @staticmethod
+    def _import_spec_fields(spec: SampleImportSpec) -> dict:
+        fields: dict = {"accession": spec.accession, "sample_type": spec.sample_type}
+        if spec.name is not None:
+            fields["name"] = spec.name
+        if spec.organism_id is not None:
+            fields["organism"] = spec.organism_id
+        if spec.metadata:
+            fields["metadata"] = spec.metadata
+        return fields
+
     def _create_metadata_attribute(self, item: dict) -> MetadataAttribute:
         item["required_for_sample_types"] = [
             SampleTypeId(link["sample_type_identifier"])
diff --git a/source/cli.rst b/source/cli.rst
index 029c86a..153d227 100644
--- a/source/cli.rst
+++ b/source/cli.rst
@@ -367,6 +367,138 @@ authentication failure; otherwise the standard mapping above.
     $ flowbio samples upload-batch --sheet ./samples.csv --sample-type RNA-Seq --json
     {"uploaded": [{"row_number": 1, "name": "liver_r1", "sample_id": "samp_1"}], "failed": [], "skipped": [], "counts": {"uploaded": 1, "failed": 0, "skipped": 0}}
 
+``samples import``
+~~~~~~~~~~~~~~~~~~
+
+Kick off a batch import of samples from public-repository accessions (SRR/
+ERR/DRR run or SRX/ERX/DRX experiment accessions) — no files to upload
+yourself.
+
+::
+
+    flowbio samples import --sheet PATH
+
+Run ``flowbio samples import --help`` for the full option list. The sheet is
+a CSV with required ``accession``/``sample_type`` columns, plus optional
+``name``/``organism`` and metadata columns (there is no ``batch-template``
+equivalent for it, since it has no reads files or project field). ``name``
+defaults to the accession when omitted. There is deliberately no
+``--sample-type`` flag: the sheet's own column is the only way to supply a
+sample type, so a mixed-type sheet needs no special handling and a
+single-type sheet just repeats the same value down the column. Every value
+is sent as-is (surrounding whitespace trimmed, including in header names);
+the accession format, sample type, and metadata rules are validated
+**server-side**. This command only checks
+that the sheet is a readable ``.csv`` and that every row has an accession
+and a sample type — a blank cell in either column rejects the whole sheet
+up front rather than shipping something the server would just reject
+anyway. A row with every cell blank (e.g. a trailing comma-only line some
+spreadsheet exports leave below the data) is skipped rather than treated as
+a row missing values. Rows are counted from ``1`` for the first data row,
+after the header (the same convention as ``upload-batch``'s ``row_number``).
+
+Unlike ``upload-batch``, a metadata column named ``<identifier>__annotation``
+is **not** given any special handling here — it is forwarded as an ordinary
+metadata key, which the server does not recognise, so it is silently
+ignored rather than attached as an annotation or rejected.
+
+Every row is submitted **together as one server-side job**. This command
+does **not wait for it to finish** — it reports the job's id and initial
+status (almost always ``"RUNNING"``) and returns immediately. Check on it
+with ``samples import-status --job-id ID``; polling (if you want it) is up
+to you, e.g. in a shell loop. Building the same thing directly against the
+library instead of the CLI? See :ref:`sample-imports`.
+
+**Output** — human: a confirmation line with the job id and a pointer to
+``import-status``. ``--json``: the created job as a single document —
+``id``, ``status``, ``created``/``started``/``finished`` (ISO 8601
+timestamps, ``null`` if not yet reached), ``accessions``, ``sample_ids``
+(empty until the job completes), ``execution_id``, ``error``.
+
+**Exit codes** — ``0`` the job was created (regardless of its eventual
+outcome — check that with ``import-status``); ``2`` the sheet isn't a
+readable ``.csv``, has no rows, or has a row with no accession or no
+sample type; ``1`` the API rejected the batch (e.g. unknown sample type,
+missing required metadata, an unsupported accession format — these come
+back as an HTTP ``422``; ``5`` in the unlikely case it answers ``400``
+instead); ``3`` authentication failure; otherwise the standard mapping
+above.
+
+**Example**
+
+``accessions.csv``:
+
+.. code-block:: text
+
+    accession,sample_type,name,organism
+    ERR1160845,RNA-Seq,liver_r1,Hs
+    ERR10677146,RNA-Seq,,
+
+.. code-block:: bash
+
+    $ flowbio samples import --sheet ./accessions.csv
+    Started import job 42 for 2 accession(s) (status: RUNNING). Check progress with 'flowbio samples import-status --job-id 42'.
+
+    $ flowbio samples import --sheet ./accessions.csv --json
+    {"id": 42, "status": "RUNNING", "created": "2024-04-05T19:34:38Z", "started": null, "finished": null, "accessions": ["ERR1160845", "ERR10677146"], "sample_ids": [], "execution_id": null, "error": null}
+
+``samples import-status``
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Fetch and report the current state of a ``samples import`` job.
+
+::
+
+    flowbio samples import-status --job-id ID
+
+Read-only — checking a job's status never changes it. There is no built-in
+polling; run this again (or wrap it in your own loop) until ``status`` leaves
+``"RUNNING"``. Gate the loop on ``flowbio``'s own exit code (assigning the
+document, not a value piped through ``jq``, in the ``while`` condition) so a
+transient failure (auth, network) breaks the loop instead of being read as
+``"RUNNING"``:
+
+.. code-block:: bash
+
+    while out=$(flowbio samples import-status --job-id 42 --json); do
+        [ "$(printf '%s' "$out" | jq -r .status)" = RUNNING ] || break
+        sleep 30
+    done
+    printf '%s' "$out" | jq -r .status    # COMPLETED / FAILED; empty if the command errored
+
+**Output** — human: a one-line summary including the sample ids and when it
+finished on ``"COMPLETED"``, when it started — or, if it hasn't yet, when it
+was created — on ``"RUNNING"``, or — on ``"FAILED"`` — when it finished plus
+the error as a separate advisory on stderr (so it isn't printed twice).
+``--json``: the job as a single document — ``id``, ``status``,
+``created``/``started``/``finished`` (ISO 8601 timestamps, useful for judging
+how long a job has been running when you've resumed polling one from
+elsewhere), ``accessions``, ``sample_ids``, ``execution_id``, ``error``.
+``--json`` never prints prose to stderr (or anywhere but that one stdout
+document); the failure reason on a ``"FAILED"`` job is the document's
+``error`` field, not a separate message.
+
+**Exit codes** — ``0`` the job was fetched and is ``"RUNNING"`` or
+``"COMPLETED"``; ``1`` either the job was fetched but is ``"FAILED"``, or the
+request itself failed (e.g. a transient server error) — human mode
+distinguishes them (an ``Error:`` line means the request failed; a
+``Job N: FAILED.`` line means the job did), and under ``--json`` a request
+failure's document is on stderr while a ``FAILED`` job's is on stdout like
+any other successful fetch. The loop above still needs ``--json``/``jq`` to
+tell ``"RUNNING"`` from ``"COMPLETED"``, since both exit ``0``; ``4`` no job
+with that id exists; ``3`` authentication failure; otherwise the standard
+mapping above.
+
+**Example**
+
+.. code-block:: bash
+
+    $ flowbio samples import-status --job-id 42
+    Job 42: COMPLETED (finished 2024-04-05 19:38:20 UTC). Sample ids: 101, 102.
+
+    $ flowbio samples import-status --job-id 42 --json
+    {"id": 42, "status": "COMPLETED", "created": "2024-04-05T19:34:38Z", "started": "2024-04-05T19:34:40Z", "finished": "2024-04-05T19:38:20Z", "accessions": ["ERR1160845", "ERR10677146"], "sample_ids": [101, 102], "execution_id": 7, "error": null}
+
 ``api get``
 ~~~~~~~~~~~
 
diff --git a/source/v2/samples.rst b/source/v2/samples.rst
index 6fb5dd2..3ebeafd 100644
--- a/source/v2/samples.rst
+++ b/source/v2/samples.rst
@@ -166,6 +166,48 @@ To reject the upload when warnings are present, set
 This raises :class:`~flowbio.v2.exceptions.AnnotationValidationError`
 if the annotation has any warnings.
 
+.. _sample-imports:
+
+Importing samples from public repositories
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Use :meth:`~flowbio.v2.samples.SampleResource.import_samples` to create
+samples directly from public-repository run or experiment accessions
+(SRR/ERR/DRR or SRX/ERX/DRX), instead of uploading files yourself. Every
+accession submitted together is tracked as a single job::
+
+    from flowbio.v2 import SampleImportSpec
+
+    job = client.samples.import_samples([
+        SampleImportSpec(accession="ERR1160845", sample_type="RNA-Seq"),
+        SampleImportSpec(
+            accession="ERR10677146",
+            sample_type="RNA-Seq",
+            metadata={"strandedness": "reverse"},
+        ),
+    ])
+
+The job starts out ``"RUNNING"``. Poll it with
+:meth:`~flowbio.v2.samples.SampleResource.get_import` until its status
+leaves ``"RUNNING"`` — bound the wait so a stuck job doesn't loop forever::
+
+    import time
+
+    deadline = time.monotonic() + 1800  # 30 minutes
+    while job.status == "RUNNING" and time.monotonic() < deadline:
+        time.sleep(5)
+        job = client.samples.get_import(job.id)
+
+    if job.status == "COMPLETED":
+        print(f"Imported samples: {job.sample_ids}")
+    elif job.status == "FAILED":
+        print(f"Import failed: {job.error}")
+    else:
+        print(f"Import did not finish within the deadline (status: {job.status})")
+
+``job.accessions`` and ``job.sample_ids`` correspond positionally once the
+job has completed.
+
 API Reference
 -------------
 
@@ -185,4 +227,13 @@ Models
 
 .. autopydantic_model:: flowbio.v2.samples.Organism
 
-.. autopydantic_model:: flowbio.v2.samples.MultiplexedUpload
\ No newline at end of file
+.. autopydantic_model:: flowbio.v2.samples.MultiplexedUpload
+
+.. autoclass:: flowbio.v2.samples.SampleImportSpec
+   :members:
+
+.. autopydantic_model:: flowbio.v2.samples.SampleImportJob
+
+.. autodata:: flowbio.v2.samples.SampleImportJobId
+
+.. autodata:: flowbio.v2.samples.SampleImportStatus
diff --git a/tests/unit/cli/test_accession_sheet.py b/tests/unit/cli/test_accession_sheet.py
new file mode 100644
index 0000000..64fac0f
--- /dev/null
+++ b/tests/unit/cli/test_accession_sheet.py
@@ -0,0 +1,265 @@
+import csv
+from pathlib import Path
+
+import pytest
+
+from flowbio.cli._accession_sheet import AccessionSheetRow, parse_accession_sheet
+from flowbio.cli._exit_codes import CliUsageError
+from flowbio.v2.samples import SampleImportSpec
+
+HEADERS = ["accession", "name", "organism", "sample_type", "cell_type", "source", "source__annotation"]
+
+
+def _write_sheet(
+    directory: Path, *records: dict[str, str], headers: list[str] = HEADERS,
+) -> Path:
+    path = directory / "sheet.csv"
+    with path.open("w", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=headers)
+        writer.writeheader()
+        writer.writerows(records)
+    return path
+
+
+def _record(**overrides: str) -> dict[str, str]:
+    record = {"accession": "ERR1160845", "sample_type": "rna_seq"}
+    record.update(overrides)
+    return record
+
+
+class TestParseAccessionSheet:
+
+    def test_accession_is_passed_through_unchanged(self, tmp_path: Path) -> None:
+        sheet = parse_accession_sheet(
+            _write_sheet(tmp_path, _record(accession="err1160845")),
+        )
+
+        assert sheet.rows[0].accession == "err1160845"
+
+    def test_empty_cells_omitted_from_metadata(self, tmp_path: Path) -> None:
+        sheet = parse_accession_sheet(_write_sheet(
+            tmp_path,
+            _record(cell_type="", source="blood"),
+        ))
+
+        assert sheet.rows[0].metadata == {"source": "blood"}
+
+    def test_name_and_organism_are_optional(self, tmp_path: Path) -> None:
+        sheet = parse_accession_sheet(
+            _write_sheet(tmp_path, _record()),
+        )
+
+        assert sheet.rows[0].name is None
+        assert sheet.rows[0].organism is None
+
+    def test_name_and_organism_are_parsed_when_present(self, tmp_path: Path) -> None:
+        sheet = parse_accession_sheet(_write_sheet(
+            tmp_path,
+            _record(name="liver_r1", organism="Hs"),
+        ))
+
+        assert sheet.rows[0].name == "liver_r1"
+        assert sheet.rows[0].organism == "Hs"
+
+    def test_sample_type_is_parsed(self, tmp_path: Path) -> None:
+        sheet = parse_accession_sheet(_write_sheet(
+            tmp_path, _record(sample_type="chip_seq"),
+        ))
+
+        assert sheet.rows[0].sample_type == "chip_seq"
+
+    def test_utf8_bom_is_stripped_from_first_header(self, tmp_path: Path) -> None:
+        path = tmp_path / "sheet.csv"
+        with path.open("w", newline="", encoding="utf-8-sig") as handle:
+            writer = csv.DictWriter(handle, fieldnames=HEADERS)
+            writer.writeheader()
+            writer.writerow(_record())
+
+        sheet = parse_accession_sheet(path)
+
+        assert sheet.rows[0].accession == "ERR1160845"
+
+    def test_row_numbers_are_one_based(self, tmp_path: Path) -> None:
+        sheet = parse_accession_sheet(_write_sheet(
+            tmp_path,
+            _record(accession="ERR1160845"),
+            _record(accession="ERR10677146"),
+        ))
+
+        assert [row.row_number for row in sheet.rows] == [1, 2]
+
+    def test_header_with_spaces_after_commas_is_still_recognised(self, tmp_path: Path) -> None:
+        path = tmp_path / "sheet.csv"
+        path.write_text("accession, sample_type, name\nERR1160845, rna_seq, liver_r1\n")
+
+        sheet = parse_accession_sheet(path)
+
+        assert sheet.rows[0].accession == "ERR1160845"
+        assert sheet.rows[0].sample_type == "rna_seq"
+        assert sheet.rows[0].name == "liver_r1"
+        assert sheet.rows[0].metadata == {}
+
+    def test_non_csv_xlsx_rejected_with_export_message(self, tmp_path: Path) -> None:
+        xlsx = tmp_path / "sheet.xlsx"
+        xlsx.write_bytes(b"PK")
+
+        with pytest.raises(CliUsageError, match="CSV"):
+            parse_accession_sheet(xlsx)
+
+    def test_tsv_sheet_rejected(self, tmp_path: Path) -> None:
+        tsv = tmp_path / "sheet.tsv"
+        tsv.write_text("accession\nERR1160845\n")
+
+        with pytest.raises(CliUsageError, match="CSV"):
+            parse_accession_sheet(tsv)
+
+    def test_missing_file_is_usage_error(self, tmp_path: Path) -> None:
+        with pytest.raises(CliUsageError):
+            parse_accession_sheet(tmp_path / "absent.csv")
+
+    def test_header_only_sheet_is_usage_error(self, tmp_path: Path) -> None:
+        with pytest.raises(CliUsageError, match="no rows"):
+            parse_accession_sheet(_write_sheet(tmp_path))
+
+    def test_row_with_blank_accession_is_usage_error(self, tmp_path: Path) -> None:
+        with pytest.raises(CliUsageError, match=r"data row\(s\) 2 has no accession"):
+            parse_accession_sheet(_write_sheet(
+                tmp_path,
+                _record(accession="ERR1160845"),
+                _record(accession=""),
+            ))
+
+    def test_row_with_no_accession_column_is_usage_error(self, tmp_path: Path) -> None:
+        path = tmp_path / "sheet.csv"
+        with path.open("w", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=["run", "name", "sample_type"])
+            writer.writeheader()
+            writer.writerow({"run": "ERR1160845", "name": "liver_r1", "sample_type": "rna_seq"})
+
+        with pytest.raises(CliUsageError):
+            parse_accession_sheet(path)
+
+    def test_usage_error_names_every_row_missing_an_accession(self, tmp_path: Path) -> None:
+        with pytest.raises(CliUsageError, match=r"data row\(s\) 1, 3 have no accession"):
+            parse_accession_sheet(_write_sheet(
+                tmp_path,
+                _record(accession=""),
+                _record(accession="ERR1"),
+                _record(accession=""),
+            ))
+
+    def test_row_with_blank_sample_type_is_usage_error(self, tmp_path: Path) -> None:
+        with pytest.raises(CliUsageError, match=r"data row\(s\) 1 has no sample_type"):
+            parse_accession_sheet(_write_sheet(tmp_path, _record(sample_type="")))
+
+    def test_row_with_no_sample_type_column_is_usage_error(self, tmp_path: Path) -> None:
+        path = tmp_path / "sheet.csv"
+        with path.open("w", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=["accession", "name"])
+            writer.writeheader()
+            writer.writerow({"accession": "ERR1160845", "name": "liver_r1"})
+
+        with pytest.raises(CliUsageError):
+            parse_accession_sheet(path)
+
+    def test_sheet_broken_in_both_columns_reports_both_in_one_error(
+        self, tmp_path: Path,
+    ) -> None:
+        with pytest.raises(CliUsageError) as excinfo:
+            parse_accession_sheet(_write_sheet(
+                tmp_path,
+                _record(accession=""),
+                _record(accession="ERR2", sample_type=""),
+            ))
+
+        assert "data row(s) 1 has no accession" in str(excinfo.value)
+        assert "data row(s) 2 has no sample_type" in str(excinfo.value)
+
+    def test_trailing_blank_row_is_skipped(self, tmp_path: Path) -> None:
+        # A comma-only line, e.g. one a spreadsheet export leaves below the
+        # data — not a row with data but no accession, so it isn't an error.
+        sheet = parse_accession_sheet(_write_sheet(
+            tmp_path,
+            _record(accession="ERR1"),
+            {},
+        ))
+
+        assert [row.accession for row in sheet.rows] == ["ERR1"]
+
+    def test_sheet_of_only_blank_rows_is_usage_error(self, tmp_path: Path) -> None:
+        with pytest.raises(CliUsageError, match="no rows"):
+            parse_accession_sheet(_write_sheet(tmp_path, {}, {}))
+
+    def test_blank_row_in_the_middle_does_not_shift_later_row_numbers(
+        self, tmp_path: Path,
+    ) -> None:
+        with pytest.raises(CliUsageError, match=r"data row\(s\) 3 has no accession"):
+            parse_accession_sheet(_write_sheet(
+                tmp_path,
+                _record(accession="ERR1"),
+                {},
+                _record(accession=""),
+            ))
+
+
+def test_row_rejects_empty_accession_by_construction() -> None:
+    with pytest.raises(ValueError, match="accession"):
+        AccessionSheetRow(
+            row_number=1,
+            accession="",
+            name=None,
+            organism=None,
+            sample_type="rna_seq",
+            metadata={},
+        )
+
+
+def test_row_rejects_empty_sample_type_by_construction() -> None:
+    with pytest.raises(ValueError, match="sample_type"):
+        AccessionSheetRow(
+            row_number=1,
+            accession="ERR1160845",
+            name=None,
+            organism=None,
+            sample_type="",
+            metadata={},
+        )
+
+
+class TestAccessionSheetRowToSpec:
+
+    def _row(self, tmp_path: Path, **overrides: str):
+        sheet = parse_accession_sheet(_write_sheet(tmp_path, _record(**overrides)))
+        return sheet.rows[0]
+
+    def test_uses_the_row_sample_type(self, tmp_path: Path) -> None:
+        row = self._row(tmp_path, sample_type="chip_seq")
+
+        spec = row.to_spec()
+
+        assert spec == SampleImportSpec(accession="ERR1160845", sample_type="chip_seq")
+
+    def test_carries_name_organism_and_metadata(self, tmp_path: Path) -> None:
+        row = self._row(tmp_path, name="liver_r1", organism="Hs", cell_type="Neuron")
+
+        spec = row.to_spec()
+
+        assert spec == SampleImportSpec(
+            accession="ERR1160845",
+            sample_type="rna_seq",
+            name="liver_r1",
+            organism_id="Hs",
+            metadata={"cell_type": "Neuron"},
+        )
+
+    def test_annotation_suffixed_column_is_forwarded_as_a_plain_metadata_key(
+        self, tmp_path: Path,
+    ) -> None:
+        # Unlike upload-batch, there's no special handling of `<id>__annotation`
+        # here — it's just another metadata column, and the server does not
+        # recognise it as one (see cli.rst).
+        row = self._row(tmp_path, source="blood", source__annotation="left lobe")
+
+        spec = row.to_spec()
+
+        assert spec.metadata == {"source": "blood", "source__annotation": "left lobe"}
diff --git a/tests/unit/cli/test_samples.py b/tests/unit/cli/test_samples.py
index 5c461c1..3f04081 100644
--- a/tests/unit/cli/test_samples.py
+++ b/tests/unit/cli/test_samples.py
@@ -13,6 +13,7 @@
 ANNOTATION_TEMPLATE_URL = f"{DEFAULT_BASE_URL}/annotation"
 ANNOTATION_UPLOAD_URL = f"{DEFAULT_BASE_URL}/upload/annotation"
 MULTIPLEXED_UPLOAD_URL = f"{DEFAULT_BASE_URL}/upload/multiplexed"
+SAMPLE_IMPORTS_URL = f"{DEFAULT_BASE_URL}/v2/sample-imports"
 TOKEN = "test.token"
 
 
@@ -923,3 +924,623 @@ def test_non_csv_sheet_is_usage_error(self, run_cli, tmp_path: Path) -> None:
         assert result.exit_code == 2
         assert upload.call_count == 0
         assert "CSV" in result.stderr
+
+
+IMPORT_HEADERS = ["accession", "name", "organism", "sample_type", "cell_type", "source", "source__annotation"]
+
+
+def _write_import_sheet(directory: Path, *records: dict[str, str]) -> Path:
+    path = directory / "accessions.csv"
+    with path.open("w", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=IMPORT_HEADERS)
+        writer.writeheader()
+        writer.writerows(records)
+    return path
+
+
+def _import_record(**overrides: str) -> dict[str, str]:
+    record = {"accession": "ERR1", "sample_type": "rna_seq"}
+    record.update(overrides)
+    return record
+
+
+def _job_json(
+    job_id: int,
+    status: str,
+    accessions: list[str],
+    sample_ids: list[int] | None = None,
+    error: str | None = None,
+    execution_id: int | None = 7,
+) -> dict:
+    return {
+        "id": job_id,
+        "status": status,
+        "created": 1700000000,
+        "started": 1700000001 if status != "RUNNING" else None,
+        "finished": 1700000002 if status in ("COMPLETED", "FAILED") else None,
+        "accessions": accessions,
+        "sample_ids": sample_ids or [],
+        "execution_id": execution_id,
+        "error": error,
+    }
+
+
+class TestSamplesImport:
+
+    @respx.mock
+    def test_kicks_off_job_and_reports_id_without_waiting(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json(
+                42, "RUNNING", ["ERR1", "ERR2"],
+            )),
+        )
+        sheet = _write_import_sheet(
+            tmp_path,
+            _import_record(accession="ERR1", cell_type="Neuron"),
+            _import_record(accession="ERR2", cell_type="Fibroblast"),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 0
+        assert route.call_count == 1
+        assert "42" in result.stdout
+        assert "import-status" in result.stdout
+
+    @respx.mock
+    def test_json_document_reports_job_fields(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json(
+                42, "RUNNING", ["ERR1"],
+            )),
+        )
+        sheet = _write_import_sheet(tmp_path, _import_record(cell_type="Neuron"))
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet), "--json",
+        )
+
+        assert result.exit_code == 0
+        document = json.loads(result.stdout)
+        assert result.stdout.count("\n") == 1
+        assert document == {
+            "id": 42,
+            "status": "RUNNING",
+            "created": "2023-11-14T22:13:20Z",
+            "started": None,
+            "finished": None,
+            "accessions": ["ERR1"],
+            "sample_ids": [],
+            "execution_id": 7,
+            "error": None,
+        }
+
+    @respx.mock
+    def test_sends_every_row_without_local_validation(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        # No metadata/sample-type endpoints are mocked: if the command called
+        # client.samples.get_metadata_attributes() or otherwise validated
+        # locally, respx would fail the test for an unmocked request.
+        route = respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json(
+                1, "RUNNING", ["not-an-accession", "ERR1", "ERR1"],
+            )),
+        )
+        sheet = _write_import_sheet(
+            tmp_path,
+            _import_record(accession="not-an-accession", sample_type="bogus"),
+            _import_record(accession="ERR1"),
+            _import_record(accession="ERR1"),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 0
+        payload = json.loads(route.calls[0].request.content)
+        assert [entry["accession"] for entry in payload["imports"]] == [
+            "not-an-accession", "ERR1", "ERR1",
+        ]
+
+    @respx.mock
+    def test_sends_name_organism_and_metadata_in_payload(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json(
+                1, "RUNNING", ["ERR1"],
+            )),
+        )
+        sheet = _write_import_sheet(tmp_path, _import_record(
+            name="liver_r1", organism="Hs", cell_type="Neuron",
+        ))
+
+        run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        payload = json.loads(route.calls[0].request.content)
+        assert payload == {
+            "imports": [{
+                "accession": "ERR1",
+                "sample_type": "rna_seq",
+                "name": "liver_r1",
+                "organism": "Hs",
+                "metadata": {"cell_type": "Neuron"},
+            }],
+        }
+
+    @respx.mock
+    def test_each_row_uses_its_own_sample_type(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json(
+                1, "RUNNING", ["ERR1", "ERR2"],
+            )),
+        )
+        sheet = _write_import_sheet(
+            tmp_path,
+            _import_record(accession="ERR1", sample_type="chip_seq"),
+            _import_record(accession="ERR2", sample_type="atac_seq"),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 0
+        payload = json.loads(route.calls[0].request.content)
+        sample_types = [entry["sample_type"] for entry in payload["imports"]]
+        assert sample_types == ["chip_seq", "atac_seq"]
+
+    @respx.mock
+    def test_trailing_blank_row_is_skipped(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json=_job_json(
+                1, "RUNNING", ["ERR1"],
+            )),
+        )
+        sheet = _write_import_sheet(
+            tmp_path, _import_record(accession="ERR1"), {},
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 0
+        payload = json.loads(route.calls[0].request.content)
+        assert [entry["accession"] for entry in payload["imports"]] == ["ERR1"]
+
+    @respx.mock
+    def test_api_rejection_propagates_as_error(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL).mock(
+            return_value=httpx.Response(
+                HTTPStatus.UNPROCESSABLE_ENTITY,
+                json={"error": "sample type 'bogus' does not exist"},
+            ),
+        )
+        sheet = _write_import_sheet(tmp_path, _import_record(sample_type="bogus"))
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 1
+        assert route.call_count == 1
+        assert "bogus" in result.stderr
+
+    @respx.mock
+    def test_non_csv_sheet_is_usage_error(self, run_cli, tmp_path: Path) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        xlsx = tmp_path / "sheet.xlsx"
+        xlsx.write_bytes(b"PK\x03\x04")
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(xlsx),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "CSV" in result.stderr
+
+    def test_missing_sheet_is_usage_error(self, run_cli) -> None:
+        result = run_cli("--token", TOKEN, "samples", "import")
+
+        assert result.exit_code == 2
+
+    @respx.mock
+    def test_header_only_sheet_is_usage_error(self, run_cli, tmp_path: Path) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        sheet = _write_import_sheet(tmp_path)
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "no rows" in result.stderr
+
+    @respx.mock
+    def test_row_missing_sample_type_is_usage_error(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        sheet = _write_import_sheet(
+            tmp_path,
+            _import_record(accession="ERR1"),
+            _import_record(accession="ERR2", sample_type=""),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "data row(s) 2 has no sample_type" in result.stderr
+
+    @respx.mock
+    def test_sheet_with_only_blank_accessions_is_usage_error(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        sheet = _write_import_sheet(
+            tmp_path, _import_record(accession=""), _import_record(accession=""),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "data row(s) 1, 2 have no accession" in result.stderr
+
+    @respx.mock
+    def test_mixed_valid_and_blank_accession_rows_is_usage_error(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        # A blank accession is rejected outright rather than silently
+        # skipped, even when the rest of the sheet is otherwise fine.
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        sheet = _write_import_sheet(
+            tmp_path,
+            _import_record(accession="ERR1"),
+            _import_record(accession=""),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "data row(s) 2 has no accession" in result.stderr
+
+    @respx.mock
+    def test_sheet_broken_in_both_columns_reports_both_in_one_error(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        sheet = _write_import_sheet(
+            tmp_path,
+            _import_record(accession=""),
+            _import_record(accession="ERR2", sample_type=""),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "data row(s) 1 has no accession" in result.stderr
+        assert "data row(s) 2 has no sample_type" in result.stderr
+
+    @respx.mock
+    def test_sheet_with_no_accession_column_is_usage_error(
+        self, run_cli, tmp_path: Path,
+    ) -> None:
+        route = respx.post(SAMPLE_IMPORTS_URL)
+        sheet = tmp_path / "accessions.csv"
+        with sheet.open("w", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=["run", "name", "sample_type"])
+            writer.writeheader()
+            writer.writerow({"run": "ERR1160845", "name": "liver_r1", "sample_type": "rna_seq"})
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import", "--sheet", str(sheet),
+        )
+
+        assert result.exit_code == 2
+        assert route.call_count == 0
+        assert "data row(s) 1 has no accession" in result.stderr
+
+
+class TestSamplesImportStatus:
+
+    @respx.mock
+    def test_reports_running_job(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "RUNNING", ["ERR1", "ERR2"],
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "42" in result.stdout
+        assert "RUNNING" in result.stdout
+
+    @respx.mock
+    def test_running_job_with_started_reports_when_it_started(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42, "status": "RUNNING", "created": 1700000000,
+                "started": 1700000001, "finished": None,
+                "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "started 2023-11-14" in result.stdout
+
+    @respx.mock
+    def test_started_with_non_utc_offset_is_reported_in_utc(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42, "status": "RUNNING", "created": 1700000000,
+                "started": "2024-04-05T19:34:38+02:00", "finished": None,
+                "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "started 2024-04-05 17:34:38 UTC" in result.stdout
+
+    @respx.mock
+    def test_started_with_no_offset_is_treated_as_utc(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42, "status": "RUNNING", "created": 1700000000,
+                "started": "2024-04-05T19:34:38", "finished": None,
+                "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "started 2024-04-05 19:34:38 UTC" in result.stdout
+
+    @respx.mock
+    def test_naive_started_is_reported_as_utc_in_json_too(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42, "status": "RUNNING", "created": 1700000000,
+                "started": "2024-04-05T19:34:38", "finished": None,
+                "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json",
+        )
+
+        assert result.exit_code == 0
+        document = json.loads(result.stdout)
+        assert document["started"] == "2024-04-05T19:34:38Z"
+
+    @respx.mock
+    def test_non_utc_offset_started_is_reported_as_utc_in_json_too(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42, "status": "RUNNING", "created": 1700000000,
+                "started": "2024-04-05T19:34:38+02:00", "finished": None,
+                "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json",
+        )
+
+        assert result.exit_code == 0
+        document = json.loads(result.stdout)
+        assert document["started"] == "2024-04-05T17:34:38Z"
+
+    @respx.mock
+    def test_running_job_falls_back_to_created_when_not_started(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42, "status": "RUNNING", "created": 1700000000,
+                "started": None, "finished": None,
+                "accessions": ["ERR1"], "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "created 2023-11-14" in result.stdout
+
+    @respx.mock
+    def test_completed_job_reports_when_it_finished(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "COMPLETED", ["ERR1"], [101],
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "finished 2023-11-14" in result.stdout
+
+    @respx.mock
+    def test_reports_completed_job_with_sample_ids(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "COMPLETED", ["ERR1", "ERR2"], [101, 102],
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "101" in result.stdout
+        assert "102" in result.stdout
+
+    @respx.mock
+    def test_reports_failed_job_with_error(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "FAILED", ["ERR1"], error="download failed",
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 1
+        assert "FAILED" in result.stdout
+        assert "download failed" not in result.stdout
+        assert "download failed" in result.stderr
+
+    @respx.mock
+    def test_failed_job_json_mode_has_error_on_stdout_only(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "FAILED", ["ERR1"], error="download failed",
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json",
+        )
+
+        assert result.exit_code == 1
+        assert result.stderr == ""
+        document = json.loads(result.stdout)
+        assert document["error"] == "download failed"
+
+    def test_non_numeric_job_id_reports_clear_message(self, run_cli) -> None:
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "not-a-number",
+        )
+
+        assert result.exit_code == 2
+        assert "_job_id" not in result.stderr
+        assert "job id" in result.stderr.lower()
+
+    @respx.mock
+    def test_completed_job_with_no_sample_ids_reports_none(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "COMPLETED", ["ERR1"], [],
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42",
+        )
+
+        assert result.exit_code == 0
+        assert "none" in result.stdout
+
+    @respx.mock
+    def test_json_document_matches_job_shape(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "COMPLETED", ["ERR1"], [101],
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json",
+        )
+
+        assert result.exit_code == 0
+        document = json.loads(result.stdout)
+        assert result.stdout.count("\n") == 1
+        assert document == {
+            "id": 42,
+            "status": "COMPLETED",
+            "created": "2023-11-14T22:13:20Z",
+            "started": "2023-11-14T22:13:21Z",
+            "finished": "2023-11-14T22:13:22Z",
+            "accessions": ["ERR1"],
+            "sample_ids": [101],
+            "execution_id": 7,
+            "error": None,
+        }
+
+    @respx.mock
+    def test_reports_job_with_no_execution_yet(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json=_job_json(
+                42, "RUNNING", ["ERR1"], execution_id=None,
+            )),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "42", "--json",
+        )
+
+        assert result.exit_code == 0
+        document = json.loads(result.stdout)
+        assert document["execution_id"] is None
+
+    @respx.mock
+    def test_unknown_job_id_is_not_found(self, run_cli) -> None:
+        respx.get(f"{SAMPLE_IMPORTS_URL}/999").mock(
+            return_value=httpx.Response(
+                HTTPStatus.NOT_FOUND, json={"error": "sample import 999 does not exist"},
+            ),
+        )
+
+        result = run_cli(
+            "--token", TOKEN, "samples", "import-status", "--job-id", "999",
+        )
+
+        assert result.exit_code == 4
+
+    def test_missing_job_id_is_usage_error(self, run_cli) -> None:
+        result = run_cli("--token", TOKEN, "samples", "import-status")
+
+        assert result.exit_code == 2
diff --git a/tests/unit/v2/test_package_exports.py b/tests/unit/v2/test_package_exports.py
index 2bb0a12..06bff5a 100644
--- a/tests/unit/v2/test_package_exports.py
+++ b/tests/unit/v2/test_package_exports.py
@@ -20,6 +20,25 @@ def test_username_password_credentials_is_exported(self) -> None:
 
         assert UsernamePasswordCredentials is DirectCredentials
 
+    def test_sample_import_types_are_exported(self) -> None:
+        from flowbio.v2 import (
+            SampleImportJob,
+            SampleImportJobId,
+            SampleImportSpec,
+            SampleImportStatus,
+        )
+        from flowbio.v2.samples import (
+            SampleImportJob as DirectJob,
+            SampleImportJobId as DirectJobId,
+            SampleImportSpec as DirectSpec,
+            SampleImportStatus as DirectStatus,
+        )
+
+        assert SampleImportSpec is DirectSpec
+        assert SampleImportJob is DirectJob
+        assert SampleImportJobId is DirectJobId
+        assert SampleImportStatus is DirectStatus
+
 
 class TestSubmoduleExports:
 
diff --git a/tests/unit/v2/test_samples.py b/tests/unit/v2/test_samples.py
index cc42ad7..59cade8 100644
--- a/tests/unit/v2/test_samples.py
+++ b/tests/unit/v2/test_samples.py
@@ -1,3 +1,5 @@
+import json
+from datetime import datetime, timezone
 from http import HTTPStatus
 from pathlib import Path
 from unittest.mock import ANY, patch
@@ -10,6 +12,7 @@
 from flowbio.v2.exceptions import (
     AnnotationValidationError,
     BadRequestError,
+    FlowApiError,
     NotFoundError,
 )
 from flowbio.v2.samples import (
@@ -17,6 +20,9 @@
     MultiplexedUpload,
     Organism,
     Project,
+    SampleImportJob,
+    SampleImportJobId,
+    SampleImportSpec,
     SampleResource,
     SampleType,
     Sample,
@@ -1049,3 +1055,229 @@ def test_chunked_multiplexed_upload(self, tmp_path: Path) -> None:
 
         assert mux_route.call_count == 3
         assert result.data_ids == ["mux_1"]
+
+
+class TestImportSamples:
+
+    @respx.mock
+    def test_posts_imports_and_parses_job(self) -> None:
+        route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json={
+                "id": 42,
+                "status": "RUNNING",
+                "created": 1700000000,
+                "started": None,
+                "finished": None,
+                "accessions": ["ERR1160845"],
+                "sample_ids": [],
+                "execution_id": None,
+                "error": None,
+            }),
+        )
+
+        client = Client()
+        result = client.samples.import_samples([
+            SampleImportSpec(accession="ERR1160845", sample_type="rna_seq"),
+        ])
+
+        assert result == SampleImportJob(
+            id=SampleImportJobId(42),
+            status="RUNNING",
+            created=1700000000,
+            started=None,
+            finished=None,
+            accessions=["ERR1160845"],
+            sample_ids=[],
+            execution_id=None,
+            error=None,
+        )
+        assert route.call_count == 1
+
+    @respx.mock
+    def test_sends_accession_and_sample_type(self) -> None:
+        route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json={
+                "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1"],
+                "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        client = Client()
+        client.samples.import_samples([
+            SampleImportSpec(accession="ERR1", sample_type="rna_seq"),
+        ])
+
+        payload = json.loads(route.calls[0].request.content)
+        assert payload == {
+            "imports": [{"accession": "ERR1", "sample_type": "rna_seq"}],
+        }
+
+    @respx.mock
+    def test_sends_optional_fields_when_present(self) -> None:
+        route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json={
+                "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1"],
+                "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        client = Client()
+        client.samples.import_samples([
+            SampleImportSpec(
+                accession="ERR1",
+                sample_type="rna_seq",
+                name="my_sample",
+                organism_id="Hs",
+                metadata={"strandedness": "reverse"},
+            ),
+        ])
+
+        payload = json.loads(route.calls[0].request.content)
+        assert payload == {
+            "imports": [{
+                "accession": "ERR1",
+                "sample_type": "rna_seq",
+                "name": "my_sample",
+                "organism": "Hs",
+                "metadata": {"strandedness": "reverse"},
+            }],
+        }
+
+    @respx.mock
+    def test_sends_multiple_imports_in_one_request(self) -> None:
+        route = respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock(
+            return_value=httpx.Response(HTTPStatus.CREATED, json={
+                "id": 1, "status": "RUNNING", "created": 1700000000, "accessions": ["ERR1", "ERR2"],
+                "sample_ids": [], "execution_id": None, "error": None,
+            }),
+        )
+
+        client = Client()
+        client.samples.import_samples([
+            SampleImportSpec(accession="ERR1", sample_type="rna_seq"),
+            SampleImportSpec(accession="ERR2", sample_type="rna_seq"),
+        ])
+
+        payload = json.loads(route.calls[0].request.content)
+        assert [entry["accession"] for entry in payload["imports"]] == ["ERR1", "ERR2"]
+        assert route.call_count == 1
+
+    @respx.mock
+    def test_raises_flow_api_error_on_validation_failure(self) -> None:
+        respx.post(f"{DEFAULT_BASE_URL}/v2/sample-imports").mock(
+            return_value=httpx.Response(
+                HTTPStatus.UNPROCESSABLE_ENTITY,
+                json={"error": "at least one accession is required"},
+            ),
+        )
+
+        client = Client()
+
+        with pytest.raises(FlowApiError) as exc_info:
+            client.samples.import_samples([])
+
+        assert exc_info.value.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+
+
+class TestGetImport:
+
+    @respx.mock
+    def test_parses_completed_job(self) -> None:
+        respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42,
+                "status": "COMPLETED",
+                "created": 1700000000,
+                "started": 1700000001,
+                "finished": 1700000002,
+                "accessions": ["ERR1160845", "ERR10677146"],
+                "sample_ids": [101, 102],
+                "execution_id": 7,
+                "error": None,
+            }),
+        )
+
+        client = Client()
+        result = client.samples.get_import(SampleImportJobId(42))
+
+        assert result == SampleImportJob(
+            id=SampleImportJobId(42),
+            status="COMPLETED",
+            created=1700000000,
+            started=1700000001,
+            finished=1700000002,
+            accessions=["ERR1160845", "ERR10677146"],
+            sample_ids=[101, 102],
+            execution_id=7,
+            error=None,
+        )
+
+    @respx.mock
+    def test_parses_failed_job_with_error(self) -> None:
+        respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42,
+                "status": "FAILED",
+                "created": 1700000000,
+                "started": 1700000001,
+                "finished": 1700000002,
+                "accessions": ["ERR1160845"],
+                "sample_ids": [],
+                "execution_id": 7,
+                "error": "download failed: connection reset",
+            }),
+        )
+
+        client = Client()
+        result = client.samples.get_import(SampleImportJobId(42))
+
+        assert result.status == "FAILED"
+        assert result.error == "download failed: connection reset"
+
+    @respx.mock
+    def test_raises_not_found_for_unknown_job(self) -> None:
+        respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/999").mock(
+            return_value=httpx.Response(
+                HTTPStatus.NOT_FOUND, json={"error": "sample import 999 does not exist"},
+            ),
+        )
+
+        client = Client()
+
+        with pytest.raises(NotFoundError):
+            client.samples.get_import(SampleImportJobId(999))
+
+    @respx.mock
+    def test_parses_job_with_only_id_and_status(self) -> None:
+        respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42,
+                "status": "RUNNING",
+            }),
+        )
+
+        client = Client()
+        result = client.samples.get_import(SampleImportJobId(42))
+
+        assert result.created is None
+        assert result.started is None
+        assert result.finished is None
+        assert result.accessions == []
+        assert result.sample_ids == []
+        assert result.execution_id is None
+        assert result.error is None
+
+    @respx.mock
+    def test_naive_timestamp_is_treated_as_utc(self) -> None:
+        respx.get(f"{DEFAULT_BASE_URL}/v2/sample-imports/42").mock(
+            return_value=httpx.Response(HTTPStatus.OK, json={
+                "id": 42,
+                "status": "RUNNING",
+                "started": "2024-04-05T19:34:38",
+            }),
+        )
+
+        client = Client()
+        result = client.samples.get_import(SampleImportJobId(42))
+
+        assert result.started == datetime(2024, 4, 5, 19, 34, 38, tzinfo=timezone.utc)