Release

README.md

@@ -57,14 +57,21 @@ See [mcp-ynab.com](https://mcp-ynab.com) for config file locations and troublesh
 | **Plans** | `list_plans`, `get_plan`, `get_plan_settings` |
 | **Accounts** | `list_accounts`, `get_account`, `create_account` |
 | **Categories** | `list_categories`, `get_category`, `create_category`, `update_category`, `create_category_group`, `update_category_group`, `get_category_for_month`, `update_category_for_month` |
-| **Payees** | `list_payees`, `get_payee`, `update_payee` |
+| **Payees** | `list_payees`, `get_payee`, `create_payee`, `update_payee` |
 | **Payee Locations** | `list_payee_locations`, `get_payee_location`, `get_payee_locations_by_payee` |
 | **Months** | `list_months`, `get_month` |
 | **Money Movements** | `list_money_movements`, `get_money_movements_for_month`, `list_money_movement_groups`, `get_money_movement_groups_for_month` |
 | **Transactions** | `list_transactions`, `get_transaction`, `get_transactions_by_account`, `get_transactions_by_category`, `get_transactions_by_month`, `get_transactions_by_payee`, `search_transactions`, `create_transaction`, `create_transactions`, `update_transaction`, `update_transactions`, `delete_transaction`, `import_transactions` |
 | **Scheduled** | `list_scheduled_transactions`, `get_scheduled_transaction`, `create_scheduled_transaction`, `update_scheduled_transaction`, `delete_scheduled_transaction` |
 | **Analytics** | `get_money_flow`, `get_spending_by_category` |
 
+### Read-only mode
+
+Set `YNAB_READ_ONLY=true` in the server's `env` to skip registering every
+mutating tool (create/update/delete/import). Clients only ever see the read
+and analytics tools, so an assistant can analyze your budget without being
+able to change it.
+
 ### Field selection
 
 Every tool that returns a model accepts an optional `exclude_fields` list. By

src/cache/service.py

@@ -554,6 +554,11 @@ async def get_payee(self, payee_id: str, plan_id: str) -> Payee:
         await self._set_cached_model(cache_key, payee, self.settings.ttl_single_entity)
         return payee
 
+    async def create_payee(self, payee: dict, plan_id: str) -> Payee:
+        p = await self.client.create_payee(payee, plan_id)
+        await self.delta.invalidate_knowledge(plan_id, ENDPOINT_PAYEES)
+        return p
+
     async def update_payee(self, payee_id: str, payee: dict, plan_id: str) -> Payee:
         p = await self.client.update_payee(payee_id, payee, plan_id)
         await self.delta.invalidate_knowledge(plan_id, ENDPOINT_PAYEES)

src/config.py

@@ -25,6 +25,7 @@ class Settings(BaseSettings):
     )
 
     ynab_api_key: str = Field(alias="YNAB_API_KEY")
+    read_only: bool = Field(default=False, alias="YNAB_READ_ONLY")
     cache_db_path: str = Field(default_factory=_default_db_path)
     http_timeout: float = 30.0
 

src/server/__init__.py

@@ -16,7 +16,7 @@
     create_category_group, update_category_group,
     get_category_for_month, update_category_for_month,
 )
-from src.server.payees import list_payees, get_payee, update_payee
+from src.server.payees import list_payees, get_payee, create_payee, update_payee
 from src.server.payee_locations import (
     list_payee_locations, get_payee_location, get_payee_locations_by_payee,
 )
@@ -44,7 +44,7 @@
     "list_categories", "get_category", "create_category", "update_category",
     "create_category_group", "update_category_group",
     "get_category_for_month", "update_category_for_month",
-    "list_payees", "get_payee", "update_payee",
+    "list_payees", "get_payee", "create_payee", "update_payee",
     "list_payee_locations", "get_payee_location", "get_payee_locations_by_payee",
     "list_money_movements", "get_money_movements_for_month",
     "list_money_movement_groups", "get_money_movement_groups_for_month",

src/server/_shared.py

@@ -84,6 +84,17 @@ def dollars_to_milliunits(amount: float) -> int:
     return round(amount * 1000)
 
 
+def write_tool():
+    """Register a tool unless the server is in read-only mode.
+
+    Set YNAB_READ_ONLY=true to skip registering every mutating tool, so
+    clients only ever see the read tools.
+    """
+    if settings.read_only:
+        return lambda func: func
+    return mcp.tool()
+
+
 # Nested default excludes: when a parent model contains a list of nested models,
 # the nested model's default excludes need to be applied explicitly because
 # Pydantic's model_dump(exclude=set) only handles top-level fields.

