From 7ab9f5750088ecc9133fa50c438a30b05e4b4ebd Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 01:31:12 +0000
Subject: [PATCH 1/9] feat: add project config and NocoBase HTTP client

---
 nocobase_crm_mcp/config.py        | 50 ++++++++++++++++-
 nocobase_crm_mcp/nocobase_http.py | 83 +++++++++++++++++++++++++++
 pyproject.toml                    |  2 +
 tests/test_nocobase_http.py       | 93 +++++++++++++++++++++++++++++++
 uv.lock                           |  8 ++-
 5 files changed, 234 insertions(+), 2 deletions(-)
 create mode 100644 nocobase_crm_mcp/nocobase_http.py
 create mode 100644 tests/test_nocobase_http.py

diff --git a/nocobase_crm_mcp/config.py b/nocobase_crm_mcp/config.py
index 3eee589..3b7f1c1 100644
--- a/nocobase_crm_mcp/config.py
+++ b/nocobase_crm_mcp/config.py
@@ -2,11 +2,21 @@ from __future__ import annotations
 
 import json
 import os
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Any
 
 
+PROJECT_PROGRESS_VALUES = {
+    "询价阶段": "fk9xj8akn46",
+    "已报价": "k4w2vcz819w",
+    "送样": "g95vqeetu83",
+    "已定点": "fe0peg505d7",
+    "SOP": "a9t9awhtt3r",
+    "终止": "a9t9awhtsfasg3r",
+}
+
+
 @dataclass(frozen=True)
 class CustomerConfig:
     resource: str = "customers"
@@ -109,6 +119,22 @@ class UserConfig:
             object.__setattr__(self, "match_fields", ["username", "nickname", "email", "f_yh_english_name"])
 
 
+@dataclass(frozen=True)
+class ProjectConfig:
+    resource: str = "t_projects"
+    id_field: str = "id"
+    name_field: str = "f_project_name"
+    customer_field: str = "client_ID"
+    inquiry_date_field: str = "xm_project_inquiry_date"
+    progress_field: str = "f_projects_progress"
+    applications_field: str = "f_projects_applications"
+    part_numbers_field: str = "f_projects_part_numbers"
+    period_usage_field: str = "f_projects_period_usage"
+    details_field: str = "f_projects_details"
+    attachment_field: str = "f_projects_attachments"
+    progress_values: dict[str, str] = field(default_factory=lambda: PROJECT_PROGRESS_VALUES.copy())
+
+
 @dataclass(frozen=True)
 class CrmConfig:
     env_name: str = "nb-mcp"
@@ -117,6 +143,7 @@ class CrmConfig:
     contact: ContactConfig = ContactConfig()
     followup: FollowupConfig = FollowupConfig()
     user: UserConfig = UserConfig()
+    project: ProjectConfig = ProjectConfig()
 
 
 def _load_json(path: Path) -> dict[str, Any]:
@@ -208,6 +235,26 @@ def _user_config(data: dict[str, Any] | None) -> UserConfig:
     )
 
 
+def _project_config(data: dict[str, Any] | None) -> ProjectConfig:
+    if not data:
+        return ProjectConfig()
+    base = ProjectConfig()
+    return replace(
+        base,
+        resource=str(_pick(data, "resource") or base.resource),
+        id_field=str(_pick(data, "idField", "id_field") or base.id_field),
+        name_field=str(_pick(data, "nameField", "name_field") or base.name_field),
+        customer_field=str(_pick(data, "customerField", "customer_field") or base.customer_field),
+        inquiry_date_field=str(_pick(data, "inquiryDateField", "inquiry_date_field") or base.inquiry_date_field),
+        progress_field=str(_pick(data, "progressField", "progress_field") or base.progress_field),
+        applications_field=str(_pick(data, "applicationsField", "applications_field") or base.applications_field),
+        part_numbers_field=str(_pick(data, "partNumbersField", "part_numbers_field") or base.part_numbers_field),
+        period_usage_field=str(_pick(data, "periodUsageField", "period_usage_field") or base.period_usage_field),
+        details_field=str(_pick(data, "detailsField", "details_field") or base.details_field),
+        attachment_field=str(_pick(data, "attachmentField", "attachment_field") or base.attachment_field),
+    )
+
+
 def load_config(path: str | Path | None = None) -> CrmConfig:
     configured_path = path or os.environ.get("NOCOBASE_MCP_CONFIG")
     raw: dict[str, Any] = {}
@@ -228,4 +275,5 @@ def load_config(path: str | Path | None = None) -> CrmConfig:
         contact=_contact_config(collections.get("contacts")),
         followup=_followup_config(collections.get("followups")),
         user=_user_config(raw.get("users") or raw.get("user")),
+        project=_project_config(raw.get("projects")),
     )
