feat: add transaction splitting tools

CLAUDE.md

@@ -171,10 +171,11 @@ refactor: split server.py into modular components (auth, tools, models)
 - Structured logging with `structlog` for debugging
 - Environment variables: `MONARCH_EMAIL`, `MONARCH_PASSWORD`, `MONARCH_MFA_SECRET`
 
-**Complete Monarch Money API Coverage (19 Tools)**
+**Complete Monarch Money API Coverage (21 Tools)**
 - **Core**: `get_accounts`, `get_transactions`, `get_budgets`, `get_cashflow`
 - **Categories**: `get_transaction_categories`
 - **Transactions**: `create_transaction`, `update_transaction`, `update_transactions_bulk`, `search_transactions`
+- **Splits**: `get_transaction_splits`, `update_transaction_splits` (full-replace; empty list removes all splits)
 - **Investments**: `get_account_holdings` (requires `account_id`), `get_account_history`
 - **Banking**: `get_institutions`, `refresh_accounts`
 - **Planning**: `get_recurring_transactions`, `set_budget_amount`
@@ -254,8 +255,8 @@ Published to PyPI as `monarch-mcp-jamiew` and to the MCP Registry as `io.github.
 - **✅ Complete API Coverage**: All 14 Monarch Money API methods as tools
 
 #### Quality Metrics (Updated May 2026)
-- **202 passing tests** with comprehensive coverage including analytics, search, bulk operations, structured output, completions, resource templates, and progress
-- **19 tools** (all returning typed Pydantic models / structured output), 3 static resources + 2 resource templates, 4 prompts
+- **206 passing tests** with comprehensive coverage including analytics, search, bulk operations, splits, structured output, completions, resource templates, and progress
+- **21 tools** (all returning typed Pydantic models / structured output), 3 static resources + 2 resource templates, 4 prompts
 - **MyPy clean** under the repo's strict config (no `Any` at non-boundaries, no `as`)
 - **Security**: Proper session handling and MFA support
 - **Modern stack**: FastMCP 1.12.2, Pydantic, structlog, pytest
@@ -363,7 +364,7 @@ Published to PyPI as `monarch-mcp-jamiew` and to the MCP Registry as `io.github.
 ### 🔄 REMAINING LOW PRIORITY TASKS
 
 #### 7. Code Architecture & Organization
-**Current State**: Single file with 19 tools, comprehensive tests
+**Current State**: Single file with 21 tools, comprehensive tests
 **Remaining Work:**
 - Split into modules: `auth.py`, `tools.py`, `models.py`, `config.py` (optional - current structure works well)
 - Implement Pydantic Settings for configuration management
@@ -399,7 +400,7 @@ Published to PyPI as `monarch-mcp-jamiew` and to the MCP Registry as `io.github.
 5. **Phase 5 (Intelligence)**: ML features, advanced analytics, financial insights
 6. **Phase 6 (Ecosystem)**: MCP extensions, developer tools, architectural improvements
 
-**Current Status** (Updated May 2026): Production-ready with 202 passing tests, 19 intelligent tools, comprehensive analytics, robust error handling, and enhanced reliability. Recent MCP modernization: every tool returns structured output (outputSchema + structured content with a text fallback), tools/resources/prompts carry human-friendly `title`s, parameterized resource templates (`accounts://{account_id}/holdings|history`), argument completions for prompts/templates, and Context-based progress reporting on the batch tools. Earlier features: `update_transactions_bulk()` for parallel batch updates, `search_transactions`, result-size tracking; fixes for the auth retry bug, date serialization, broken pipes, and date parsing. Note: `get_account_holdings` now requires an `account_id` (the underlying library always did).
+**Current Status** (Updated June 2026): Production-ready with 206 passing tests, 21 intelligent tools (including transaction splitting via `get_transaction_splits` / `update_transaction_splits`), comprehensive analytics, robust error handling, and enhanced reliability. Recent MCP modernization: every tool returns structured output (outputSchema + structured content with a text fallback), tools/resources/prompts carry human-friendly `title`s, parameterized resource templates (`accounts://{account_id}/holdings|history`), argument completions for prompts/templates, and Context-based progress reporting on the batch tools. Earlier features: `update_transactions_bulk()` for parallel batch updates, `search_transactions`, result-size tracking; fixes for the auth retry bug, date serialization, broken pipes, and date parsing. Note: `get_account_holdings` now requires an `account_id` (the underlying library always did).
 
 ## Upstream Library & Fork Landscape
 
