diff --git a/nocobase_crm_mcp/projects.py b/nocobase_crm_mcp/projects.py
new file mode 100644
index 0000000..6d48f2a
--- /dev/null
+++ b/nocobase_crm_mcp/projects.py
@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from datetime import date
+from typing import Any
+
+from .config import ProjectConfig
+from .nocobase_http import NocoBaseHttpError
+
+
+def _configured_scalar_fields(config: ProjectConfig) -> tuple[str, ...]:
+    return (
+        config.name_field,
+        config.customer_field,
+        config.inquiry_date_field,
+        config.progress_field,
+        config.applications_field,
+        config.part_numbers_field,
+        config.period_usage_field,
+        config.details_field,
+    )
+
+
+PROJECT_WHITELIST = _configured_scalar_fields(ProjectConfig())
+
+
+class _InvalidProjectInput(ValueError):
+    pass
+
+
+class _InvalidProgress(ValueError):
+    pass
+
+
+class ProjectService:
+    def __init__(self, config: ProjectConfig, nb_client: Any, http_client: Any) -> None:
+        self.config = config
+        self.nb_client = nb_client
+        self.http_client = http_client
+        self._whitelist = _configured_scalar_fields(config)
+
+    def create_project(
+        self,
+        *,
+        name: str,
+        customer_id: Any | None = None,
+        inquiry_date: str | None = None,
+        progress: str | None = None,
+        applications: str | None = None,
+        part_numbers: str | None = None,
+        period_usage: str | None = None,
+        details: str | None = None,
+        attachment_ids: Sequence[Any] | None = None,
+    ) -> dict[str, Any]:
+        try:
+            values = self._project_values(
+                name=name,
+                customer_id=customer_id,
+                inquiry_date=inquiry_date,
+                progress=progress,
+                applications=applications,
+                part_numbers=part_numbers,
+                period_usage=period_usage,
+                details=details,
+                require_name=True,
+            )
+            attachment_ids = _deduplicated_attachment_ids(attachment_ids)
+        except _InvalidProgress:
+            return _error("invalid_progress", "progress must be one of the configured project progress labels")
+        except _InvalidProjectInput as error:
+            return _error("invalid_project_input", str(error))
+
+        response = self.nb_client.create_record(self.config.resource, values, self._whitelist)
+        project_id = _project_id(response, self.config.id_field)
+        if not attachment_ids:
+            return {"project_id": project_id}
+
+        try:
+            self.http_client.add_attachment_ids(project_id, attachment_ids)
+        except Exception as error:
+            return _attachment_error(project_id, error)
+        return {"project_id": project_id}
+
+    def update_project(
+        self,
+        project_id: Any,
+        *,
+        name: str | None = None,
+        customer_id: Any | None = None,
+        inquiry_date: str | None = None,
+        progress: str | None = None,
+        applications: str | None = None,
+        part_numbers: str | None = None,
+        period_usage: str | None = None,
+        details: str | None = None,
+        add_attachment_ids: Sequence[Any] | None = None,
+        remove_attachment_ids: Sequence[Any] | None = None,
+    ) -> dict[str, Any]:
+        try:
+            values = self._project_values(
+                name=name,
+                customer_id=customer_id,
+                inquiry_date=inquiry_date,
+                progress=progress,
+                applications=applications,
+                part_numbers=part_numbers,
+                period_usage=period_usage,
+                details=details,
+                require_name=False,
+            )
+            add_attachment_ids = _deduplicated_attachment_ids(add_attachment_ids)
+            remove_attachment_ids = _deduplicated_attachment_ids(remove_attachment_ids)
+            if _overlap(add_attachment_ids, remove_attachment_ids):
+                raise _InvalidProjectInput("attachment IDs cannot be added and removed in the same request")
+            if not values and not add_attachment_ids and not remove_attachment_ids:
+                raise _InvalidProjectInput("update request must include project fields or attachment IDs")
+        except _InvalidProgress:
+            return _error("invalid_progress", "progress must be one of the configured project progress labels")
+        except _InvalidProjectInput as error:
+            return _error("invalid_project_input", str(error))
+
+        if values:
+            self.nb_client.update_record(self.config.resource, project_id, values, self._whitelist)
+
+        try:
+            if add_attachment_ids:
+                self.http_client.add_attachment_ids(project_id, add_attachment_ids)
+            if remove_attachment_ids:
+                self.http_client.remove_attachment_ids(project_id, remove_attachment_ids)
+        except Exception as error:
+            return _attachment_error(project_id, error)
+        return {"project_id": project_id}
+
+    def _project_values(
+        self,
+        *,
+        name: str | None,
+        customer_id: Any | None,
+        inquiry_date: str | None,
+        progress: str | None,
+        applications: str | None,
+        part_numbers: str | None,
+        period_usage: str | None,
+        details: str | None,
+        require_name: bool,
+    ) -> dict[str, Any]:
+        values: dict[str, Any] = {}
+        if name is not None:
+            if not isinstance(name, str) or not (normalized_name := name.strip()):
+                raise _InvalidProjectInput("name must not be blank")
+            values[self.config.name_field] = normalized_name
+        elif require_name:
+            raise _InvalidProjectInput("name is required")
+
+        if customer_id is not None:
+            values[self.config.customer_field] = customer_id
+        if inquiry_date is not None:
+            if not isinstance(inquiry_date, str):
+                raise _InvalidProjectInput("inquiry_date must be a YYYY-MM-DD date")
+            try:
+                parsed_date = date.fromisoformat(inquiry_date)
+                if parsed_date.isoformat() != inquiry_date:
+                    raise ValueError("date must use YYYY-MM-DD format")
+                values[self.config.inquiry_date_field] = parsed_date.isoformat()
+            except ValueError as error:
+                raise _InvalidProjectInput("inquiry_date must be a YYYY-MM-DD date") from error
+        if progress is not None:
+            if not isinstance(progress, str) or progress not in self.config.progress_values:
+                raise _InvalidProgress
+            values[self.config.progress_field] = self.config.progress_values[progress]
+
+        for value, field in (
+            (applications, self.config.applications_field),
+            (part_numbers, self.config.part_numbers_field),
+            (period_usage, self.config.period_usage_field),
+            (details, self.config.details_field),
+        ):
+            if value is not None:
+                values[field] = value
+        return values
+
+
+def _deduplicated_attachment_ids(attachment_ids: Sequence[Any] | None) -> list[Any]:
+    if attachment_ids is None:
+        return []
+    if isinstance(attachment_ids, (str, bytes)) or not isinstance(attachment_ids, Sequence):
+        raise _InvalidProjectInput("attachment IDs must be a list")
+
+    unique_ids: list[Any] = []
+    for attachment_id in attachment_ids:
+        if attachment_id is None or isinstance(attachment_id, (Mapping, Sequence)):
+            raise _InvalidProjectInput("attachment IDs must be scalar values")
+        if attachment_id not in unique_ids:
+            unique_ids.append(attachment_id)
+    return unique_ids
+
+
+def _overlap(left: Sequence[Any], right: Sequence[Any]) -> bool:
+    return any(left_id == right_id for left_id in left for right_id in right)
+
+
+def _project_id(response: Mapping[str, Any], id_field: str) -> Any:
+    data = response.get("data")
+    if not isinstance(data, Mapping) or id_field not in data:
+        raise ValueError(f"NocoBase create response did not contain {id_field!r}")
+    return data[id_field]
+
+
+def _attachment_error(project_id: Any, error: Exception) -> dict[str, Any]:
+    if isinstance(error, NocoBaseHttpError):
+        if error.status_code == 404:
+            code = "attachment_not_found"
+        elif error.status_code in {401, 403}:
+            code = "nb_upload_unauthorized"
+        else:
+            code = "nb_upload_failed"
+    else:
+        code = "nb_upload_failed"
+    return {"project_id": project_id, "error": {"code": code, "message": str(error)}}
+
+
+def _error(code: str, message: str) -> dict[str, Any]:
+    return {"error": {"code": code, "message": message}}
diff --git a/tests/test_projects.py b/tests/test_projects.py
new file mode 100644
index 0000000..3e45ca7
--- /dev/null
+++ b/tests/test_projects.py
@@ -0,0 +1,160 @@
+import unittest
+
+from nocobase_crm_mcp.config import ProjectConfig
+from nocobase_crm_mcp.nocobase_http import NocoBaseHttpError
+from nocobase_crm_mcp.projects import PROJECT_WHITELIST, ProjectService
+
+
+class FakeNbClient:
+    def __init__(self) -> None:
+        self.create_calls: list[tuple[str, dict, tuple[str, ...]]] = []
+        self.update_calls: list[tuple[str, object, dict, tuple[str, ...]]] = []
+
+    def create_record(self, resource, values, whitelist):
+        self.create_calls.append((resource, dict(values), tuple(whitelist)))
+        return {"data": {"id": 41}}
+
+    def update_record(self, resource, project_id, values, whitelist):
+        self.update_calls.append((resource, project_id, dict(values), tuple(whitelist)))
+        return {"data": {"id": project_id}}
+
+
+class FakeHttpClient:
+    def __init__(self) -> None:
+        self.add_calls: list[tuple[object, list[object]]] = []
+        self.remove_calls: list[tuple[object, list[object]]] = []
+        self.add_error: Exception | None = None
+        self.remove_error: Exception | None = None
+
+    def add_attachment_ids(self, project_id, attachment_ids) -> None:
+        self.add_calls.append((project_id, list(attachment_ids)))
+        if self.add_error:
+            raise self.add_error
+
+    def remove_attachment_ids(self, project_id, attachment_ids) -> None:
+        self.remove_calls.append((project_id, list(attachment_ids)))
+        if self.remove_error:
+            raise self.remove_error
+
+
+class ProjectServiceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.nb = FakeNbClient()
+        self.http = FakeHttpClient()
+        self.config = ProjectConfig()
+        self.service = ProjectService(self.config, self.nb, self.http)
+
+    def test_create_maps_fixed_fields_then_adds_attachment_ids(self) -> None:
+        result = self.service.create_project(
+            name=" EV inverter ", customer_id=3, progress="SOP", attachment_ids=[8, 9]
+        )
+
+        self.assertEqual(
+            self.nb.create_calls,
+            [
+                (
+                    "t_projects",
+                    {
+                        "f_project_name": "EV inverter",
+                        "client_ID": 3,
+                        "f_projects_progress": "a9t9awhtt3r",
+                    },
+                    PROJECT_WHITELIST,
+                )
+            ],
+        )
+        self.assertEqual(self.http.add_calls, [(41, [8, 9])])
+        self.assertEqual(result["project_id"], 41)
+
+    def test_whitelist_contains_only_the_configured_scalar_fields(self) -> None:
+        self.service.create_project(name="x")
+
+        self.assertEqual(
+            PROJECT_WHITELIST,
+            (
+                self.config.name_field,
+                self.config.customer_field,
+                self.config.inquiry_date_field,
+                self.config.progress_field,
+                self.config.applications_field,
+                self.config.part_numbers_field,
+                self.config.period_usage_field,
+                self.config.details_field,
+            ),
+        )
+        self.assertNotIn(self.config.attachment_field, PROJECT_WHITELIST)
+
+    def test_update_rejects_overlapping_attachment_lists_before_writing(self) -> None:
+        result = self.service.update_project(41, add_attachment_ids=[8], remove_attachment_ids=[8])
+
+        self.assertEqual(result["error"]["code"], "invalid_project_input")
+        self.assertEqual(self.nb.update_calls, [])
+        self.assertEqual(self.http.add_calls, [])
+        self.assertEqual(self.http.remove_calls, [])
+
+    def test_invalid_date_progress_and_blank_name_have_stable_codes(self) -> None:
+        self.assertEqual(
+            self.service.create_project(name="x", inquiry_date="2026/07/12")["error"]["code"],
+            "invalid_project_input",
+        )
+        self.assertEqual(
+            self.service.create_project(name="x", inquiry_date="20260712")["error"]["code"],
+            "invalid_project_input",
+        )
+        self.assertEqual(
+            self.service.create_project(name="x", progress="草稿")["error"]["code"],
+            "invalid_progress",
+        )
+        self.assertEqual(self.service.create_project(name=" \t ")["error"]["code"], "invalid_project_input")
+
+    def test_update_rejects_an_empty_request_before_writing(self) -> None:
+        result = self.service.update_project(41)
+
+        self.assertEqual(result["error"]["code"], "invalid_project_input")
+        self.assertEqual(self.nb.update_calls, [])
+
+    def test_update_writes_scalars_then_deduplicated_attachment_changes_in_order(self) -> None:
+        result = self.service.update_project(
+            41,
+            inquiry_date="2026-07-12",
+            add_attachment_ids=[8, 8, 9],
+            remove_attachment_ids=[10, 10],
+        )
+
+        self.assertEqual(
+            self.nb.update_calls,
+            [
+                (
+                    "t_projects",
+                    41,
+                    {"xm_project_inquiry_date": "2026-07-12"},
+                    PROJECT_WHITELIST,
+                )
+            ],
+        )
+        self.assertEqual(self.http.add_calls, [(41, [8, 9])])
+        self.assertEqual(self.http.remove_calls, [(41, [10])])
+        self.assertEqual(result, {"project_id": 41})
+
+    def test_association_errors_keep_project_id_and_map_to_stable_codes(self) -> None:
+        cases = [
+            (404, "attachment_not_found"),
+            (401, "nb_upload_unauthorized"),
+            (403, "nb_upload_unauthorized"),
+            (500, "nb_upload_failed"),
+        ]
+
+        for status_code, error_code in cases:
+            with self.subTest(status_code=status_code):
+                self.setUp()
+                self.http.add_error = NocoBaseHttpError(status_code, "attachment request failed")
+
+                result = self.service.create_project(name="x", attachment_ids=[8])
+
+                self.assertEqual(result["project_id"], 41)
+                self.assertEqual(result["error"]["code"], error_code)
+                self.assertEqual(len(self.nb.create_calls), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()