diff --git a/nocobase_crm_mcp/nocobase_http.py b/nocobase_crm_mcp/nocobase_http.py
new file mode 100644
index 0000000..3fc8849
--- /dev/null
+++ b/nocobase_crm_mcp/nocobase_http.py
@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Sequence
+
+import httpx
+
+
+class NocoBaseHttpError(RuntimeError):
+    def __init__(self, status_code: int, message: str) -> None:
+        self.status_code = status_code
+        super().__init__(message)
+
+
+class NocoBaseHttpClient:
+    """Small synchronous client for the fixed NocoBase attachment endpoints."""
+
+    _ATTACHMENT_FIELD = "t_projects.f_projects_attachments"
+
+    def __init__(
+        self,
+        api_base_url: str,
+        jwt: str,
+        timeout: float = 30.0,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        self.api_base_url = api_base_url.strip().rstrip("/")
+        self.jwt = jwt.strip()
+        if not self.api_base_url:
+            raise ValueError("NocoBase API base URL must not be blank")
+        if not self.jwt:
+            raise ValueError("NocoBase JWT must not be blank")
+        self._client = httpx.Client(
+            base_url=self.api_base_url,
+            headers={"Authorization": f"Bearer {self.jwt}"},
+            timeout=timeout,
+            transport=transport,
+        )
+
+    def upload_attachment(self, path: Path, filename: str, mimetype: str | None) -> dict[str, Any]:
+        with Path(path).open("rb") as handle:
+            response = self._client.post(
+                "/attachments:create",
+                params={"attachmentField": self._ATTACHMENT_FIELD},
+                files={"file": (filename, handle, mimetype or "application/octet-stream")},
+            )
+        payload = self._json_response(response)
+        data = payload.get("data")
+        if not isinstance(data, dict):
+            raise NocoBaseHttpError(response.status_code, "NocoBase response did not contain an attachment object")
+        return data
+
+    def add_attachment_ids(self, project_id: Any, attachment_ids: Sequence[Any]) -> None:
+        self._post_attachment_ids(project_id, attachment_ids, "add")
+
+    def remove_attachment_ids(self, project_id: Any, attachment_ids: Sequence[Any]) -> None:
+        self._post_attachment_ids(project_id, attachment_ids, "remove")
+
+    def _post_attachment_ids(self, project_id: Any, attachment_ids: Sequence[Any], action: str) -> None:
+        response = self._client.post(
+            f"/t_projects/{project_id}/f_projects_attachments:{action}",
+            json={"tk": list(attachment_ids)},
+        )
+        self._raise_for_error(response)
+
+    def _json_response(self, response: httpx.Response) -> dict[str, Any]:
+        self._raise_for_error(response)
+        try:
+            payload = response.json()
+        except ValueError as error:
+            raise NocoBaseHttpError(response.status_code, "NocoBase response was not valid JSON") from error
+        if not isinstance(payload, dict):
+            raise NocoBaseHttpError(response.status_code, "NocoBase response was not a JSON object")
+        return payload
+
+    def _raise_for_error(self, response: httpx.Response) -> None:
+        if response.is_success:
+            return
+        message = response.text.strip() or response.reason_phrase or "NocoBase request failed"
+        raise NocoBaseHttpError(response.status_code, self._redact(message))
+
+    def _redact(self, value: str) -> str:
+        return value.replace(self.jwt, "<redacted>")
diff --git a/pyproject.toml b/pyproject.toml
index ed8869c..273d533 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,9 @@ version = "0.1.0"
 description = "Read-only MCP server for querying NocoBase CRM data through nb CLI."
 requires-python = ">=3.12"
 dependencies = [
+    "httpx>=0.27,<1",
     "mcp[cli]>=1.27,<2",
+    "python-multipart>=0.0.9,<1",
 ]
 
 [project.scripts]
diff --git a/tests/test_nocobase_http.py b/tests/test_nocobase_http.py
new file mode 100644
index 0000000..080ec06
--- /dev/null
+++ b/tests/test_nocobase_http.py
@@ -0,0 +1,93 @@
+import json
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import httpx
+
+from nocobase_crm_mcp.config import load_config
+from nocobase_crm_mcp.nocobase_http import NocoBaseHttpClient, NocoBaseHttpError
+
+
+class ProjectConfigTests(unittest.TestCase):
+    def test_load_config_parses_project_defaults(self) -> None:
+        with TemporaryDirectory() as directory:
+            path = Path(directory) / "config.json"
+            path.write_text(json.dumps({"projects": {"resource": "t_projects"}}), encoding="utf-8")
+
+            config = load_config(path)
+
+        self.assertEqual(config.project.resource, "t_projects")
+        self.assertEqual(config.project.name_field, "f_project_name")
+        self.assertEqual(config.project.customer_field, "client_ID")
+        self.assertEqual(config.project.inquiry_date_field, "xm_project_inquiry_date")
+        self.assertEqual(config.project.progress_field, "f_projects_progress")
+        self.assertEqual(config.project.applications_field, "f_projects_applications")
+        self.assertEqual(config.project.part_numbers_field, "f_projects_part_numbers")
+        self.assertEqual(config.project.period_usage_field, "f_projects_period_usage")
+        self.assertEqual(config.project.details_field, "f_projects_details")
+        self.assertEqual(config.project.attachment_field, "f_projects_attachments")
+        self.assertEqual(config.project.progress_values["SOP"], "a9t9awhtt3r")
+
+
+class NocoBaseHttpClientTests(unittest.TestCase):
+    def test_upload_posts_file_with_attachment_field(self) -> None:
+        seen: dict[str, str] = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen["url"] = str(request.url)
+            seen["auth"] = request.headers["authorization"]
+            return httpx.Response(200, json={"data": {"id": 42}})
+
+        with TemporaryDirectory() as directory:
+            path = Path(directory) / "quote.pdf"
+            path.write_bytes(b"pdf")
+            client = NocoBaseHttpClient(
+                "https://nb.example/api/", "jwt", transport=httpx.MockTransport(handler)
+            )
+
+            self.assertEqual(client.upload_attachment(path, "quote.pdf", "application/pdf")["id"], 42)
+
+        self.assertTrue(seen["url"].endswith("/attachments:create?attachmentField=t_projects.f_projects_attachments"))
+        self.assertEqual(seen["auth"], "Bearer jwt")
+
+    def test_attachment_associations_post_ids_to_fixed_relation(self) -> None:
+        requests: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            requests.append(request)
+            return httpx.Response(204)
+
+        client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
+        client.add_attachment_ids(7, [8, 9])
+        client.remove_attachment_ids(7, [10])
+
+        self.assertEqual(str(requests[0].url), "https://nb.example/api/t_projects/7/f_projects_attachments:add")
+        self.assertEqual(json.loads(requests[0].content), {"tk": [8, 9]})
+        self.assertEqual(str(requests[1].url), "https://nb.example/api/t_projects/7/f_projects_attachments:remove")
+        self.assertEqual(json.loads(requests[1].content), {"tk": [10]})
+
+    def test_rejects_blank_upload_credentials(self) -> None:
+        with self.assertRaises(ValueError):
+            NocoBaseHttpClient(" ", "jwt")
+        with self.assertRaises(ValueError):
+            NocoBaseHttpClient("https://nb.example/api", " ")
+
+    def test_http_error_redacts_jwt(self) -> None:
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(401, text="Bearer super-secret-token is invalid")
+
+        client = NocoBaseHttpClient(
+            "https://nb.example/api", "super-secret-token", transport=httpx.MockTransport(handler)
+        )
+
+        with self.assertRaises(NocoBaseHttpError) as caught:
+            client.add_attachment_ids(7, [8])
+
+        self.assertEqual(caught.exception.status_code, 401)
+        self.assertNotIn("super-secret-token", str(caught.exception))
+        self.assertIn("<redacted>", str(caught.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/uv.lock b/uv.lock
index 6421824..453aebc 100644
--- a/uv.lock
+++ b/uv.lock
@@ -352,11 +352,17 @@ name = "nocobase-crm-mcp"
 version = "0.1.0"
 source = { editable = "." }
 dependencies = [
+    { name = "httpx" },
     { name = "mcp", extra = ["cli"] },
+    { name = "python-multipart" },
 ]
 
 [package.metadata]
-requires-dist = [{ name = "mcp", extras = ["cli"], specifier = ">=1.27,<2" }]
+requires-dist = [
+    { name = "httpx", specifier = ">=0.27,<1" },
+    { name = "mcp", extras = ["cli"], specifier = ">=1.27,<2" },
+    { name = "python-multipart", specifier = ">=0.0.9,<1" },
+]
 
 [[package]]
 name = "pycparser"

From e213a91fea89487205357a89f6d7a157f7994023 Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 01:36:20 +0000
Subject: [PATCH 2/9] feat: add NocoBase project record writes

---
 nocobase_crm_mcp/nb_client.py | 28 +++++++++++++++++++++
 tests/test_nb_client.py       | 46 +++++++++++++++++++++++++++++++++++
 2 files changed, 74 insertions(+)

diff --git a/nocobase_crm_mcp/nb_client.py b/nocobase_crm_mcp/nb_client.py
index ec19f72..17b827a 100644
--- a/nocobase_crm_mcp/nb_client.py
+++ b/nocobase_crm_mcp/nb_client.py
@@ -89,6 +89,34 @@ class NbClient:
         self._append_repeated(command, "--except", except_fields)
         return self._run_json(command)
 
+    def create_record(
+        self,
+        resource: str,
+        values: Mapping[str, Any],
+        whitelist: Sequence[str],
+    ) -> dict[str, Any]:
+        if not values:
+            raise ValueError("values must not be empty")
+        command = self._resource_command("create", resource)
+        self._append_json(command, "--values", values)
+        self._append_repeated(command, "--whitelist", whitelist)
+        return self._run_json(command)
+
+    def update_record(
+        self,
+        resource: str,
+        record_id: Any,
+        values: Mapping[str, Any],
+        whitelist: Sequence[str],
+    ) -> dict[str, Any]:
+        if not values:
+            raise ValueError("values must not be empty")
+        command = self._resource_command("update", resource)
+        self._append_json_or_scalar(command, "--filter-by-tk", record_id)
+        self._append_json(command, "--values", values)
+        self._append_repeated(command, "--whitelist", whitelist)
+        return self._run_json(command)
+
     def query_records(
         self,
         resource: str,
diff --git a/tests/test_nb_client.py b/tests/test_nb_client.py
index 1a4122c..f0245a0 100644
--- a/tests/test_nb_client.py
+++ b/tests/test_nb_client.py
@@ -86,6 +86,52 @@ class NbClientTests(unittest.TestCase):
         self.assertIn("--orders", commands[1])
         self.assertIn("--timezone", commands[1])
 
+    def test_create_record_uses_values_and_whitelist(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{"id":7}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+        client.create_record("t_projects", {"f_project_name": "EV inverter"}, ["f_project_name"])
+
+        self.assertEqual(commands[0][:4], ["nb", "api", "resource", "create"])
+        self.assertEqual(
+            json.loads(commands[0][commands[0].index("--values") + 1]),
+            {"f_project_name": "EV inverter"},
+        )
+        self.assertEqual(commands[0][commands[0].index("--whitelist") + 1], "f_project_name")
+
+    def test_update_record_uses_filter_by_tk(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{"id":7}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+        client.update_record("t_projects", 7, {"client_ID": 3}, ["client_ID"])
+
+        self.assertEqual(commands[0][:4], ["nb", "api", "resource", "update"])
+        self.assertEqual(commands[0][commands[0].index("--filter-by-tk") + 1], "7")
+
+    def test_record_writes_reject_empty_values_without_running_command(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+
+        with self.assertRaises(ValueError):
+            client.create_record("t_projects", {}, ["f_project_name"])
+        with self.assertRaises(ValueError):
+            client.update_record("t_projects", 7, {}, ["f_project_name"])
+
+        self.assertEqual(commands, [])
+
     def test_list_collection_fields_uses_data_modeling_command(self) -> None:
         commands: list[list[str]] = []
 

From 4ab02025c71df9c8912ba359efe939c32e7ee3b5 Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 01:40:17 +0000
Subject: [PATCH 3/9] fix: reject nested project write values

---
 nocobase_crm_mcp/nb_client.py | 15 +++++++++++----
 tests/test_nb_client.py       | 25 +++++++++++++++++++++++++
 2 files changed, 36 insertions(+), 4 deletions(-)

diff --git a/nocobase_crm_mcp/nb_client.py b/nocobase_crm_mcp/nb_client.py
index 17b827a..5eaf694 100644
--- a/nocobase_crm_mcp/nb_client.py
+++ b/nocobase_crm_mcp/nb_client.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 import os
 import subprocess
+from collections.abc import Set as AbstractSet
 from pathlib import Path
 from typing import Any, Callable, Mapping, Sequence
 
@@ -95,8 +96,7 @@ class NbClient:
         values: Mapping[str, Any],
         whitelist: Sequence[str],
     ) -> dict[str, Any]:
-        if not values:
-            raise ValueError("values must not be empty")
+        self._validate_scalar_values(values)
         command = self._resource_command("create", resource)
         self._append_json(command, "--values", values)
         self._append_repeated(command, "--whitelist", whitelist)
@@ -109,8 +109,7 @@ class NbClient:
         values: Mapping[str, Any],
         whitelist: Sequence[str],
     ) -> dict[str, Any]:
-        if not values:
-            raise ValueError("values must not be empty")
+        self._validate_scalar_values(values)
         command = self._resource_command("update", resource)
         self._append_json_or_scalar(command, "--filter-by-tk", record_id)
         self._append_json(command, "--values", values)
@@ -301,3 +300,11 @@ class NbClient:
         if value is None or value == "":
             return
         command.extend([flag, str(value)])
+
+    @staticmethod
+    def _validate_scalar_values(values: Mapping[str, Any]) -> None:
+        if not values:
+            raise ValueError("values must not be empty")
+        for field, value in values.items():
+            if isinstance(value, (Mapping, list, AbstractSet, tuple)):
+                raise ValueError(f"values[{field!r}] must be scalar")
diff --git a/tests/test_nb_client.py b/tests/test_nb_client.py
index f0245a0..d274be0 100644
--- a/tests/test_nb_client.py
+++ b/tests/test_nb_client.py
@@ -132,6 +132,31 @@ class NbClientTests(unittest.TestCase):
 
         self.assertEqual(commands, [])
 
+    def test_record_writes_reject_nested_values_without_running_command(self) -> None:
+        commands: list[list[str]] = []
+
+        def runner(cmd, **kwargs):
+            commands.append(cmd)
+            return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{}}', stderr="")
+
+        client = NbClient(env_name="nb-mcp", runner=runner)
+        nested_values = [
+            {"client": {"id": 3}},
+            {"client_ids": [3]},
+            {"client_ids": {3}},
+            {"client_ids": (3,)},
+        ]
+
+        for values in nested_values:
+            with self.subTest(operation="create", values=values):
+                with self.assertRaises(ValueError):
+                    client.create_record("t_projects", values, list(values))
+            with self.subTest(operation="update", values=values):
+                with self.assertRaises(ValueError):
+                    client.update_record("t_projects", 7, values, list(values))
+
+        self.assertEqual(commands, [])
+
     def test_list_collection_fields_uses_data_modeling_command(self) -> None:
         commands: list[list[str]] = []
 

From 6eff1aba5972b5aad4939f59f04850839929bcef Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 01:47:43 +0000
Subject: [PATCH 4/9] feat: add fixed project write service

---
 nocobase_crm_mcp/projects.py | 223 +++++++++++++++++++++++++++++++++++
 tests/test_projects.py       | 160 +++++++++++++++++++++++++
 2 files changed, 383 insertions(+)
 create mode 100644 nocobase_crm_mcp/projects.py
 create mode 100644 tests/test_projects.py

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()

From 731a41c652088b7a9b30ad4af0762370c3f1e0fc Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 01:55:59 +0000
Subject: [PATCH 5/9] feat: add authenticated attachment uploads

---
 nocobase_crm_mcp/uploads.py | 112 ++++++++++++++++++++++++++
 tests/test_uploads.py       | 155 ++++++++++++++++++++++++++++++++++++
 2 files changed, 267 insertions(+)
 create mode 100644 nocobase_crm_mcp/uploads.py
 create mode 100644 tests/test_uploads.py

diff --git a/nocobase_crm_mcp/uploads.py b/nocobase_crm_mcp/uploads.py
new file mode 100644
index 0000000..f6a4ee6
--- /dev/null
+++ b/nocobase_crm_mcp/uploads.py
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import fnmatch
+import tempfile
+from collections.abc import Iterable, Mapping, Sequence
+from pathlib import Path
+from typing import Any
+
+from starlette.datastructures import UploadFile
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+from .nocobase_http import NocoBaseHttpError
+
+
+_COPY_CHUNK_BYTES = 1024 * 1024
+
+
+class UploadEndpoint:
+    """ASGI endpoint that forwards one bounded multipart file to NocoBase."""
+
+    def __init__(
+        self,
+        http_client: Any,
+        max_bytes: int,
+        temp_dir: str | Path,
+        allowed_mime_types: Sequence[str] | None = None,
+    ) -> None:
+        self._http_client = http_client
+        self._max_bytes = max_bytes
+        self._temp_dir = Path(temp_dir)
+        self._allowed_mime_types = tuple(allowed_mime_types) if allowed_mime_types is not None else None
+
+    async def __call__(self, request: Request) -> JSONResponse:
+        if request.method != "POST":
+            return _error_response("invalid_upload", "Uploads require POST", 405)
+        if not _is_multipart(request):
+            return _error_response("invalid_upload", "Uploads require multipart form data", 400)
+
+        try:
+            form = await request.form()
+        except Exception:
+            return _error_response("invalid_upload", "Could not read upload form data", 400)
+
+        uploads = [(name, value) for name, value in form.multi_items() if isinstance(value, UploadFile)]
+        file_values = form.getlist("file")
+        if len(uploads) != 1 or len(file_values) != 1 or uploads[0][0] != "file":
+            await _close_uploads(value for _, value in uploads)
+            return _error_response("invalid_upload", "Provide exactly one file upload", 400)
+
+        upload = file_values[0]
+        if not isinstance(upload, UploadFile):
+            return _error_response("invalid_upload", "Provide exactly one file upload", 400)
+
+        filename = upload.filename or ""
+        mimetype = upload.content_type
+        if self._allowed_mime_types is not None and not any(
+            fnmatch.fnmatch(mimetype or "", allowed) for allowed in self._allowed_mime_types
+        ):
+            await upload.close()
+            return _error_response("unsupported_media_type", "This media type is not allowed", 415)
+
+        temporary_path: Path | None = None
+        try:
+            with tempfile.NamedTemporaryFile(delete=False, dir=self._temp_dir) as temporary:
+                temporary_path = Path(temporary.name)
+                size = 0
+                while chunk := await upload.read(_COPY_CHUNK_BYTES):
+                    size += len(chunk)
+                    if size > self._max_bytes:
+                        return _error_response("upload_too_large", "Upload exceeds the configured size limit", 413)
+                    temporary.write(chunk)
+
+            attachment = self._http_client.upload_attachment(temporary_path, filename, mimetype)
+            attachment_id = _attachment_id(attachment)
+            if attachment_id is None:
+                return _error_response("nb_upload_failed", "NocoBase attachment upload failed", 502)
+            return JSONResponse(
+                {"upload_id": attachment_id, "filename": filename, "size": size, "mimetype": mimetype}
+            )
+        except NocoBaseHttpError as error:
+            code = "nb_upload_unauthorized" if error.status_code in {401, 403} else "nb_upload_failed"
+            return _error_response(code, "NocoBase attachment upload failed", 502)
+        except Exception:
+            return _error_response("nb_upload_failed", "NocoBase attachment upload failed", 502)
+        finally:
+            await upload.close()
+            if temporary_path is not None:
+                temporary_path.unlink(missing_ok=True)
+
+
+def _is_multipart(request: Request) -> bool:
+    content_type = request.headers.get("content-type", "")
+    return content_type.split(";", 1)[0].strip().lower() == "multipart/form-data"
+
+
+def _attachment_id(attachment: Any) -> int | None:
+    if not isinstance(attachment, Mapping):
+        return None
+    attachment_id = attachment.get("id")
+    if isinstance(attachment_id, int) and not isinstance(attachment_id, bool):
+        return attachment_id
+    return None
+
+
+async def _close_uploads(uploads: Iterable[UploadFile]) -> None:
+    for upload in uploads:
+        await upload.close()
+
+
+def _error_response(code: str, message: str, status_code: int) -> JSONResponse:
+    return JSONResponse({"error": {"code": code, "message": message}}, status_code=status_code)
diff --git a/tests/test_uploads.py b/tests/test_uploads.py
new file mode 100644
index 0000000..5f196e6
--- /dev/null
+++ b/tests/test_uploads.py
@@ -0,0 +1,155 @@
+import gc
+import unittest
+import warnings
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import httpx
+from starlette.applications import Starlette
+from starlette.routing import Route
+
+from nocobase_crm_mcp.nocobase_http import NocoBaseHttpError
+from nocobase_crm_mcp.uploads import UploadEndpoint
+
+
+AUTH = {"Authorization": "Bearer secret"}
+
+
+class FakeHttpClient:
+    def __init__(self) -> None:
+        self.calls: list[tuple[Path, str, str | None, bytes]] = []
+        self.error: Exception | None = None
+
+    def upload_attachment(self, path: Path, filename: str, mimetype: str | None) -> dict[str, object]:
+        self.calls.append((path, filename, mimetype, path.read_bytes()))
+        if self.error:
+            raise self.error
+        return {"id": 42}
+
+
+class UploadEndpointTests(unittest.IsolatedAsyncioTestCase):
+    async def _client_for(
+        self,
+        http_client: FakeHttpClient,
+        temp_dir: str,
+        *,
+        max_bytes: int = 5,
+        allowed_mime_types: tuple[str, ...] | None = None,
+    ) -> httpx.AsyncClient:
+        endpoint = UploadEndpoint(http_client, max_bytes, temp_dir, allowed_mime_types)
+
+        async def upload_route(request):
+            return await endpoint(request)
+
+        app = Starlette(
+            routes=[
+                Route(
+                    "/uploads",
+                    upload_route,
+                    methods=["GET", "POST"],
+                )
+            ]
+        )
+        return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver")
+
+    async def test_upload_returns_nocobase_attachment_id_and_removes_temp_file(self) -> None:
+        http_client = FakeHttpClient()
+        with TemporaryDirectory() as temp_dir:
+            async with await self._client_for(http_client, temp_dir) as client:
+                response = await client.post(
+                    "/uploads",
+                    headers=AUTH,
+                    files={"file": ("quote.pdf", b"pdf", "application/pdf")},
+                )
+
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(
+                response.json(),
+                {"upload_id": 42, "filename": "quote.pdf", "size": 3, "mimetype": "application/pdf"},
+            )
+            self.assertEqual(http_client.calls[0][1:], ("quote.pdf", "application/pdf", b"pdf"))
+            self.assertEqual(list(Path(temp_dir).iterdir()), [])
+
+    async def test_upload_rejects_missing_multiple_and_over_limit_files(self) -> None:
+        http_client = FakeHttpClient()
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", ResourceWarning)
+            with TemporaryDirectory() as temp_dir:
+                async with await self._client_for(http_client, temp_dir) as client:
+                    missing = await client.post("/uploads", headers=AUTH, files={})
+                    multiple = await client.post(
+                        "/uploads",
+                        headers=AUTH,
+                        files=[
+                            ("file", ("one.pdf", b"one", "application/pdf")),
+                            ("file", ("two.pdf", b"two", "application/pdf")),
+                        ],
+                    )
+                    over_limit = await client.post(
+                        "/uploads",
+                        headers=AUTH,
+                        files={"file": ("big.bin", b"x" * 6, "application/octet-stream")},
+                    )
+                temp_paths = list(Path(temp_dir).iterdir())
+            gc.collect()
+
+        self.assertEqual(missing.status_code, 400)
+        self.assertEqual(missing.json()["error"]["code"], "invalid_upload")
+        self.assertEqual(multiple.status_code, 400)
+        self.assertEqual(multiple.json()["error"]["code"], "invalid_upload")
+        self.assertEqual(over_limit.status_code, 413)
+        self.assertEqual(over_limit.json()["error"]["code"], "upload_too_large")
+        self.assertEqual(http_client.calls, [])
+        self.assertEqual(temp_paths, [])
+        self.assertFalse(any(issubclass(warning.category, ResourceWarning) for warning in caught))
+
+    async def test_upload_rejects_disallowed_mime_type_before_calling_nocobase(self) -> None:
+        http_client = FakeHttpClient()
+        with TemporaryDirectory() as temp_dir:
+            async with await self._client_for(
+                http_client, temp_dir, allowed_mime_types=("image/*", "application/pdf")
+            ) as client:
+                response = await client.post(
+                    "/uploads",
+                    headers=AUTH,
+                    files={"file": ("notes.txt", b"notes", "text/plain")},
+                )
+
+            self.assertEqual(response.status_code, 415)
+            self.assertEqual(response.json()["error"]["code"], "unsupported_media_type")
+            self.assertEqual(http_client.calls, [])
+            self.assertEqual(list(Path(temp_dir).iterdir()), [])
+
+    async def test_upload_maps_upstream_errors_and_removes_temp_file(self) -> None:
+        cases = ((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), TemporaryDirectory() as temp_dir:
+                http_client = FakeHttpClient()
+                http_client.error = NocoBaseHttpError(status_code, "upstream failure")
+                async with await self._client_for(http_client, temp_dir) as client:
+                    response = await client.post(
+                        "/uploads",
+                        headers=AUTH,
+                        files={"file": ("quote.pdf", b"pdf", "application/pdf")},
+                    )
+
+                self.assertEqual(response.status_code, 502)
+                self.assertEqual(response.json()["error"]["code"], error_code)
+                self.assertEqual(list(Path(temp_dir).iterdir()), [])
+
+    async def test_upload_requires_post_and_multipart_form_data(self) -> None:
+        http_client = FakeHttpClient()
+        with TemporaryDirectory() as temp_dir:
+            async with await self._client_for(http_client, temp_dir) as client:
+                get_response = await client.get("/uploads", headers=AUTH)
+                json_response = await client.post("/uploads", headers=AUTH, json={"file": "not-a-file"})
+
+            self.assertEqual(get_response.status_code, 405)
+            self.assertEqual(json_response.status_code, 400)
+            self.assertEqual(json_response.json()["error"]["code"], "invalid_upload")
+            self.assertEqual(list(Path(temp_dir).iterdir()), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

From dae2784e2b4255577d21eb0b6c5949d0204f2d54 Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 02:04:31 +0000
Subject: [PATCH 6/9] feat: expose project writes and upload endpoint

---
 README.md                     |  45 +++++++++++--
 config.example.json           |  13 ++++
 nocobase_crm_mcp/server.py    | 123 ++++++++++++++++++++++++++++++++--
 scripts/manage_mcp_service.py |   8 ++-
 tests/test_mcp_service.py     |  11 +++
 tests/test_server.py          |  64 +++++++++++++++++-
 6 files changed, 250 insertions(+), 14 deletions(-)

diff --git a/README.md b/README.md
index 5d68b95..c219455 100644
--- a/README.md
+++ b/README.md
@@ -1,8 +1,8 @@
 # NocoBase CRM MCP
 
-Read-only MCP server for remote agents that need customer, contact, follow-up, and CRM overview data from NocoBase.
+MCP server for remote agents that need customer, contact, follow-up, CRM overview, and tightly scoped project write access in NocoBase.
 
-The server calls the official `nb api resource list|get|query` commands. It does not expose create, update, destroy, or arbitrary command execution tools.
+The default stdio server remains read-only. Its managed HTTP mode adds only fixed-schema project create/update tools and authenticated attachment uploads; it never exposes arbitrary collections, fields, commands, or delete operations.
 
 ## Configure
 
@@ -64,6 +64,35 @@ The `owner` parameter on `crm_overview`, `list_customers`, and `list_followups`
 
 Use `crm_filter_options` to discover status/source values and owner candidates before constructing filtered queries. List tools clamp `limit` to `1..100` and normalize negative offsets to `0`.
 
+## Project writes and uploads
+
+Managed HTTP mode exposes two additional MCP tools:
+
+- `create_project(name, customer_id?, inquiry_date?, progress?, applications?, part_numbers?, period_usage?, details?, attachment_ids?)`
+- `update_project(project_id, name?, customer_id?, inquiry_date?, progress?, applications?, part_numbers?, period_usage?, details?, add_attachment_ids?, remove_attachment_ids?)`
+
+They target only `t_projects` and the fixed field mapping in `projects` in `config.json`. A project name is required for creation; `inquiry_date` uses `YYYY-MM-DD`; progress accepts only `询价阶段`, `已报价`, `送样`, `已定点`, `SOP`, or `终止`. Attachments are added or removed incrementally—there is no replacement attachment-list operation.
+
+The same HTTP service exposes authenticated `POST /uploads`. Submit exactly one multipart field named `file`; the default single-file limit is 25 MiB (`NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400`). A successful upload returns an `upload_id`, which callers can pass to either project tool. The upload route and `/mcp` use the same MCP bearer token.
+
+HTTP write mode also requires these server-side NocoBase credentials. Do not put them in `config.json`, pass them to MCP clients, or log them:
+
+- `NOCOBASE_MCP_UPLOAD_API_BASE_URL` (for example `https://nocobase.example/api`)
+- `NOCOBASE_MCP_UPLOAD_JWT` (a NocoBase user JWT)
+
+`scripts/mcp_service.sh start` creates `.runtime/nocobase-crm-mcp.env` with mode `0600`, its parent directory with mode `0700`, and only the non-secret upload size setting. Add the API URL and JWT to that private env file before starting or restarting the managed service. HTTP write mode fails at startup with the missing variable name when either credential is absent.
+
+Operational flow:
+
+```bash
+NOCOBASE_MCP_UPLOAD_API_BASE_URL='https://nocobase.example/api'
+NOCOBASE_MCP_UPLOAD_JWT='<user-jwt>'
+bash scripts/mcp_service.sh start
+curl -X POST -H "Authorization: Bearer <mcp-token>" -F "file=@quote.pdf" http://<host>:10087/uploads
+```
+
+Set the first two values in `.runtime/nocobase-crm-mcp.env` (rather than in client configuration) before the service starts. An uploaded attachment that is not yet associated with a project is intentionally retained; this release does not auto-clean orphan attachments. It also provides no delete operation for projects or attachments.
+
 ## Permissions
 
 Runtime tools are read-only, but the role behind `nb-mcp` needs these read permissions:
@@ -79,7 +108,15 @@ nb api data-modeling collections fields list --collection-name <collection> --en
 
 Use a super-admin or configuration-capable role for this one-time discovery step. The remote Agent does not need data-modeling access after `config.json` is generated.
 
-Do not grant write permissions such as collection apply/destroy, fields apply, or business data create/update/destroy for this MCP.
+For the read-only stdio/default server, do not grant write permissions such as collection apply/destroy, fields apply, or business data create/update/destroy.
+
+For the managed HTTP write mode only, the NocoBase JWT role additionally needs:
+
+- `t_projects`: create and update
+- `attachments`: upload/create
+- `t_projects.f_projects_attachments`: add and remove attachment relations
+
+Do not grant project or attachment delete permissions. This server does not offer deletes, and it retains uploaded but unassociated attachments.
 
 ## Remote HTTP MCP
 
@@ -104,7 +141,7 @@ For a managed user-level systemd service:
 bash scripts/mcp_service.sh start
 ```
 
-The first `start` creates `.runtime/nocobase-crm-mcp.env` with a random bearer token automatically. Use `new-token` only when rotating the token later.
+The first `start` creates `.runtime/nocobase-crm-mcp.env` with a random bearer token and the non-secret 25 MiB upload size setting automatically. Before the managed HTTP service can start, add `NOCOBASE_MCP_UPLOAD_API_BASE_URL` and `NOCOBASE_MCP_UPLOAD_JWT` to that private file. Use `new-token` only when rotating the token later.
 
 Show the saved URL/header/client config:
 
diff --git a/config.example.json b/config.example.json
index 9699f1c..602926b 100644
--- a/config.example.json
+++ b/config.example.json
@@ -6,6 +6,19 @@
     "fields": ["id", "username", "nickname", "email", "f_yh_english_name"],
     "matchFields": ["username", "nickname", "email", "f_yh_english_name"]
   },
+  "projects": {
+    "resource": "t_projects",
+    "idField": "id",
+    "nameField": "f_project_name",
+    "customerField": "client_ID",
+    "inquiryDateField": "xm_project_inquiry_date",
+    "progressField": "f_projects_progress",
+    "applicationsField": "f_projects_applications",
+    "partNumbersField": "f_projects_part_numbers",
+    "periodUsageField": "f_projects_period_usage",
+    "detailsField": "f_projects_details",
+    "attachmentField": "f_projects_attachments"
+  },
   "collections": {
     "customers": {
       "resource": "t_nt3pjadelqi",
diff --git a/nocobase_crm_mcp/server.py b/nocobase_crm_mcp/server.py
index 29e9a51..d5350ef 100644
--- a/nocobase_crm_mcp/server.py
+++ b/nocobase_crm_mcp/server.py
@@ -3,9 +3,13 @@ from __future__ import annotations
 import argparse
 import hmac
 import os
+import tempfile
 from pathlib import Path
 from typing import Annotated, Any
 
+from starlette.applications import Starlette
+from starlette.routing import Mount, Route
+
 try:
     from pydantic import Field
 except ModuleNotFoundError:
@@ -24,6 +28,9 @@ except ModuleNotFoundError:
 from .config import load_config
 from .crm import CrmService
 from .nb_client import NbClient
+from .nocobase_http import NocoBaseHttpClient
+from .projects import ProjectService
+from .uploads import UploadEndpoint
 
 
 DEFAULT_HTTP_HOST = "0.0.0.0"
@@ -31,6 +38,10 @@ DEFAULT_HTTP_PORT = 10087
 DEFAULT_HTTP_PATH = "/mcp"
 DEFAULT_TOKEN_ENV = "NOCOBASE_MCP_TOKEN"
 ENABLE_CONFIG_TOOLS_ENV = "NOCOBASE_MCP_ENABLE_CONFIG_TOOLS"
+UPLOAD_API_BASE_URL_ENV = "NOCOBASE_MCP_UPLOAD_API_BASE_URL"
+UPLOAD_JWT_ENV = "NOCOBASE_MCP_UPLOAD_JWT"
+UPLOAD_MAX_BYTES_ENV = "NOCOBASE_MCP_UPLOAD_MAX_BYTES"
+DEFAULT_UPLOAD_MAX_BYTES = 25 * 1024 * 1024
 
 
 def _bearer_token(authorization: str) -> str:
@@ -93,9 +104,17 @@ def _make_service(config_path: str | Path | None = None) -> CrmService:
     return CrmService(config, client)
 
 
+def _safe_tool_result(operation: Any, /, *args: Any, **kwargs: Any) -> dict[str, Any]:
+    try:
+        return operation(*args, **kwargs)
+    except Exception:
+        return {"error": {"code": "project_write_failed", "message": "Project write failed"}}
+
+
 def create_server(
     *,
     crm_service: CrmService | None = None,
+    project_service: ProjectService | None = None,
     config_path: str | Path | None = None,
     readonly: bool = True,
     host: str = DEFAULT_HTTP_HOST,
@@ -106,8 +125,9 @@ def create_server(
 ) -> Any:
     if FastMCP is None:
         raise RuntimeError("Missing MCP SDK. Install dependencies with: uv sync")
-    del readonly
     service = crm_service or _make_service(config_path)
+    if not readonly and project_service is None:
+        raise ValueError("project_service is required when readonly is false")
     transport_security = None
     if TransportSecuritySettings is not None and (allowed_hosts is not None or allowed_origins is not None):
         transport_security = TransportSecuritySettings(
@@ -237,6 +257,64 @@ def create_server(
             offset=offset,
         )
 
+    if not readonly:
+
+        @server.tool()
+        def create_project(
+            name: Annotated[str, Field(description="Project name. It must not be blank.")],
+            customer_id: Annotated[Any | None, Field(description="Optional NocoBase customer ID.")] = None,
+            inquiry_date: Annotated[str | None, Field(description="Optional project inquiry date in YYYY-MM-DD format.")] = None,
+            progress: Annotated[str | None, Field(description="Optional configured Chinese project progress label.")] = None,
+            applications: Annotated[str | None, Field(description="Optional project applications.")] = None,
+            part_numbers: Annotated[str | None, Field(description="Optional project part numbers.")] = None,
+            period_usage: Annotated[str | None, Field(description="Optional project period usage.")] = None,
+            details: Annotated[str | None, Field(description="Optional project details.")] = None,
+            attachment_ids: Annotated[list[Any] | None, Field(description="Optional uploaded attachment IDs to associate.")] = None,
+        ) -> dict[str, Any]:
+            """Create one fixed-schema NocoBase project and optionally associate uploaded attachments."""
+            return _safe_tool_result(
+                project_service.create_project,
+                name=name,
+                customer_id=customer_id,
+                inquiry_date=inquiry_date,
+                progress=progress,
+                applications=applications,
+                part_numbers=part_numbers,
+                period_usage=period_usage,
+                details=details,
+                attachment_ids=attachment_ids,
+            )
+
+        @server.tool()
+        def update_project(
+            project_id: Annotated[Any, Field(description="NocoBase project ID.")],
+            name: Annotated[str | None, Field(description="Optional project name.")] = None,
+            customer_id: Annotated[Any | None, Field(description="Optional NocoBase customer ID.")] = None,
+            inquiry_date: Annotated[str | None, Field(description="Optional project inquiry date in YYYY-MM-DD format.")] = None,
+            progress: Annotated[str | None, Field(description="Optional configured Chinese project progress label.")] = None,
+            applications: Annotated[str | None, Field(description="Optional project applications.")] = None,
+            part_numbers: Annotated[str | None, Field(description="Optional project part numbers.")] = None,
+            period_usage: Annotated[str | None, Field(description="Optional project period usage.")] = None,
+            details: Annotated[str | None, Field(description="Optional project details.")] = None,
+            add_attachment_ids: Annotated[list[Any] | None, Field(description="Attachment IDs to add incrementally.")] = None,
+            remove_attachment_ids: Annotated[list[Any] | None, Field(description="Attachment IDs to remove incrementally.")] = None,
+        ) -> dict[str, Any]:
+            """Update fixed project fields and incrementally add or remove uploaded attachments."""
+            return _safe_tool_result(
+                project_service.update_project,
+                project_id,
+                name=name,
+                customer_id=customer_id,
+                inquiry_date=inquiry_date,
+                progress=progress,
+                applications=applications,
+                part_numbers=part_numbers,
+                period_usage=period_usage,
+                details=details,
+                add_attachment_ids=add_attachment_ids,
+                remove_attachment_ids=remove_attachment_ids,
+            )
+
     return server
 
 
@@ -247,8 +325,33 @@ def _read_token_from_env(env_name: str) -> str:
     return token
 
 
-def create_http_app(server: Any, token: str) -> Any:
-    return BearerTokenAuthMiddleware(server.streamable_http_app(), token)
+def create_http_app(server: Any, token: str, upload_endpoint: UploadEndpoint | None = None) -> Any:
+    if upload_endpoint is None:
+        return BearerTokenAuthMiddleware(server.streamable_http_app(), token)
+    app = Starlette(
+        routes=[
+            Route("/uploads", endpoint=upload_endpoint, methods=["POST"]),
+            Mount("/", app=server.streamable_http_app()),
+        ]
+    )
+    return BearerTokenAuthMiddleware(app, token)
+
+
+def _read_upload_environment() -> tuple[str, str, int]:
+    api_base_url = os.environ.get(UPLOAD_API_BASE_URL_ENV, "").strip()
+    if not api_base_url:
+        raise RuntimeError(f"{UPLOAD_API_BASE_URL_ENV} must be set for HTTP write mode")
+    jwt = os.environ.get(UPLOAD_JWT_ENV, "").strip()
+    if not jwt:
+        raise RuntimeError(f"{UPLOAD_JWT_ENV} must be set for HTTP write mode")
+    raw_max_bytes = os.environ.get(UPLOAD_MAX_BYTES_ENV, str(DEFAULT_UPLOAD_MAX_BYTES)).strip()
+    try:
+        max_bytes = int(raw_max_bytes)
+    except ValueError as error:
+        raise RuntimeError(f"{UPLOAD_MAX_BYTES_ENV} must be a positive integer") from error
+    if max_bytes <= 0:
+        raise RuntimeError(f"{UPLOAD_MAX_BYTES_ENV} must be a positive integer")
+    return api_base_url, jwt, max_bytes
 
 
 def run_http_server(
@@ -264,16 +367,24 @@ def run_http_server(
     import uvicorn
 
     token = _read_token_from_env(token_env)
+    api_base_url, jwt, max_bytes = _read_upload_environment()
+    config = load_config(config_path)
+    nb_client = NbClient(env_name=config.env_name, data_source=config.data_source)
+    crm_service = CrmService(config, nb_client)
+    http_client = NocoBaseHttpClient(api_base_url, jwt)
+    project_service = ProjectService(config.project, nb_client, http_client)
     server = create_server(
-        config_path=config_path,
-        readonly=True,
+        crm_service=crm_service,
+        project_service=project_service,
+        readonly=False,
         host=host,
         port=port,
         streamable_http_path=streamable_http_path,
         allowed_hosts=allowed_hosts,
         allowed_origins=allowed_origins,
     )
-    app = create_http_app(server, token)
+    upload_endpoint = UploadEndpoint(http_client, max_bytes=max_bytes, temp_dir=Path(tempfile.gettempdir()))
+    app = create_http_app(server, token, upload_endpoint=upload_endpoint)
     uvicorn.run(app, host=host, port=port, log_level="info")
 
 
diff --git a/scripts/manage_mcp_service.py b/scripts/manage_mcp_service.py
index 05ee77b..529e34a 100755
--- a/scripts/manage_mcp_service.py
+++ b/scripts/manage_mcp_service.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Manage the user-level read-only HTTP NocoBase CRM MCP systemd service."""
+"""Manage the user-level HTTP NocoBase CRM MCP systemd service."""
 
 from __future__ import annotations
 
@@ -26,6 +26,7 @@ DEFAULT_NB_ENV = "nb-mcp"
 DEFAULT_CONFIG_PATH = "config.json"
 DEFAULT_TOKEN_BYTES = 32
 DEFAULT_UV_CACHE_DIR = "/tmp/uv-cache"
+DEFAULT_UPLOAD_MAX_BYTES = 25 * 1024 * 1024
 
 
 @dataclass(frozen=True)
@@ -87,7 +88,7 @@ def render_service_unit(
     )
     return (
         "[Unit]\n"
-        "Description=NocoBase CRM readonly HTTP MCP server\n"
+        "Description=NocoBase CRM HTTP MCP server\n"
         "After=network-online.target\n"
         "Wants=network-online.target\n"
         "\n"
@@ -125,6 +126,7 @@ def write_token_env_file(
         env_file.write(f"NOCOBASE_MCP_PATH={path}\n")
         env_file.write(f"NOCOBASE_MCP_ENV={nb_env}\n")
         env_file.write(f"NOCOBASE_MCP_CONFIG={config_path}\n")
+        env_file.write(f"NOCOBASE_MCP_UPLOAD_MAX_BYTES={DEFAULT_UPLOAD_MAX_BYTES}\n")
     env_path.chmod(0o600)
 
 
@@ -364,7 +366,7 @@ def service_status(*, project_root: Path) -> str:
 
 
 def main(argv: Sequence[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description="Manage the readonly HTTP NocoBase CRM MCP user service.")
+    parser = argparse.ArgumentParser(description="Manage the HTTP NocoBase CRM MCP user service.")
     subparsers = parser.add_subparsers(dest="command", required=True)
 
     new_token_parser = subparsers.add_parser("new-token", help="Generate a new token and print remote MCP client configuration.")
diff --git a/tests/test_mcp_service.py b/tests/test_mcp_service.py
index 163e7f0..758acda 100644
--- a/tests/test_mcp_service.py
+++ b/tests/test_mcp_service.py
@@ -75,6 +75,17 @@ class McpServiceConfigTests(unittest.TestCase):
             self.assertEqual(stat.S_IMODE(os.stat(env_path).st_mode), 0o600)
             self.assertEqual(stat.S_IMODE(os.stat(env_path.parent).st_mode), 0o700)
 
+    def test_generated_env_has_size_but_no_upload_jwt(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            env_path = Path(tmp) / "mcp.env"
+
+            manage_mcp_service.write_token_env_file(env_path, "token")
+
+            content = env_path.read_text(encoding="utf-8")
+            self.assertIn("NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400\n", content)
+            self.assertNotIn("NOCOBASE_MCP_UPLOAD_JWT", content)
+            self.assertNotIn("NOCOBASE_MCP_UPLOAD_API_BASE_URL", content)
+
     def test_render_remote_config_prints_url_and_authorization_header(self) -> None:
         output = manage_mcp_service.render_remote_config(
             hosts=["192.168.1.10"],
diff --git a/tests/test_server.py b/tests/test_server.py
index 5908aa1..f52ec74 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -2,7 +2,16 @@ import asyncio
 import unittest
 from unittest.mock import patch
 
-from nocobase_crm_mcp.server import BearerTokenAuthMiddleware, FastMCP, create_http_app, create_server
+from nocobase_crm_mcp.server import (
+    DEFAULT_HTTP_HOST,
+    DEFAULT_HTTP_PATH,
+    DEFAULT_HTTP_PORT,
+    BearerTokenAuthMiddleware,
+    FastMCP,
+    create_http_app,
+    create_server,
+    run_http_server,
+)
 
 
 async def _ok_app(scope, receive, send):
@@ -59,6 +68,20 @@ class FakeCrmService:
         return {"results": []}
 
 
+class FakeProjectService:
+    def create_project(self, **kwargs):
+        return {"project_id": 1}
+
+    def update_project(self, project_id, **kwargs):
+        return {"project_id": project_id}
+
+
+class FakeUploadEndpoint:
+    async def __call__(self, scope, receive, send):
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"uploaded"})
+
+
 class ServerTests(unittest.TestCase):
     def test_bearer_token_middleware_rejects_missing_or_wrong_token(self) -> None:
         app = BearerTokenAuthMiddleware(_ok_app, "secret")
@@ -103,6 +126,45 @@ class ServerTests(unittest.TestCase):
             },
         )
 
+    def test_write_server_exposes_fixed_project_tools(self) -> None:
+        if FastMCP is None:
+            self.skipTest("MCP SDK is not installed")
+
+        server = create_server(
+            crm_service=FakeCrmService(),
+            project_service=FakeProjectService(),
+            readonly=False,
+        )
+
+        names = {tool.name for tool in asyncio.run(server.list_tools())}
+
+        self.assertTrue({"create_project", "update_project"}.issubset(names))
+        self.assertNotIn("destroy_project", names)
+
+    def test_upload_route_requires_the_same_bearer_token(self) -> None:
+        if FastMCP is None:
+            self.skipTest("MCP SDK is not installed")
+
+        server = create_server(crm_service=FakeCrmService(), readonly=True)
+        app = create_http_app(server, "secret", upload_endpoint=FakeUploadEndpoint())
+
+        self.assertEqual(run_asgi_request(app, path="/uploads")[0]["status"], 401)
+
+    def test_http_write_mode_requires_upload_api_base_url(self) -> None:
+        with patch.dict("os.environ", {"NOCOBASE_MCP_TOKEN": "secret"}, clear=True):
+            with patch("uvicorn.run") as uvicorn_run:
+                with self.assertRaisesRegex(RuntimeError, "NOCOBASE_MCP_UPLOAD_API_BASE_URL"):
+                    run_http_server(
+                        config_path=None,
+                        host=DEFAULT_HTTP_HOST,
+                        port=DEFAULT_HTTP_PORT,
+                        token_env="NOCOBASE_MCP_TOKEN",
+                        streamable_http_path=DEFAULT_HTTP_PATH,
+                        allowed_hosts=None,
+                        allowed_origins=None,
+                    )
+                uvicorn_run.assert_not_called()
+
     def test_config_template_tool_is_opt_in(self) -> None:
         if FastMCP is None:
             self.skipTest("MCP SDK is not installed")

From 546da0ad455322c189060d6953741068aa6bce51 Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 02:09:09 +0000
Subject: [PATCH 7/9] fix: preserve private upload settings on token rotation

---
 README.md                     | 20 ++++++++++++++------
 scripts/manage_mcp_service.py | 28 ++++++++++++++++++++++++++++
 tests/test_mcp_service.py     | 30 ++++++++++++++++++++++++++++++
 3 files changed, 72 insertions(+), 6 deletions(-)

diff --git a/README.md b/README.md
index c219455..3c11c10 100644
--- a/README.md
+++ b/README.md
@@ -80,18 +80,26 @@ HTTP write mode also requires these server-side NocoBase credentials. Do not put
 - `NOCOBASE_MCP_UPLOAD_API_BASE_URL` (for example `https://nocobase.example/api`)
 - `NOCOBASE_MCP_UPLOAD_JWT` (a NocoBase user JWT)
 
-`scripts/mcp_service.sh start` creates `.runtime/nocobase-crm-mcp.env` with mode `0600`, its parent directory with mode `0700`, and only the non-secret upload size setting. Add the API URL and JWT to that private env file before starting or restarting the managed service. HTTP write mode fails at startup with the missing variable name when either credential is absent.
+`scripts/mcp_service.sh start` creates `.runtime/nocobase-crm-mcp.env` with mode `0600`, its parent directory with mode `0700`, and only the non-secret upload size setting. HTTP write mode fails at startup with the missing variable name when either credential is absent.
 
-Operational flow:
+First-install flow:
 
 ```bash
-NOCOBASE_MCP_UPLOAD_API_BASE_URL='https://nocobase.example/api'
-NOCOBASE_MCP_UPLOAD_JWT='<user-jwt>'
+# Creates the private env file; the service is expected to fail until credentials are added.
+bash scripts/mcp_service.sh start
+
+# Add server-side NocoBase credentials to the systemd EnvironmentFile.
+cat >> .runtime/nocobase-crm-mcp.env <<'EOF'
+NOCOBASE_MCP_UPLOAD_API_BASE_URL=https://nocobase.example/api
+NOCOBASE_MCP_UPLOAD_JWT=<user-jwt>
+EOF
+
+# Start the service with the newly configured private env file.
 bash scripts/mcp_service.sh start
 curl -X POST -H "Authorization: Bearer <mcp-token>" -F "file=@quote.pdf" http://<host>:10087/uploads
 ```
 
-Set the first two values in `.runtime/nocobase-crm-mcp.env` (rather than in client configuration) before the service starts. An uploaded attachment that is not yet associated with a project is intentionally retained; this release does not auto-clean orphan attachments. It also provides no delete operation for projects or attachments.
+The NocoBase credentials belong in `.runtime/nocobase-crm-mcp.env`, not in shell-local variables or client configuration. Token rotation preserves these operator-managed env lines. An uploaded attachment that is not yet associated with a project is intentionally retained; this release does not auto-clean orphan attachments. It also provides no delete operation for projects or attachments.
 
 ## Permissions
 
@@ -141,7 +149,7 @@ For a managed user-level systemd service:
 bash scripts/mcp_service.sh start
 ```
 
-The first `start` creates `.runtime/nocobase-crm-mcp.env` with a random bearer token and the non-secret 25 MiB upload size setting automatically. Before the managed HTTP service can start, add `NOCOBASE_MCP_UPLOAD_API_BASE_URL` and `NOCOBASE_MCP_UPLOAD_JWT` to that private file. Use `new-token` only when rotating the token later.
+The first `start` creates `.runtime/nocobase-crm-mcp.env` with a random bearer token and the non-secret 25 MiB upload size setting automatically, then fails until the operator adds `NOCOBASE_MCP_UPLOAD_API_BASE_URL` and `NOCOBASE_MCP_UPLOAD_JWT` to that private file. Run `start` again after adding them. Use `new-token` only when rotating the token later; it preserves operator-managed env lines.
 
 Show the saved URL/header/client config:
 
diff --git a/scripts/manage_mcp_service.py b/scripts/manage_mcp_service.py
index 529e34a..3e55c84 100755
--- a/scripts/manage_mcp_service.py
+++ b/scripts/manage_mcp_service.py
@@ -27,6 +27,17 @@ DEFAULT_CONFIG_PATH = "config.json"
 DEFAULT_TOKEN_BYTES = 32
 DEFAULT_UV_CACHE_DIR = "/tmp/uv-cache"
 DEFAULT_UPLOAD_MAX_BYTES = 25 * 1024 * 1024
+MANAGED_ENV_NAMES = frozenset(
+    {
+        "NOCOBASE_MCP_TOKEN",
+        "NOCOBASE_MCP_HOST",
+        "NOCOBASE_MCP_PORT",
+        "NOCOBASE_MCP_PATH",
+        "NOCOBASE_MCP_ENV",
+        "NOCOBASE_MCP_CONFIG",
+        "NOCOBASE_MCP_UPLOAD_MAX_BYTES",
+    }
+)
 
 
 @dataclass(frozen=True)
@@ -116,6 +127,7 @@ def write_token_env_file(
     nb_env: str = DEFAULT_NB_ENV,
     config_path: str = DEFAULT_CONFIG_PATH,
 ) -> None:
+    preserved_lines = _operator_env_lines(env_path)
     env_path.parent.mkdir(parents=True, exist_ok=True)
     env_path.parent.chmod(0o700)
     fd = os.open(env_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
@@ -127,9 +139,25 @@ def write_token_env_file(
         env_file.write(f"NOCOBASE_MCP_ENV={nb_env}\n")
         env_file.write(f"NOCOBASE_MCP_CONFIG={config_path}\n")
         env_file.write(f"NOCOBASE_MCP_UPLOAD_MAX_BYTES={DEFAULT_UPLOAD_MAX_BYTES}\n")
+        if preserved_lines:
+            env_file.write("\n")
+            env_file.write("\n".join(preserved_lines))
+            env_file.write("\n")
     env_path.chmod(0o600)
 
 
+def _operator_env_lines(env_path: Path) -> list[str]:
+    if not env_path.exists():
+        return []
+    preserved_lines: list[str] = []
+    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
+        key, separator, _value = raw_line.strip().partition("=")
+        if separator and key.strip() in MANAGED_ENV_NAMES:
+            continue
+        preserved_lines.append(raw_line)
+    return preserved_lines
+
+
 def read_service_env_file(env_path: Path) -> dict[str, str]:
     if not env_path.exists():
         raise RuntimeError(
diff --git a/tests/test_mcp_service.py b/tests/test_mcp_service.py
index 758acda..f80340e 100644
--- a/tests/test_mcp_service.py
+++ b/tests/test_mcp_service.py
@@ -122,6 +122,36 @@ class McpServiceConfigTests(unittest.TestCase):
             self.assertIn("http://10.0.0.2:10087/mcp", output)
             run_systemctl.assert_not_called()
 
+    def test_new_token_preserves_operator_upload_settings_and_other_env_lines(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            project_root = Path(tmp) / "project"
+            project_root.mkdir()
+            config_home = Path(tmp) / "config"
+            env_path = project_root / ".runtime" / "nocobase-crm-mcp.env"
+            manage_mcp_service.write_token_env_file(env_path, "old-token")
+            with env_path.open("a", encoding="utf-8") as env_file:
+                env_file.write("# operator-managed upload credentials\n")
+                env_file.write("NOCOBASE_MCP_UPLOAD_API_BASE_URL=https://nb.example/api\n")
+                env_file.write("NOCOBASE_MCP_UPLOAD_JWT=operator-jwt\n")
+                env_file.write("EXTRA_OPERATOR_SETTING=preserve-me\n")
+
+            with (
+                patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
+                patch.object(manage_mcp_service, "generate_token", return_value="new-token"),
+                patch.object(manage_mcp_service, "detect_remote_hosts", return_value=[]),
+                patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
+            ):
+                manage_mcp_service.new_token(project_root=project_root)
+
+            content = env_path.read_text(encoding="utf-8")
+            self.assertIn("NOCOBASE_MCP_TOKEN=new-token\n", content)
+            self.assertIn("# operator-managed upload credentials\n", content)
+            self.assertIn("NOCOBASE_MCP_UPLOAD_API_BASE_URL=https://nb.example/api\n", content)
+            self.assertIn("NOCOBASE_MCP_UPLOAD_JWT=operator-jwt\n", content)
+            self.assertIn("EXTRA_OPERATOR_SETTING=preserve-me\n", content)
+            self.assertEqual(stat.S_IMODE(os.stat(env_path).st_mode), 0o600)
+            run_systemctl.assert_not_called()
+
     def test_start_service_uses_saved_token_and_restarts_service(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             project_root = Path(tmp) / "project"

From 87a26196dde86df089ad6fb6b82facf53fad0fdf Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 02:21:19 +0000
Subject: [PATCH 8/9] fix: harden project writes and uploads

---
 nocobase_crm_mcp/nocobase_http.py | 12 ++++++-
 nocobase_crm_mcp/projects.py      | 21 ++++++++++--
 nocobase_crm_mcp/uploads.py       | 53 +++++++++++++++++++++++++++---
 tests/test_nocobase_http.py       | 25 ++++++++++++++
 tests/test_projects.py            | 39 ++++++++++++++++++++++
 tests/test_server.py              | 27 ++++++++++++++--
 tests/test_uploads.py             | 54 +++++++++++++++++++++++++++++++
 7 files changed, 221 insertions(+), 10 deletions(-)

diff --git a/nocobase_crm_mcp/nocobase_http.py b/nocobase_crm_mcp/nocobase_http.py
index 3fc8849..b0eb21b 100644
--- a/nocobase_crm_mcp/nocobase_http.py
+++ b/nocobase_crm_mcp/nocobase_http.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from pathlib import Path
 from typing import Any, Sequence
+from urllib.parse import quote
 
 import httpx
 
@@ -57,8 +58,9 @@ class NocoBaseHttpClient:
         self._post_attachment_ids(project_id, attachment_ids, "remove")
 
     def _post_attachment_ids(self, project_id: Any, attachment_ids: Sequence[Any], action: str) -> None:
+        project_id_segment = _project_id_segment(project_id)
         response = self._client.post(
-            f"/t_projects/{project_id}/f_projects_attachments:{action}",
+            f"/t_projects/{project_id_segment}/f_projects_attachments:{action}",
             json={"tk": list(attachment_ids)},
         )
         self._raise_for_error(response)
@@ -81,3 +83,11 @@ class NocoBaseHttpClient:
 
     def _redact(self, value: str) -> str:
         return value.replace(self.jwt, "<redacted>")
+
+
+def _project_id_segment(project_id: Any) -> str:
+    if isinstance(project_id, bool) or not isinstance(project_id, (str, int)):
+        raise ValueError("project_id must be a scalar identifier")
+    if isinstance(project_id, str) and not project_id.strip():
+        raise ValueError("project_id must not be blank")
+    return quote(str(project_id), safe="")
diff --git a/nocobase_crm_mcp/projects.py b/nocobase_crm_mcp/projects.py
index 6d48f2a..e2ba1af 100644
--- a/nocobase_crm_mcp/projects.py
+++ b/nocobase_crm_mcp/projects.py
@@ -97,6 +97,7 @@ class ProjectService:
         remove_attachment_ids: Sequence[Any] | None = None,
     ) -> dict[str, Any]:
         try:
+            _validate_identifier(project_id, "project_id")
             values = self._project_values(
                 name=name,
                 customer_id=customer_id,
@@ -153,6 +154,7 @@ class ProjectService:
             raise _InvalidProjectInput("name is required")
 
         if customer_id is not None:
+            _validate_identifier(customer_id, "customer_id")
             values[self.config.customer_field] = customer_id
         if inquiry_date is not None:
             if not isinstance(inquiry_date, str):
@@ -165,7 +167,9 @@ class ProjectService:
             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:
+            if not isinstance(progress, str):
+                raise _InvalidProjectInput("progress must be a scalar value")
+            if progress not in self.config.progress_values:
                 raise _InvalidProgress
             values[self.config.progress_field] = self.config.progress_values[progress]
 
@@ -176,6 +180,8 @@ class ProjectService:
             (details, self.config.details_field),
         ):
             if value is not None:
+                if not isinstance(value, str):
+                    raise _InvalidProjectInput(f"{field} must be a scalar value")
                 values[field] = value
         return values
 
@@ -188,7 +194,11 @@ def _deduplicated_attachment_ids(attachment_ids: Sequence[Any] | None) -> list[A
 
     unique_ids: list[Any] = []
     for attachment_id in attachment_ids:
-        if attachment_id is None or isinstance(attachment_id, (Mapping, Sequence)):
+        try:
+            _validate_identifier(attachment_id, "attachment IDs")
+        except _InvalidProjectInput as error:
+            raise _InvalidProjectInput("attachment IDs must be scalar values") from error
+        if attachment_id is None:
             raise _InvalidProjectInput("attachment IDs must be scalar values")
         if attachment_id not in unique_ids:
             unique_ids.append(attachment_id)
@@ -206,6 +216,13 @@ def _project_id(response: Mapping[str, Any], id_field: str) -> Any:
     return data[id_field]
 
 
+def _validate_identifier(value: Any, name: str) -> None:
+    if isinstance(value, bool) or not isinstance(value, (str, int)):
+        raise _InvalidProjectInput(f"{name} must be a scalar identifier")
+    if isinstance(value, str) and not value.strip():
+        raise _InvalidProjectInput(f"{name} must not be blank")
+
+
 def _attachment_error(project_id: Any, error: Exception) -> dict[str, Any]:
     if isinstance(error, NocoBaseHttpError):
         if error.status_code == 404:
diff --git a/nocobase_crm_mcp/uploads.py b/nocobase_crm_mcp/uploads.py
index f6a4ee6..0a2d202 100644
--- a/nocobase_crm_mcp/uploads.py
+++ b/nocobase_crm_mcp/uploads.py
@@ -14,6 +14,11 @@ from .nocobase_http import NocoBaseHttpError
 
 
 _COPY_CHUNK_BYTES = 1024 * 1024
+_MULTIPART_OVERHEAD_BYTES = 64 * 1024
+
+
+class _UploadTooLarge(ValueError):
+    pass
 
 
 class UploadEndpoint:
@@ -38,17 +43,20 @@ class UploadEndpoint:
             return _error_response("invalid_upload", "Uploads require multipart form data", 400)
 
         try:
+            request = await _bounded_request(request, self._max_bytes + _MULTIPART_OVERHEAD_BYTES)
             form = await request.form()
+        except _UploadTooLarge:
+            return _error_response("upload_too_large", "Upload exceeds the configured size limit", 413)
         except Exception:
             return _error_response("invalid_upload", "Could not read upload form data", 400)
 
-        uploads = [(name, value) for name, value in form.multi_items() if isinstance(value, UploadFile)]
-        file_values = form.getlist("file")
-        if len(uploads) != 1 or len(file_values) != 1 or uploads[0][0] != "file":
+        items = list(form.multi_items())
+        uploads = [(name, value) for name, value in items if isinstance(value, UploadFile)]
+        if len(items) != 1 or len(uploads) != 1 or uploads[0][0] != "file":
             await _close_uploads(value for _, value in uploads)
             return _error_response("invalid_upload", "Provide exactly one file upload", 400)
 
-        upload = file_values[0]
+        upload = uploads[0][1]
         if not isinstance(upload, UploadFile):
             return _error_response("invalid_upload", "Provide exactly one file upload", 400)
 
@@ -94,6 +102,43 @@ def _is_multipart(request: Request) -> bool:
     return content_type.split(";", 1)[0].strip().lower() == "multipart/form-data"
 
 
+async def _bounded_request(request: Request, max_body_bytes: int) -> Request:
+    content_length = request.headers.get("content-length")
+    if content_length is not None:
+        try:
+            if int(content_length) > max_body_bytes:
+                raise _UploadTooLarge
+        except ValueError:
+            pass
+
+    received_bytes = 0
+    chunks: list[bytes] = []
+    while True:
+        message = await request.receive()
+        if message.get("type") == "http.request":
+            body = message.get("body", b"")
+            received_bytes += len(body)
+            if received_bytes > max_body_bytes:
+                raise _UploadTooLarge
+            chunks.append(body)
+            if not message.get("more_body", False):
+                break
+        elif message.get("type") == "http.disconnect":
+            break
+
+    body = b"".join(chunks)
+    replayed = False
+
+    async def receive() -> dict[str, Any]:
+        nonlocal replayed
+        if replayed:
+            return {"type": "http.disconnect"}
+        replayed = True
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    return Request(request.scope, receive=receive)
+
+
 def _attachment_id(attachment: Any) -> int | None:
     if not isinstance(attachment, Mapping):
         return None
diff --git a/tests/test_nocobase_http.py b/tests/test_nocobase_http.py
index 080ec06..c033c0b 100644
--- a/tests/test_nocobase_http.py
+++ b/tests/test_nocobase_http.py
@@ -67,6 +67,31 @@ class NocoBaseHttpClientTests(unittest.TestCase):
         self.assertEqual(str(requests[1].url), "https://nb.example/api/t_projects/7/f_projects_attachments:remove")
         self.assertEqual(json.loads(requests[1].content), {"tk": [10]})
 
+    def test_attachment_association_encodes_project_id_as_one_path_segment(self) -> None:
+        requests: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            requests.append(request)
+            return httpx.Response(204)
+
+        client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
+        client.add_attachment_ids("../project/7?#", [8])
+
+        self.assertEqual(
+            str(requests[0].url),
+            "https://nb.example/api/t_projects/..%2Fproject%2F7%3F%23/f_projects_attachments:add",
+        )
+
+    def test_attachment_association_rejects_non_scalar_project_id_before_request(self) -> None:
+        client = NocoBaseHttpClient(
+            "https://nb.example/api",
+            "jwt",
+            transport=httpx.MockTransport(lambda request: self.fail("request must not be sent")),
+        )
+
+        with self.assertRaisesRegex(ValueError, "project_id"):
+            client.add_attachment_ids({"id": 7}, [8])
+
     def test_rejects_blank_upload_credentials(self) -> None:
         with self.assertRaises(ValueError):
             NocoBaseHttpClient(" ", "jwt")
diff --git a/tests/test_projects.py b/tests/test_projects.py
index 3e45ca7..9095b3b 100644
--- a/tests/test_projects.py
+++ b/tests/test_projects.py
@@ -113,6 +113,45 @@ class ProjectServiceTests(unittest.TestCase):
         self.assertEqual(result["error"]["code"], "invalid_project_input")
         self.assertEqual(self.nb.update_calls, [])
 
+    def test_create_rejects_nested_scalar_inputs_before_writing(self) -> None:
+        cases = {
+            "name": {"name": "x"},
+            "customer_id": {"id": 3},
+            "inquiry_date": ["2026-07-12"],
+            "progress": ("SOP",),
+            "applications": ["EV"],
+            "part_numbers": ("123",),
+            "period_usage": {"annual"},
+            "details": {"note": "x"},
+        }
+
+        for field, value in cases.items():
+            with self.subTest(field=field):
+                self.setUp()
+                arguments = {"name": "x", **{field: value}}
+                result = self.service.create_project(**arguments)
+
+                self.assertEqual(result["error"]["code"], "invalid_project_input")
+                self.assertEqual(self.nb.create_calls, [])
+                self.assertEqual(self.http.add_calls, [])
+
+    def test_update_rejects_nested_project_id_before_any_client_call(self) -> None:
+        for project_id in ({"id": 41}, [41], {41}, (41,)):
+            with self.subTest(project_id=project_id):
+                self.setUp()
+                result = self.service.update_project(project_id, name="x", add_attachment_ids=[8])
+
+                self.assertEqual(result["error"]["code"], "invalid_project_input")
+                self.assertEqual(self.nb.update_calls, [])
+                self.assertEqual(self.http.add_calls, [])
+
+    def test_create_rejects_nested_attachment_id_before_any_client_call(self) -> None:
+        result = self.service.create_project(name="x", attachment_ids=[{"id": 8}])
+
+        self.assertEqual(result["error"]["code"], "invalid_project_input")
+        self.assertEqual(self.nb.create_calls, [])
+        self.assertEqual(self.http.add_calls, [])
+
     def test_update_writes_scalars_then_deduplicated_attachment_changes_in_order(self) -> None:
         result = self.service.update_project(
             41,
diff --git a/tests/test_server.py b/tests/test_server.py
index f52ec74..aa9cdaf 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -2,6 +2,9 @@ import asyncio
 import unittest
 from unittest.mock import patch
 
+import httpx
+from starlette.responses import JSONResponse
+
 from nocobase_crm_mcp.server import (
     DEFAULT_HTTP_HOST,
     DEFAULT_HTTP_PATH,
@@ -77,9 +80,8 @@ class FakeProjectService:
 
 
 class FakeUploadEndpoint:
-    async def __call__(self, scope, receive, send):
-        await send({"type": "http.response.start", "status": 200, "headers": []})
-        await send({"type": "http.response.body", "body": b"uploaded"})
+    async def __call__(self, request):
+        return JSONResponse({"uploaded": True})
 
 
 class ServerTests(unittest.TestCase):
@@ -150,6 +152,25 @@ class ServerTests(unittest.TestCase):
 
         self.assertEqual(run_asgi_request(app, path="/uploads")[0]["status"], 401)
 
+    def test_upload_route_accepts_the_matching_bearer_token(self) -> None:
+        if FastMCP is None:
+            self.skipTest("MCP SDK is not installed")
+
+        server = create_server(crm_service=FakeCrmService(), readonly=True)
+
+        async def upload_endpoint(request):
+            return JSONResponse({"uploaded": True})
+
+        app = create_http_app(server, "secret", upload_endpoint=upload_endpoint)
+
+        async def request_upload() -> httpx.Response:
+            async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as client:
+                return await client.post("/uploads", headers={"Authorization": "Bearer secret"})
+
+        response = asyncio.run(request_upload())
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json(), {"uploaded": True})
+
     def test_http_write_mode_requires_upload_api_base_url(self) -> None:
         with patch.dict("os.environ", {"NOCOBASE_MCP_TOKEN": "secret"}, clear=True):
             with patch("uvicorn.run") as uvicorn_run:
diff --git a/tests/test_uploads.py b/tests/test_uploads.py
index 5f196e6..1e7a117 100644
--- a/tests/test_uploads.py
+++ b/tests/test_uploads.py
@@ -3,6 +3,7 @@ import unittest
 import warnings
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from unittest.mock import patch
 
 import httpx
 from starlette.applications import Starlette
@@ -103,6 +104,59 @@ class UploadEndpointTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(temp_paths, [])
         self.assertFalse(any(issubclass(warning.category, ResourceWarning) for warning in caught))
 
+    async def test_upload_rejects_non_file_form_fields(self) -> None:
+        http_client = FakeHttpClient()
+        with TemporaryDirectory() as temp_dir:
+            async with await self._client_for(http_client, temp_dir) as client:
+                response = await client.post(
+                    "/uploads",
+                    files={"file": ("quote.pdf", b"pdf", "application/pdf")},
+                    data={"description": "quote"},
+                )
+
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(response.json()["error"]["code"], "invalid_upload")
+            self.assertEqual(http_client.calls, [])
+            self.assertEqual(list(Path(temp_dir).iterdir()), [])
+
+    async def test_upload_rejects_oversized_misreported_chunked_body_before_multipart_parsing(self) -> None:
+        http_client = FakeHttpClient()
+        endpoint = UploadEndpoint(http_client, max_bytes=5, temp_dir="/tmp")
+        parsed = False
+
+        async def forbidden_form(self):
+            nonlocal parsed
+            parsed = True
+            self.fail("multipart parsing must not begin for an oversized body")
+
+        body = b"x" * (5 + 65537)
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/uploads",
+            "headers": [
+                (b"content-type", b"multipart/form-data; boundary=test"),
+                (b"content-length", b"1"),
+            ],
+        }
+        messages = [
+            {"type": "http.request", "body": body[:10], "more_body": True},
+            {"type": "http.request", "body": body[10:], "more_body": False},
+        ]
+
+        async def receive():
+            return messages.pop(0)
+
+        from starlette.requests import Request
+
+        starlette_request = Request(scope, receive=receive)
+        with patch("nocobase_crm_mcp.uploads.Request.form", new=forbidden_form):
+            response = await endpoint(starlette_request)
+
+        self.assertEqual(response.status_code, 413)
+        self.assertFalse(parsed)
+        self.assertEqual(http_client.calls, [])
+
     async def test_upload_rejects_disallowed_mime_type_before_calling_nocobase(self) -> None:
         http_client = FakeHttpClient()
         with TemporaryDirectory() as temp_dir:

From b51be25e2e5ad3b886a47e1159f112fdb8de510d Mon Sep 17 00:00:00 2001
From: nac <nac@nac.local>
Date: Mon, 13 Jul 2026 02:24:58 +0000
Subject: [PATCH 9/9] fix: reject dot project ids

---
 nocobase_crm_mcp/nocobase_http.py | 13 ++++++++++---
 nocobase_crm_mcp/projects.py      | 11 +++++++++--
 tests/test_nocobase_http.py       | 11 +++++++++++
 tests/test_projects.py            | 10 ++++++++++
 4 files changed, 40 insertions(+), 5 deletions(-)

diff --git a/nocobase_crm_mcp/nocobase_http.py b/nocobase_crm_mcp/nocobase_http.py
index b0eb21b..020a2fc 100644
--- a/nocobase_crm_mcp/nocobase_http.py
+++ b/nocobase_crm_mcp/nocobase_http.py
@@ -86,8 +86,15 @@ class NocoBaseHttpClient:
 
 
 def _project_id_segment(project_id: Any) -> str:
+    return quote(validate_project_id(project_id), safe="")
+
+
+def validate_project_id(project_id: Any) -> str:
     if isinstance(project_id, bool) or not isinstance(project_id, (str, int)):
         raise ValueError("project_id must be a scalar identifier")
-    if isinstance(project_id, str) and not project_id.strip():
-        raise ValueError("project_id must not be blank")
-    return quote(str(project_id), safe="")
+    if isinstance(project_id, str):
+        if not project_id.strip():
+            raise ValueError("project_id must not be blank")
+        if project_id in {".", ".."}:
+            raise ValueError("project_id must not be a dot path segment")
+    return str(project_id)
diff --git a/nocobase_crm_mcp/projects.py b/nocobase_crm_mcp/projects.py
index e2ba1af..f71c280 100644
--- a/nocobase_crm_mcp/projects.py
+++ b/nocobase_crm_mcp/projects.py
@@ -5,7 +5,7 @@ from datetime import date
 from typing import Any
 
 from .config import ProjectConfig
-from .nocobase_http import NocoBaseHttpError
+from .nocobase_http import NocoBaseHttpError, validate_project_id
 
 
 def _configured_scalar_fields(config: ProjectConfig) -> tuple[str, ...]:
@@ -97,7 +97,7 @@ class ProjectService:
         remove_attachment_ids: Sequence[Any] | None = None,
     ) -> dict[str, Any]:
         try:
-            _validate_identifier(project_id, "project_id")
+            _validate_project_id(project_id)
             values = self._project_values(
                 name=name,
                 customer_id=customer_id,
@@ -223,6 +223,13 @@ def _validate_identifier(value: Any, name: str) -> None:
         raise _InvalidProjectInput(f"{name} must not be blank")
 
 
+def _validate_project_id(project_id: Any) -> None:
+    try:
+        validate_project_id(project_id)
+    except ValueError as error:
+        raise _InvalidProjectInput(str(error)) from error
+
+
 def _attachment_error(project_id: Any, error: Exception) -> dict[str, Any]:
     if isinstance(error, NocoBaseHttpError):
         if error.status_code == 404:
diff --git a/tests/test_nocobase_http.py b/tests/test_nocobase_http.py
index c033c0b..e7c6f59 100644
--- a/tests/test_nocobase_http.py
+++ b/tests/test_nocobase_http.py
@@ -92,6 +92,17 @@ class NocoBaseHttpClientTests(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "project_id"):
             client.add_attachment_ids({"id": 7}, [8])
 
+    def test_attachment_association_rejects_dot_path_project_ids_before_request(self) -> None:
+        client = NocoBaseHttpClient(
+            "https://nb.example/api",
+            "jwt",
+            transport=httpx.MockTransport(lambda request: self.fail("request must not be sent")),
+        )
+
+        for project_id in (".", ".."):
+            with self.subTest(project_id=project_id), self.assertRaisesRegex(ValueError, "project_id"):
+                client.add_attachment_ids(project_id, [8])
+
     def test_rejects_blank_upload_credentials(self) -> None:
         with self.assertRaises(ValueError):
             NocoBaseHttpClient(" ", "jwt")
diff --git a/tests/test_projects.py b/tests/test_projects.py
index 9095b3b..a99f075 100644
--- a/tests/test_projects.py
+++ b/tests/test_projects.py
@@ -145,6 +145,16 @@ class ProjectServiceTests(unittest.TestCase):
                 self.assertEqual(self.nb.update_calls, [])
                 self.assertEqual(self.http.add_calls, [])
 
+    def test_update_rejects_dot_path_project_id_before_any_client_call(self) -> None:
+        for project_id in (".", ".."):
+            with self.subTest(project_id=project_id):
+                self.setUp()
+                result = self.service.update_project(project_id, name="x", add_attachment_ids=[8])
+
+                self.assertEqual(result["error"]["code"], "invalid_project_input")
+                self.assertEqual(self.nb.update_calls, [])
+                self.assertEqual(self.http.add_calls, [])
+
     def test_create_rejects_nested_attachment_id_before_any_client_call(self) -> None:
         result = self.service.create_project(name="x", attachment_ids=[{"id": 8}])