diff --git a/AGENTS.md b/AGENTS.md
index 49e0ad2..c466d1b 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -23,6 +23,7 @@ connector plugins and Jinja2 generation. Use `uv`; the CLI entry point is `datam
 
 ## Relevant entry points
 
+- Name new branches with the `feature/` prefix.
 - `src/datam8/cmd/`, `app.py`: CLI commands and registration.
 - `src/datam8/api/`: app/server construction, routes and response envelopes.
 - `src/datam8/source.py`, `plugins/base.py`, `plugins/manager.py`: source metadata,
diff --git a/docs/backend-contract.md b/docs/backend-contract.md
index 4ebfb24..394b16c 100644
--- a/docs/backend-contract.md
+++ b/docs/backend-contract.md
@@ -41,14 +41,14 @@ Conversion failures identify the affected path and stop loading.
 The current `create_app().openapi()` exposes these domains (no `/api` prefix):
 
 - `GET /health` (204), `GET /version` (`schemaVersion`, `appVersion`), `GET /config`.
-- `GET /solution`, `GET /solution/full`.
+- `GET|PATCH /solution`, `GET /solution/full`.
 - `GET /entities`, `GET|PUT|PATCH|DELETE /entities/{locator}`, `PUT /entities/clone`,
   `POST /entities/rename`, `POST /entities/move`, `POST /entities/move-single`.
 - `POST /model/save`, `POST /model/reload`, `GET /model/unsaved`.
 - `POST /model/generate`: optional body `{ "target": "...", "cleanOutput": true,
   "payloads": [] }`; synchronous response `{ "target": "...", "outputPath": "...",
   "message": null }`. Uses the currently loaded solution.
-- `/functions/*` for function retrieval, updates and moves.
+- `/functions/*` for function retrieval, creation, updates, deletion and moves.
 - `GET /plugins`, `POST /plugins/reload`, `GET /plugins/{plugin_id}` and its
   `/ui-schema`, `/data-type-mappings`, `/connection-properties` subpaths.
 - `POST /secrets/check`, `PUT /secrets/set`.
@@ -117,6 +117,16 @@ editor. Runtime configuration changes and process failure may require a restart.
   the model move fails. Saving a deleted folder also removes child functions.
 - Function source paths are relative to their model entity. Absolute/drive-qualified
   paths, traversal, empty segments and symlink escapes are rejected.
+- `POST /functions/{modelEntityId}` accepts `{ "stepNo": 1, "name": "Clean",
+  "source": "clean.py", "sourceCode": "..." }`; it creates the transformation and
+  source file and persists the model. Sources must be relative `.py` paths inside the
+  entity directory, and an existing file is never overwritten.
+- `DELETE /functions/{modelEntityId}/{stepNo|name}` removes the selected transformation,
+  persists the model and deletes its local source file only when no remaining function
+  references it.
+- `PATCH /solution` accepts a partial solution object and writes the validated settings
+  to the `.dm8s` file. `schemaVersion` is managed by migrations. The API reloads the
+  active solution after a successful update and returns 409 when model changes are pending.
 - A bare function filename first resolves beside the entity JSON, then in its
   same-named subdirectory for older solutions. Renaming an entity keeps function
   files in place and persists an explicit relative path when that fallback was used.
diff --git a/docs/cli-agent-contract.md b/docs/cli-agent-contract.md
index 08f2be4..45f1e54 100644
--- a/docs/cli-agent-contract.md
+++ b/docs/cli-agent-contract.md
@@ -136,6 +136,8 @@ All commands below support `--solution` and `--json`:
   Update uses the same code input and a step/name selector. Move takes a relative
   destination. Delete removes the transformation and only its unshared local file.
   Generic metadata editing does not imply function file creation/deletion.
+- `solution update --data JSON` (or `--file FILE`, with `-` for stdin) applies a partial
+  solution settings update and writes the `.dm8s` file.
 - `plugins list`, `plugins show ID`, `plugins ui-schema ID`,
   `plugins data-type-mappings ID` and `plugins connection-properties ID` inspect
   connector configuration without creating a connection.
diff --git a/src/datam8/api/routes/functions.py b/src/datam8/api/routes/functions.py
index ade7990..9a2d55f 100644
--- a/src/datam8/api/routes/functions.py
+++ b/src/datam8/api/routes/functions.py
@@ -27,6 +27,7 @@
 
 from datam8 import factory, functions
 from datam8.model import EntityWrapper
+from datam8.operations import ModelSession, Operation
 from datam8_model.model import ModelEntity
 
 from .responses import (
@@ -103,6 +104,50 @@ class UpdateFunctionBody(BaseModel):
     source_code: Annotated[str, Field(alias="sourceCode")]
 
 
+class CreateFunctionBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_no: Annotated[int, Field(alias="stepNo", ge=1)]
+    name: str
+    source: str
+    source_code: Annotated[str, Field(alias="sourceCode")]
+
+
+@functions_router.post("/{modelEntityId:int}")
+async def create_function(modelEntityId: int, body: CreateFunctionBody) -> Response:
+    wrapper = _get_wrapper_by_id(modelEntityId)
+    operation = Operation(
+        op="functions.create",
+        locator=str(wrapper.locator),
+        data={
+            "stepNo": body.step_no,
+            "name": body.name,
+            "kind": "function",
+            "function": {"source": body.source},
+        },
+        source_code=body.source_code,
+    )
+    ModelSession(model=factory.get_model()).execute(operation)
+    return Response204NoContent()
+
+
+@functions_router.delete("/{modelEntityId:int}/{stepNo:int}")
+async def delete_function_with_step_no(modelEntityId: int, stepNo: int) -> Response:
+    wrapper = _get_wrapper_by_id(modelEntityId)
+    ModelSession(model=factory.get_model()).execute(
+        Operation(op="functions.delete", locator=str(wrapper.locator), step_no=stepNo)
+    )
+    return Response204NoContent()
+
+
+@functions_router.delete("/{modelEntityId:int}/{name:str}")
+async def delete_function_with_name(modelEntityId: int, name: str) -> Response:
+    wrapper = _get_wrapper_by_id(modelEntityId)
+    ModelSession(model=factory.get_model()).execute(
+        Operation(op="functions.delete", locator=str(wrapper.locator), name=name)
+    )
+    return Response204NoContent()
+
+
 def _update_function(id: int, step_no_or_name: str | int, source_code: str) -> None:
     wrapper = _get_wrapper_by_id(id)
     functions.update_function(wrapper, step_no_or_name, source_code)
diff --git a/src/datam8/api/routes/solution.py b/src/datam8/api/routes/solution.py
index c710617..21901bc 100644
--- a/src/datam8/api/routes/solution.py
+++ b/src/datam8/api/routes/solution.py
@@ -16,10 +16,13 @@
 # You should have received a copy of the GNU General Public License
 # along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-from fastapi import APIRouter
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
-from datam8 import factory, model
+from datam8 import config, factory, model
+from datam8.operations import update_solution_file
 from datam8_model import base as b
 from datam8_model import solution as s
 
@@ -31,6 +34,19 @@ async def get_solution() -> s.Solution:
     return factory.get_model().solution
 
 
+@solution_router.patch("")
+async def update_solution(body: dict[str, Any]) -> s.Solution:
+    model_ = factory.get_model()
+    changed, deleted = model_.get_unsaved_entities()
+    if changed or deleted:
+        raise HTTPException(status_code=409, detail="Save or discard pending model changes first")
+
+    update_solution_file(config.solution_path, body)
+    factory._model = await factory.load_model(config.solution_path)
+    factory._plugin_manager = None
+    return factory.get_model().solution
+
+
 class DumpSolutionResponse(BaseModel):
     solution: s.Solution
     base_entities: list[model.EntityWrapperVariant]
diff --git a/src/datam8/cmd/solution.py b/src/datam8/cmd/solution.py
index 123b43d..64cf30e 100644
--- a/src/datam8/cmd/solution.py
+++ b/src/datam8/cmd/solution.py
@@ -21,9 +21,10 @@
 import typer
 
 from datam8 import config, opts, parser
+from datam8.operations import update_solution_file
 
 from . import common
-from .entities import Fields
+from .entities import Data, Fields, File
 
 app = typer.Typer(name="solution", help="Read the solution descriptor", no_args_is_help=True)
 
@@ -49,3 +50,23 @@ def show(
 
         rich.print(data)
     return {"item": data}
+
+
+@app.command()
+@common.output
+def update(
+    solution_path: opts.SolutionPath,
+    data: Data = None,
+    file: File = None,
+    json_output: opts.JsonOutput = False,
+    log_level: opts.LogLevel = opts.LogLevels.WARNING,
+    version: opts.Version = False,
+):
+    """Update solution settings from a partial JSON object."""
+    common.main_callback(solution_path, log_level, version)
+    result = {"item": update_solution_file(config.solution_path, common.input_value(data, file))}
+    if not json_output:
+        import rich
+
+        rich.print(result["item"])
+    return result
diff --git a/src/datam8/operations.py b/src/datam8/operations.py
index acb51e3..0d5331a 100644
--- a/src/datam8/operations.py
+++ b/src/datam8/operations.py
@@ -25,14 +25,15 @@
 from pathlib import Path, PureWindowsPath
 from typing import Any, Literal, cast
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator
 
 from datam8 import config, errors, factory, functions, source
 from datam8.model import EntityWrapper
-from datam8.model.model import class_from_type
+from datam8.model.model import Model, class_from_type
 from datam8.plugins.navigation import normalize_source_item
 from datam8_model import base as b
 from datam8_model import model as m
+from datam8_model import solution as s
 
 
 class Operation(BaseModel):
@@ -139,6 +140,30 @@ class Batch(BaseModel):
     operations: list[Operation]
 
 
+def update_solution_file(solution_path: Path, patch: Any) -> s.Solution:
+    """Validate and persist a partial solution descriptor update."""
+    if not isinstance(patch, dict) or not patch:
+        raise errors.Datam8ValidationError(message="Solution update requires a non-empty object")
+    if "schemaVersion" in patch or "schema_version" in patch:
+        raise errors.Datam8ValidationError(message="schemaVersion is managed by migrations")
+
+    from datam8 import parser
+    from datam8.model.model import MODEL_DUMP_OPTIONS
+
+    current = parser.parse_solution_file(solution_path)
+    data = current.model_dump(mode="json", by_alias=True, exclude_unset=True)
+    data.update(patch)
+    try:
+        updated = s.Solution.model_validate(data)
+    except ValidationError as err:
+        raise errors.Datam8ValidationError(
+            message="Invalid solution update",
+            details=err.errors(include_input=False, include_url=False),
+        ) from err
+    updated.to_json_file(solution_path, "w", MODEL_DUMP_OPTIONS)
+    return updated
+
+
 def select_key(value: Any, segment: str | int | dict[str, Any]) -> str | int:
     if isinstance(segment, dict):
         if not isinstance(value, list):
@@ -218,11 +243,16 @@ def references(node):
 
 
 class ModelSession:
-    def __init__(self, solution: Path):
-        config.set_solution(solution)
-        config.lazy = True
-        self.model = factory.create_model()
-        factory._plugin_manager = None
+    def __init__(self, solution: Path | None = None, *, model: Model | None = None):
+        if model is None:
+            if solution is None:
+                raise ValueError("A solution path or loaded model is required")
+            config.set_solution(solution)
+            config.lazy = True
+            self.model = factory.create_model()
+            factory._plugin_manager = None
+        else:
+            self.model = model
 
     def show(self, selector: str, by="locator", at=None, fields: str | None = None):
         from datam8 import opts
@@ -470,9 +500,10 @@ def function_source(self, wrapper, transformation):
             PureWindowsPath(name).drive
             or Path(name).is_absolute()
             or any(p in ("", ".", "..") for p in parts)
+            or Path(name.replace("\\", "/")).suffix.lower() != ".py"
         ):
             raise errors.Datam8ValidationError(
-                message="Function source must be relative to its entity directory"
+                message="Function source must be a relative .py path inside its entity directory"
             )
         path = functions._build_source_file_path(wrapper, transformation)
         if not path.is_relative_to(wrapper.source_file.parent.resolve()):
diff --git a/tests/test_016_function_files.py b/tests/test_016_function_files.py
index 2b0211f..bc9ff7a 100644
--- a/tests/test_016_function_files.py
+++ b/tests/test_016_function_files.py
@@ -241,3 +241,42 @@ def test_move_does_not_delete_unreferenced_legacy_fallback(function_solution):
     assert client.post("/model/save").status_code == 204
     assert legacy.read_text() == "# unreferenced fallback"
     assert (root / "Model/Core/Archive/clean.py").exists()
+
+
+def test_api_function_create_and_delete_use_cli_operations(function_solution):
+    client, root, _ = function_solution("clean.py", missing=True)
+    response = client.post(
+        "/functions/1",
+        json={
+            "stepNo": 4,
+            "name": "Created",
+            "source": "created.py",
+            "sourceCode": "def transform(df):\n    return df\n",
+        },
+    )
+    assert response.status_code == 204, response.text
+    source = root / "Model/Core/created.py"
+    assert source.read_text(encoding="utf-8").startswith("def transform")
+    assert client.get("/functions/1/4").json()["item"]["source_code"] == source.read_text()
+    saved = json.loads((root / "Model/Core/Customer.json").read_text())
+    assert any(item["name"] == "Created" for item in saved["transformations"])
+
+    response = client.delete("/functions/1/Created")
+    assert response.status_code == 204, response.text
+    assert not source.exists()
+    saved = json.loads((root / "Model/Core/Customer.json").read_text())
+    assert not any(item["name"] == "Created" for item in saved["transformations"])
+
+
+def test_api_function_create_rejects_collision_and_non_python_source(function_solution):
+    client, root, _ = function_solution("clean.py", missing=True)
+    source = root / "Model/Core/created.py"
+    source.write_text("# preserve", encoding="utf-8")
+    body = {"stepNo": 4, "name": "Created", "source": "created.py", "sourceCode": "new"}
+    response = client.post("/functions/1", json=body)
+    assert response.status_code == 409
+    assert source.read_text(encoding="utf-8") == "# preserve"
+
+    response = client.post("/functions/1", json={**body, "source": "created.txt"})
+    assert response.status_code == 400
+    assert source.read_text(encoding="utf-8") == "# preserve"
diff --git a/tests/test_091_agent_cli.py b/tests/test_091_agent_cli.py
index af2373d..3fc5f23 100644
--- a/tests/test_091_agent_cli.py
+++ b/tests/test_091_agent_cli.py
@@ -258,6 +258,49 @@ def test_noninteractive_init_root_options_and_missing_solution(tmp_path, monkeyp
     run(path, "init", "CLI", code=2)
 
 
+def test_solution_update_cli_and_api_share_validation(workspace, monkeypatch):
+    patch = {
+        "generatorTargets": [
+            {"name": "secondary", "sourcePath": "Generate2", "outputPath": "Output2"}
+        ]
+    }
+    cli_result = run(workspace, "solution", "update", "--data", json.dumps(patch))
+    assert cli_result["item"]["generatorTargets"][0]["name"] == "secondary"
+    original = workspace.read_text(encoding="utf-8")
+    invalid = run(
+        workspace,
+        "solution",
+        "update",
+        "--data",
+        '{"schemaVersion":"invalid"}',
+        code=2,
+    )
+    assert invalid["error"]["code"] == "validation_error"
+    assert workspace.read_text(encoding="utf-8") == original
+
+    config.set_solution(workspace)
+    config.lazy = True
+    monkeypatch.setattr(factory, "_model", None)
+    factory.create_model()
+    client = TestClient(create_app())
+    response = client.patch("/solution", json={"schemaVersion": "invalid"})
+    assert response.status_code == 400
+    assert workspace.read_text(encoding="utf-8") == original
+
+    response = client.patch(
+        "/solution",
+        json={
+            "generatorTargets": [
+                {"name": "api", "sourcePath": "GenerateApi", "outputPath": "OutputApi"}
+            ]
+        },
+    )
+    assert response.status_code == 200, response.text
+    assert response.json()["generatorTargets"][0]["name"] == "api"
+    assert factory.get_model().solution.generatorTargets[0].name == "api"
+    assert json.loads(workspace.read_text(encoding="utf-8"))["generatorTargets"][0]["name"] == "api"
+
+
 def test_secret_stdin_batch_and_plugin_json(workspace, monkeypatch):
     from datam8 import secrets
     from datam8.cmd import secret as secret_commands