@@ -415,7 +416,7 @@ This MCP server is a thin wrapper over a Python Monarch Money client. That clien
 
 **Our pin:** `pyproject.toml` → `[tool.uv.sources]` pins `monarchmoneycommunity` to a **specific commit SHA** (the fork's `dev` HEAD), not a moving branch, for reproducible builds. When bumping, update the SHA *and* the comment date there.
 
-**Unused capabilities in the fork we already depend on** (zero new dependencies — just need new `@mcp.tool()` wrappers in `server.py`): transaction tags (`get/set/create_transaction_tag`), splits (`get/update_transaction_splits`), `find_duplicate_transactions`, `get_transaction_details`, `get_cashflow_summary`, `get_subscription_details`, `get_credit_history`, `delete_transaction`, `create_transaction_category`, `update_account`, `request_accounts_refresh_and_wait`.
+**Unused capabilities in the fork we already depend on** (zero new dependencies — just need new `@mcp.tool()` wrappers in `server.py`): transaction tags (`get/set/create_transaction_tag`), `find_duplicate_transactions`, `get_transaction_details`, `get_cashflow_summary`, `get_subscription_details`, `get_credit_history`, `delete_transaction`, `create_transaction_category`, `update_account`, `request_accounts_refresh_and_wait`.
 
 **`keithah/monarchmoney-enhanced` (worth exploring in a followup, needs testing):** adds whole capability areas neither our fork nor the parent has, several of which map onto TODOs above — a transaction **rules engine** (categorization/amount/ignore rules + apply-to-existing), a built-in **caching layer** (`preload_cache`, `clear_cache`, cache metrics), **proactive session management** (`validate_session`, `ensure_valid_session`, `is_session_stale`), **goals**, **bills** (`get_bills`), **merchants**, and **insights** (`get_insights`, `get_net_worth_history`, `get_investment_performance`, `get_credit_score`). Caveat: it is **not** a strict superset — our current fork has a few methods it lacks (`upload_attachment`, `reset_budget`, flex-budget methods, `get_credit_history`). So adopting it is a real decision (switch dependency vs. cherry-pick specific GraphQL queries), not a drop-in — needs hands-on testing against a live account first.
 

server.py

@@ -576,6 +576,33 @@ class TransactionResult(MMModel):
     transaction: JsonValue
 
 
+class TransactionSplit(BaseModel):
+    """One leg of a split transaction.
+
+    The split amounts must sum to the parent transaction's amount (Monarch
+    validates this and rejects the update otherwise). Amounts keep the parent's
+    sign convention — expenses are negative, income positive.
+    """
+
+    amount: float
+    category_id: str | None = None
+    merchant_name: str | None = None
+    notes: str | None = None
+
+
+class TransactionSplitsResult(MMModel):
+    transaction_id: str
+    has_split_transactions: bool
+    splits: list[JsonValue]
+
+
+class UpdateSplitsResult(MMModel):
+    transaction_id: str
+    has_split_transactions: bool
+    splits: list[JsonValue]
+    message: str
+
+
 class BulkSummary(BaseModel):
     total: int
     succeeded: int
@@ -1877,6 +1904,123 @@ async def update_single(update_data: dict[str, Any]) -> BulkItemResult:
         raise
 
 
+@mcp.tool(annotations=READONLY, title="Get Transaction Splits")
+@track_usage
+async def get_transaction_splits(transaction_id: str) -> TransactionSplitsResult:
+    """Get the split legs of a transaction.
+
+    Splitting lets a single transaction be divided across multiple categories
+    (e.g. a Target run that is part groceries, part household). This returns the
+    current split legs, if any.
+
+    Args:
+        transaction_id: ID of the transaction to inspect
+
+    Returns:
+        The transaction id, whether it currently has splits, and the list of
+        split legs (each with its own amount, category, merchant, and notes).
+        ``splits`` is empty for an un-split transaction.
+    """
+    await ensure_authenticated()
+
+    try:
+        result = await api_call_with_retry("get_transaction_splits", transaction_id=transaction_id)
+        result = convert_dates_to_strings(result)
+        transaction = result.get("getTransaction") or {} if isinstance(result, dict) else {}
+        splits = transaction.get("splitTransactions") or []
+        return TransactionSplitsResult(
+            transaction_id=transaction_id,
+            has_split_transactions=bool(splits),
+            splits=splits,
+        )
+    except Exception as e:
+        log.error("Failed to get transaction splits", error=str(e), transaction_id=transaction_id)
+        raise
+
+
+@mcp.tool(annotations=WRITE_IDEMPOTENT, title="Update Transaction Splits")
+@track_usage
+async def update_transaction_splits(transaction_id: str, splits: list[TransactionSplit]) -> UpdateSplitsResult:
+    """Create, replace, or remove the splits on a transaction.
+
+    This is a full replacement: the splits you pass become the transaction's
+    complete set of split legs, replacing any that exist. Pass an empty list to
+    remove all splits and restore the transaction to a single un-split entry.
+
+    Args:
+        transaction_id: ID of the transaction to split (required)
+        splits: The complete set of split legs. Each leg has:
+            - amount (required): Leg amount, using the parent's sign convention
+              (expenses negative, income positive). All leg amounts MUST sum to
+              the parent transaction's amount or Monarch rejects the update.
+            - category_id (optional): Category for this leg
+            - merchant_name (optional): Merchant display name for this leg;
+              defaults to the parent merchant when omitted
+            - notes (optional): Per-leg memo
+            Pass an empty list to delete all existing splits.
+
+    Example:
+        Split a -100.00 transaction into groceries and household:
+            transaction_id="txn_123"
+            splits=[
+                {"amount": -70.00, "category_id": "cat_groceries", "notes": "Food"},
+                {"amount": -30.00, "category_id": "cat_household"},
+            ]
+
+    Returns:
+        The transaction id, whether it now has splits, the resulting split legs,
+        and a human-readable summary message.
+    """
+    await ensure_authenticated()
+
+    try:
+        # Translate our snake_case inputs into the camelCase shape the API expects.
+        split_data: list[dict[str, Any]] = []
+        for split in splits:
+            entry: dict[str, Any] = {"amount": split.amount}
+            # Always send merchantName (empty string => inherit the parent merchant),
+            # matching the documented split payload shape.
+            entry["merchantName"] = split.merchant_name if split.merchant_name is not None else ""
+            if split.category_id is not None:
+                entry["categoryId"] = split.category_id
+            if split.notes is not None:
+                entry["notes"] = split.notes
+            split_data.append(entry)
+
+        log.info("updating_transaction_splits", transaction_id=transaction_id, split_count=len(split_data))
+
+        result = await asyncio.wait_for(
+            api_call_with_retry("update_transaction_splits", transaction_id=transaction_id, split_data=split_data),
+            timeout=30.0,
+        )
+        result = convert_dates_to_strings(result)
+
+        payload = result.get("updateTransactionSplit") or {} if isinstance(result, dict) else {}
+        errors = payload.get("errors")
+        if errors:
+            raise ValueError(f"Monarch rejected the split update: {errors}")
+
+        transaction = payload.get("transaction") or {}
+        result_splits = transaction.get("splitTransactions") or []
+        message = (
+            f"Removed all splits from transaction {transaction_id}"
+            if not split_data
+            else f"Set {len(result_splits)} split(s) on transaction {transaction_id}"
+        )
+        return UpdateSplitsResult(
+            transaction_id=transaction_id,
+            has_split_transactions=bool(transaction.get("hasSplitTransactions")),
+            splits=result_splits,
+            message=message,
+        )
+    except asyncio.TimeoutError as e:
+        log.error("update_transaction_splits_timeout", transaction_id=transaction_id)
+        raise ValueError("Split update timed out after 30 seconds. Please try again.") from e
+    except Exception as e:
+        log.error("update_transaction_splits_failed", transaction_id=transaction_id, error=str(e))
+        raise
+
+
 @mcp.tool(annotations=READONLY, title="Get Account Holdings")
 @track_usage
 async def get_account_holdings(account_id: str) -> HoldingsResult:

tests/test_splits.py

@@ -0,0 +1,125 @@
+"""Tests for transaction split tools: get_transaction_splits and update_transaction_splits.
+
+Uses the ``mock_api`` fixture (see conftest.py) so tests are offline: it patches
+``ensure_authenticated`` and ``api_call_with_retry``.
+"""
+
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+import server
+
+
+def _split_response(splits: list[dict[str, Any]]) -> dict[str, Any]:
+    """Shape returned by the library's get_transaction_splits."""
+    return {
+        "getTransaction": {
+            "id": "txn_1",
+            "amount": -100.0,
+            "splitTransactions": splits,
+        }
+    }
+
+
+def _update_response(splits: list[dict[str, Any]], has_splits: bool) -> dict[str, Any]:
+    """Shape returned by the library's update_transaction_splits."""
+    return {
+        "updateTransactionSplit": {
+            "errors": None,
+            "transaction": {
+                "id": "txn_1",
+                "hasSplitTransactions": has_splits,
+                "splitTransactions": splits,
+            },
+        }
+    }
+
+
+class TestGetTransactionSplits:
+    @pytest.mark.asyncio
+    async def test_returns_split_legs(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = _split_response(
+            [
+                {"id": "s1", "amount": -70.0, "category": {"id": "cat_1", "name": "Groceries"}},
+                {"id": "s2", "amount": -30.0, "category": {"id": "cat_2", "name": "Household"}},
+            ]
+        )
+        result = await server.get_transaction_splits(transaction_id="txn_1")
+        assert result.transaction_id == "txn_1"
+        assert result.has_split_transactions is True
+        assert len(result.splits) == 2
+        mock_api.assert_awaited_once_with("get_transaction_splits", transaction_id="txn_1")
+
+    @pytest.mark.asyncio
+    async def test_unsplit_transaction_has_empty_splits(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = _split_response([])
+        result = await server.get_transaction_splits(transaction_id="txn_1")
+        assert result.has_split_transactions is False
+        assert result.splits == []
+
+    @pytest.mark.asyncio
+    async def test_missing_transaction_key_is_safe(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = {"getTransaction": None}
+        result = await server.get_transaction_splits(transaction_id="txn_1")
+        assert result.has_split_transactions is False
+        assert result.splits == []
+
+
+class TestUpdateTransactionSplits:
+    @pytest.mark.asyncio
+    async def test_creates_splits_and_translates_payload(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = _update_response(
+            [{"id": "s1", "amount": -70.0}, {"id": "s2", "amount": -30.0}], has_splits=True
+        )
+        result = await server.update_transaction_splits(
+            transaction_id="txn_1",
+            splits=[
+                server.TransactionSplit(amount=-70.0, category_id="cat_1", notes="Food"),
+                server.TransactionSplit(amount=-30.0, category_id="cat_2"),
+            ],
+        )
+        assert result.has_split_transactions is True
+        assert len(result.splits) == 2
+        assert "Set 2 split(s)" in result.message
+
+        # Inputs are translated into the camelCase shape the API expects.
+        _, kwargs = mock_api.await_args
+        sent = kwargs["split_data"]
+        assert sent[0] == {"amount": -70.0, "merchantName": "", "categoryId": "cat_1", "notes": "Food"}
+        assert sent[1] == {"amount": -30.0, "merchantName": "", "categoryId": "cat_2"}
+
+    @pytest.mark.asyncio
+    async def test_passes_merchant_name_when_provided(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = _update_response([{"id": "s1", "amount": -100.0}], has_splits=True)
+        await server.update_transaction_splits(
+            transaction_id="txn_1",
+            splits=[server.TransactionSplit(amount=-100.0, merchant_name="Corner Deli")],
+        )
+        _, kwargs = mock_api.await_args
+        assert kwargs["split_data"][0]["merchantName"] == "Corner Deli"
+
+    @pytest.mark.asyncio
+    async def test_empty_list_removes_all_splits(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = _update_response([], has_splits=False)
+        result = await server.update_transaction_splits(transaction_id="txn_1", splits=[])
+        assert result.has_split_transactions is False
+        assert result.splits == []
+        assert "Removed all splits" in result.message
+        _, kwargs = mock_api.await_args
+        assert kwargs["split_data"] == []
+
+    @pytest.mark.asyncio
+    async def test_api_payload_errors_raise(self, mock_api: AsyncMock) -> None:
+        mock_api.return_value = {
+            "updateTransactionSplit": {
+                "errors": {"message": "Split amounts must sum to the transaction total"},
+                "transaction": None,
+            }
+        }
+        with pytest.raises(ValueError, match="Monarch rejected the split update"):
+            await server.update_transaction_splits(
+                transaction_id="txn_1",
+                splits=[server.TransactionSplit(amount=-10.0, category_id="cat_1")],
+            )

tests/test_structured_output.py

@@ -21,8 +21,8 @@ def test_every_tool_advertises_output_schema() -> None:
     assert without_schema == []
 
 
-def test_tool_count_is_nineteen() -> None:
-    assert len(mgr.list_tools()) == 19
+def test_tool_count_is_twenty_one() -> None:
+    assert len(mgr.list_tools()) == 21
 
 
 @pytest.mark.parametrize(

tests/test_tool_coverage.py

@@ -51,6 +51,14 @@ def _side_effect(method_name: str, *args: Any, **kwargs: Any) -> Any:
         id="create_transaction",
     ),
     pytest.param(lambda: server.update_transaction(transaction_id="txn_1", notes="memo"), id="update_transaction"),
+    pytest.param(lambda: server.get_transaction_splits(transaction_id="txn_1"), id="get_transaction_splits"),
+    pytest.param(
+        lambda: server.update_transaction_splits(
+            transaction_id="txn_1",
+            splits=[server.TransactionSplit(amount=-10.0, category_id="cat_1")],
+        ),
+        id="update_transaction_splits",
+    ),
     pytest.param(lambda: server.get_account_holdings(account_id="acc_1"), id="get_account_holdings"),
     pytest.param(lambda: server.get_account_history(account_id="acc_1"), id="get_account_history"),
     pytest.param(lambda: server.get_institutions(), id="get_institutions"),