diff --git a/yunke-server/app/api/__init__.py b/yunke-server/app/api/__init__.py
new file mode 100644
index 0000000..c7446fa
--- /dev/null
+++ b/yunke-server/app/api/__init__.py
@@ -0,0 +1,6 @@
+"""FastAPI layer for the yunke data-mirror viewing API.
+
+Read-only HTTP surface over the mirrored DB. Public entrypoint is
+:mod:`app.api.main` (``app = FastAPI(...)``). Routers live under
+:mod:`app.api.routers` and share dependencies from :mod:`app.api.deps`.
+"""
diff --git a/yunke-server/app/api/deps.py b/yunke-server/app/api/deps.py
new file mode 100644
index 0000000..1562052
--- /dev/null
+++ b/yunke-server/app/api/deps.py
@@ -0,0 +1,10 @@
+"""FastAPI dependencies for the yunke data-mirror API.
+
+Re-exports :func:`app.db.base.get_session` so route code (and tests) depend on
+a single API-layer seam. Tests override ``app.api.deps.get_session`` to swap in
+an isolated in-memory SQLite session.
+"""
+
+from app.db.base import get_session
+
+__all__ = ["get_session"]
diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py
new file mode 100644
index 0000000..dcce284
--- /dev/null
+++ b/yunke-server/app/api/main.py
@@ -0,0 +1,37 @@
+"""FastAPI app for the yunke data-mirror viewing API.
+
+Run locally with::
+
+    uvicorn app.api.main:app --reload
+
+Includes:
+* ``GET /api/health``  -> ``{"status": "ok"}`` liveness probe.
+* ``GET /api/customers`` (see :mod:`app.api.routers.customer`).
+
+CORS is configured permissively (wildcard origin, no credentials) so the local
+frontend can call the API during development. Auth / finer CORS are LATER tasks.
+"""
+
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.api.routers import customer
+
+app = FastAPI(title="yunke data mirror")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(customer.router)
+
+
+@app.get("/api/health")
+async def health() -> dict[str, str]:
+    """Trivial liveness probe."""
+    return {"status": "ok"}
diff --git a/yunke-server/app/api/routers/__init__.py b/yunke-server/app/api/routers/__init__.py
new file mode 100644
index 0000000..b95c69c
--- /dev/null
+++ b/yunke-server/app/api/routers/__init__.py
@@ -0,0 +1 @@
+"""HTTP routers for the yunke data-mirror API."""
diff --git a/yunke-server/app/api/routers/customer.py b/yunke-server/app/api/routers/customer.py
new file mode 100644
index 0000000..8db9a6f
--- /dev/null
+++ b/yunke-server/app/api/routers/customer.py
@@ -0,0 +1,70 @@
+"""GET /api/customers — paginated, filterable customer list.
+
+Read-only query over the mirrored ``customer`` table. Supports filtering by
+``channel``, ``owner_account_id`` (direct columns) and ``department_id``
+(resolved through ``customer.user_id`` -> ``staff.dept_id``). Results are
+ordered by ``last_chat_time desc`` (nulls last) then ``friend_id`` so the
+most-recently-active contacts surface first.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+import sqlalchemy as sa
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_session
+from app.api.schemas import CustomerOut, Page
+from app.db.models import Customer, Staff
+
+router = APIRouter(prefix="/api", tags=["customers"])
+
+SessionDep = Annotated[AsyncSession, Depends(get_session)]
+PageParam = Annotated[int, Query(ge=1)]
+PageSizeParam = Annotated[int, Query(ge=1, le=100)]
+
+
+@router.get("/customers")
+async def list_customers(
+    session: SessionDep,
+    page: PageParam = 1,
+    page_size: PageSizeParam = 20,
+    channel: Annotated[str | None, Query()] = None,
+    department_id: Annotated[str | None, Query()] = None,
+    owner_account_id: Annotated[str | None, Query()] = None,
+) -> Page[CustomerOut]:
+    """List customers with optional filters and pagination."""
+    conditions = []
+    if channel is not None:
+        conditions.append(Customer.channel == channel)
+    if owner_account_id is not None:
+        conditions.append(Customer.owner_account_id == owner_account_id)
+    if department_id is not None:
+        # Customer has no dept column; resolve via the owning staff member.
+        dept_user_ids = sa.select(Staff.user_id).where(
+            Staff.dept_id == department_id
+        )
+        conditions.append(Customer.user_id.in_(dept_user_ids))
+
+    total_stmt = sa.select(sa.func.count()).select_from(Customer)
+    if conditions:
+        total_stmt = total_stmt.where(*conditions)
+    total = (await session.execute(total_stmt)).scalar_one()
+
+    list_stmt = sa.select(Customer)
+    if conditions:
+        list_stmt = list_stmt.where(*conditions)
+    list_stmt = (
+        list_stmt.order_by(
+            sa.nullslast(Customer.last_chat_time.desc()),
+            Customer.friend_id.asc(),
+        )
+        .limit(page_size)
+        .offset((page - 1) * page_size)
+    )
+
+    rows = (await session.execute(list_stmt)).scalars().all()
+    items = [CustomerOut.model_validate(row) for row in rows]
+    return Page(items=items, total=total, page=page, page_size=page_size)
diff --git a/yunke-server/app/api/schemas.py b/yunke-server/app/api/schemas.py
new file mode 100644
index 0000000..5563009
--- /dev/null
+++ b/yunke-server/app/api/schemas.py
@@ -0,0 +1,43 @@
+"""Pydantic v2 response models for the yunke data-mirror API."""
+
+from __future__ import annotations
+
+from typing import Generic, TypeVar
+
+from pydantic import BaseModel, ConfigDict
+
+T = TypeVar("T")
+
+
+class CustomerOut(BaseModel):
+    """Public Customer view — a subset of :class:`app.db.models.Customer`.
+
+    Only the fields the viewing frontend needs today are exposed; the verbatim
+    ``raw_json`` blob and internal audit columns stay server-side.
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    friend_id: str
+    channel: str
+    owner_account_id: str | None = None
+    friend_nickname: str | None = None
+    friend_remark: str | None = None
+    real_name: str | None = None
+    phone: str | None = None
+    region: str | None = None
+    sale_name: str | None = None
+    dept_name: str | None = None
+    head_image_url: str | None = None
+    last_chat_time: int | None = None
+    message_count: int | None = None
+    version: str | None = None
+
+
+class Page(BaseModel, Generic[T]):
+    """Generic pagination envelope returned by list endpoints."""
+
+    items: list[T]
+    total: int
+    page: int
+    page_size: int
diff --git a/yunke-server/tests/test_api_customer.py b/yunke-server/tests/test_api_customer.py
new file mode 100644
index 0000000..8b546e4
--- /dev/null
+++ b/yunke-server/tests/test_api_customer.py
@@ -0,0 +1,218 @@
+"""Tests for app.api — minimal customer list API + health check.
+
+Drives the FastAPI app over an in-process transport (``httpx.AsyncClient`` with
+``ASGITransport``) — no real network socket. The ``get_session`` dependency is
+overridden to yield an isolated in-memory async SQLite session. ``StaticPool``
+keeps a single connection behind every session so all requests inside a test
+see the same seeded schema/rows.
+
+Coverage (Task 1.8 acceptance):
+* ``GET /api/health`` -> 200 ``{"status": "ok"}``.
+* ``GET /api/customers?channel=wechat`` -> ``total == 2`` and correct shape.
+* ``GET /api/customers?channel=whatsapp`` -> ``total == 1``.
+* Pagination: ``page_size=1&page=1`` -> 1 item, ``total`` still 3.
+* Ordering: ``last_chat_time desc`` (nulls last) then ``friend_id``.
+* Documented filters ``owner_account_id`` and ``department_id`` actually filter.
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.api import deps
+from app.api.main import app
+from app.db.base import Base
+from app.db.models import Customer, Staff
+
+
+@pytest.fixture
+async def client():
+    """Yield an ``AsyncClient`` against ``app`` with a seeded in-memory DB.
+
+    Seeds three Customers (two wechat, one whatsapp) plus two Staff rows so the
+    ``department_id`` filter (Customer.user_id -> Staff.dept_id) is exercisable.
+    """
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    async with SessionLocal() as session:
+        session.add_all(
+            [
+                Customer(
+                    friend_id="wx_a",
+                    channel="wechat",
+                    owner_account_id="wx_owner_1",
+                    user_id="u1",
+                    friend_nickname="Alice",
+                    last_chat_time=2000,
+                    message_count=10,
+                    version="v1",
+                ),
+                Customer(
+                    friend_id="wx_b",
+                    channel="wechat",
+                    owner_account_id="wx_owner_1",
+                    user_id="u2",
+                    friend_nickname="Bob",
+                    last_chat_time=1000,
+                    version="v1",
+                ),
+                Customer(
+                    friend_id="wa_c",
+                    channel="whatsapp",
+                    owner_account_id="wa_owner_1",
+                    user_id="u3",
+                    friend_nickname="Carol",
+                    last_chat_time=None,
+                    version="v1",
+                ),
+                Staff(user_id="u1", dept_id="d100"),
+                Staff(user_id="u2", dept_id="d200"),
+            ]
+        )
+        await session.commit()
+
+    async def _override_session():
+        async with SessionLocal() as s:
+            yield s
+
+    app.dependency_overrides[deps.get_session] = _override_session
+    try:
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            yield ac
+    finally:
+        app.dependency_overrides.clear()
+        await engine.dispose()
+
+
+async def test_health_returns_ok(client: AsyncClient):
+    """``GET /api/health`` -> 200 ``{"status": "ok"}``."""
+    r = await client.get("/api/health")
+    assert r.status_code == 200
+    assert r.json() == {"status": "ok"}
+
+
+async def test_list_default_page_returns_all(client: AsyncClient):
+    """No query params -> page 1, page_size 20, all 3 rows."""
+    r = await client.get("/api/customers")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["page"] == 1
+    assert body["page_size"] == 20
+    assert body["total"] == 3
+    assert len(body["items"]) == 3
+    # Shape: every documented CustomerOut field is present.
+    item = body["items"][0]
+    for key in (
+        "friend_id",
+        "channel",
+        "owner_account_id",
+        "friend_nickname",
+        "friend_remark",
+        "real_name",
+        "phone",
+        "region",
+        "sale_name",
+        "dept_name",
+        "head_image_url",
+        "last_chat_time",
+        "message_count",
+        "version",
+    ):
+        assert key in item
+
+
+async def test_list_filters_by_channel_wechat(client: AsyncClient):
+    """``?channel=wechat`` -> total 2, both wechat rows."""
+    r = await client.get(
+        "/api/customers", params={"channel": "wechat", "page": 1, "page_size": 10}
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 2
+    assert len(body["items"]) == 2
+    assert {it["friend_id"] for it in body["items"]} == {"wx_a", "wx_b"}
+    assert all(it["channel"] == "wechat" for it in body["items"])
+
+
+async def test_list_filters_by_channel_whatsapp(client: AsyncClient):
+    """``?channel=whatsapp`` -> total 1."""
+    r = await client.get("/api/customers", params={"channel": "whatsapp"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 1
+    assert body["items"][0]["friend_id"] == "wa_c"
+
+
+async def test_pagination_page_size_one(client: AsyncClient):
+    """``page_size=1&page=1`` -> 1 item but ``total`` still reflects all rows."""
+    r = await client.get(
+        "/api/customers", params={"page": 1, "page_size": 1}
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 3
+    assert len(body["items"]) == 1
+
+
+async def test_order_is_last_chat_time_desc_nulls_last(client: AsyncClient):
+    """Order: ``last_chat_time desc`` (nulls last), then ``friend_id``.
+
+    Seeded rows have last_chat_time 2000 / 1000 / None -> expected order
+    ``[wx_a, wx_b, wa_c]`` (the whatsapp row with NULL last_chat_time lands
+    last even though it has no timestamp).
+    """
+    r = await client.get(
+        "/api/customers", params={"page": 1, "page_size": 20}
+    )
+    assert r.status_code == 200
+    ids = [it["friend_id"] for it in r.json()["items"]]
+    assert ids == ["wx_a", "wx_b", "wa_c"]
+
+
+async def test_filter_by_owner_account_id(client: AsyncClient):
+    """``?owner_account_id=wa_owner_1`` -> only the whatsapp customer."""
+    r = await client.get(
+        "/api/customers", params={"owner_account_id": "wa_owner_1"}
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 1
+    assert body["items"][0]["friend_id"] == "wa_c"
+
+
+async def test_filter_by_department_id(client: AsyncClient):
+    """``?department_id=d100`` -> customers whose ``user_id`` maps to that dept.
+
+    Customer ``wx_a`` has ``user_id=u1``; Staff ``u1`` belongs to ``d100``.
+    """
+    r = await client.get(
+        "/api/customers", params={"department_id": "d100"}
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 1
+    assert body["items"][0]["friend_id"] == "wx_a"
+
+
+async def test_filters_compose_channel_and_owner(client: AsyncClient):
+    """Two filters AND together: channel=wechat + owner=wx_owner_1 -> 2 rows."""
+    r = await client.get(
+        "/api/customers",
+        params={"channel": "wechat", "owner_account_id": "wx_owner_1"},
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 2
+    assert {it["friend_id"] for it in body["items"]} == {"wx_a", "wx_b"}