src/server/accounts.py

@@ -19,7 +19,7 @@ async def list_accounts(
     return serialize_list(accounts, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def create_account(
     plan_id: str,

src/server/categories.py

@@ -43,7 +43,7 @@ async def get_category(
     return serialize(cat, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def create_category(
     plan_id: str,
@@ -82,7 +82,7 @@ async def create_category(
     return serialize(cat, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_category(
     plan_id: str,
@@ -124,7 +124,7 @@ async def update_category(
     return serialize(cat, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def create_category_group(
     plan_id: str,
@@ -144,7 +144,7 @@ async def create_category_group(
     return serialize(group, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_category_group(
     plan_id: str,
@@ -190,7 +190,7 @@ async def get_category_for_month(
     return serialize(cat, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_category_for_month(
     category_id: str,

src/server/payees.py

@@ -37,7 +37,27 @@ async def get_payee(
     return serialize(payee, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
+@_shared.handle_errors
+async def create_payee(
+    plan_id: str,
+    name: str,
+    exclude_fields: list[str] | None = None,
+) -> str:
+    """Create a new payee.
+
+    Args:
+        plan_id: The plan ID (use list_plans to find available IDs)
+        name: The name of the new payee (max 500 characters)
+        exclude_fields: Optional list of field names to exclude from the response.
+            If omitted, the model's default exclude list is used (see FIELDS.md).
+            Pass [] to return all fields. Pass a custom list to override the default.
+    """
+    payee = await _shared.cache.create_payee({"name": name}, plan_id)
+    return serialize(payee, exclude_fields=exclude_fields)
+
+
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_payee(
     plan_id: str,

src/server/scheduled.py

@@ -39,7 +39,7 @@ async def get_scheduled_transaction(
     return serialize(txn, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def create_scheduled_transaction(
     plan_id: str,
@@ -99,7 +99,7 @@ async def create_scheduled_transaction(
     return serialize(txn, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_scheduled_transaction(
     plan_id: str,
@@ -163,7 +163,7 @@ async def update_scheduled_transaction(
     return serialize(txn, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def delete_scheduled_transaction(
     scheduled_transaction_id: str,

src/server/transactions.py

@@ -215,7 +215,7 @@ async def search_transactions(
     return serialize_list(matches, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def create_transaction(
     plan_id: str,
@@ -281,7 +281,7 @@ async def create_transaction(
     return serialize(txn, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def create_transactions(
     plan_id: str,
@@ -336,7 +336,7 @@ async def create_transactions(
     return serialize_list(txns, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_transaction(
     plan_id: str,
@@ -408,7 +408,7 @@ async def update_transaction(
     return serialize(txn, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def delete_transaction(
     transaction_id: str,
@@ -428,7 +428,7 @@ async def delete_transaction(
     return serialize(txn, exclude_fields=exclude_fields)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def import_transactions(plan_id: str) -> str:
     """Import available transactions on all linked accounts for the given plan.
@@ -444,7 +444,7 @@ async def import_transactions(plan_id: str) -> str:
     return json.dumps({"transaction_ids": transaction_ids}, indent=2)
 
 
-@_shared.mcp.tool()
+@_shared.write_tool()
 @_shared.handle_errors
 async def update_transactions(
     plan_id: str,

src/ynab_client.py

@@ -324,7 +324,7 @@ async def update_category_group(
         self, category_group_id: str, category_group: dict, plan_id: str
     ) -> CategoryGroup:
         data = await self._patch(
-            f"/plans/{plan_id}/categories/groups/{category_group_id}",
+            f"/plans/{plan_id}/category_groups/{category_group_id}",
             json={"category_group": category_group},
         )
         return CategoryGroup.model_validate(data["data"]["category_group"])
@@ -364,6 +364,13 @@ async def get_payees(
         knowledge = data["data"]["server_knowledge"]
         return payees, knowledge
 
+    async def create_payee(self, payee: dict, plan_id: str) -> Payee:
+        data = await self._post(
+            f"/plans/{plan_id}/payees",
+            json={"payee": payee},
+        )
+        return Payee.model_validate(data["data"]["payee"])
+
     async def get_payee(self, payee_id: str, plan_id: str) -> Payee:
         data = await self._get(f"/plans/{plan_id}/payees/{payee_id}")
         return Payee.model_validate(data["data"]["payee"])

tests/test_client.py

@@ -270,6 +270,22 @@ async def test_returns_payees(self, client):
         assert payees[0].name == "Amazon"
 
 
+class TestCreatePayee:
+    @pytest.mark.asyncio
+    async def test_creates_payee(self, client):
+        data = {"data": {"payee": {"id": "p1", "name": "New Store"}, "server_knowledge": 4}}
+        client._client = AsyncMock()
+        client._client.post = AsyncMock(return_value=_mock_response(data, status_code=201))
+
+        result = await client.create_payee({"name": "New Store"}, "b1")
+        assert result.id == "p1"
+        assert result.name == "New Store"
+
+        post_call = client._client.post.call_args
+        assert post_call[0][0] == "/plans/b1/payees"
+        assert post_call[1]["json"] == {"payee": {"name": "New Store"}}
+
+
 # ── Months ────────────────────────────────────────────────────
 
 
@@ -348,6 +364,18 @@ async def test_transaction_by_category_url(self, client):
         url = client._client.get.call_args[0][0]
         assert url == "/plans/b1/categories/c1/transactions"
 
+    @pytest.mark.asyncio
+    async def test_update_category_group_url(self, client):
+        data = {"data": {"category_group": {
+            "id": "g1", "name": "Bills", "hidden": False, "deleted": False, "categories": [],
+        }}}
+        client._client = AsyncMock()
+        client._client.patch = AsyncMock(return_value=_mock_response(data))
+
+        await client.update_category_group("g1", {"name": "Bills"}, "b1")
+        url = client._client.patch.call_args[0][0]
+        assert url == "/plans/b1/category_groups/g1"
+
     @pytest.mark.asyncio
     async def test_month_category_url(self, client):
         data = {"data": {"category": {"id": "c1", "category_group_id": "g1", "name": "Food"}}}

tests/test_server.py

@@ -484,6 +484,19 @@ async def test_returns_payees(self, mock_cache):
         assert result[0]["name"] == "Amazon"
 
 
+class TestCreatePayee:
+    @pytest.mark.asyncio
+    async def test_creates_payee(self, mock_cache):
+        from src.server import create_payee
+
+        mock_cache.create_payee = AsyncMock(return_value=_make_payee(name="New Store"))
+        result = json.loads(await create_payee(plan_id="bud-1", name="New Store"))
+        assert result["name"] == "New Store"
+
+        call_args = mock_cache.create_payee.call_args
+        assert call_args[0][0] == {"name": "New Store"}
+
+
 # ── Month Tools ───────────────────────────────────────────────
 
 
@@ -833,3 +846,47 @@ async def test_returns_user(self, mock_cache):
         mock_cache.get_user = AsyncMock(return_value=User(id="user-123"))
         result = json.loads(await get_user())
         assert result["id"] == "user-123"
+
+
+# ── Read-Only Mode ────────────────────────────────────────────
+
+
+class TestReadOnlyMode:
+    def test_write_tool_skips_registration_when_read_only(self):
+        from src.server import _shared
+
+        async def dummy_write_tool() -> str:
+            return ""
+
+        with patch.object(_shared.settings, "read_only", True):
+            decorated = _shared.write_tool()(dummy_write_tool)
+
+        assert decorated is dummy_write_tool
+
+    @pytest.mark.asyncio
+    async def test_write_tool_registers_when_not_read_only(self):
+        from src.server import _shared
+
+        async def dummy_registered_tool() -> str:
+            return ""
+
+        with patch.object(_shared.settings, "read_only", False):
+            _shared.write_tool()(dummy_registered_tool)
+
+        names = [t.name for t in await _shared.mcp.list_tools()]
+        assert "dummy_registered_tool" in names
+
+    @pytest.mark.asyncio
+    async def test_mutating_tools_use_write_tool(self):
+        """Every create/update/delete/import tool must go through write_tool()."""
+        import re
+        from pathlib import Path
+
+        server_dir = Path(__file__).parent.parent / "src" / "server"
+        pattern = re.compile(
+            r"@_shared\.mcp\.tool\(\)\n@_shared\.handle_errors\n"
+            r"async def (create_|update_|delete_|import_)\w+"
+        )
+        for module in server_dir.glob("*.py"):
+            leaks = pattern.findall(module.read_text())
+            assert not leaks, f"{module.name} registers mutating tools unconditionally: {leaks}"