diff --git a/app/services/accessory_service.py b/app/services/accessory_service.py
new file mode 100644
index 0000000..0f878ba
--- /dev/null
+++ b/app/services/accessory_service.py
@@ -0,0 +1,339 @@
+import re
+from collections import defaultdict
+from datetime import datetime
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.models.accessory import AccessoryType, FinishedGoodsAccessory
+from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
+from app.schemas.accessory import (
+    AccessoryCreateRequest,
+    AccessoryTypeCreateRequest,
+    AccessoryTypeUpdateRequest,
+    AccessoryUpdateRequest,
+)
+
+PAICHAN_PATTERN = re.compile(r"^[A-Z]{1,2}\d{5}(-J)?$")
+
+
+# ── Error classes ──
+
+
+class InvalidPaichanError(Exception):
+    pass
+
+
+class PaichanNotFoundError(Exception):
+    pass
+
+
+class AccessoryNotFoundError(Exception):
+    pass
+
+
+class AccessoryAlreadyBoxedError(Exception):
+    pass
+
+
+class AccessoryDuplicateLocationError(Exception):
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
+class AccessoryAlreadyOffShelfError(Exception):
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
+class AccessoryTypeNotFoundError(Exception):
+    pass
+
+
+class DuplicateAccessoryTypeNameError(Exception):
+    pass
+
+
+# ── Helpers ──
+
+
+def _is_transit_location(location_code: str) -> bool:
+    return location_code.startswith("TRANS-")
+
+
+def _erp_paicha_items_sql(db: Session) -> str:
+    """Reuse the same ERP query pattern as location_service."""
+    dialect_name = db.bind.dialect.name if db.bind is not None else ""
+    if dialect_name == "postgresql":
+        return (
+            'SELECT "总排号", "工令号", "数量" '
+            'FROM "ERPAuto"."vw_productionContractData" '
+            'WHERE "排产号" = :paicha_no'
+        )
+    return (
+        "SELECT [总排号], [工令号], [数量] "
+        "FROM [ERPAuto].[vw_productionContractData] "
+        "WHERE [排产号] = :paicha_no"
+    )
+
+
+def _is_accessory_boxed(db: Session, accessory: FinishedGoodsAccessory) -> bool:
+    """Check whether the accessory is already boxed (EXISTS in box_item as 'accessory')."""
+    boxed = (
+        db.query(FinishedGoodsBoxItem)
+        .join(FinishedGoodsBox, FinishedGoodsBoxItem.box_id == FinishedGoodsBox.id)
+        .filter(
+            FinishedGoodsBoxItem.zongpai_no == accessory.zongpai_no,
+            FinishedGoodsBox.paichan_no == accessory.paichan_no,
+            FinishedGoodsBoxItem.item_type == "accessory",
+        )
+        .first()
+    )
+    return boxed is not None
+
+
+def _accessory_to_dict(
+    accessory: FinishedGoodsAccessory, is_boxed: bool = False
+) -> dict:
+    return {
+        "id": accessory.id,
+        "paichan_no": accessory.paichan_no,
+        "zongpai_no": accessory.zongpai_no,
+        "accessory_type": accessory.accessory_type,
+        "quantity": accessory.quantity,
+        "location_code": accessory.location_code,
+        "is_boxed": is_boxed,
+        "created_at": accessory.created_at.isoformat(),
+    }
+
+
+# ── Work order query ──
+
+
+def query_work_orders(db: Session, paichan_no: str) -> dict:
+    """Query ERP for work orders grouped by work_order_no for a given paichan_no."""
+    paichan_no = paichan_no.strip().upper()
+    if not PAICHAN_PATTERN.match(paichan_no):
+        raise InvalidPaichanError()
+
+    rows = db.execute(
+        text(_erp_paicha_items_sql(db)), {"paicha_no": paichan_no}
+    ).fetchall()
+
+    if not rows:
+        raise PaichanNotFoundError()
+
+    grouped: dict[str, list[str]] = defaultdict(list)
+    for row in rows:
+        zongpai_no = row[0]
+        work_order_no = row[1]
+        grouped[work_order_no].append(zongpai_no)
+
+    work_orders = [
+        {"work_order_no": wo_no, "zongpai_nos": zongpai_nos}
+        for wo_no, zongpai_nos in grouped.items()
+    ]
+
+    return {
+        "paichan_no": paichan_no,
+        "work_orders": work_orders,
+    }
+
+
+# ── Accessory CRUD ──
+
+
+def list_accessories(db: Session, paichan_no: str) -> dict:
+    """List all accessories for a given paichan_no, including is_boxed status."""
+    accessories = (
+        db.query(FinishedGoodsAccessory)
+        .filter(FinishedGoodsAccessory.paichan_no == paichan_no)
+        .order_by(FinishedGoodsAccessory.id)
+        .all()
+    )
+
+    items = []
+    for acc in accessories:
+        boxed = _is_accessory_boxed(db, acc)
+        items.append(_accessory_to_dict(acc, is_boxed=boxed))
+
+    return {
+        "paichan_no": paichan_no,
+        "items": items,
+    }
+
+
+def create_accessory(
+    db: Session, req: AccessoryCreateRequest
+) -> FinishedGoodsAccessory:
+    """Create a new accessory record after validating against ERP data."""
+    # 1. Validate paichan_no exists in ERP
+    rows = db.execute(
+        text(_erp_paicha_items_sql(db)), {"paicha_no": req.paichan_no}
+    ).fetchall()
+
+    if not rows:
+        raise PaichanNotFoundError()
+
+    # 2. Validate zongpai_no belongs to this paichan_no in ERP
+    valid_zongpai_nos = {row[0] for row in rows}
+    if req.zongpai_no not in valid_zongpai_nos:
+        raise ValueError(
+            f"zongpai_no {req.zongpai_no} does not belong to paichan_no {req.paichan_no}"
+        )
+
+    # 3. Create the record
+    record = FinishedGoodsAccessory(
+        paichan_no=req.paichan_no,
+        zongpai_no=req.zongpai_no,
+        accessory_type=req.accessory_type,
+        quantity=req.quantity,
+        location_code=req.location_code,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+    return record
+
+
+def update_accessory(
+    db: Session, accessory_id: int, req: AccessoryUpdateRequest
+) -> dict:
+    """Partially update an accessory record with shelving logic."""
+    accessory = (
+        db.query(FinishedGoodsAccessory)
+        .filter(FinishedGoodsAccessory.id == accessory_id)
+        .first()
+    )
+    if accessory is None:
+        raise AccessoryNotFoundError()
+
+    # Check if boxed
+    if _is_accessory_boxed(db, accessory):
+        raise AccessoryAlreadyBoxedError()
+
+    # Shelving logic when location_code changes
+    if req.location_code is not None and accessory.location_code is not None:
+        existing_loc = accessory.location_code
+        new_loc = req.location_code
+
+        if existing_loc == new_loc:
+            # No change needed for location
+            pass
+        elif _is_transit_location(existing_loc):
+            raise AccessoryAlreadyOffShelfError(
+                location_code=existing_loc,
+                registered_at=accessory.created_at.isoformat(),
+            )
+        elif _is_transit_location(new_loc):
+            # Off-shelf/down-shelf: normal operation
+            accessory.location_code = new_loc
+            accessory.created_at = datetime.now()
+        else:
+            # Both are normal locations → duplicate
+            raise AccessoryDuplicateLocationError(
+                location_code=existing_loc,
+                registered_at=accessory.created_at.isoformat(),
+            )
+    elif req.location_code is not None and accessory.location_code is None:
+        # Not yet shelved, just set the location
+        accessory.location_code = req.location_code
+
+    # Update other fields if provided
+    if req.accessory_type is not None:
+        accessory.accessory_type = req.accessory_type
+    if req.quantity is not None:
+        accessory.quantity = req.quantity
+
+    db.commit()
+    db.refresh(accessory)
+    return _accessory_to_dict(accessory, is_boxed=_is_accessory_boxed(db, accessory))
+
+
+def delete_accessory(db: Session, accessory_id: int) -> dict:
+    """Delete an accessory record if it is not boxed."""
+    accessory = (
+        db.query(FinishedGoodsAccessory)
+        .filter(FinishedGoodsAccessory.id == accessory_id)
+        .first()
+    )
+    if accessory is None:
+        raise AccessoryNotFoundError()
+
+    if _is_accessory_boxed(db, accessory):
+        raise AccessoryAlreadyBoxedError()
+
+    db.delete(accessory)
+    db.commit()
+    return {"id": accessory_id, "deleted": True}
+
+
+# ── Accessory type management ──
+
+
+def list_accessory_types(db: Session) -> dict:
+    """List all accessory types ordered by sort_order."""
+    types = (
+        db.query(AccessoryType)
+        .order_by(AccessoryType.sort_order, AccessoryType.id)
+        .all()
+    )
+    return {
+        "types": [
+            {"id": t.id, "name": t.name, "sort_order": t.sort_order} for t in types
+        ]
+    }
+
+
+def create_accessory_type(db: Session, req: AccessoryTypeCreateRequest) -> dict:
+    """Create a new accessory type. Name must be unique."""
+    existing = db.query(AccessoryType).filter(AccessoryType.name == req.name).first()
+    if existing is not None:
+        raise DuplicateAccessoryTypeNameError()
+
+    record = AccessoryType(
+        name=req.name,
+        sort_order=req.sort_order,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+    return {"id": record.id, "name": record.name, "sort_order": record.sort_order}
+
+
+def update_accessory_type(
+    db: Session, type_id: int, req: AccessoryTypeUpdateRequest
+) -> dict:
+    """Partially update an accessory type."""
+    record = db.query(AccessoryType).filter(AccessoryType.id == type_id).first()
+    if record is None:
+        raise AccessoryTypeNotFoundError()
+
+    # If name is being changed, check for duplicate
+    if req.name is not None and req.name != record.name:
+        duplicate = (
+            db.query(AccessoryType).filter(AccessoryType.name == req.name).first()
+        )
+        if duplicate is not None:
+            raise DuplicateAccessoryTypeNameError()
+        record.name = req.name
+
+    if req.sort_order is not None:
+        record.sort_order = req.sort_order
+
+    db.commit()
+    db.refresh(record)
+    return {"id": record.id, "name": record.name, "sort_order": record.sort_order}
+
+
+def delete_accessory_type(db: Session, type_id: int) -> dict:
+    """Delete an accessory type."""
+    record = db.query(AccessoryType).filter(AccessoryType.id == type_id).first()
+    if record is None:
+        raise AccessoryTypeNotFoundError()
+
+    db.delete(record)
+    db.commit()
+    return {"id": type_id, "deleted": True}