This is page 1 of 2. Use http://codebase.md/chrisguidry/you-need-an-mcp?lines=true&page={x} to view the full context.
# Directory Structure
```
├── .envrc
├── .gitignore
├── .pre-commit-config.yaml
├── .python-version
├── CLAUDE.md
├── DESIGN.md
├── models.py
├── plans
│ └── caching.md
├── pyproject.toml
├── README.md
├── repository.py
├── server.py
├── tests
│ ├── assertions.py
│ ├── conftest.py
│ ├── test_accounts.py
│ ├── test_assertions.py
│ ├── test_budget_months.py
│ ├── test_categories.py
│ ├── test_payees.py
│ ├── test_repository.py
│ ├── test_scheduled_transactions.py
│ ├── test_transactions.py
│ ├── test_updates.py
│ └── test_utilities.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.12.8
2 |
```
--------------------------------------------------------------------------------
/.envrc:
--------------------------------------------------------------------------------
```
1 | if [ -f .venv/bin/activate ]; then
2 | source .venv/bin/activate
3 | fi
4 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | __pycache__/
2 | *.py[oc]
3 | build/
4 | dist/
5 | wheels/
6 | *.egg-info
7 | .venv
8 | .coverage
9 | .python-version
10 |
```
--------------------------------------------------------------------------------
/.pre-commit-config.yaml:
--------------------------------------------------------------------------------
```yaml
1 | repos:
2 | - repo: https://github.com/pre-commit/pre-commit-hooks
3 | rev: v4.5.0
4 | hooks:
5 | - id: trailing-whitespace
6 | - id: end-of-file-fixer
7 | - id: check-merge-conflict
8 | - id: check-yaml
9 | - id: check-toml
10 | - id: check-added-large-files
11 |
12 | - repo: https://github.com/astral-sh/ruff-pre-commit
13 | rev: v0.12.0
14 | hooks:
15 | - id: ruff-format
16 | - id: ruff
17 | args: [--fix, --exit-non-zero-on-fix]
18 |
19 | - repo: https://github.com/pre-commit/mirrors-mypy
20 | rev: v1.16.1
21 | hooks:
22 | - id: mypy
23 | additional_dependencies:
24 | - types-requests
25 | - pytest
26 | - fastmcp
27 | - pydantic
28 | - ynab
29 | args: [--strict, --show-error-codes]
30 | files: ^(server\.py|models\.py|tests/.+\.py)$
31 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # you-need-an-mcp
2 |
3 | An MCP server providing LLMs access to a YNAB budget.
4 |
5 | ## Setup
6 |
7 | ### 1. Install Dependencies
8 |
9 | ```bash
10 | uv sync
11 | ```
12 |
13 | ### 2. Get YNAB Access Token
14 |
15 | To use this MCP server, you need a YNAB Personal Access Token:
16 |
17 | 1. Log into your YNAB account at https://app.youneedabudget.com
18 | 2. Go to **Account Settings** (click your email in the top right corner)
19 | 3. Click on **Developer Settings** in the left sidebar
20 | 4. Click **New Token**
21 | 5. Enter a token name (e.g., "MCP Server")
22 | 6. Click **Generate**
23 | 7. Copy the generated token (you won't be able to see it again)
24 |
25 | ### 3. Set Environment Variables
26 |
27 | ```bash
28 | export YNAB_ACCESS_TOKEN=your_token_here
29 | ```
30 |
31 | Optionally, set a default budget ID to avoid having to specify it in every call:
32 |
33 | ```bash
34 | export YNAB_DEFAULT_BUDGET=your_budget_id_here
35 | ```
36 |
37 | ### 4. Run the Server
38 |
39 | ```bash
40 | uv run python server.py
41 | ```
42 |
43 | ## Available Tools
44 |
45 | - `list_budgets()` - Returns all your YNAB budgets
46 | - `list_accounts(budget_id=None, limit=100, offset=0, include_closed=False)` - Returns accounts with pagination and filtering
47 | - `list_categories(budget_id=None, limit=50, offset=0, include_hidden=False)` - Returns categories with pagination and filtering
48 | - `list_category_groups(budget_id=None)` - Returns category groups with totals (lighter weight overview)
49 |
50 | ### Pagination
51 |
52 | The `list_accounts` and `list_categories` tools support pagination. Use the `offset` parameter to get subsequent pages:
53 | - First page: `list_categories(limit=50, offset=0)`
54 | - Second page: `list_categories(limit=50, offset=50)`
55 | - Check `pagination.has_more` to see if there are more results
56 |
57 | ## Security Note
58 |
59 | Keep your YNAB access token secure and never commit it to version control. The token provides read access to all your budget data.
60 |
```
--------------------------------------------------------------------------------
/CLAUDE.md:
--------------------------------------------------------------------------------
```markdown
1 | # CLAUDE.md
2 |
3 | ## Project Overview
4 |
5 | MCP (Model Context Protocol) server for YNAB (You Need A Budget) using FastMCP. Provides structured access to YNAB financial data.
6 |
7 | ## Essential Commands
8 |
9 | ```bash
10 | # Development
11 | uv sync --group dev # Install all dependencies
12 | pytest --cov=server --cov=models --cov-report=term-missing # Run tests with coverage
13 | fastmcp run server.py:mcp # Run MCP server
14 |
15 | # Authentication (required)
16 | export YNAB_ACCESS_TOKEN=your_token_here
17 | export YNAB_BUDGET=your_budget_id_here # Required
18 | ```
19 |
20 | ## Critical Design Principles
21 |
22 | 1. **100% test coverage is mandatory** - No exceptions. All new code must have complete coverage.
23 | 2. **All listing tools must implement pagination** - Use `limit` and `offset` parameters.
24 | 3. **Automatically filter out deleted/hidden/closed data** - Only show active, relevant data.
25 | 4. **Use real YNAB SDK models in tests** - Mock only the API calls, not the data structures.
26 | 5. **Handle milliunits properly** - 1000 milliunits = 1 currency unit.
27 | 6. **Try to avoid using abbreviations in names**
28 |
29 | ## Product Philosophy
30 |
31 | - **Household finance assistant** - Help heads of household be more effective with YNAB
32 | - **Insights and notifications** - Surface important financial patterns and alerts
33 | - **Safe evolution** - Currently read-only, will add careful mutations (transactions, imports)
34 | - **Natural language friendly** - Enable text-based transaction entry and import assistance
35 | - **User-friendly defaults** - Sensible limits, current month defaults, active data only
36 | - **Performance conscious** - Pagination prevents token overflow, efficient payee search
37 |
38 | ## Security & Privacy
39 |
40 | - **Never log financial amounts or account numbers** - Use debug logs carefully
41 | - **Sanitize error messages** - Don't expose internal IDs or sensitive details
42 | - **Token safety** - Never commit or expose YNAB access tokens
43 | - **Fail securely** - Return generic errors for auth failures
44 |
45 | ## Tool Documentation
46 |
47 | **Critical**: Tool docstrings directly influence how LLMs use the MCP server. Every parameter must have clear descriptions explaining valid values and defaults. Bad docs = bad AI behavior.
48 |
49 | ## Testing Guidelines
50 |
51 | - Always run the full test suite after changes: `uv run pytest`
52 | - Use FastMCP's testing pattern with direct client-server testing
53 | - Mock YNAB API calls, but use real YNAB model instances
54 | - Verify pagination works correctly for all listing endpoints
55 | - Prefer not to use test classes, just simple test functions. Test organization should come through new files
56 |
57 | ## Architecture
58 |
59 | - `server.py` - MCP server implementation using `@mcp.tool()` decorators
60 | - `models.py` - Pydantic models matching YNAB's data structures
61 | - `DESIGN.md` - Detailed use cases and design philosophy
62 | - Uses context managers for YNAB client lifecycle
63 | - Returns structured JSON with consistent pagination format
64 | - Handle YNAB API errors gracefully with user-friendly messages
65 |
```
--------------------------------------------------------------------------------
/tests/test_assertions.py:
--------------------------------------------------------------------------------
```python
1 | """Test assertion helpers."""
2 |
3 | import pytest
4 | from assertions import extract_response_data
5 |
6 |
7 | def test_extract_response_data_invalid_type() -> None:
8 | """Test that extract_response_data raises TypeError for invalid input."""
9 | with pytest.raises(TypeError, match="Expected CallToolResult with content"):
10 | extract_response_data("invalid_input")
11 |
12 |
13 | def test_extract_response_data_invalid_list() -> None:
14 | """Test that extract_response_data raises TypeError for old list format."""
15 | with pytest.raises(TypeError, match="Expected CallToolResult with content"):
16 | extract_response_data([])
17 |
```
--------------------------------------------------------------------------------
/tests/assertions.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test assertion helpers for YNAB MCP Server tests.
3 |
4 | This module provides helper functions for common test assertions and response
5 | parsing to reduce boilerplate in test files.
6 | """
7 |
8 | import json
9 | from typing import Any
10 |
11 | from mcp.types import TextContent
12 |
13 |
14 | def extract_response_data(result: Any) -> dict[str, Any]:
15 | """Extract JSON data from MCP client response."""
16 | # Handle FastMCP CallToolResult format
17 | if not hasattr(result, "content"):
18 | raise TypeError(f"Expected CallToolResult with content, got {type(result)}")
19 |
20 | content = result.content
21 | assert len(content) == 1
22 | response_data: dict[str, Any] | None = (
23 | json.loads(content[0].text) if isinstance(content[0], TextContent) else None
24 | )
25 | assert response_data is not None
26 | return response_data
27 |
28 |
29 | def assert_pagination_info(
30 | pagination: dict[str, Any],
31 | *,
32 | total_count: int,
33 | limit: int,
34 | offset: int = 0,
35 | has_more: bool = False,
36 | ) -> None:
37 | """Assert pagination info matches expected values."""
38 | assert pagination["total_count"] == total_count
39 | assert pagination["limit"] == limit
40 | assert pagination["offset"] == offset
41 | assert pagination["has_more"] == has_more
42 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "you-need-an-mcp"
3 | version = "0.1.0"
4 | description = "Add your description here"
5 | readme = "README.md"
6 | requires-python = ">=3.12"
7 | dependencies = ["fastmcp>=2.11.3", "pydantic>=2.11.5", "ynab>=1.4.0"]
8 |
9 | [dependency-groups]
10 | dev = [
11 | "pytest>=8.4.0",
12 | "pytest-asyncio>=1.0.0",
13 | "pytest-cov>=6.2.1",
14 | "pytest-xdist>=3.7.0",
15 | "pytest-env>=1.1.5",
16 | "mypy>=1.8.0",
17 | "pre-commit>=3.6.0",
18 | "ruff>=0.12.0",
19 | ]
20 |
21 | [tool.pytest.ini_options]
22 | addopts = [
23 | "--cov=server",
24 | "--cov=models",
25 | "--cov=tests/",
26 | "--cov-branch",
27 | "--cov-report=term-missing",
28 | "--cov-fail-under=100",
29 | "--strict-markers",
30 | "--strict-config",
31 | "-Werror",
32 | ]
33 | asyncio_mode = "auto"
34 | testpaths = ["tests"]
35 | filterwarnings = ["error"]
36 | env = [
37 | "YNAB_BUDGET=test_budget_id",
38 | "YNAB_ACCESS_TOKEN=test_token_123",
39 | ]
40 |
41 | [tool.mypy]
42 | python_version = "3.12"
43 | strict = true
44 |
45 | [[tool.mypy.overrides]]
46 | module = ["ynab", "ynab.*"]
47 | implicit_reexport = true
48 |
49 | [tool.ruff]
50 | target-version = "py312"
51 |
52 | [tool.ruff.lint]
53 | select = [
54 | "E", # pycodestyle errors
55 | "W", # pycodestyle warnings
56 | "F", # pyflakes
57 | "I", # isort
58 | "B", # flake8-bugbear
59 | "UP", # pyupgrade (enforce modern python syntax)
60 | "RUF", # ruff-specific rules
61 | ]
62 | ignore = []
63 |
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test fixtures for YNAB MCP Server tests.
3 |
4 | This module contains pytest fixtures for testing without calling the actual YNAB API.
5 | """
6 |
7 | import sys
8 | from collections.abc import AsyncGenerator, Generator
9 | from datetime import date
10 | from pathlib import Path
11 | from typing import Any
12 | from unittest.mock import MagicMock, Mock, patch
13 |
14 | import fastmcp
15 | import pytest
16 | import ynab
17 | from fastmcp.client import Client, FastMCPTransport
18 |
19 | # Add parent directory to path to import server module
20 | sys.path.insert(0, str(Path(__file__).parent.parent))
21 | import server
22 |
23 |
24 | @pytest.fixture
25 | def mock_environment_variables(monkeypatch: pytest.MonkeyPatch) -> None:
26 | """Mock environment variables for testing."""
27 | monkeypatch.setenv("YNAB_ACCESS_TOKEN", "test_token_123")
28 | monkeypatch.setenv("YNAB_BUDGET", "test_budget_id")
29 |
30 |
31 | @pytest.fixture
32 | def ynab_client(mock_environment_variables: None) -> Generator[MagicMock, None, None]:
33 | """Mock YNAB client with proper autospec for testing."""
34 | mock_client = MagicMock(spec=ynab.ApiClient)
35 | mock_client.__enter__.return_value = mock_client
36 | mock_client.__exit__.return_value = None
37 | yield mock_client
38 |
39 |
40 | @pytest.fixture
41 | def mock_repository() -> Generator[MagicMock, None, None]:
42 | """Mock the repository to prevent API calls during testing."""
43 | with patch("server._repository") as mock_repo:
44 | yield mock_repo
45 |
46 |
47 | @pytest.fixture
48 | def categories_api(ynab_client: MagicMock) -> Generator[MagicMock, None, None]:
49 | mock_api = Mock(spec=ynab.CategoriesApi)
50 | with patch("ynab.CategoriesApi", return_value=mock_api):
51 | yield mock_api
52 |
53 |
54 | @pytest.fixture
55 | async def mcp_client() -> AsyncGenerator[Client[FastMCPTransport], None]:
56 | """Mock MCP client with proper autospec for testing."""
57 | async with fastmcp.Client(server.mcp) as client:
58 | yield client
59 |
60 |
61 | # Test data factories
62 | def create_ynab_account(
63 | *,
64 | id: str = "acc-1",
65 | name: str = "Test Account",
66 | account_type: ynab.AccountType = ynab.AccountType.CHECKING,
67 | on_budget: bool = True,
68 | closed: bool = False,
69 | balance: int = 100_000,
70 | deleted: bool = False,
71 | **kwargs: Any,
72 | ) -> ynab.Account:
73 | """Create a YNAB Account for testing with sensible defaults."""
74 | return ynab.Account(
75 | id=id,
76 | name=name,
77 | type=account_type,
78 | on_budget=on_budget,
79 | closed=closed,
80 | note=kwargs.get("note"),
81 | balance=balance,
82 | cleared_balance=kwargs.get("cleared_balance", balance - 5_000),
83 | uncleared_balance=kwargs.get("uncleared_balance", 5_000),
84 | transfer_payee_id=kwargs.get("transfer_payee_id"),
85 | direct_import_linked=kwargs.get("direct_import_linked", False),
86 | direct_import_in_error=kwargs.get("direct_import_in_error", False),
87 | last_reconciled_at=kwargs.get("last_reconciled_at"),
88 | debt_original_balance=kwargs.get("debt_original_balance"),
89 | debt_interest_rates=kwargs.get("debt_interest_rates"),
90 | debt_minimum_payments=kwargs.get("debt_minimum_payments"),
91 | debt_escrow_amounts=kwargs.get("debt_escrow_amounts"),
92 | deleted=deleted,
93 | )
94 |
95 |
96 | def create_ynab_payee(
97 | *,
98 | id: str = "payee-1",
99 | name: str = "Test Payee",
100 | deleted: bool = False,
101 | **kwargs: Any,
102 | ) -> ynab.Payee:
103 | """Create a YNAB Payee for testing with sensible defaults."""
104 | return ynab.Payee(
105 | id=id,
106 | name=name,
107 | transfer_account_id=kwargs.get("transfer_account_id"),
108 | deleted=deleted,
109 | )
110 |
111 |
112 | def create_ynab_category(
113 | *,
114 | id: str = "cat-1",
115 | name: str = "Test Category",
116 | category_group_id: str = "group-1",
117 | hidden: bool = False,
118 | deleted: bool = False,
119 | budgeted: int = 50_000,
120 | activity: int = -30_000,
121 | balance: int = 20_000,
122 | **kwargs: Any,
123 | ) -> ynab.Category:
124 | """Create a YNAB Category for testing with sensible defaults."""
125 | return ynab.Category(
126 | id=id,
127 | category_group_id=category_group_id,
128 | category_group_name=kwargs.get("category_group_name"),
129 | name=name,
130 | hidden=hidden,
131 | original_category_group_id=kwargs.get("original_category_group_id"),
132 | note=kwargs.get("note"),
133 | budgeted=budgeted,
134 | activity=activity,
135 | balance=balance,
136 | goal_type=kwargs.get("goal_type"),
137 | goal_needs_whole_amount=kwargs.get("goal_needs_whole_amount"),
138 | goal_day=kwargs.get("goal_day"),
139 | goal_cadence=kwargs.get("goal_cadence"),
140 | goal_cadence_frequency=kwargs.get("goal_cadence_frequency"),
141 | goal_creation_month=kwargs.get("goal_creation_month"),
142 | goal_target=kwargs.get("goal_target"),
143 | goal_target_month=kwargs.get("goal_target_month"),
144 | goal_percentage_complete=kwargs.get("goal_percentage_complete"),
145 | goal_months_to_budget=kwargs.get("goal_months_to_budget"),
146 | goal_under_funded=kwargs.get("goal_under_funded"),
147 | goal_overall_funded=kwargs.get("goal_overall_funded"),
148 | goal_overall_left=kwargs.get("goal_overall_left"),
149 | deleted=deleted,
150 | )
151 |
152 |
153 | def create_ynab_transaction(
154 | *,
155 | id: str = "txn-1",
156 | transaction_date: date = date(2024, 1, 15),
157 | amount: int = -50_000,
158 | account_id: str = "acc-1",
159 | deleted: bool = False,
160 | **kwargs: Any,
161 | ) -> ynab.TransactionDetail:
162 | """Create a YNAB TransactionDetail for testing with sensible defaults."""
163 | return ynab.TransactionDetail(
164 | id=id,
165 | date=transaction_date,
166 | amount=amount,
167 | memo=kwargs.get("memo"),
168 | cleared=kwargs.get("cleared", ynab.TransactionClearedStatus.CLEARED),
169 | approved=kwargs.get("approved", True),
170 | flag_color=kwargs.get("flag_color"),
171 | account_id=account_id,
172 | account_name=kwargs.get("account_name", "Test Account"),
173 | payee_id=kwargs.get("payee_id"),
174 | payee_name=kwargs.get("payee_name"),
175 | category_id=kwargs.get("category_id"),
176 | category_name=kwargs.get("category_name"),
177 | transfer_account_id=kwargs.get("transfer_account_id"),
178 | transfer_transaction_id=kwargs.get("transfer_transaction_id"),
179 | matched_transaction_id=kwargs.get("matched_transaction_id"),
180 | import_id=kwargs.get("import_id"),
181 | import_payee_name=kwargs.get("import_payee_name"),
182 | import_payee_name_original=kwargs.get("import_payee_name_original"),
183 | debt_transaction_type=kwargs.get("debt_transaction_type"),
184 | deleted=deleted,
185 | subtransactions=kwargs.get("subtransactions", []),
186 | )
187 |
```
--------------------------------------------------------------------------------
/tests/test_categories.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test category-related MCP tools.
3 | """
4 |
5 | from unittest.mock import MagicMock
6 |
7 | import ynab
8 | from assertions import assert_pagination_info, extract_response_data
9 | from conftest import create_ynab_category
10 | from fastmcp.client import Client, FastMCPTransport
11 |
12 |
13 | async def test_list_categories_success(
14 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
15 | ) -> None:
16 | """Test successful category listing."""
17 | visible_category = create_ynab_category(
18 | id="cat-1",
19 | name="Groceries",
20 | note="Food shopping",
21 | goal_type="TB",
22 | goal_target=100_000,
23 | goal_percentage_complete=50,
24 | )
25 |
26 | hidden_category = create_ynab_category(
27 | id="cat-hidden",
28 | name="Hidden Category",
29 | hidden=True, # Should be excluded
30 | budgeted=10_000,
31 | activity=0,
32 | balance=10_000,
33 | )
34 |
35 | category_group = ynab.CategoryGroupWithCategories(
36 | id="group-1",
37 | name="Monthly Bills",
38 | hidden=False,
39 | deleted=False,
40 | categories=[visible_category, hidden_category],
41 | )
42 |
43 | # Mock repository to return category groups
44 | mock_repository.get_category_groups.return_value = [category_group]
45 |
46 | result = await mcp_client.call_tool("list_categories", {})
47 | response_data = extract_response_data(result)
48 |
49 | # Should only include visible category
50 | categories = response_data["categories"]
51 | assert len(categories) == 1
52 | assert categories[0]["id"] == "cat-1"
53 | assert categories[0]["name"] == "Groceries"
54 | assert categories[0]["category_group_name"] == "Monthly Bills"
55 |
56 | assert_pagination_info(
57 | response_data["pagination"],
58 | total_count=1,
59 | limit=50,
60 | has_more=False,
61 | )
62 |
63 |
64 | async def test_list_category_groups_success(
65 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
66 | ) -> None:
67 | """Test successful category group listing."""
68 |
69 | category = ynab.Category(
70 | id="cat-1",
71 | category_group_id="group-1",
72 | category_group_name="Monthly Bills",
73 | name="Test Category",
74 | hidden=False,
75 | original_category_group_id=None,
76 | note=None,
77 | budgeted=50000,
78 | activity=-30000,
79 | balance=20000,
80 | goal_type=None,
81 | goal_needs_whole_amount=None,
82 | goal_day=None,
83 | goal_cadence=None,
84 | goal_cadence_frequency=None,
85 | goal_creation_month=None,
86 | goal_target=None,
87 | goal_target_month=None,
88 | goal_percentage_complete=None,
89 | goal_months_to_budget=None,
90 | goal_under_funded=None,
91 | goal_overall_funded=None,
92 | goal_overall_left=None,
93 | deleted=False,
94 | )
95 |
96 | category_group = ynab.CategoryGroupWithCategories(
97 | id="group-1",
98 | name="Monthly Bills",
99 | hidden=False,
100 | deleted=False,
101 | categories=[category],
102 | )
103 |
104 | # Mock repository to return category groups
105 | mock_repository.get_category_groups.return_value = [category_group]
106 |
107 | result = await mcp_client.call_tool("list_category_groups", {})
108 |
109 | groups_data = extract_response_data(result)
110 | # Should return a list of category groups
111 | assert isinstance(groups_data, list)
112 | assert len(groups_data) == 1
113 | group = groups_data[0]
114 | assert group["id"] == "group-1"
115 | assert group["name"] == "Monthly Bills"
116 |
117 |
118 | async def test_list_categories_filters_deleted_and_hidden(
119 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
120 | ) -> None:
121 | """Test that list_categories automatically filters out deleted and hidden."""
122 |
123 | # Active category (should be included)
124 | mock_active_category = ynab.Category(
125 | id="cat-active",
126 | name="Active Category",
127 | category_group_id="group-1",
128 | hidden=False,
129 | deleted=False,
130 | note="Active",
131 | budgeted=10000,
132 | activity=-5000,
133 | balance=5000,
134 | goal_type=None,
135 | goal_target=None,
136 | goal_percentage_complete=None,
137 | goal_under_funded=None,
138 | goal_creation_month=None,
139 | goal_target_month=None,
140 | goal_overall_funded=None,
141 | goal_overall_left=None,
142 | )
143 |
144 | # Hidden category (should be excluded)
145 | mock_hidden_category = ynab.Category(
146 | id="cat-hidden",
147 | name="Hidden Category",
148 | category_group_id="group-1",
149 | hidden=True,
150 | deleted=False,
151 | note="Hidden",
152 | budgeted=0,
153 | activity=0,
154 | balance=0,
155 | goal_type=None,
156 | goal_target=None,
157 | goal_percentage_complete=None,
158 | goal_under_funded=None,
159 | goal_creation_month=None,
160 | goal_target_month=None,
161 | goal_overall_funded=None,
162 | goal_overall_left=None,
163 | )
164 |
165 | # Deleted category (should be excluded)
166 | mock_deleted_category = ynab.Category(
167 | id="cat-deleted",
168 | name="Deleted Category",
169 | category_group_id="group-1",
170 | hidden=False,
171 | deleted=True,
172 | note="Deleted",
173 | budgeted=0,
174 | activity=0,
175 | balance=0,
176 | goal_type=None,
177 | goal_target=None,
178 | goal_percentage_complete=None,
179 | goal_under_funded=None,
180 | goal_creation_month=None,
181 | goal_target_month=None,
182 | goal_overall_funded=None,
183 | goal_overall_left=None,
184 | )
185 |
186 | category_group = ynab.CategoryGroupWithCategories(
187 | id="group-1",
188 | name="Monthly Bills",
189 | hidden=False,
190 | deleted=False,
191 | categories=[
192 | mock_active_category,
193 | mock_hidden_category,
194 | mock_deleted_category,
195 | ],
196 | )
197 |
198 | # Mock repository to return category groups
199 | mock_repository.get_category_groups.return_value = [category_group]
200 |
201 | result = await mcp_client.call_tool("list_categories", {})
202 |
203 | response_data = extract_response_data(result)
204 | # Should only include the active category
205 | assert len(response_data["categories"]) == 1
206 | assert response_data["categories"][0]["id"] == "cat-active"
207 | assert response_data["categories"][0]["name"] == "Active Category"
208 |
209 |
210 | async def test_list_category_groups_filters_deleted(
211 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
212 | ) -> None:
213 | """Test that list_category_groups automatically filters out deleted groups."""
214 |
215 | # Active group (should be included)
216 | active_group = ynab.CategoryGroupWithCategories(
217 | id="group-active",
218 | name="Active Group",
219 | hidden=False,
220 | deleted=False,
221 | categories=[],
222 | )
223 |
224 | # Deleted group (should be excluded)
225 | deleted_group = ynab.CategoryGroupWithCategories(
226 | id="group-deleted",
227 | name="Deleted Group",
228 | hidden=False,
229 | deleted=True,
230 | categories=[],
231 | )
232 |
233 | # Mock repository to return category groups
234 | mock_repository.get_category_groups.return_value = [active_group, deleted_group]
235 |
236 | result = await mcp_client.call_tool("list_category_groups", {})
237 |
238 | response_data = extract_response_data(result)
239 | # Should only include the active group
240 | assert isinstance(response_data, list)
241 | assert len(response_data) == 1
242 | group = response_data[0]
243 | assert group["id"] == "group-active"
244 | assert group["name"] == "Active Group"
245 |
```
--------------------------------------------------------------------------------
/tests/test_utilities.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test utility functions in server module.
3 | """
4 |
5 | from datetime import date, datetime
6 | from decimal import Decimal
7 | from unittest.mock import patch
8 |
9 | import pytest
10 |
11 | import server
12 |
13 |
14 | def test_decimal_precision_milliunits_conversion() -> None:
15 | """Test that milliunits conversion maintains Decimal precision."""
16 | # Test various milliunits values that could lose precision with floats
17 | test_cases = [
18 | (123456, Decimal("123.456")), # Regular amount
19 | (1, Decimal("0.001")), # Smallest unit
20 | (999, Decimal("0.999")), # Just under 1
21 | (1000, Decimal("1")), # Exactly 1
22 | (1001, Decimal("1.001")), # Just over 1
23 | (999999999, Decimal("999999.999")), # Large amount
24 | (-50000, Decimal("-50")), # Negative amount
25 | (0, Decimal("0")), # Zero
26 | ]
27 |
28 | for milliunits, expected in test_cases:
29 | from models import milliunits_to_currency
30 |
31 | result = milliunits_to_currency(milliunits)
32 | assert result == expected, (
33 | f"Failed for {milliunits}: got {result}, expected {expected}"
34 | )
35 | # Ensure result is actually a Decimal, not float
36 | assert isinstance(result, Decimal), (
37 | f"Result {result} is not a Decimal but {type(result)}"
38 | )
39 |
40 |
41 | def test_milliunits_to_currency_valid_input() -> None:
42 | """Test milliunits conversion with valid input."""
43 | from models import milliunits_to_currency
44 |
45 | result = milliunits_to_currency(123456)
46 | assert result == Decimal("123.456")
47 |
48 |
49 | def test_milliunits_to_currency_none_input() -> None:
50 | """Test milliunits conversion with None input raises TypeError."""
51 | from typing import Any
52 |
53 | with pytest.raises(TypeError):
54 | from models import milliunits_to_currency
55 |
56 | none_value: Any = None
57 | milliunits_to_currency(none_value)
58 |
59 |
60 | def test_milliunits_to_currency_zero() -> None:
61 | """Test milliunits conversion with zero."""
62 | from models import milliunits_to_currency
63 |
64 | result = milliunits_to_currency(0)
65 | assert result == Decimal("0")
66 |
67 |
68 | def test_milliunits_to_currency_negative() -> None:
69 | """Test milliunits conversion with negative value."""
70 | from models import milliunits_to_currency
71 |
72 | result = milliunits_to_currency(-50000)
73 | assert result == Decimal("-50")
74 |
75 |
76 | def test_convert_month_to_date_with_date_object() -> None:
77 | """Test convert_month_to_date with date object returns unchanged."""
78 | test_date = date(2024, 3, 15)
79 | result = server.convert_month_to_date(test_date)
80 | assert result == test_date
81 |
82 |
83 | def test_convert_month_to_date_with_current() -> None:
84 | """Test convert_month_to_date with 'current' returns current month date."""
85 | with patch("server.datetime") as mock_datetime:
86 | mock_datetime.now.return_value = datetime(2024, 9, 20, 16, 45, 0)
87 |
88 | result = server.convert_month_to_date("current")
89 | assert result == date(2024, 9, 1)
90 |
91 |
92 | def test_convert_month_to_date_with_last_and_next() -> None:
93 | """Test convert_month_to_date with 'last' and 'next' literals."""
94 | # Test normal month (June -> May and July)
95 | with patch("server.datetime") as mock_datetime:
96 | mock_datetime.now.return_value = datetime(2024, 6, 15, 10, 30, 0)
97 |
98 | result_last = server.convert_month_to_date("last")
99 | assert result_last == date(2024, 5, 1)
100 |
101 | result_next = server.convert_month_to_date("next")
102 | assert result_next == date(2024, 7, 1)
103 |
104 | # Test January edge case (January -> December previous year)
105 | with patch("server.datetime") as mock_datetime:
106 | mock_datetime.now.return_value = datetime(2024, 1, 10, 14, 45, 0)
107 |
108 | result_last = server.convert_month_to_date("last")
109 | assert result_last == date(2023, 12, 1)
110 |
111 | result_next = server.convert_month_to_date("next")
112 | assert result_next == date(2024, 2, 1)
113 |
114 | # Test December edge case (December -> January next year)
115 | with patch("server.datetime") as mock_datetime:
116 | mock_datetime.now.return_value = datetime(2024, 12, 25, 9, 15, 0)
117 |
118 | result_last = server.convert_month_to_date("last")
119 | assert result_last == date(2024, 11, 1)
120 |
121 | result_next = server.convert_month_to_date("next")
122 | assert result_next == date(2025, 1, 1)
123 |
124 |
125 | def test_convert_month_to_date_invalid_value() -> None:
126 | """Test convert_month_to_date with invalid value raises error."""
127 | from typing import Any
128 |
129 | with pytest.raises(ValueError, match="Invalid month value: invalid"):
130 | invalid_value: Any = "invalid"
131 | server.convert_month_to_date(invalid_value)
132 |
133 |
134 | def test_convert_transaction_to_model_basic() -> None:
135 | """Test Transaction.from_ynab with basic transaction."""
136 | import ynab
137 |
138 | from models import Transaction
139 |
140 | txn = ynab.TransactionDetail(
141 | id="txn-123",
142 | date=date(2024, 6, 15),
143 | amount=-50000,
144 | memo="Test transaction",
145 | cleared=ynab.TransactionClearedStatus.CLEARED,
146 | approved=True,
147 | flag_color=ynab.TransactionFlagColor.RED,
148 | account_id="acc-1",
149 | payee_id="payee-1",
150 | category_id="cat-1",
151 | transfer_account_id=None,
152 | transfer_transaction_id=None,
153 | matched_transaction_id=None,
154 | import_id=None,
155 | import_payee_name=None,
156 | import_payee_name_original=None,
157 | debt_transaction_type=None,
158 | deleted=False,
159 | account_name="Checking",
160 | payee_name="Test Payee",
161 | category_name="Test Category",
162 | subtransactions=[],
163 | )
164 |
165 | result = Transaction.from_ynab(txn)
166 |
167 | assert result.id == "txn-123"
168 | assert result.date == date(2024, 6, 15)
169 | assert result.amount == Decimal("-50")
170 | assert result.account_name == "Checking"
171 | assert result.payee_name == "Test Payee"
172 | assert result.category_name == "Test Category"
173 | assert result.subtransactions is None
174 |
175 |
176 | def test_convert_transaction_to_model_without_optional_attributes() -> None:
177 | """Test Transaction.from_ynab with minimal TransactionDetail."""
178 | import ynab
179 |
180 | from models import Transaction
181 |
182 | minimal_txn = ynab.TransactionDetail(
183 | id="txn-456",
184 | date=date(2024, 6, 16),
185 | amount=-25000,
186 | memo="Minimal transaction",
187 | cleared=ynab.TransactionClearedStatus.UNCLEARED,
188 | approved=True,
189 | flag_color=None,
190 | account_id="acc-2",
191 | payee_id="payee-2",
192 | category_id="cat-2",
193 | transfer_account_id=None,
194 | transfer_transaction_id=None,
195 | matched_transaction_id=None,
196 | import_id=None,
197 | import_payee_name=None,
198 | import_payee_name_original=None,
199 | debt_transaction_type=None,
200 | deleted=False,
201 | account_name="Test Account 2",
202 | payee_name="Test Payee 2",
203 | category_name="Test Category 2",
204 | subtransactions=[],
205 | )
206 |
207 | result = Transaction.from_ynab(minimal_txn)
208 |
209 | assert result.id == "txn-456"
210 | assert result.account_name == "Test Account 2"
211 | assert result.payee_name == "Test Payee 2"
212 | assert result.category_name == "Test Category 2"
213 |
214 |
215 | def test_milliunits_to_currency_from_models() -> None:
216 | """Test milliunits_to_currency function from models module."""
217 | from models import milliunits_to_currency
218 |
219 | assert milliunits_to_currency(50000) == Decimal("50")
220 | assert milliunits_to_currency(-25000) == Decimal("-25")
221 | assert milliunits_to_currency(1000) == Decimal("1")
222 | assert milliunits_to_currency(0) == Decimal("0")
223 |
```
--------------------------------------------------------------------------------
/tests/test_accounts.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test account-related MCP tools.
3 | """
4 |
5 | from unittest.mock import MagicMock
6 |
7 | import ynab
8 | from assertions import assert_pagination_info, extract_response_data
9 | from conftest import create_ynab_account
10 | from fastmcp.client import Client, FastMCPTransport
11 |
12 |
13 | async def test_list_accounts_success(
14 | mock_repository: MagicMock,
15 | mcp_client: Client[FastMCPTransport],
16 | ) -> None:
17 | """Test successful account listing."""
18 | open_account = create_ynab_account(
19 | id="acc-1",
20 | name="Checking",
21 | account_type=ynab.AccountType.CHECKING,
22 | note="Main account",
23 | )
24 |
25 | closed_account = create_ynab_account(
26 | id="acc-2",
27 | name="Savings",
28 | account_type=ynab.AccountType.SAVINGS,
29 | closed=True, # Should be excluded
30 | balance=0,
31 | )
32 |
33 | # Mock repository to return test accounts
34 | mock_repository.get_accounts.return_value = [open_account, closed_account]
35 |
36 | result = await mcp_client.call_tool("list_accounts", {})
37 | response_data = extract_response_data(result)
38 |
39 | # Should only include open account
40 | accounts = response_data["accounts"]
41 | assert len(accounts) == 1
42 | assert accounts[0]["id"] == "acc-1"
43 | assert accounts[0]["name"] == "Checking"
44 |
45 | assert_pagination_info(
46 | response_data["pagination"],
47 | total_count=1,
48 | limit=100,
49 | has_more=False,
50 | )
51 |
52 |
53 | async def test_list_accounts_filters_closed_accounts(
54 | mock_repository: MagicMock,
55 | mcp_client: Client[FastMCPTransport],
56 | ) -> None:
57 | """Test that list_accounts automatically excludes closed accounts."""
58 | open_checking = create_ynab_account(
59 | id="acc-1",
60 | name="Checking",
61 | account_type=ynab.AccountType.CHECKING,
62 | closed=False,
63 | )
64 |
65 | closed_savings = create_ynab_account(
66 | id="acc-2",
67 | name="Old Savings",
68 | account_type=ynab.AccountType.SAVINGS,
69 | closed=True,
70 | )
71 |
72 | open_credit = create_ynab_account(
73 | id="acc-3",
74 | name="Credit Card",
75 | account_type=ynab.AccountType.CREDITCARD,
76 | closed=False,
77 | )
78 |
79 | mock_repository.get_accounts.return_value = [
80 | open_checking,
81 | closed_savings,
82 | open_credit,
83 | ]
84 |
85 | result = await mcp_client.call_tool("list_accounts", {})
86 | response_data = extract_response_data(result)
87 |
88 | # Should only include open accounts
89 | accounts = response_data["accounts"]
90 | assert len(accounts) == 2
91 |
92 | account_names = [acc["name"] for acc in accounts]
93 | assert "Checking" in account_names
94 | assert "Credit Card" in account_names
95 | assert "Old Savings" not in account_names # Closed account excluded
96 |
97 |
98 | async def test_list_accounts_pagination(
99 | mock_repository: MagicMock,
100 | mcp_client: Client[FastMCPTransport],
101 | ) -> None:
102 | """Test account listing with pagination."""
103 | accounts = []
104 | for i in range(5):
105 | accounts.append(
106 | create_ynab_account(
107 | id=f"acc-{i}",
108 | name=f"Account {i}",
109 | closed=False,
110 | )
111 | )
112 |
113 | mock_repository.get_accounts.return_value = accounts
114 |
115 | # Test first page
116 | result = await mcp_client.call_tool("list_accounts", {"limit": 2, "offset": 0})
117 | response_data = extract_response_data(result)
118 |
119 | assert len(response_data["accounts"]) == 2
120 | assert_pagination_info(
121 | response_data["pagination"],
122 | total_count=5,
123 | limit=2,
124 | has_more=True,
125 | )
126 |
127 | # Test second page
128 | result = await mcp_client.call_tool("list_accounts", {"limit": 2, "offset": 2})
129 | response_data = extract_response_data(result)
130 |
131 | assert len(response_data["accounts"]) == 2
132 | assert_pagination_info(
133 | response_data["pagination"],
134 | total_count=5,
135 | limit=2,
136 | offset=2,
137 | has_more=True,
138 | )
139 |
140 |
141 | async def test_list_accounts_with_repository_sync(
142 | mock_repository: MagicMock,
143 | mcp_client: Client[FastMCPTransport],
144 | ) -> None:
145 | """Test that list_accounts triggers repository sync when needed."""
146 | account = create_ynab_account(id="acc-1", name="Test Account")
147 |
148 | # Mock repository to return empty initially (not initialized)
149 | mock_repository.get_accounts.return_value = [account]
150 |
151 | result = await mcp_client.call_tool("list_accounts", {})
152 | response_data = extract_response_data(result)
153 |
154 | # Verify repository was called
155 | mock_repository.get_accounts.assert_called_once()
156 |
157 | # Verify account data was returned
158 | assert len(response_data["accounts"]) == 1
159 | assert response_data["accounts"][0]["id"] == "acc-1"
160 |
161 |
162 | async def test_list_accounts_account_types(
163 | mock_repository: MagicMock,
164 | mcp_client: Client[FastMCPTransport],
165 | ) -> None:
166 | """Test that different account types are handled correctly."""
167 | checking_account = create_ynab_account(
168 | id="acc-checking",
169 | name="My Checking",
170 | account_type=ynab.AccountType.CHECKING,
171 | on_budget=True,
172 | )
173 |
174 | savings_account = create_ynab_account(
175 | id="acc-savings",
176 | name="Emergency Fund",
177 | account_type=ynab.AccountType.SAVINGS,
178 | on_budget=True,
179 | )
180 |
181 | credit_card = create_ynab_account(
182 | id="acc-credit",
183 | name="Visa Card",
184 | account_type=ynab.AccountType.CREDITCARD,
185 | on_budget=True,
186 | )
187 |
188 | investment_account = create_ynab_account(
189 | id="acc-investment",
190 | name="401k",
191 | account_type=ynab.AccountType.OTHERASSET,
192 | on_budget=False, # Typically off-budget
193 | )
194 |
195 | mock_repository.get_accounts.return_value = [
196 | checking_account,
197 | savings_account,
198 | credit_card,
199 | investment_account,
200 | ]
201 |
202 | result = await mcp_client.call_tool("list_accounts", {})
203 | response_data = extract_response_data(result)
204 |
205 | # All account types should be included (none are closed)
206 | accounts = response_data["accounts"]
207 | assert len(accounts) == 4
208 |
209 | # Verify account types are preserved
210 | account_types = {acc["id"]: acc["type"] for acc in accounts}
211 | assert account_types["acc-checking"] == "checking"
212 | assert account_types["acc-savings"] == "savings"
213 | assert account_types["acc-credit"] == "creditCard"
214 | assert account_types["acc-investment"] == "otherAsset"
215 |
216 | # Verify on_budget status
217 | on_budget_status = {acc["id"]: acc["on_budget"] for acc in accounts}
218 | assert on_budget_status["acc-checking"] is True
219 | assert on_budget_status["acc-investment"] is False
220 |
221 |
222 | async def test_list_accounts_with_debt_fields(
223 | mock_repository: MagicMock,
224 | mcp_client: Client[FastMCPTransport],
225 | ) -> None:
226 | """Test that debt-related fields are properly included for debt accounts."""
227 | # Create a mortgage account with debt fields
228 | mortgage_account = create_ynab_account(
229 | id="acc-mortgage",
230 | name="Home Mortgage",
231 | account_type=ynab.AccountType.MORTGAGE,
232 | on_budget=False,
233 | balance=-250_000_000, # -$250,000 in milliunits
234 | debt_interest_rates={
235 | "2024-01-01": 3375,
236 | "2024-07-01": 3250,
237 | }, # 3.375%, 3.25% in milliunits
238 | debt_minimum_payments={
239 | "2024-01-01": 1500_000,
240 | "2024-07-01": 1450_000,
241 | }, # $1500, $1450 in milliunits
242 | debt_escrow_amounts={
243 | "2024-01-01": 300_000,
244 | "2024-07-01": 325_000,
245 | }, # $300, $325 in milliunits
246 | )
247 |
248 | # Create a credit card with empty debt fields
249 | credit_card = create_ynab_account(
250 | id="acc-credit",
251 | name="Visa Card",
252 | account_type=ynab.AccountType.CREDITCARD,
253 | on_budget=True,
254 | balance=-2500_000, # -$2,500 in milliunits
255 | debt_interest_rates={}, # Empty for credit cards
256 | debt_minimum_payments={},
257 | debt_escrow_amounts={},
258 | )
259 |
260 | # Create a regular checking account without debt fields
261 | checking_account = create_ynab_account(
262 | id="acc-checking",
263 | name="Checking",
264 | account_type=ynab.AccountType.CHECKING,
265 | balance=5000_000, # $5,000 in milliunits
266 | )
267 |
268 | mock_repository.get_accounts.return_value = [
269 | mortgage_account,
270 | credit_card,
271 | checking_account,
272 | ]
273 |
274 | result = await mcp_client.call_tool("list_accounts", {})
275 | response_data = extract_response_data(result)
276 |
277 | accounts = response_data["accounts"]
278 | assert len(accounts) == 3
279 |
280 | # Find mortgage account and verify debt fields
281 | mortgage = next(acc for acc in accounts if acc["id"] == "acc-mortgage")
282 | assert mortgage["debt_interest_rates"] == {
283 | "2024-01-01": "0.03375", # 3.375% as decimal
284 | "2024-07-01": "0.0325", # 3.25% as decimal
285 | }
286 | assert mortgage["debt_minimum_payments"] == {
287 | "2024-01-01": "1500",
288 | "2024-07-01": "1450",
289 | }
290 | assert mortgage["debt_escrow_amounts"] == {
291 | "2024-01-01": "300",
292 | "2024-07-01": "325",
293 | }
294 |
295 | # Verify credit card has null debt fields (empty dicts become None)
296 | credit = next(acc for acc in accounts if acc["id"] == "acc-credit")
297 | assert credit["debt_interest_rates"] is None
298 | assert credit["debt_minimum_payments"] is None
299 | assert credit["debt_escrow_amounts"] is None
300 |
301 | # Verify checking account has null debt fields
302 | checking = next(acc for acc in accounts if acc["id"] == "acc-checking")
303 | assert checking["debt_interest_rates"] is None
304 | assert checking["debt_minimum_payments"] is None
305 | assert checking["debt_escrow_amounts"] is None
306 |
```
--------------------------------------------------------------------------------
/tests/test_payees.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test suite for payee-related functionality in YNAB MCP Server.
3 | """
4 |
5 | from unittest.mock import MagicMock
6 |
7 | import ynab
8 | from assertions import extract_response_data
9 | from conftest import create_ynab_payee
10 | from fastmcp.client import Client, FastMCPTransport
11 |
12 |
13 | async def test_list_payees_success(
14 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
15 | ) -> None:
16 | """Test successful payee listing."""
17 |
18 | payee1 = create_ynab_payee(id="payee-1", name="Amazon")
19 | payee2 = create_ynab_payee(id="payee-2", name="Whole Foods")
20 |
21 | # Deleted payee should be excluded by default
22 | payee_deleted = create_ynab_payee(
23 | id="payee-deleted",
24 | name="Closed Store",
25 | deleted=True,
26 | )
27 |
28 | # Transfer payee
29 | payee_transfer = create_ynab_payee(
30 | id="payee-transfer",
31 | name="Transfer : Savings",
32 | transfer_account_id="acc-savings",
33 | )
34 |
35 | # Mock repository to return test payees
36 | mock_repository.get_payees.return_value = [
37 | payee2,
38 | payee1,
39 | payee_deleted,
40 | payee_transfer,
41 | ]
42 |
43 | result = await mcp_client.call_tool("list_payees", {})
44 | response_data = extract_response_data(result)
45 |
46 | # Should have 3 payees (deleted one excluded)
47 | assert len(response_data["payees"]) == 3
48 |
49 | # Should be sorted by name
50 | assert response_data["payees"][0]["name"] == "Amazon"
51 | assert response_data["payees"][1]["name"] == "Transfer : Savings"
52 | assert response_data["payees"][2]["name"] == "Whole Foods"
53 |
54 | # Check transfer payee details
55 | transfer_payee = response_data["payees"][1]
56 | assert transfer_payee["id"] == "payee-transfer"
57 |
58 | # Check pagination
59 | assert response_data["pagination"]["total_count"] == 3
60 | assert response_data["pagination"]["has_more"] is False
61 |
62 |
63 | async def test_list_payees_pagination(
64 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
65 | ) -> None:
66 | """Test payee listing with pagination."""
67 |
68 | # Create multiple payees
69 | payees = []
70 | for i in range(5):
71 | payee = ynab.Payee(
72 | id=f"payee-{i}",
73 | name=f"Store {i:02d}", # Store 00, Store 01, etc. for predictable sorting
74 | transfer_account_id=None,
75 | deleted=False,
76 | )
77 | payees.append(payee)
78 |
79 | mock_repository.get_payees.return_value = payees
80 |
81 | # Test first page
82 | result = await mcp_client.call_tool("list_payees", {"limit": 2, "offset": 0})
83 |
84 | response_data = extract_response_data(result)
85 | assert response_data is not None
86 | assert len(response_data["payees"]) == 2
87 | assert response_data["pagination"]["total_count"] == 5
88 | assert response_data["pagination"]["has_more"] is True
89 |
90 | # Should be sorted alphabetically
91 | assert response_data["payees"][0]["name"] == "Store 00"
92 | assert response_data["payees"][1]["name"] == "Store 01"
93 |
94 |
95 | async def test_list_payees_filters_deleted(
96 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
97 | ) -> None:
98 | """Test that list_payees automatically filters out deleted payees."""
99 |
100 | # Active payee (should be included)
101 | payee_active = ynab.Payee(
102 | id="payee-active",
103 | name="Active Store",
104 | transfer_account_id=None,
105 | deleted=False,
106 | )
107 |
108 | # Deleted payee (should be excluded)
109 | payee_deleted = ynab.Payee(
110 | id="payee-deleted",
111 | name="Deleted Store",
112 | transfer_account_id=None,
113 | deleted=True,
114 | )
115 |
116 | mock_repository.get_payees.return_value = [payee_active, payee_deleted]
117 |
118 | result = await mcp_client.call_tool("list_payees", {})
119 |
120 | response_data = extract_response_data(result)
121 | assert response_data is not None
122 | # Should only include the active payee
123 | assert len(response_data["payees"]) == 1
124 | assert response_data["payees"][0]["name"] == "Active Store"
125 | assert response_data["payees"][0]["id"] == "payee-active"
126 |
127 |
128 | async def test_find_payee_filters_deleted(
129 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
130 | ) -> None:
131 | """Test that find_payee automatically filters out deleted payees."""
132 |
133 | # Both payees have "amazon" in name, but one is deleted
134 | payee_active = ynab.Payee(
135 | id="payee-active", name="Amazon", transfer_account_id=None, deleted=False
136 | )
137 |
138 | payee_deleted = ynab.Payee(
139 | id="payee-deleted",
140 | name="Amazon Prime",
141 | transfer_account_id=None,
142 | deleted=True,
143 | )
144 |
145 | mock_repository.get_payees.return_value = [payee_active, payee_deleted]
146 |
147 | result = await mcp_client.call_tool("find_payee", {"name_search": "amazon"})
148 |
149 | response_data = extract_response_data(result)
150 | assert response_data is not None
151 | # Should only find the active Amazon payee, not the deleted one
152 | assert len(response_data["payees"]) == 1
153 | assert response_data["payees"][0]["name"] == "Amazon"
154 | assert response_data["payees"][0]["id"] == "payee-active"
155 |
156 |
157 | async def test_find_payee_success(
158 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
159 | ) -> None:
160 | """Test successful payee search by name."""
161 |
162 | # Create payees with different names for searching
163 | payees = [
164 | ynab.Payee(
165 | id="payee-amazon",
166 | name="Amazon",
167 | transfer_account_id=None,
168 | deleted=False,
169 | ),
170 | ynab.Payee(
171 | id="payee-amazon-web",
172 | name="Amazon Web Services",
173 | transfer_account_id=None,
174 | deleted=False,
175 | ),
176 | ynab.Payee(
177 | id="payee-starbucks",
178 | name="Starbucks",
179 | transfer_account_id=None,
180 | deleted=False,
181 | ),
182 | ynab.Payee(
183 | id="payee-grocery",
184 | name="Whole Foods Market",
185 | transfer_account_id=None,
186 | deleted=False,
187 | ),
188 | ynab.Payee(
189 | id="payee-deleted",
190 | name="Amazon Prime",
191 | transfer_account_id=None,
192 | deleted=True,
193 | ),
194 | ]
195 |
196 | mock_repository.get_payees.return_value = payees
197 |
198 | # Test searching for "amazon" (case-insensitive)
199 | result = await mcp_client.call_tool("find_payee", {"name_search": "amazon"})
200 |
201 | response_data = extract_response_data(result)
202 | assert response_data is not None
203 | # Should find Amazon and Amazon Web Services, but not deleted Amazon Prime
204 | assert len(response_data["payees"]) == 2
205 | assert response_data["pagination"]["total_count"] == 2
206 | assert response_data["pagination"]["has_more"] is False
207 |
208 | # Should be sorted alphabetically
209 | payee_names = [p["name"] for p in response_data["payees"]]
210 | assert payee_names == ["Amazon", "Amazon Web Services"]
211 |
212 |
213 | async def test_find_payee_case_insensitive(
214 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
215 | ) -> None:
216 | """Test that payee search is case-insensitive."""
217 |
218 | payees = [
219 | ynab.Payee(
220 | id="payee-1",
221 | name="Starbucks Coffee",
222 | transfer_account_id=None,
223 | deleted=False,
224 | )
225 | ]
226 |
227 | mock_repository.get_payees.return_value = payees
228 |
229 | # Test various case combinations
230 | search_terms_matches = [
231 | ("STARBUCKS", 1),
232 | ("starbucks", 1),
233 | ("StArBuCkS", 1),
234 | ("coffee", 1),
235 | ("COFFEE", 1),
236 | ("nonexistent", 0), # This will test the else branch
237 | ]
238 |
239 | for search_term, expected_count in search_terms_matches:
240 | result = await mcp_client.call_tool("find_payee", {"name_search": search_term})
241 |
242 | response_data = extract_response_data(result)
243 | assert len(response_data["payees"]) == expected_count
244 | if expected_count > 0:
245 | assert response_data["payees"][0]["name"] == "Starbucks Coffee"
246 |
247 |
248 | async def test_find_payee_limit(
249 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
250 | ) -> None:
251 | """Test payee search with limit parameter."""
252 |
253 | # Create multiple payees with "store" in the name
254 | payees = []
255 | for i in range(5):
256 | payees.append(
257 | ynab.Payee(
258 | id=f"payee-{i}",
259 | name=f"Store {i:02d}", # Store 00, Store 01, etc.
260 | transfer_account_id=None,
261 | deleted=False,
262 | )
263 | )
264 |
265 | mock_repository.get_payees.return_value = payees
266 |
267 | # Test with limit of 2
268 | result = await mcp_client.call_tool(
269 | "find_payee", {"name_search": "store", "limit": 2}
270 | )
271 |
272 | response_data = extract_response_data(result)
273 | assert response_data is not None
274 | assert len(response_data["payees"]) == 2
275 | assert response_data["pagination"]["total_count"] == 5
276 | assert response_data["pagination"]["has_more"] is True
277 |
278 | # Should be first 2 in alphabetical order
279 | assert response_data["payees"][0]["name"] == "Store 00"
280 | assert response_data["payees"][1]["name"] == "Store 01"
281 |
282 |
283 | async def test_find_payee_no_matches(
284 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
285 | ) -> None:
286 | """Test payee search with no matching results."""
287 |
288 | payees = [
289 | ynab.Payee(
290 | id="payee-1", name="Starbucks", transfer_account_id=None, deleted=False
291 | )
292 | ]
293 |
294 | mock_repository.get_payees.return_value = payees
295 |
296 | result = await mcp_client.call_tool("find_payee", {"name_search": "nonexistent"})
297 |
298 | response_data = extract_response_data(result)
299 | assert response_data is not None
300 | assert len(response_data["payees"]) == 0
301 | assert response_data["pagination"]["total_count"] == 0
302 | assert response_data["pagination"]["has_more"] is False
303 |
```
--------------------------------------------------------------------------------
/plans/caching.md:
--------------------------------------------------------------------------------
```markdown
1 | # YNAB Local Repository with Differential Sync
2 |
3 | ## Overview
4 |
5 | A local-first repository pattern that serves YNAB data from memory while using background differential sync to maintain consistency. The repository mirrors YNAB data locally, enabling instant reads without API latency.
6 |
7 | ## Core Design Principles
8 |
9 | 1. **Local-First**: All reads from in-memory repository, zero API calls during normal operation
10 | 2. **Differential Sync**: Use YNAB's `server_knowledge` to fetch only changes, not full datasets
11 | 3. **Repository Pattern**: Speaks only YNAB SDK models, no MCP-specific types
12 | 4. **Single Budget**: Server operates on one budget specified by `YNAB_BUDGET` env var
13 | 5. **Background Sync**: Updates happen out-of-band, never blocking MCP tool calls
14 |
15 | ## Key Constraints
16 |
17 | - **YNAB is source of truth**: Local repository is read-only mirror, never modifies data
18 | - **Eventually consistent**: Repository converges to YNAB state within sync interval
19 | - **Handle stale knowledge**: When server returns 409, fall back to full refresh
20 | - **Thread-safe**: Multiple MCP tools can read concurrently during sync
21 | - **Memory-only initially**: Start with dicts, defer persistence to later
22 |
23 | ## Repository Interface
24 |
25 | ```python
26 | class YNABRepository:
27 | """Local repository for YNAB data with background differential sync."""
28 |
29 | def __init__(self, budget_id: str, api_client_factory: Callable):
30 | self.budget_id = budget_id # Set once from YNAB_BUDGET env var
31 | self.api_client_factory = api_client_factory
32 |
33 | # In-memory storage - simple dicts
34 | self._data: dict[str, list] = {} # entity_type -> list of entities
35 | self._server_knowledge: dict[str, int] = {} # entity_type -> server_knowledge
36 | self._lock = threading.RLock()
37 | self._last_sync: datetime | None = None
38 |
39 | # Data access - returns YNAB SDK models directly
40 | def get_accounts(self) -> list[ynab.Account]:
41 | def get_categories(self) -> list[ynab.CategoryGroupWithCategories]:
42 | def get_transactions(self, since_date: date | None = None) -> list[ynab.TransactionDetail]:
43 | def get_payees(self) -> list[ynab.Payee]:
44 | def get_budget_month(self, month: date) -> ynab.MonthDetail:
45 |
46 | # Sync management
47 | def sync(self) -> None: # Fetch deltas and update repository
48 | def needs_sync(self) -> bool: # Check if sync is needed
49 | def last_sync_time(self) -> datetime | None:
50 | ```
51 |
52 | ## How Differential Sync Works
53 |
54 | ### Initial Load
55 | ```python
56 | # First call without last_knowledge_of_server
57 | response = api.get_accounts(budget_id)
58 | # Returns: all accounts + server_knowledge: 100
59 | self._data["accounts"] = response.data.accounts
60 | self._server_knowledge["accounts"] = response.data.server_knowledge
61 | ```
62 |
63 | ### Delta Sync
64 | ```python
65 | # Subsequent calls with last_knowledge_of_server
66 | response = api.get_accounts(budget_id, last_knowledge_of_server=100)
67 | # Returns: only changed accounts + server_knowledge: 101
68 | # Apply changes: add/update/remove based on response
69 | ```
70 |
71 | ### Applying Deltas
72 | ```python
73 | def apply_deltas(current: list, deltas: list) -> list:
74 | entity_map = {e.id: e for e in current}
75 |
76 | for delta in deltas:
77 | if delta.deleted:
78 | entity_map.pop(delta.id, None)
79 | else:
80 | entity_map[delta.id] = delta # Add or update
81 |
82 | return list(entity_map.values())
83 | ```
84 |
85 | ## Budget Configuration Change
86 |
87 | ### Environment Variables
88 | - **OLD**: `YNAB_DEFAULT_BUDGET` (optional, with fallback logic)
89 | - **NEW**: `YNAB_BUDGET` (required for server startup)
90 |
91 | The server will fail to start if `YNAB_BUDGET` is not set, making configuration explicit and removing ambiguity.
92 |
93 | ### Tool Signature Simplification
94 | Remove `budget_id` parameter from all MCP tools since the server operates on a single budget:
95 |
96 | ```python
97 | # Before
98 | @mcp.tool()
99 | def list_accounts(budget_id: str | None = None, limit: int = 100, offset: int = 0):
100 | budget_id = budget_id_or_default(budget_id)
101 | ...
102 |
103 | # After
104 | @mcp.tool()
105 | def list_accounts(limit: int = 100, offset: int = 0):
106 | # No budget_id needed - using server's configured budget
107 | ...
108 | ```
109 |
110 | This change applies to all tools: `list_accounts`, `list_categories`, `list_transactions`, `list_payees`, `get_budget_month`, etc.
111 |
112 | Additionally, the `list_budgets` tool becomes unnecessary and should be removed since the server operates on a single configured budget.
113 |
114 | ### MCP Instructions Update
115 | Simplify the MCP server instructions to remove budget_id complexity:
116 |
117 | ```python
118 | mcp = FastMCP[None](
119 | name="YNAB",
120 | instructions="""
121 | Access to your YNAB budget data including accounts, categories, and transactions.
122 | The server operates on the budget configured via YNAB_BUDGET environment variable.
123 | All data is served from a local repository that syncs with YNAB in the background.
124 | """
125 | )
126 | ```
127 |
128 | ## Integration with MCP Tools
129 |
130 | ### Current Pattern (Direct API with budget_id)
131 | ```python
132 | @mcp.tool()
133 | def list_accounts(budget_id: str | None = None):
134 | budget_id = budget_id_or_default(budget_id)
135 | with get_ynab_client() as api_client:
136 | accounts_api = ynab.AccountsApi(api_client)
137 | response = accounts_api.get_accounts(budget_id)
138 | # Process and return
139 | ```
140 |
141 | ### New Pattern (Repository without budget_id)
142 | ```python
143 | # Global repository instance for the configured budget
144 | _repository: YNABRepository | None = None
145 |
146 | def get_repository() -> YNABRepository:
147 | global _repository
148 | if _repository is None:
149 | budget_id = os.environ["YNAB_BUDGET"] # Required at startup
150 | _repository = YNABRepository(
151 | budget_id=budget_id,
152 | api_client_factory=get_ynab_client
153 | )
154 | # Initial sync to populate
155 | _repository.sync()
156 | return _repository
157 |
158 | @mcp.tool()
159 | def list_accounts(limit: int = 100, offset: int = 0): # No budget_id parameter
160 | repo = get_repository()
161 |
162 | # Trigger background sync if needed (non-blocking)
163 | if repo.needs_sync():
164 | threading.Thread(target=repo.sync).start()
165 |
166 | # Return data instantly from repository
167 | accounts = repo.get_accounts()
168 | # Apply existing filtering/pagination
169 | return process_accounts(accounts)
170 | ```
171 |
172 | ## Critical Implementation Details
173 |
174 | ### Entity Types to Sync
175 | - `accounts` - All accounts in budget
176 | - `categories` - Category groups with nested categories
177 | - `transactions` - Transaction history (consider date limits)
178 | - `payees` - All payees
179 | - `scheduled_transactions` - Scheduled/recurring transactions
180 | - `budget_months` - Month-specific budget data (current/last/next)
181 |
182 | ### Thread Safety
183 | - Use `threading.RLock()` for all repository data access
184 | - Sync updates entire entity list atomically
185 | - Reads can happen during sync (old data until sync completes)
186 |
187 | ### Error Handling
188 | - **Network failure**: Continue serving stale data, retry sync later
189 | - **409 (stale knowledge)**: Clear entity type, fetch all without last_knowledge
190 | - **429 (rate limit)**: YNAB allows 200 requests/hour per token. Use exponential backoff, track request count
191 | - **Invalid token**: Fail gracefully, log error, serve cached data
192 |
193 | ### Memory Management
194 | - Typical budget: ~1-5MB in memory
195 | - Consider transaction date limits (e.g., last 2 years only)
196 | - Clear old budget month data (keep current + last + next)
197 |
198 | ## Benefits
199 |
200 | - **Performance**: Sub-millisecond reads vs 100-500ms API calls
201 | - **Reliability**: Works offline, degrades gracefully
202 | - **Efficiency**: 60-80% fewer API calls after initial sync
203 | - **User Experience**: Instant responses in MCP tools
204 |
205 | ## Future Considerations
206 |
207 | - **Persistence**: SQLite for data survival across restarts
208 | - **Selective Sync**: Only sync entity types actually used
209 | - **Smart Scheduling**: Sync more frequently during business hours
210 | - **Multi-Budget**: Support switching between budgets efficiently
211 |
212 | ## Migration Notes
213 |
214 | ### ✅ Completed Breaking Changes
215 | 1. **Environment variable**: `YNAB_DEFAULT_BUDGET` → `YNAB_BUDGET` (now required) ✅
216 | 2. **Tool signatures**: Remove `budget_id` parameter from all tools ✅
217 | 3. **Tool removal**: Delete `list_budgets` tool entirely ✅
218 | 4. **Error handling**: Server fails to start without `YNAB_BUDGET` ✅
219 | 5. **Test infrastructure**: Updated with pytest-env for environment variable support ✅
220 |
221 | ### User Impact
222 | - Users must set `YNAB_BUDGET` before starting the server ✅
223 | - LLMs no longer need to handle budget selection logic ✅
224 | - Simpler, cleaner tool interfaces without optional budget_id parameters ✅
225 |
226 | ### Implementation Status
227 | - **Phase 0: Budget ID Removal** ✅ COMPLETED
228 | - All 57 tests passing with 100% coverage
229 | - Clean foundation ready for repository pattern implementation
230 |
231 | - **Phase 1: Repository Pattern** ✅ COMPLETED
232 | - ✅ YNABRepository class created with differential sync
233 | - ✅ Thread-safe data access with RLock
234 | - ✅ Delta application for add/update/delete operations
235 | - ✅ Lazy initialization per entity type
236 | - ✅ Server integration - all tools use repository
237 | - ✅ Background sync (non-blocking, triggered when data is stale)
238 | - ✅ needs_sync() method for staleness detection
239 | - ✅ Proper error handling (ConflictException, 429 rate limiting, fallback)
240 | - ✅ Initial population at server startup
241 | - ✅ Python logging with structured error handling
242 | - ✅ Test coverage migration (all 97 tests passing with 100% coverage)
243 |
244 | - **Phase 2: Test Quality Improvements** ✅ COMPLETED
245 | - ✅ Hoisted all inline imports to top of test files
246 | - ✅ Removed unhelpful comments that just repeated code
247 | - ✅ Fixed poor test patterns (replaced try/except: pass with pytest.raises)
248 | - ✅ Consolidated duplicate test helper functions into conftest.py
249 | - ✅ Eliminated code duplication across 8+ test files
250 | - ✅ Maintained 100% test coverage throughout cleanup
251 |
252 | ## Success Criteria
253 |
254 | 1. MCP tools never wait for API calls during normal operation
255 | 2. Repository stays synchronized within 5 minutes of YNAB changes
256 | 3. All existing MCP tool functionality works unchanged (except budget_id removal)
257 | 4. Memory usage stays under 10MB for typical budgets
258 | 5. Graceful degradation when YNAB API is unavailable
259 |
```
--------------------------------------------------------------------------------
/tests/test_updates.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for update functionality in YNAB MCP Server.
3 |
4 | Tests the update_category_budget and update_transaction tools.
5 | """
6 |
7 | from datetime import date
8 | from typing import Any
9 | from unittest.mock import MagicMock
10 |
11 | import ynab
12 | from assertions import extract_response_data
13 | from fastmcp.client import Client, FastMCPTransport
14 |
15 |
16 | def create_ynab_category(
17 | *,
18 | id: str = "cat-1",
19 | category_group_id: str = "group-1",
20 | budgeted: int = 100_000, # $100.00
21 | activity: int = -50_000, # -$50.00
22 | balance: int = 50_000, # $50.00
23 | **kwargs: Any,
24 | ) -> ynab.Category:
25 | """Create a YNAB Category for testing with sensible defaults."""
26 | return ynab.Category(
27 | id=id,
28 | category_group_id=category_group_id,
29 | category_group_name=kwargs.get("category_group_name", "Test Group"),
30 | name=kwargs.get("name", "Test Category"),
31 | hidden=kwargs.get("hidden", False),
32 | original_category_group_id=kwargs.get("original_category_group_id"),
33 | note=kwargs.get("note"),
34 | budgeted=budgeted,
35 | activity=activity,
36 | balance=balance,
37 | goal_type=kwargs.get("goal_type"),
38 | goal_needs_whole_amount=kwargs.get("goal_needs_whole_amount"),
39 | goal_day=kwargs.get("goal_day"),
40 | goal_cadence=kwargs.get("goal_cadence"),
41 | goal_cadence_frequency=kwargs.get("goal_cadence_frequency"),
42 | goal_creation_month=kwargs.get("goal_creation_month"),
43 | goal_target=kwargs.get("goal_target"),
44 | goal_target_month=kwargs.get("goal_target_month"),
45 | goal_percentage_complete=kwargs.get("goal_percentage_complete"),
46 | goal_months_to_budget=kwargs.get("goal_months_to_budget"),
47 | goal_under_funded=kwargs.get("goal_under_funded"),
48 | goal_overall_funded=kwargs.get("goal_overall_funded"),
49 | goal_overall_left=kwargs.get("goal_overall_left"),
50 | deleted=kwargs.get("deleted", False),
51 | )
52 |
53 |
54 | def create_ynab_transaction_detail(
55 | *,
56 | id: str = "txn-1",
57 | date: date = date(2024, 1, 15),
58 | amount: int = -50_000, # -$50.00
59 | account_id: str = "acc-1",
60 | **kwargs: Any,
61 | ) -> ynab.TransactionDetail:
62 | """Create a YNAB TransactionDetail for testing with sensible defaults."""
63 | return ynab.TransactionDetail(
64 | id=id,
65 | date=date,
66 | amount=amount,
67 | memo=kwargs.get("memo"),
68 | cleared=kwargs.get("cleared", ynab.TransactionClearedStatus.CLEARED),
69 | approved=kwargs.get("approved", True),
70 | flag_color=kwargs.get("flag_color"),
71 | account_id=account_id,
72 | account_name=kwargs.get("account_name", "Test Account"),
73 | payee_id=kwargs.get("payee_id"),
74 | payee_name=kwargs.get("payee_name"),
75 | category_id=kwargs.get("category_id"),
76 | category_name=kwargs.get("category_name"),
77 | transfer_account_id=kwargs.get("transfer_account_id"),
78 | transfer_transaction_id=kwargs.get("transfer_transaction_id"),
79 | matched_transaction_id=kwargs.get("matched_transaction_id"),
80 | import_id=kwargs.get("import_id"),
81 | import_payee_name=kwargs.get("import_payee_name"),
82 | import_payee_name_original=kwargs.get("import_payee_name_original"),
83 | debt_transaction_type=kwargs.get("debt_transaction_type"),
84 | deleted=kwargs.get("deleted", False),
85 | subtransactions=kwargs.get("subtransactions", []),
86 | )
87 |
88 |
89 | async def test_update_category_budget_success(
90 | mock_environment_variables: None,
91 | categories_api: MagicMock,
92 | mock_repository: MagicMock,
93 | mcp_client: Client[FastMCPTransport],
94 | ) -> None:
95 | """Test successful category budget update."""
96 |
97 | # Create the updated category that will be returned
98 | updated_category = create_ynab_category(
99 | id="cat-groceries",
100 | category_group_id="group-everyday",
101 | name="Groceries",
102 | budgeted=200_000, # $200.00 (new budgeted amount)
103 | activity=-150_000, # -$150.00
104 | balance=50_000, # $50.00
105 | )
106 |
107 | # Mock repository methods
108 | mock_repository.update_month_category.return_value = updated_category
109 |
110 | # Mock the categories response for group names
111 | category_group = ynab.CategoryGroupWithCategories(
112 | id="group-everyday",
113 | name="Everyday Expenses",
114 | hidden=False,
115 | deleted=False,
116 | categories=[updated_category],
117 | )
118 |
119 | # Mock repository to return category groups
120 | mock_repository.get_category_groups.return_value = [category_group]
121 |
122 | # Execute the tool
123 | result = await mcp_client.call_tool(
124 | "update_category_budget",
125 | {
126 | "category_id": "cat-groceries",
127 | "budgeted": "200.00",
128 | "month": "current",
129 | },
130 | )
131 |
132 | # Verify the response
133 | category_data = extract_response_data(result)
134 |
135 | assert category_data["id"] == "cat-groceries"
136 | assert category_data["name"] == "Groceries"
137 | assert category_data["category_group_name"] == "Everyday Expenses"
138 | assert category_data["budgeted"] == "200" # $200.00
139 | assert category_data["activity"] == "-150" # -$150.00
140 | assert category_data["balance"] == "50" # $50.00
141 |
142 | # Verify the repository was called correctly
143 | mock_repository.update_month_category.assert_called_once()
144 | call_args = mock_repository.update_month_category.call_args
145 | assert call_args[0][0] == "cat-groceries" # category_id
146 | assert call_args[0][1].year == 2025 # current month (from date.today())
147 | assert call_args[0][2] == 200_000 # budgeted_milliunits
148 |
149 |
150 | async def test_update_transaction_success(
151 | mock_environment_variables: None,
152 | mock_repository: MagicMock,
153 | mcp_client: Client[FastMCPTransport],
154 | ) -> None:
155 | """Test successful transaction update."""
156 |
157 | # Create the updated transaction that will be returned
158 | updated_transaction = create_ynab_transaction_detail(
159 | id="txn-123",
160 | date=date(2024, 1, 15),
161 | amount=-75_000, # -$75.00
162 | account_id="acc-checking",
163 | account_name="Checking",
164 | payee_id="payee-amazon",
165 | payee_name="Amazon",
166 | category_id="cat-household", # Updated category
167 | category_name="Household Items", # Updated category name
168 | memo="Amazon purchase - household items", # Updated memo
169 | cleared=ynab.TransactionClearedStatus.CLEARED,
170 | approved=True,
171 | )
172 |
173 | # Mock the existing transaction response (what we fetch before updating)
174 | original_transaction = create_ynab_transaction_detail(
175 | id="txn-123",
176 | date=date(2024, 1, 15),
177 | amount=-75_000, # -$75.00
178 | account_id="acc-checking",
179 | account_name="Checking",
180 | payee_id="payee-amazon",
181 | payee_name="Amazon",
182 | category_id="cat-food", # Original category
183 | category_name="Food", # Original category name
184 | memo="Amazon purchase", # Original memo
185 | cleared=ynab.TransactionClearedStatus.CLEARED,
186 | approved=True,
187 | )
188 |
189 | # Mock the API to return the transaction directly via the repository
190 | mock_repository.get_transaction_by_id.return_value = original_transaction
191 | mock_repository.update_transaction.return_value = updated_transaction
192 |
193 | # Execute the tool
194 | result = await mcp_client.call_tool(
195 | "update_transaction",
196 | {
197 | "transaction_id": "txn-123",
198 | "category_id": "cat-household",
199 | "memo": "Amazon purchase - household items",
200 | },
201 | )
202 |
203 | # Verify the response
204 | transaction_data = extract_response_data(result)
205 |
206 | assert transaction_data["id"] == "txn-123"
207 | assert transaction_data["amount"] == "-75" # -$75.00
208 | assert transaction_data["category_id"] == "cat-household"
209 | assert transaction_data["category_name"] == "Household Items"
210 | assert transaction_data["memo"] == "Amazon purchase - household items"
211 | assert transaction_data["cleared"] == "cleared"
212 |
213 | # Verify the repository was called correctly
214 | mock_repository.get_transaction_by_id.assert_called_once_with("txn-123")
215 | mock_repository.update_transaction.assert_called_once()
216 |
217 |
218 | async def test_update_category_budget_with_specific_month(
219 | mock_environment_variables: None,
220 | categories_api: MagicMock,
221 | mcp_client: Client[FastMCPTransport],
222 | ) -> None:
223 | """Test category budget update for a specific month."""
224 |
225 | updated_category = create_ynab_category(
226 | id="cat-dining",
227 | name="Dining Out",
228 | budgeted=150_000, # $150.00
229 | )
230 |
231 | save_response = ynab.SaveCategoryResponse(
232 | data=ynab.SaveCategoryResponseData(
233 | category=updated_category, server_knowledge=0
234 | )
235 | )
236 | categories_api.update_month_category.return_value = save_response
237 |
238 | # Mock categories response for group names
239 | category_group = ynab.CategoryGroupWithCategories(
240 | id="group-1",
241 | name="Fun Money",
242 | hidden=False,
243 | deleted=False,
244 | categories=[updated_category],
245 | )
246 | categories_response = ynab.CategoriesResponse(
247 | data=ynab.CategoriesResponseData(
248 | category_groups=[category_group], server_knowledge=0
249 | )
250 | )
251 | categories_api.get_categories.return_value = categories_response
252 |
253 | # Execute with specific date
254 | result = await mcp_client.call_tool(
255 | "update_category_budget",
256 | {
257 | "category_id": "cat-dining",
258 | "budgeted": "150.00",
259 | "month": "2024-03-01", # Specific month
260 | },
261 | )
262 |
263 | # Verify the response
264 | category_data = extract_response_data(result)
265 | assert category_data["budgeted"] == "150"
266 |
267 | # Verify correct month was passed to API
268 | call_args = categories_api.update_month_category.call_args[0]
269 | month_arg = call_args[1]
270 | assert month_arg == date(2024, 3, 1)
271 |
272 |
273 | async def test_update_transaction_minimal_fields(
274 | mock_environment_variables: None,
275 | mock_repository: MagicMock,
276 | mcp_client: Client[FastMCPTransport],
277 | ) -> None:
278 | """Test transaction update with only category change."""
279 |
280 | # Mock the existing transaction response (what we fetch before updating)
281 | original_transaction = create_ynab_transaction_detail(
282 | id="txn-456",
283 | category_id="cat-food",
284 | category_name="Food",
285 | )
286 |
287 | updated_transaction = create_ynab_transaction_detail(
288 | id="txn-456",
289 | category_id="cat-gas",
290 | category_name="Gas & Fuel",
291 | )
292 |
293 | # Mock the repository methods
294 | mock_repository.get_transaction_by_id.return_value = original_transaction
295 | mock_repository.update_transaction.return_value = updated_transaction
296 |
297 | # Execute with only category_id change
298 | result = await mcp_client.call_tool(
299 | "update_transaction",
300 | {
301 | "transaction_id": "txn-456",
302 | "category_id": "cat-gas",
303 | },
304 | )
305 |
306 | # Verify the response
307 | transaction_data = extract_response_data(result)
308 | assert transaction_data["category_id"] == "cat-gas"
309 |
310 | # Verify the repository was called correctly
311 | mock_repository.get_transaction_by_id.assert_called_once_with("txn-456")
312 | mock_repository.update_transaction.assert_called_once()
313 |
314 |
315 | async def test_update_transaction_with_payee(
316 | mock_environment_variables: None,
317 | mock_repository: MagicMock,
318 | mcp_client: Client[FastMCPTransport],
319 | ) -> None:
320 | """Test transaction update with payee_id to cover all branches."""
321 |
322 | # Mock the existing transaction response (what we fetch before updating)
323 | original_transaction = create_ynab_transaction_detail(
324 | id="txn-789",
325 | amount=-25_500, # -$25.50
326 | payee_id="payee-generic",
327 | payee_name="Generic Store",
328 | memo="Store purchase",
329 | )
330 |
331 | updated_transaction = create_ynab_transaction_detail(
332 | id="txn-789",
333 | amount=-25_500, # -$25.50
334 | payee_id="payee-starbucks",
335 | payee_name="Starbucks",
336 | memo="Coffee run",
337 | )
338 |
339 | # Mock the repository methods
340 | mock_repository.get_transaction_by_id.return_value = original_transaction
341 | mock_repository.update_transaction.return_value = updated_transaction
342 |
343 | # Execute with payee_id
344 | result = await mcp_client.call_tool(
345 | "update_transaction",
346 | {
347 | "transaction_id": "txn-789",
348 | "payee_id": "payee-starbucks",
349 | "memo": "Coffee run",
350 | },
351 | )
352 |
353 | # Verify the response
354 | transaction_data = extract_response_data(result)
355 | assert transaction_data["payee_id"] == "payee-starbucks"
356 | assert transaction_data["memo"] == "Coffee run"
357 | assert transaction_data["amount"] == "-25.5"
358 |
359 | # Verify the repository was called correctly
360 | mock_repository.get_transaction_by_id.assert_called_once_with("txn-789")
361 | mock_repository.update_transaction.assert_called_once()
362 |
```
--------------------------------------------------------------------------------
/DESIGN.md:
--------------------------------------------------------------------------------
```markdown
1 | # DESIGN.md
2 |
3 | This document outlines the driving use cases and design philosophy for the YNAB MCP Server, focused on helping heads of household manage family finances more effectively.
4 |
5 | ## Division of Responsibilities
6 |
7 | **MCP Server provides**:
8 | - Structured access to YNAB data with consistent formatting
9 | - Efficient filtering and search capabilities
10 | - Pagination for large datasets
11 | - Clean abstractions over YNAB's API complexity
12 |
13 | **LLM provides**:
14 | - Natural language understanding
15 | - Pattern recognition and analysis
16 | - Intelligent summarization
17 | - Actionable recommendations
18 | - Context-aware interpretation
19 |
20 | ## Use Case Format Guide
21 |
22 | Each use case follows this structure for clarity:
23 | - **User says**: Natural language examples
24 | - **Why it matters**: User context and pain points
25 | - **MCP Server Role**: Which tools to use and what data they provide
26 | - **LLM Role**: Analysis and intelligence needed
27 | - **Example Implementation Flow**: Step-by-step approach (when helpful)
28 | - **Edge Cases**: YNAB-specific gotchas to handle (when applicable)
29 |
30 | ## Core Use Cases
31 |
32 | ### 1. Weekly Family Finance Check-ins
33 | **User says**: "How are we doing on our budget this month? Which categories are we overspending?"
34 |
35 | **Why it matters**: Busy parents need quick weekly snapshots without opening YNAB. They want conversational summaries that highlight what needs attention.
36 |
37 | **MCP Server Role**:
38 | - `get_budget_month()` - Provides current month's budgeted amounts, activity, and balances
39 | - `list_categories()` - Gets category structure for organization
40 | - Returns clean, structured data with proper currency formatting
41 |
42 | **LLM Role**:
43 | - Interprets which categories are overspent/underspent
44 | - Prioritizes what needs attention
45 | - Suggests fund reallocation
46 | - Generates conversational summary
47 |
48 | **Example Implementation Flow**:
49 | 1. Call `get_budget_month()` to get current state
50 | 2. Identify categories where `balance < 0` (overspent)
51 | 3. Find categories with available funds (`balance > 0`)
52 | 4. Prioritize by spending velocity and days left in month
53 | 5. Format as conversational response with specific suggestions
54 |
55 | **Edge Cases**:
56 | - Handle credit card payment categories differently
57 | - Consider "Inflow: Ready to Assign" as special
58 | - Account for scheduled transactions not yet posted
59 |
60 | ### 2. Kid-Related Expense Tracking
61 | **User says**: "How much have we spent on soccer this year?" or "Show me all transactions for Emma's activities"
62 |
63 | **Why it matters**: Parents track expenses by child or activity for budgeting, tax deductions, or custody arrangements. Finding these across multiple categories and payees is tedious in YNAB.
64 |
65 | **MCP Server Role**:
66 | - `find_payee()` - Searches for "soccer", "dance academy", etc.
67 | - `list_transactions()` - Filters by payee_id, date ranges
68 | - Handles pagination for large transaction sets
69 | - Returns transaction details with amounts and dates
70 |
71 | **LLM Role**:
72 | - Identifies relevant search terms from natural language
73 | - Aggregates totals across multiple payees/categories
74 | - Groups related expenses
75 | - Formats results for specific use (taxes, custody docs)
76 |
77 | **Key Implementation Notes**:
78 | - May need multiple payee searches (e.g., "Soccer Club", "Soccer Store", "Soccer Camp")
79 | - Consider memo fields for additional context
80 | - Date ranges should align with tax year or custody period
81 |
82 | ### 3. Subscription and Recurring Expense Audits
83 | **User says**: "What subscriptions are we paying for?" or "List all our monthly recurring expenses"
84 |
85 | **Why it matters**: Subscription creep affects every family. Parents need to identify forgotten subscriptions and understand their true monthly commitments.
86 |
87 | **MCP Server Role**:
88 | - `list_transactions()` - Provides transaction history with dates
89 | - `list_payees()` - Gets payee details for merchant identification
90 | - Efficient pagination for analyzing patterns over time
91 |
92 | **LLM Role**:
93 | - Pattern recognition to identify recurring transactions
94 | - Frequency analysis (monthly, annual, etc.)
95 | - Grouping by merchant
96 | - Flagging unusual patterns or new subscriptions
97 |
98 | ### 4. Pre-Shopping Budget Checks
99 | **User says**: "Can we afford to spend $300 at Costco today?" or "How much grocery money do we have left?"
100 |
101 | **Why it matters**: Quick budget checks before shopping trips prevent overspending and the stress of moving money after the fact.
102 |
103 | **MCP Server Role**:
104 | - `get_budget_month()` - Current balances for relevant categories
105 | - `list_categories()` - Category relationships and groupings
106 | - Real-time accurate balance data
107 |
108 | **LLM Role**:
109 | - Maps "Costco shopping" to relevant categories (groceries, household, etc.)
110 | - Calculates total available across multiple categories
111 | - Suggests reallocation strategies
112 | - Provides go/no-go recommendation
113 |
114 | ### 5. Financial Partnership Transparency
115 | **User says**: "Give me a simple summary of our finances" or "Are we okay financially?" or "Did that Amazon return get credited?"
116 |
117 | **Why it matters**: In many couples, one person manages the detailed budget while their partner needs simple, reassuring visibility without YNAB complexity. Partners want to understand the big picture and verify specific transactions without learning budgeting software.
118 |
119 | **MCP Server Role**:
120 | - `get_budget_month()` - Overall budget health data
121 | - `list_accounts()` - Account balances for net worth
122 | - `list_transactions()` - Recent transaction verification
123 | - `find_payee()` - Quick lookup for specific merchants
124 |
125 | **LLM Role**:
126 | - Translates budget complexity into simple terms
127 | - Provides reassuring summaries ("Yes, you're on track")
128 | - Answers specific concerns without overwhelming detail
129 | - Bridges the knowledge gap between budget manager and partner
130 |
131 | ### 6. End-of-Month Category Sweep
132 | **User says**: "Which categories have money left over?" or "Help me zero out my budget"
133 |
134 | **Why it matters**: YNAB's zero-based budgeting requires monthly cleanup. Parents need quick identification of surplus funds and smart reallocation suggestions.
135 |
136 | **MCP Server Role**:
137 | - `get_budget_month()` - All category balances for current month
138 | - `list_category_groups()` - Organized view of budget structure
139 | - Accurate to-the-penny balance data
140 |
141 | **LLM Role**:
142 | - Identifies categories with positive balances
143 | - Analyzes historical spending to suggest reallocations
144 | - Prioritizes based on upcoming needs
145 | - Future: Generates reallocation transactions
146 |
147 | ### 7. Emergency Fund Reality Checks
148 | **User says**: "How many months could we survive on our emergency fund?" or "What's our true available emergency money?"
149 |
150 | **Why it matters**: Provides peace of mind by calculating realistic burn rates based on essential expenses and actual family spending patterns.
151 |
152 | **MCP Server Role**:
153 | - `list_accounts()` - Gets emergency fund account balances
154 | - `list_transactions()` - Historical spending data for analysis
155 | - `list_categories()` - Category structure for expense classification
156 |
157 | **LLM Role**:
158 | - Categorizes expenses as essential vs. discretionary
159 | - Calculates average monthly burn rate
160 | - Projects survival duration
161 | - Scenario modeling based on different assumptions
162 |
163 | ### 8. Financial Scenario Planning
164 | **User says**: "What if I get a 10% raise?" or "Can we afford a $400/month car payment?" or "What happens if we have another baby?"
165 |
166 | **Why it matters**: Families need to model major financial decisions before committing. They want to understand how changes in income, expenses, or family size would impact their budget without actually making changes in YNAB.
167 |
168 | **MCP Server Role**:
169 | - `get_budget_month()` - Current budget as baseline
170 | - `list_transactions()` - Historical spending patterns
171 | - `list_categories()` - Understanding fixed vs. variable expenses
172 | - `list_accounts()` - Current financial position
173 |
174 | **LLM Role**:
175 | - Models income changes across categories
176 | - Projects new expense impacts
177 | - Identifies categories that would need adjustment
178 | - Calculates how long until savings goals are met
179 | - Suggests budget reallocations for new scenarios
180 |
181 | **Future MCP Tools Needed**:
182 | - `create_budget_scenario()` - Clone budget for what-if analysis
183 | - `get_category_spending_history()` - Trends over time for better projections
184 |
185 | ### 9. AI-Assisted Payee Cleanup
186 | **User says**: "Help me clean up my payees" or "Find duplicate payees and suggest how to merge them" or "Which payees should be renamed for consistency?"
187 |
188 | **Why it matters**: YNAB imports create messy payee lists with variations like "AMZN MKTP US*M123", "Amazon.com", "AMAZON PRIME". Manual cleanup in YNAB's UI is tedious and time-consuming. Parents need consistent payee names for accurate spending analysis and budget reports.
189 |
190 | **MCP Server Role**:
191 | - `list_payees()` - Get all payees with transaction counts
192 | - `list_transactions()` - Analyze transaction patterns per payee
193 | - Returns payee names, IDs, and usage statistics
194 |
195 | **LLM Role**:
196 | - Pattern recognition to identify similar payees (fuzzy matching)
197 | - Groups variations of the same merchant (e.g., "Amazon", "AMZN", "Amazon Prime")
198 | - Suggests canonical names based on clarity and frequency
199 | - Identifies rarely-used payees that could be merged
200 | - Prioritizes high-impact cleanups (most transactions affected)
201 | - Generates step-by-step cleanup instructions for YNAB UI
202 |
203 | **Example Implementation Flow**:
204 | 1. Call `list_payees()` to get all payees with pagination
205 | 2. Use NLP/fuzzy matching to identify potential duplicate groups
206 | 3. For each group, analyze transaction patterns to confirm similarity
207 | 4. Suggest a canonical name (prefer clear, human-readable versions)
208 | 5. Calculate impact (number of transactions that would be affected)
209 | 6. Present recommendations ranked by impact with manual cleanup steps
210 |
211 | **Edge Cases**:
212 | - Transfer payees (contain account names) need special handling
213 | - Starting Balance and Manual Balance Adjustment are system payees
214 | - Credit card payment payees follow specific patterns
215 | - Venmo/PayPal transactions may need memo analysis for better grouping
216 | - Consider frequency of use when suggesting merge direction
217 |
218 | **SDK Limitations (as of 2025-06)**:
219 | - ❌ No payee merging/combining operations in YNAB API
220 | - ❌ No payee deletion capability
221 | - ❌ No bulk payee operations
222 | - ✅ Only individual payee renaming supported via `update_payee()`
223 |
224 | **Current Implementation Approach**:
225 | This use case is **advisory only** due to API limitations. The MCP server can identify duplicate patterns and suggest cleanup strategies, but users must manually execute merges in YNAB's web interface. A future `rename_payee()` tool could automate the renaming step after manual merging.
226 |
227 | ## Design Principles for Use Cases
228 |
229 | 1. **Conversational First**: Every query should feel natural to speak or type
230 | 2. **Context Aware**: Understand "we", "our", "the kids" in the context of a family
231 | 3. **Action Oriented**: Don't just report data, suggest next steps
232 | 4. **Time Sensitive**: Respect that parents are asking between activities
233 | 5. **Trust Building**: Be transparent about calculations and assumptions
234 |
235 | ## Future Use Case Directions
236 |
237 | ### Receipt/Transaction Quick Entry
238 | **User says**: "Add Costco $127.43 groceries and household"
239 |
240 | **Future MCP Tools Needed**:
241 | - `create_transaction()` - Add new transactions with splits
242 | - `get_recent_payees()` - Smart payee matching
243 | - `suggest_categories()` - Based on payee history
244 |
245 | ### Bill Reminders
246 | **User says**: "What bills are due this week?"
247 |
248 | **MCP Server Role**:
249 | - `list_scheduled_transactions()` - Future tool for recurring transactions
250 | - Current workaround: Analyze transaction history for patterns
251 |
252 | ### Import Assistance
253 | **User says**: "Help me categorize these Venmo transactions"
254 |
255 | **Future MCP Tools Needed**:
256 | - `import_transactions()` - Bulk import capability
257 | - `update_transaction()` - Modify imported transactions
258 | - `match_payees()` - Fuzzy matching for payee cleanup
259 |
260 | ## Tool Implementation Status
261 |
262 | ### Currently Implemented (11 tools)
263 | - ✅ `list_budgets()` - All use cases
264 | - ✅ `list_accounts()` - Emergency fund calculations
265 | - ✅ `list_categories()` - Budget structure understanding
266 | - ✅ `list_category_groups()` - Efficient category overview
267 | - ✅ `get_budget_month()` - Weekly check-ins, category sweep
268 | - ✅ `get_month_category_by_id()` - Specific category details
269 | - ✅ `list_transactions()` - Expense tracking, subscriptions, transparency
270 | - ✅ `list_payees()` - Payee analysis
271 | - ✅ `find_payee()` - Efficient payee search
272 | - ✅ `list_scheduled_transactions()` - Bill reminders, recurring expenses
273 |
274 | ### Recently Implemented
275 | - ✅ `list_scheduled_transactions()` - Bill reminders, recurring expenses
276 | - Supports all major use cases: subscription audits, bill reminders, recurring expense analysis
277 | - Comprehensive filtering: account, category, payee, frequency, upcoming days, amount range
278 | - Full pagination support following existing patterns
279 | - Consistent field naming with regular transactions using shared base model
280 | - 100% test coverage with extensive edge case testing
281 |
282 | ### Planned Tools (SDK-supported)
283 | - 🔄 `create_transaction()` - Quick entry
284 | - 🔄 `update_transaction()` - Import assistance
285 | - 🔄 `import_transactions()` - Bulk import
286 |
287 | ### Optimization Considerations
288 | The SDK offers specialized transaction endpoints that could optimize specific use cases:
289 | - `get_transactions_by_account()` - Direct account filtering
290 | - `get_transactions_by_category()` - Direct category filtering
291 | - `get_transactions_by_month()` - Month-specific queries
292 | - `get_transactions_by_payee()` - Direct payee filtering
293 |
294 | **Current approach**: Single `list_transactions()` with flexible filtering
295 | **Trade-offs**:
296 | - ✅ Simpler API surface for LLMs to learn
297 | - ✅ One tool handles all filtering combinations
298 | - ❌ Potentially less efficient for single-filter queries
299 | - ❌ May miss SDK-specific optimizations
300 |
301 | **Recommendation**: Keep the single `list_transactions()` approach because:
302 | 1. LLMs perform better with fewer, more flexible tools
303 | 2. Most use cases need multiple filters anyway (date + payee, category + amount)
304 | 3. The performance difference is negligible for household-scale data
305 | 4. Reduces tool discovery complexity for the LLM
306 |
307 | ### Creative Solutions Needed
308 | - 💡 Smart payee matching - Build on existing tools
309 | - 💡 Category suggestions - Analyze transaction history
310 | - 💡 Fuzzy payee matching - Custom logic required
311 |
312 | ## Success Metrics
313 |
314 | A use case is successful when:
315 | - It saves the user time vs. using YNAB directly
316 | - It provides insights not easily visible in YNAB's interface
317 | - It helps prevent financial stress or surprises
318 | - It works within the natural flow of family life
319 |
```
--------------------------------------------------------------------------------
/tests/test_budget_months.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test budget month and month category-related MCP tools.
3 | """
4 |
5 | from collections.abc import Generator
6 | from datetime import date
7 | from unittest.mock import MagicMock, Mock, patch
8 |
9 | import pytest
10 | import ynab
11 | from assertions import extract_response_data
12 | from fastmcp.client import Client, FastMCPTransport
13 |
14 |
15 | @pytest.fixture
16 | def months_api(ynab_client: MagicMock) -> Generator[MagicMock, None, None]:
17 | mock_api = Mock(spec=ynab.MonthsApi)
18 | with patch("ynab.MonthsApi", return_value=mock_api):
19 | yield mock_api
20 |
21 |
22 | async def test_get_budget_month_success(
23 | months_api: MagicMock,
24 | categories_api: MagicMock,
25 | mock_repository: MagicMock,
26 | mcp_client: Client[FastMCPTransport],
27 | ) -> None:
28 | """Test successful budget month retrieval."""
29 | category = ynab.Category(
30 | id="cat-1",
31 | category_group_id="group-1",
32 | category_group_name="Monthly Bills",
33 | name="Groceries",
34 | hidden=False,
35 | original_category_group_id=None,
36 | note="Food",
37 | budgeted=50000,
38 | activity=-30000,
39 | balance=20000,
40 | goal_type="TB",
41 | goal_needs_whole_amount=None,
42 | goal_day=None,
43 | goal_cadence=None,
44 | goal_cadence_frequency=None,
45 | goal_creation_month=None,
46 | goal_target=100000,
47 | goal_target_month=None,
48 | goal_percentage_complete=50,
49 | goal_months_to_budget=None,
50 | goal_under_funded=0,
51 | goal_overall_funded=None,
52 | goal_overall_left=None,
53 | deleted=False,
54 | )
55 |
56 | month = ynab.MonthDetail(
57 | month=date(2024, 1, 1),
58 | note="January budget",
59 | income=400000,
60 | budgeted=350000,
61 | activity=-200000,
62 | to_be_budgeted=50000,
63 | age_of_money=15,
64 | deleted=False,
65 | categories=[category],
66 | )
67 |
68 | # Mock repository methods
69 | mock_repository.get_budget_month.return_value = month
70 |
71 | # Mock the categories API call for getting group names
72 | category_group = ynab.CategoryGroupWithCategories(
73 | id="group-1",
74 | name="Monthly Bills",
75 | hidden=False,
76 | deleted=False,
77 | categories=[category],
78 | )
79 |
80 | # Mock repository to return category groups
81 | mock_repository.get_category_groups.return_value = [category_group]
82 |
83 | result = await mcp_client.call_tool("get_budget_month", {})
84 |
85 | response_data = extract_response_data(result)
86 | assert response_data["note"] == "January budget"
87 | assert len(response_data["categories"]) == 1
88 | assert response_data["categories"][0]["id"] == "cat-1"
89 | assert response_data["categories"][0]["category_group_name"] == "Monthly Bills"
90 |
91 |
92 | async def test_get_month_category_by_id_success(
93 | months_api: MagicMock,
94 | categories_api: MagicMock,
95 | mock_repository: MagicMock,
96 | mcp_client: Client[FastMCPTransport],
97 | ) -> None:
98 | """Test successful month category retrieval by ID."""
99 | mock_category = ynab.Category(
100 | id="cat-1",
101 | category_group_id="group-1",
102 | category_group_name="Monthly Bills",
103 | name="Groceries",
104 | hidden=False,
105 | original_category_group_id=None,
106 | note="Food",
107 | budgeted=50000,
108 | activity=-30000,
109 | balance=20000,
110 | goal_type="TB",
111 | goal_needs_whole_amount=None,
112 | goal_day=None,
113 | goal_cadence=None,
114 | goal_cadence_frequency=None,
115 | goal_creation_month=None,
116 | goal_target=100000,
117 | goal_target_month=None,
118 | goal_percentage_complete=50,
119 | goal_months_to_budget=None,
120 | goal_under_funded=0,
121 | goal_overall_funded=None,
122 | goal_overall_left=None,
123 | deleted=False,
124 | )
125 |
126 | # Mock repository method
127 | mock_repository.get_month_category_by_id.return_value = mock_category
128 |
129 | # Mock the categories API call for getting group names
130 | category_group = ynab.CategoryGroupWithCategories(
131 | id="group-1",
132 | name="Monthly Bills",
133 | hidden=False,
134 | deleted=False,
135 | categories=[mock_category],
136 | )
137 | # Mock repository to return category groups
138 | mock_repository.get_category_groups.return_value = [category_group]
139 |
140 | result = await mcp_client.call_tool(
141 | "get_month_category_by_id",
142 | {"category_id": "cat-1"},
143 | )
144 |
145 | response_data = extract_response_data(result)
146 | assert response_data["id"] == "cat-1"
147 | assert response_data["name"] == "Groceries"
148 | assert response_data["category_group_name"] == "Monthly Bills"
149 |
150 |
151 | async def test_get_month_category_by_id_default_budget(
152 | categories_api: MagicMock,
153 | mock_repository: MagicMock,
154 | mcp_client: Client[FastMCPTransport],
155 | ) -> None:
156 | """Test month category retrieval using default budget."""
157 | mock_category = ynab.Category(
158 | id="cat-2",
159 | category_group_id="group-2",
160 | category_group_name="Fun Money",
161 | name="Entertainment",
162 | hidden=False,
163 | original_category_group_id=None,
164 | note="Fun stuff",
165 | budgeted=25000,
166 | activity=-15000,
167 | balance=10000,
168 | goal_type=None,
169 | goal_needs_whole_amount=None,
170 | goal_day=None,
171 | goal_cadence=None,
172 | goal_cadence_frequency=None,
173 | goal_creation_month=None,
174 | goal_target=None,
175 | goal_target_month=None,
176 | goal_percentage_complete=None,
177 | goal_months_to_budget=None,
178 | goal_under_funded=None,
179 | goal_overall_funded=None,
180 | goal_overall_left=None,
181 | deleted=False,
182 | )
183 |
184 | # Mock repository method
185 | mock_repository.get_month_category_by_id.return_value = mock_category
186 |
187 | # Mock the categories API call for getting group names
188 | category_group = ynab.CategoryGroupWithCategories(
189 | id="group-2",
190 | name="Fun Money",
191 | hidden=False,
192 | deleted=False,
193 | categories=[mock_category],
194 | )
195 | # Mock repository to return category groups
196 | mock_repository.get_category_groups.return_value = [category_group]
197 |
198 | # Call without budget_id to test default
199 | result = await mcp_client.call_tool(
200 | "get_month_category_by_id", {"category_id": "cat-2"}
201 | )
202 |
203 | response_data = extract_response_data(result)
204 | assert response_data["id"] == "cat-2"
205 | assert response_data["name"] == "Entertainment"
206 | assert response_data["category_group_name"] == "Fun Money"
207 |
208 |
209 | async def test_get_month_category_by_id_no_groups(
210 | categories_api: MagicMock,
211 | mock_repository: MagicMock,
212 | mcp_client: Client[FastMCPTransport],
213 | ) -> None:
214 | """Test month category retrieval when no category groups exist."""
215 | mock_category = ynab.Category(
216 | id="cat-orphan",
217 | category_group_id="group-missing",
218 | category_group_name="Missing Group",
219 | name="Orphan Category",
220 | hidden=False,
221 | original_category_group_id=None,
222 | note="Category with no group",
223 | budgeted=10000,
224 | activity=-5000,
225 | balance=5000,
226 | goal_type=None,
227 | goal_needs_whole_amount=None,
228 | goal_day=None,
229 | goal_cadence=None,
230 | goal_cadence_frequency=None,
231 | goal_creation_month=None,
232 | goal_target=None,
233 | goal_target_month=None,
234 | goal_percentage_complete=None,
235 | goal_months_to_budget=None,
236 | goal_under_funded=None,
237 | goal_overall_funded=None,
238 | goal_overall_left=None,
239 | deleted=False,
240 | )
241 |
242 | # Mock repository method
243 | mock_repository.get_month_category_by_id.return_value = mock_category
244 |
245 | # Mock empty category groups response
246 | mock_repository.get_category_groups.return_value = []
247 |
248 | result = await mcp_client.call_tool(
249 | "get_month_category_by_id", {"category_id": "cat-orphan"}
250 | )
251 |
252 | response_data = extract_response_data(result)
253 | assert response_data["id"] == "cat-orphan"
254 | assert response_data["category_group_name"] is None
255 |
256 |
257 | async def test_get_month_category_by_id_category_not_in_groups(
258 | categories_api: MagicMock,
259 | mock_repository: MagicMock,
260 | mcp_client: Client[FastMCPTransport],
261 | ) -> None:
262 | """Test month category retrieval when category is not found in any group."""
263 | mock_category = ynab.Category(
264 | id="cat-notfound",
265 | category_group_id="group-old",
266 | category_group_name="Old Group",
267 | name="Not Found Category",
268 | hidden=False,
269 | original_category_group_id=None,
270 | note="Category not in groups",
271 | budgeted=5000,
272 | activity=-2000,
273 | balance=3000,
274 | goal_type=None,
275 | goal_needs_whole_amount=None,
276 | goal_day=None,
277 | goal_cadence=None,
278 | goal_cadence_frequency=None,
279 | goal_creation_month=None,
280 | goal_target=None,
281 | goal_target_month=None,
282 | goal_percentage_complete=None,
283 | goal_months_to_budget=None,
284 | goal_under_funded=None,
285 | goal_overall_funded=None,
286 | goal_overall_left=None,
287 | deleted=False,
288 | )
289 |
290 | # Create some other categories that don't match
291 | other_category1 = ynab.Category(
292 | id="cat-other1",
293 | category_group_id="group-1",
294 | category_group_name="Group 1",
295 | name="Other Category 1",
296 | hidden=False,
297 | original_category_group_id=None,
298 | note=None,
299 | budgeted=0,
300 | activity=0,
301 | balance=0,
302 | goal_type=None,
303 | goal_needs_whole_amount=None,
304 | goal_day=None,
305 | goal_cadence=None,
306 | goal_cadence_frequency=None,
307 | goal_creation_month=None,
308 | goal_target=None,
309 | goal_target_month=None,
310 | goal_percentage_complete=None,
311 | goal_months_to_budget=None,
312 | goal_under_funded=None,
313 | goal_overall_funded=None,
314 | goal_overall_left=None,
315 | deleted=False,
316 | )
317 |
318 | other_category2 = ynab.Category(
319 | id="cat-other2",
320 | category_group_id="group-2",
321 | category_group_name="Group 2",
322 | name="Other Category 2",
323 | hidden=False,
324 | original_category_group_id=None,
325 | note=None,
326 | budgeted=0,
327 | activity=0,
328 | balance=0,
329 | goal_type=None,
330 | goal_needs_whole_amount=None,
331 | goal_day=None,
332 | goal_cadence=None,
333 | goal_cadence_frequency=None,
334 | goal_creation_month=None,
335 | goal_target=None,
336 | goal_target_month=None,
337 | goal_percentage_complete=None,
338 | goal_months_to_budget=None,
339 | goal_under_funded=None,
340 | goal_overall_funded=None,
341 | goal_overall_left=None,
342 | deleted=False,
343 | )
344 |
345 | # Mock repository method
346 | mock_repository.get_month_category_by_id.return_value = mock_category
347 |
348 | # Mock category groups with categories that don't include our target
349 | category_group1 = ynab.CategoryGroupWithCategories(
350 | id="group-1",
351 | name="Group 1",
352 | hidden=False,
353 | deleted=False,
354 | categories=[other_category1],
355 | )
356 |
357 | category_group2 = ynab.CategoryGroupWithCategories(
358 | id="group-2",
359 | name="Group 2",
360 | hidden=False,
361 | deleted=False,
362 | categories=[other_category2],
363 | )
364 |
365 | # Add an empty category group to test the empty categories branch
366 | empty_group = ynab.CategoryGroupWithCategories(
367 | id="group-empty",
368 | name="Empty Group",
369 | hidden=False,
370 | deleted=False,
371 | categories=[],
372 | )
373 |
374 | # Mock repository to return category groups
375 | mock_repository.get_category_groups.return_value = [
376 | category_group1,
377 | empty_group,
378 | category_group2,
379 | ]
380 |
381 | result = await mcp_client.call_tool(
382 | "get_month_category_by_id", {"category_id": "cat-notfound"}
383 | )
384 |
385 | response_data = extract_response_data(result)
386 | assert response_data["id"] == "cat-notfound"
387 | assert response_data["category_group_name"] is None
388 |
389 |
390 | async def test_get_budget_month_with_default_budget(
391 | months_api: MagicMock,
392 | categories_api: MagicMock,
393 | mock_repository: MagicMock,
394 | mcp_client: Client[FastMCPTransport],
395 | ) -> None:
396 | """Test budget month retrieval with default budget."""
397 | category = ynab.Category(
398 | id="cat-default",
399 | category_group_id="group-default",
400 | category_group_name="Default Group",
401 | name="Default Category",
402 | hidden=False,
403 | original_category_group_id=None,
404 | note=None,
405 | budgeted=0,
406 | activity=0,
407 | balance=0,
408 | goal_type=None,
409 | goal_needs_whole_amount=None,
410 | goal_day=None,
411 | goal_cadence=None,
412 | goal_cadence_frequency=None,
413 | goal_creation_month=None,
414 | goal_target=None,
415 | goal_target_month=None,
416 | goal_percentage_complete=None,
417 | goal_months_to_budget=None,
418 | goal_under_funded=None,
419 | goal_overall_funded=None,
420 | goal_overall_left=None,
421 | deleted=False,
422 | )
423 |
424 | month = ynab.MonthDetail(
425 | month=date(2024, 2, 1),
426 | note=None,
427 | income=0,
428 | budgeted=0,
429 | activity=0,
430 | to_be_budgeted=0,
431 | age_of_money=None,
432 | deleted=False,
433 | categories=[category],
434 | )
435 |
436 | # Mock repository method
437 | mock_repository.get_budget_month.return_value = month
438 |
439 | # Mock the categories API call for getting group names
440 | category_group = ynab.CategoryGroupWithCategories(
441 | id="group-default",
442 | name="Default Group",
443 | hidden=False,
444 | deleted=False,
445 | categories=[category],
446 | )
447 | # Mock repository to return category groups
448 | mock_repository.get_category_groups.return_value = [category_group]
449 |
450 | # Call without budget_id to test default
451 | result = await mcp_client.call_tool("get_budget_month", {})
452 |
453 | response_data = extract_response_data(result)
454 | assert len(response_data["categories"]) == 1
455 | assert response_data["categories"][0]["id"] == "cat-default"
456 |
457 |
458 | async def test_get_budget_month_filters_deleted_and_hidden(
459 | months_api: MagicMock,
460 | categories_api: MagicMock,
461 | mock_repository: MagicMock,
462 | mcp_client: Client[FastMCPTransport],
463 | ) -> None:
464 | """Test that get_budget_month filters out deleted and hidden categories."""
465 | # Create active category
466 | active_category = ynab.Category(
467 | id="cat-active",
468 | category_group_id="group-1",
469 | category_group_name="Group 1",
470 | name="Active Category",
471 | hidden=False,
472 | original_category_group_id=None,
473 | note=None,
474 | budgeted=10000,
475 | activity=-5000,
476 | balance=5000,
477 | goal_type=None,
478 | goal_needs_whole_amount=None,
479 | goal_day=None,
480 | goal_cadence=None,
481 | goal_cadence_frequency=None,
482 | goal_creation_month=None,
483 | goal_target=None,
484 | goal_target_month=None,
485 | goal_percentage_complete=None,
486 | goal_months_to_budget=None,
487 | goal_under_funded=None,
488 | goal_overall_funded=None,
489 | goal_overall_left=None,
490 | deleted=False,
491 | )
492 |
493 | # Create deleted category (should be filtered out)
494 | deleted_category = ynab.Category(
495 | id="cat-deleted",
496 | category_group_id="group-1",
497 | category_group_name="Group 1",
498 | name="Deleted Category",
499 | hidden=False,
500 | original_category_group_id=None,
501 | note=None,
502 | budgeted=0,
503 | activity=0,
504 | balance=0,
505 | goal_type=None,
506 | goal_needs_whole_amount=None,
507 | goal_day=None,
508 | goal_cadence=None,
509 | goal_cadence_frequency=None,
510 | goal_creation_month=None,
511 | goal_target=None,
512 | goal_target_month=None,
513 | goal_percentage_complete=None,
514 | goal_months_to_budget=None,
515 | goal_under_funded=None,
516 | goal_overall_funded=None,
517 | goal_overall_left=None,
518 | deleted=True,
519 | )
520 |
521 | # Create hidden category (should be filtered out)
522 | hidden_category = ynab.Category(
523 | id="cat-hidden",
524 | category_group_id="group-1",
525 | category_group_name="Group 1",
526 | name="Hidden Category",
527 | hidden=True,
528 | original_category_group_id=None,
529 | note=None,
530 | budgeted=0,
531 | activity=0,
532 | balance=0,
533 | goal_type=None,
534 | goal_needs_whole_amount=None,
535 | goal_day=None,
536 | goal_cadence=None,
537 | goal_cadence_frequency=None,
538 | goal_creation_month=None,
539 | goal_target=None,
540 | goal_target_month=None,
541 | goal_percentage_complete=None,
542 | goal_months_to_budget=None,
543 | goal_under_funded=None,
544 | goal_overall_funded=None,
545 | goal_overall_left=None,
546 | deleted=False,
547 | )
548 |
549 | month = ynab.MonthDetail(
550 | month=date(2024, 1, 1),
551 | note=None,
552 | income=100000,
553 | budgeted=10000,
554 | activity=-5000,
555 | to_be_budgeted=95000,
556 | age_of_money=10,
557 | deleted=False,
558 | categories=[active_category, deleted_category, hidden_category],
559 | )
560 |
561 | # Mock repository method
562 | mock_repository.get_budget_month.return_value = month
563 |
564 | # Mock the categories API call for getting group names
565 | category_group = ynab.CategoryGroupWithCategories(
566 | id="group-1",
567 | name="Group 1",
568 | hidden=False,
569 | deleted=False,
570 | categories=[active_category, deleted_category, hidden_category],
571 | )
572 | # Mock repository to return category groups
573 | mock_repository.get_category_groups.return_value = [category_group]
574 |
575 | result = await mcp_client.call_tool("get_budget_month", {})
576 |
577 | response_data = extract_response_data(result)
578 | # Should only include the active category
579 | assert len(response_data["categories"]) == 1
580 | assert response_data["categories"][0]["id"] == "cat-active"
581 |
```
--------------------------------------------------------------------------------
/models.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Pydantic models for YNAB MCP Server responses.
3 |
4 | These models provide structured, well-documented data types for all YNAB API responses,
5 | including detailed explanations of YNAB's data model subtleties and conventions.
6 | """
7 |
8 | from __future__ import annotations
9 |
10 | import datetime
11 | from decimal import Decimal
12 | from typing import TYPE_CHECKING
13 |
14 | import ynab
15 | from pydantic import BaseModel, Field
16 |
17 | if TYPE_CHECKING: # pragma: no cover
18 | from repository import YNABRepository
19 |
20 |
21 | def milliunits_to_currency(milliunits: int, decimal_digits: int = 2) -> Decimal:
22 | """Convert YNAB milliunits to currency amount.
23 |
24 | YNAB uses milliunits where 1000 milliunits = 1 currency unit.
25 | """
26 | return Decimal(milliunits) / Decimal("1000")
27 |
28 |
29 | class PaginationInfo(BaseModel):
30 | """Pagination metadata for listing endpoints."""
31 |
32 | total_count: int = Field(..., description="Total number of items available")
33 | limit: int = Field(..., description="Maximum items per page")
34 | offset: int = Field(..., description="Number of items skipped")
35 | has_more: bool = Field(..., description="Whether more items are available")
36 |
37 |
38 | class Account(BaseModel):
39 | """A YNAB account with balance information.
40 |
41 | All amounts are in currency units with Decimal precision.
42 | """
43 |
44 | id: str = Field(..., description="Unique account identifier")
45 | name: str = Field(..., description="User-defined account name")
46 | type: str = Field(
47 | ...,
48 | description="Account type. Common values: 'checking', 'savings', 'creditCard', "
49 | "'cash', 'lineOfCredit', 'otherAsset', 'otherLiability', 'mortgage', "
50 | "'autoLoan', 'studentLoan'",
51 | )
52 | on_budget: bool = Field(
53 | ..., description="Whether this account is included in budget calculations"
54 | )
55 | closed: bool = Field(..., description="Whether this account has been closed")
56 | note: str | None = Field(None, description="User-defined account notes")
57 | balance: Decimal | None = Field(
58 | None, description="Current account balance in currency units"
59 | )
60 | cleared_balance: Decimal | None = Field(
61 | None, description="Balance of cleared transactions in currency units"
62 | )
63 | debt_interest_rates: dict[datetime.date, Decimal] | None = Field(
64 | None,
65 | description="Interest rates by date for debt accounts. Keys are dates, "
66 | "values are interest rates as decimals (e.g., 0.03375 for 3.375%)",
67 | )
68 | debt_minimum_payments: dict[datetime.date, Decimal] | None = Field(
69 | None,
70 | description="Minimum payment amounts by date for debt accounts. "
71 | "Keys are dates, values are payment amounts in currency units",
72 | )
73 | debt_escrow_amounts: dict[datetime.date, Decimal] | None = Field(
74 | None,
75 | description="Escrow amounts by date for mortgage accounts. Keys are dates, "
76 | "values are escrow amounts in currency units",
77 | )
78 |
79 | @classmethod
80 | def from_ynab(cls, account: ynab.Account) -> Account:
81 | """Convert YNAB Account object to our Account model."""
82 | # Convert debt interest rates from milliunits to decimal (e.g., 3375 -> 0.03375)
83 | debt_interest_rates = None
84 | if hasattr(account, "debt_interest_rates") and account.debt_interest_rates:
85 | debt_interest_rates = {
86 | datetime.date.fromisoformat(date_str): milliunits_to_currency(rate)
87 | / 100
88 | for date_str, rate in account.debt_interest_rates.items()
89 | }
90 |
91 | # Convert debt minimum payments from milliunits to currency
92 | debt_minimum_payments = None
93 | if hasattr(account, "debt_minimum_payments") and account.debt_minimum_payments:
94 | debt_minimum_payments = {
95 | datetime.date.fromisoformat(date_str): milliunits_to_currency(amount)
96 | for date_str, amount in account.debt_minimum_payments.items()
97 | }
98 |
99 | # Convert debt escrow amounts from milliunits to currency
100 | debt_escrow_amounts = None
101 | if hasattr(account, "debt_escrow_amounts") and account.debt_escrow_amounts:
102 | debt_escrow_amounts = {
103 | datetime.date.fromisoformat(date_str): milliunits_to_currency(amount)
104 | for date_str, amount in account.debt_escrow_amounts.items()
105 | }
106 |
107 | return cls(
108 | id=account.id,
109 | name=account.name,
110 | type=account.type,
111 | on_budget=account.on_budget,
112 | closed=account.closed,
113 | note=account.note,
114 | balance=milliunits_to_currency(account.balance)
115 | if account.balance is not None
116 | else None,
117 | cleared_balance=milliunits_to_currency(account.cleared_balance)
118 | if account.cleared_balance is not None
119 | else None,
120 | debt_interest_rates=debt_interest_rates,
121 | debt_minimum_payments=debt_minimum_payments,
122 | debt_escrow_amounts=debt_escrow_amounts,
123 | )
124 |
125 |
126 | class Category(BaseModel):
127 | """A YNAB category with budget and goal information."""
128 |
129 | id: str = Field(..., description="Unique category identifier")
130 | name: str = Field(..., description="Category name")
131 | category_group_id: str = Field(..., description="Category group ID")
132 | category_group_name: str | None = Field(None, description="Category group name")
133 | note: str | None = Field(None, description="Category notes")
134 | budgeted: Decimal | None = Field(None, description="Amount budgeted")
135 | activity: Decimal | None = Field(
136 | None,
137 | description="Spending activity (negative = spending)",
138 | )
139 | balance: Decimal | None = Field(None, description="Available balance")
140 | goal_type: str | None = Field(
141 | None,
142 | description="Goal type: NEED (refill up to X monthly - budget full target), "
143 | "TB (target balance by date), TBD (target by specific date), MF (funding)",
144 | )
145 | goal_target: Decimal | None = Field(None, description="Goal target amount")
146 | goal_percentage_complete: int | None = Field(
147 | None, description="Goal percentage complete"
148 | )
149 | goal_under_funded: Decimal | None = Field(
150 | None, description="Amount under-funded for goal"
151 | )
152 |
153 | @classmethod
154 | def from_ynab(
155 | cls, category: ynab.Category, category_group_name: str | None = None
156 | ) -> Category:
157 | """Convert YNAB Category object to our Category model.
158 |
159 | Args:
160 | category: The YNAB category object
161 | category_group_name: Optional category group name to include
162 | """
163 | return cls(
164 | id=category.id,
165 | name=category.name,
166 | category_group_id=category.category_group_id,
167 | category_group_name=category_group_name,
168 | note=category.note,
169 | budgeted=milliunits_to_currency(category.budgeted)
170 | if category.budgeted is not None
171 | else None,
172 | activity=milliunits_to_currency(category.activity)
173 | if category.activity is not None
174 | else None,
175 | balance=milliunits_to_currency(category.balance)
176 | if category.balance is not None
177 | else None,
178 | goal_type=category.goal_type,
179 | goal_target=milliunits_to_currency(category.goal_target)
180 | if category.goal_target is not None
181 | else None,
182 | goal_percentage_complete=category.goal_percentage_complete,
183 | goal_under_funded=milliunits_to_currency(category.goal_under_funded)
184 | if category.goal_under_funded is not None
185 | else None,
186 | )
187 |
188 |
189 | class CategoryGroup(BaseModel):
190 | """A YNAB category group with summary totals."""
191 |
192 | id: str = Field(..., description="Unique category group identifier")
193 | name: str = Field(..., description="Category group name")
194 | hidden: bool = Field(..., description="Whether hidden from budget view")
195 | category_count: int = Field(..., description="Number of categories in group")
196 | total_budgeted: Decimal | None = Field(None, description="Total budgeted amount")
197 | total_activity: Decimal | None = Field(None, description="Total activity")
198 | total_balance: Decimal | None = Field(None, description="Total balance")
199 |
200 | @classmethod
201 | def from_ynab(
202 | cls, category_group: ynab.CategoryGroupWithCategories
203 | ) -> CategoryGroup:
204 | """Convert YNAB CategoryGroup object to our CategoryGroup model.
205 |
206 | Calculates aggregated totals from active (non-deleted, non-hidden) categories.
207 | """
208 | # Calculate totals for the group (exclude deleted and hidden categories)
209 | active_categories = [
210 | cat
211 | for cat in category_group.categories
212 | if not cat.deleted and not cat.hidden
213 | ]
214 |
215 | total_budgeted = sum(cat.budgeted or 0 for cat in active_categories)
216 | total_activity = sum(cat.activity or 0 for cat in active_categories)
217 | total_balance = sum(cat.balance or 0 for cat in active_categories)
218 |
219 | return cls(
220 | id=category_group.id,
221 | name=category_group.name,
222 | hidden=category_group.hidden,
223 | category_count=len(active_categories),
224 | total_budgeted=milliunits_to_currency(total_budgeted),
225 | total_activity=milliunits_to_currency(total_activity),
226 | total_balance=milliunits_to_currency(total_balance),
227 | )
228 |
229 |
230 | class BudgetMonth(BaseModel):
231 | """Monthly budget summary with category details.
232 |
233 | Includes income, budgeted amounts, spending activity, and category breakdowns.
234 | """
235 |
236 | month: datetime.date | None = Field(None, description="Budget month date")
237 | note: str | None = Field(
238 | None, description="User-defined notes for this budget month"
239 | )
240 | income: Decimal | None = Field(
241 | None, description="Total income for the month in currency units"
242 | )
243 | budgeted: Decimal | None = Field(
244 | None, description="Total amount budgeted across all categories"
245 | )
246 | activity: Decimal | None = Field(
247 | None, description="Total spending activity for the month"
248 | )
249 | to_be_budgeted: Decimal | None = Field(
250 | None, description="Amount remaining to be budgeted (can be negative)"
251 | )
252 | age_of_money: int | None = Field(
253 | None,
254 | description="Age of money in days (how long money sits before being spent)",
255 | )
256 | categories: list[Category] = Field(
257 | ..., description="Categories with monthly budget data"
258 | )
259 | pagination: PaginationInfo | None = Field(
260 | None, description="Pagination information"
261 | )
262 |
263 |
264 | # Response models for tools that need pagination
265 | class AccountsResponse(BaseModel):
266 | """Response for list_accounts tool."""
267 |
268 | accounts: list[Account] = Field(..., description="List of accounts")
269 | pagination: PaginationInfo = Field(..., description="Pagination information")
270 |
271 |
272 | class CategoriesResponse(BaseModel):
273 | """Response for list_categories tool."""
274 |
275 | categories: list[Category] = Field(..., description="List of categories")
276 | pagination: PaginationInfo = Field(..., description="Pagination information")
277 |
278 |
279 | def format_flag(flag_color: str | None, flag_name: str | None) -> str | None:
280 | """Format flag as 'Name (Color)' or just color if no name."""
281 | if not flag_color:
282 | return None
283 | if flag_name:
284 | return f"{flag_name} ({flag_color.title()})"
285 | return flag_color.title()
286 |
287 |
288 | class BaseTransaction(BaseModel):
289 | """Base fields shared between Transaction and ScheduledTransaction models."""
290 |
291 | id: str = Field(..., description="Unique identifier")
292 | amount: Decimal | None = Field(
293 | None,
294 | description="Amount in currency units (negative = spending, positive = income)",
295 | )
296 | memo: str | None = Field(None, description="User-entered memo")
297 | flag: str | None = Field(
298 | None,
299 | description="Flag as 'Name (Color)' format",
300 | )
301 | account_id: str = Field(..., description="Account ID")
302 | account_name: str | None = Field(None, description="Account name")
303 | payee_id: str | None = Field(None, description="Payee ID")
304 | payee_name: str | None = Field(None, description="Payee name")
305 | category_id: str | None = Field(None, description="Category ID")
306 | category_name: str | None = Field(None, description="Category name")
307 |
308 |
309 | class Subtransaction(BaseModel):
310 | """A subtransaction within a split transaction."""
311 |
312 | id: str = Field(..., description="Unique subtransaction identifier")
313 | amount: Decimal | None = Field(None, description="Amount in currency units")
314 | memo: str | None = Field(None, description="Memo")
315 | payee_id: str | None = Field(None, description="Payee ID")
316 | payee_name: str | None = Field(None, description="Payee name")
317 | category_id: str | None = Field(None, description="Category ID")
318 | category_name: str | None = Field(None, description="Category name")
319 |
320 |
321 | class ScheduledSubtransaction(BaseModel):
322 | """A scheduled subtransaction within a split scheduled transaction."""
323 |
324 | id: str = Field(..., description="Unique scheduled subtransaction identifier")
325 | amount: Decimal | None = Field(None, description="Amount in currency units")
326 | memo: str | None = Field(None, description="Memo")
327 | payee_id: str | None = Field(None, description="Payee ID")
328 | payee_name: str | None = Field(None, description="Payee name")
329 | category_id: str | None = Field(None, description="Category ID")
330 | category_name: str | None = Field(None, description="Category name")
331 |
332 |
333 | class Transaction(BaseTransaction):
334 | """A YNAB transaction with full details."""
335 |
336 | date: datetime.date = Field(..., description="Transaction date")
337 | cleared: str = Field(..., description="Cleared status")
338 | approved: bool = Field(
339 | ...,
340 | description="Whether transaction is approved",
341 | )
342 | parent_transaction_id: str | None = Field(
343 | None, description="Parent transaction ID if this is a subtransaction"
344 | )
345 | subtransactions: list[Subtransaction] | None = Field(
346 | None, description="Subtransactions for splits"
347 | )
348 |
349 | @classmethod
350 | def from_ynab(
351 | cls,
352 | txn: ynab.TransactionDetail | ynab.HybridTransaction,
353 | repository: YNABRepository | None = None,
354 | ) -> Transaction:
355 | """Convert YNAB transaction object to our Transaction model.
356 |
357 | Args:
358 | txn: The YNAB transaction object
359 | repository: Optional repository to resolve parent transaction info
360 | """
361 | # Convert amount from milliunits
362 | amount = milliunits_to_currency(txn.amount)
363 |
364 | # Handle HybridTransaction subtransactions that need parent payee resolution
365 | payee_id = txn.payee_id
366 | payee_name = getattr(txn, "payee_name", None)
367 |
368 | # Check if this is a subtransaction that needs parent payee info
369 | if (
370 | hasattr(txn, "type")
371 | and txn.type == "subtransaction"
372 | and not payee_id
373 | and not payee_name
374 | and hasattr(txn, "parent_transaction_id")
375 | and txn.parent_transaction_id
376 | and repository
377 | ):
378 | parent_txn = repository.get_transaction_by_id(txn.parent_transaction_id)
379 | parent_payee_id = parent_txn.payee_id
380 | parent_payee_name = getattr(parent_txn, "payee_name", None)
381 | if parent_payee_id or parent_payee_name:
382 | payee_id = parent_payee_id
383 | payee_name = parent_payee_name
384 |
385 | # Handle subtransactions if present and available
386 | subtransactions = None
387 | if hasattr(txn, "subtransactions") and txn.subtransactions:
388 | subtransactions = []
389 | for sub in txn.subtransactions:
390 | if not sub.deleted:
391 | # Inherit parent payee info if subtransaction payee is null
392 | sub_payee_id = sub.payee_id if sub.payee_id else payee_id
393 | sub_payee_name = sub.payee_name if sub.payee_name else payee_name
394 |
395 | subtransactions.append(
396 | Subtransaction(
397 | id=sub.id,
398 | amount=milliunits_to_currency(sub.amount),
399 | memo=sub.memo,
400 | payee_id=sub_payee_id,
401 | payee_name=sub_payee_name,
402 | category_id=sub.category_id,
403 | category_name=sub.category_name,
404 | )
405 | )
406 |
407 | return cls(
408 | id=txn.id,
409 | date=txn.var_date,
410 | amount=amount,
411 | memo=txn.memo,
412 | cleared=txn.cleared,
413 | approved=txn.approved,
414 | flag=format_flag(txn.flag_color, getattr(txn, "flag_name", None)),
415 | account_id=txn.account_id,
416 | account_name=getattr(txn, "account_name", None),
417 | payee_id=payee_id,
418 | payee_name=payee_name,
419 | category_id=txn.category_id,
420 | category_name=getattr(txn, "category_name", None),
421 | parent_transaction_id=getattr(txn, "parent_transaction_id", None),
422 | subtransactions=subtransactions,
423 | )
424 |
425 |
426 | class ScheduledTransaction(BaseTransaction):
427 | """A YNAB scheduled transaction with frequency and timing details."""
428 |
429 | date_first: datetime.date = Field(..., description="First occurrence date")
430 | date_next: datetime.date = Field(..., description="Next occurrence date")
431 | frequency: str = Field(
432 | ...,
433 | description="Recurrence frequency",
434 | )
435 | subtransactions: list[ScheduledSubtransaction] | None = Field(
436 | None, description="Scheduled subtransactions for splits"
437 | )
438 |
439 | @classmethod
440 | def from_ynab(cls, st: ynab.ScheduledTransactionDetail) -> ScheduledTransaction:
441 | """Convert YNAB scheduled transaction to ScheduledTransaction model."""
442 | # Convert amount from milliunits
443 | amount = milliunits_to_currency(st.amount)
444 |
445 | # Handle scheduled subtransactions if present and available
446 | subtransactions = None
447 | if hasattr(st, "subtransactions") and st.subtransactions:
448 | subtransactions = []
449 | for sub in st.subtransactions:
450 | if not sub.deleted:
451 | subtransactions.append(
452 | ScheduledSubtransaction(
453 | id=sub.id,
454 | amount=milliunits_to_currency(sub.amount),
455 | memo=sub.memo,
456 | payee_id=sub.payee_id,
457 | payee_name=sub.payee_name,
458 | category_id=sub.category_id,
459 | category_name=sub.category_name,
460 | )
461 | )
462 |
463 | return cls(
464 | id=st.id,
465 | date_first=st.date_first,
466 | date_next=st.date_next,
467 | frequency=st.frequency,
468 | amount=amount,
469 | memo=st.memo,
470 | flag=format_flag(st.flag_color, getattr(st, "flag_name", None)),
471 | account_id=st.account_id,
472 | account_name=getattr(st, "account_name", None),
473 | payee_id=st.payee_id,
474 | payee_name=getattr(st, "payee_name", None),
475 | category_id=st.category_id,
476 | category_name=getattr(st, "category_name", None),
477 | subtransactions=subtransactions,
478 | )
479 |
480 |
481 | class TransactionsResponse(BaseModel):
482 | """Response for list_transactions tool."""
483 |
484 | transactions: list[Transaction] = Field(..., description="List of transactions")
485 | pagination: PaginationInfo = Field(..., description="Pagination information")
486 |
487 |
488 | class Payee(BaseModel):
489 | """A YNAB payee (person, company, or entity that receives payments)."""
490 |
491 | id: str = Field(..., description="Unique payee identifier")
492 | name: str = Field(..., description="Payee name")
493 |
494 | @classmethod
495 | def from_ynab(cls, payee: ynab.Payee) -> Payee:
496 | """Convert YNAB Payee object to our Payee model."""
497 | return cls(
498 | id=payee.id,
499 | name=payee.name,
500 | )
501 |
502 |
503 | class PayeesResponse(BaseModel):
504 | """Response for list_payees tool."""
505 |
506 | payees: list[Payee] = Field(..., description="List of payees")
507 | pagination: PaginationInfo = Field(..., description="Pagination information")
508 |
509 |
510 | class ScheduledTransactionsResponse(BaseModel):
511 | """Response for list_scheduled_transactions tool."""
512 |
513 | scheduled_transactions: list[ScheduledTransaction] = Field(
514 | ..., description="List of scheduled transactions"
515 | )
516 | pagination: PaginationInfo = Field(..., description="Pagination information")
517 |
```
--------------------------------------------------------------------------------
/repository.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | YNAB Repository with differential sync.
3 |
4 | Provides local-first access to YNAB data with background synchronization.
5 | """
6 |
7 | import logging
8 | import threading
9 | import time
10 | from collections.abc import Callable
11 | from datetime import date, datetime
12 | from typing import Any
13 |
14 | import ynab
15 | from ynab.exceptions import ApiException, ConflictException
16 |
17 | logger = logging.getLogger(__name__)
18 |
19 |
20 | class YNABRepository:
21 | """Local repository for YNAB data with background differential sync."""
22 |
23 | def __init__(self, budget_id: str, access_token: str):
24 | self.budget_id = budget_id
25 | self.configuration = ynab.Configuration(access_token=access_token)
26 |
27 | # In-memory storage - generic dict for different entity types
28 | self._data: dict[str, list[Any]] = {}
29 | self._server_knowledge: dict[str, int] = {}
30 | self._lock = threading.RLock()
31 | self._last_sync: datetime | None = None
32 |
33 | # Testing flag to disable background sync
34 | self._background_sync_enabled = True
35 |
36 | def get_accounts(self) -> list[ynab.Account]:
37 | """Get all accounts from local repository."""
38 | with self._lock:
39 | # If no data exists, do synchronous sync (first time)
40 | if "accounts" not in self._data:
41 | logger.info("No accounts data - performing initial sync")
42 | self.sync_accounts()
43 | # If data exists but is stale, trigger background sync
44 | elif self.needs_sync():
45 | logger.info("Accounts data is stale - triggering background sync")
46 | self._trigger_background_sync("accounts")
47 |
48 | return self._data.get("accounts", [])
49 |
50 | def get_payees(self) -> list[ynab.Payee]:
51 | """Get all payees from local repository."""
52 | with self._lock:
53 | # If no data exists, do synchronous sync (first time)
54 | if "payees" not in self._data:
55 | logger.info("No payees data - performing initial sync")
56 | self.sync_payees()
57 | # If data exists but is stale, trigger background sync
58 | elif self.needs_sync():
59 | logger.info("Payees data is stale - triggering background sync")
60 | self._trigger_background_sync("payees")
61 |
62 | return self._data.get("payees", [])
63 |
64 | def get_category_groups(self) -> list[ynab.CategoryGroupWithCategories]:
65 | """Get all category groups from local repository."""
66 | with self._lock:
67 | # If no data exists, do synchronous sync (first time)
68 | if "category_groups" not in self._data:
69 | logger.info("No category groups data - performing initial sync")
70 | self.sync_category_groups()
71 | # If data exists but is stale, trigger background sync
72 | elif self.needs_sync():
73 | logger.info(
74 | "Category groups data is stale - triggering background sync"
75 | )
76 | self._trigger_background_sync("category_groups")
77 |
78 | return self._data.get("category_groups", [])
79 |
80 | def get_transactions(self) -> list[ynab.TransactionDetail]:
81 | """Get all transactions from local repository."""
82 | with self._lock:
83 | # If no data exists, do synchronous sync (first time)
84 | if "transactions" not in self._data:
85 | logger.info("No transactions data - performing initial sync")
86 | self.sync_transactions()
87 | # If data exists but is stale, trigger background sync
88 | elif self.needs_sync():
89 | logger.info("Transactions data is stale - triggering background sync")
90 | self._trigger_background_sync("transactions")
91 |
92 | return self._data.get("transactions", [])
93 |
94 | def sync_accounts(self) -> None:
95 | """Sync accounts with YNAB API using differential sync."""
96 | self._sync_entity("accounts", self._sync_accounts_from_api)
97 |
98 | def sync_payees(self) -> None:
99 | """Sync payees with YNAB API using differential sync."""
100 | self._sync_entity("payees", self._sync_payees_from_api)
101 |
102 | def sync_category_groups(self) -> None:
103 | """Sync category groups with YNAB API using differential sync."""
104 | self._sync_entity("category_groups", self._sync_category_groups_from_api)
105 |
106 | def sync_transactions(self) -> None:
107 | """Sync transactions with YNAB API using differential sync."""
108 | self._sync_entity("transactions", self._sync_transactions_from_api)
109 |
110 | def _sync_accounts_from_api(
111 | self, last_knowledge: int | None
112 | ) -> tuple[list[ynab.Account], int]:
113 | """Fetch accounts from YNAB API with optional server knowledge."""
114 | with ynab.ApiClient(self.configuration) as api_client:
115 | accounts_api = ynab.AccountsApi(api_client)
116 |
117 | if last_knowledge is not None:
118 | try:
119 | # Try delta sync first
120 | response = self._handle_api_call_with_retry(
121 | lambda: accounts_api.get_accounts(
122 | self.budget_id, last_knowledge_of_server=last_knowledge
123 | )
124 | )
125 | except ConflictException as e:
126 | # Fall back to full sync on stale knowledge
127 | logger.info(
128 | f"Falling back to full accounts sync due to conflict: {e}"
129 | )
130 | response = self._handle_api_call_with_retry(
131 | lambda: accounts_api.get_accounts(self.budget_id)
132 | )
133 | except ApiException as e:
134 | # Log API error and fall back to full sync
135 | logger.warning(f"API error during accounts delta sync: {e}")
136 | response = self._handle_api_call_with_retry(
137 | lambda: accounts_api.get_accounts(self.budget_id)
138 | )
139 | except Exception as e:
140 | # Log unexpected error and re-raise
141 | logger.error(f"Unexpected error during accounts delta sync: {e}")
142 | raise
143 | else:
144 | response = self._handle_api_call_with_retry(
145 | lambda: accounts_api.get_accounts(self.budget_id)
146 | )
147 |
148 | return list(response.data.accounts), response.data.server_knowledge
149 |
150 | def _sync_payees_from_api(
151 | self, last_knowledge: int | None
152 | ) -> tuple[list[ynab.Payee], int]:
153 | """Fetch payees from YNAB API with optional server knowledge."""
154 | with ynab.ApiClient(self.configuration) as api_client:
155 | payees_api = ynab.PayeesApi(api_client)
156 |
157 | if last_knowledge is not None:
158 | try:
159 | # Try delta sync first
160 | response = self._handle_api_call_with_retry(
161 | lambda: payees_api.get_payees(
162 | self.budget_id, last_knowledge_of_server=last_knowledge
163 | )
164 | )
165 | except ConflictException as e:
166 | # Fall back to full sync on stale knowledge
167 | logger.info(
168 | f"Falling back to full payees sync due to conflict: {e}"
169 | )
170 | response = self._handle_api_call_with_retry(
171 | lambda: payees_api.get_payees(self.budget_id)
172 | )
173 | except ApiException as e:
174 | # Log API error and fall back to full sync
175 | logger.warning(f"API error during payees delta sync: {e}")
176 | response = self._handle_api_call_with_retry(
177 | lambda: payees_api.get_payees(self.budget_id)
178 | )
179 | except Exception as e:
180 | # Log unexpected error and re-raise
181 | logger.error(f"Unexpected error during payees delta sync: {e}")
182 | raise
183 | else:
184 | response = self._handle_api_call_with_retry(
185 | lambda: payees_api.get_payees(self.budget_id)
186 | )
187 |
188 | return list(response.data.payees), response.data.server_knowledge
189 |
190 | def _sync_category_groups_from_api(
191 | self, last_knowledge: int | None
192 | ) -> tuple[list[ynab.CategoryGroupWithCategories], int]:
193 | """Fetch category groups from YNAB API with optional server knowledge."""
194 | with ynab.ApiClient(self.configuration) as api_client:
195 | categories_api = ynab.CategoriesApi(api_client)
196 |
197 | if last_knowledge is not None:
198 | try:
199 | # Try delta sync first
200 | response = self._handle_api_call_with_retry(
201 | lambda: categories_api.get_categories(
202 | self.budget_id, last_knowledge_of_server=last_knowledge
203 | )
204 | )
205 | except ConflictException as e:
206 | # Fall back to full sync on stale knowledge
207 | logger.info(
208 | f"Category groups conflict, falling back to full sync: {e}"
209 | )
210 | response = self._handle_api_call_with_retry(
211 | lambda: categories_api.get_categories(self.budget_id)
212 | )
213 | except ApiException as e:
214 | # Log API error and fall back to full sync
215 | logger.warning(f"API error during category groups delta sync: {e}")
216 | response = self._handle_api_call_with_retry(
217 | lambda: categories_api.get_categories(self.budget_id)
218 | )
219 | except Exception as e:
220 | # Log unexpected error and re-raise
221 | logger.error(
222 | f"Unexpected error during category groups delta sync: {e}"
223 | )
224 | raise
225 | else:
226 | response = self._handle_api_call_with_retry(
227 | lambda: categories_api.get_categories(self.budget_id)
228 | )
229 |
230 | return list(response.data.category_groups), response.data.server_knowledge
231 |
232 | def _sync_transactions_from_api(
233 | self, last_knowledge: int | None
234 | ) -> tuple[list[ynab.TransactionDetail], int]:
235 | """Fetch transactions from YNAB API with optional server knowledge."""
236 | with ynab.ApiClient(self.configuration) as api_client:
237 | transactions_api = ynab.TransactionsApi(api_client)
238 |
239 | if last_knowledge is not None:
240 | try:
241 | # Try delta sync first
242 | response = self._handle_api_call_with_retry(
243 | lambda: transactions_api.get_transactions(
244 | self.budget_id, last_knowledge_of_server=last_knowledge
245 | )
246 | )
247 | except ConflictException as e:
248 | # Fall back to full sync on stale knowledge
249 | logger.info(
250 | f"Falling back to full transactions sync due to conflict: {e}"
251 | )
252 | response = self._handle_api_call_with_retry(
253 | lambda: transactions_api.get_transactions(self.budget_id)
254 | )
255 | except ApiException as e:
256 | # Log API error and fall back to full sync
257 | logger.warning(f"API error during transactions delta sync: {e}")
258 | response = self._handle_api_call_with_retry(
259 | lambda: transactions_api.get_transactions(self.budget_id)
260 | )
261 | except Exception as e:
262 | # Log unexpected error and re-raise
263 | logger.error(
264 | f"Unexpected error during transactions delta sync: {e}"
265 | )
266 | raise
267 | else:
268 | response = self._handle_api_call_with_retry(
269 | lambda: transactions_api.get_transactions(self.budget_id)
270 | )
271 |
272 | return list(response.data.transactions), response.data.server_knowledge
273 |
274 | def _sync_entity(
275 | self, entity_type: str, sync_func: Callable[[int | None], tuple[list[Any], int]]
276 | ) -> None:
277 | """Generic sync method for any entity type."""
278 | with self._lock:
279 | current_knowledge = self._server_knowledge.get(entity_type, 0)
280 | last_knowledge = current_knowledge if current_knowledge > 0 else None
281 |
282 | # Fetch from API
283 | entities, new_knowledge = sync_func(last_knowledge)
284 |
285 | # Apply changes
286 | if last_knowledge is not None and entity_type in self._data:
287 | # Apply delta changes
288 | self._apply_deltas(entity_type, entities)
289 | else:
290 | # Full refresh
291 | self._data[entity_type] = entities
292 |
293 | # Update metadata
294 | self._server_knowledge[entity_type] = new_knowledge
295 | self._last_sync = datetime.now()
296 |
297 | def _apply_deltas(self, entity_type: str, delta_entities: list[Any]) -> None:
298 | """Apply delta changes to an entity list."""
299 | current_entities = self._data.get(entity_type, [])
300 | entity_map = {entity.id: entity for entity in current_entities}
301 |
302 | for delta_entity in delta_entities:
303 | if hasattr(delta_entity, "deleted") and delta_entity.deleted:
304 | # Remove deleted entity
305 | entity_map.pop(delta_entity.id, None)
306 | else:
307 | # Add new or update existing entity
308 | entity_map[delta_entity.id] = delta_entity
309 |
310 | # Update the entity list
311 | self._data[entity_type] = list(entity_map.values())
312 |
313 | def is_initialized(self) -> bool:
314 | """Check if repository has been initially populated."""
315 | with self._lock:
316 | return len(self._data) > 0 or self._last_sync is not None
317 |
318 | def last_sync_time(self) -> datetime | None:
319 | """Get the last sync time."""
320 | with self._lock:
321 | return self._last_sync
322 |
323 | def needs_sync(self, max_age_minutes: int = 5) -> bool:
324 | """Check if repository needs to be synced based on staleness."""
325 | with self._lock:
326 | if self._last_sync is None:
327 | return True
328 |
329 | age_minutes = (datetime.now() - self._last_sync).total_seconds() / 60
330 | return age_minutes > max_age_minutes
331 |
332 | def _trigger_background_sync(self, entity_type: str) -> None:
333 | """Trigger background sync for a specific entity type."""
334 | if not self._background_sync_enabled:
335 | return
336 |
337 | sync_method = {
338 | "accounts": self.sync_accounts,
339 | "payees": self.sync_payees,
340 | "category_groups": self.sync_category_groups,
341 | "transactions": self.sync_transactions,
342 | }.get(entity_type)
343 |
344 | if sync_method:
345 | sync_thread = threading.Thread(
346 | target=self._background_sync_entity,
347 | args=(entity_type, sync_method),
348 | daemon=True,
349 | name=f"ynab-sync-{entity_type}",
350 | )
351 | sync_thread.start()
352 |
353 | def _background_sync_entity(
354 | self, entity_type: str, sync_method: Callable[[], None]
355 | ) -> None:
356 | """Background sync for a specific entity type with error handling."""
357 | try:
358 | logger.info(f"Starting background sync for {entity_type}")
359 | sync_method()
360 | logger.info(f"Completed background sync for {entity_type}")
361 | except Exception as e:
362 | logger.error(f"Background sync failed for {entity_type}: {e}")
363 | # Continue serving stale data on error
364 |
365 | def _handle_api_call_with_retry(
366 | self, api_call: Callable[[], Any], max_retries: int = 3
367 | ) -> Any:
368 | """Handle API call with exponential backoff for rate limiting."""
369 | for attempt in range(max_retries):
370 | try:
371 | return api_call()
372 | except ConflictException:
373 | # Let the calling method handle ConflictException for fallback logic
374 | raise
375 | except ApiException as e:
376 | if e.status == 429:
377 | # Rate limited - YNAB allows 200 requests/hour
378 | wait_time = 2**attempt
379 | logger.warning(
380 | f"Rate limited - waiting {wait_time}s (retry {attempt + 1})"
381 | )
382 | if attempt < max_retries - 1:
383 | time.sleep(wait_time)
384 | continue
385 | else:
386 | logger.error("Max retries exceeded for rate limiting")
387 | raise
388 | else:
389 | # Other API error - don't retry, let caller handle
390 | logger.error(f"API error {e.status}: {e}")
391 | raise
392 | except Exception as e:
393 | logger.error(f"Unexpected error during API call: {e}")
394 | raise
395 |
396 | def update_month_category(
397 | self, category_id: str, month: date, budgeted_milliunits: int
398 | ) -> ynab.Category:
399 | """Update a category's budget for a specific month."""
400 | with ynab.ApiClient(self.configuration) as api_client:
401 | categories_api = ynab.CategoriesApi(api_client)
402 |
403 | save_month_category = ynab.SaveMonthCategory(budgeted=budgeted_milliunits)
404 | patch_wrapper = ynab.PatchMonthCategoryWrapper(category=save_month_category)
405 |
406 | response = categories_api.update_month_category(
407 | self.budget_id, month, category_id, patch_wrapper
408 | )
409 |
410 | # Invalidate category groups cache since budget amounts changed
411 | with self._lock:
412 | if "category_groups" in self._data:
413 | del self._data["category_groups"]
414 | if "category_groups" in self._server_knowledge:
415 | del self._server_knowledge["category_groups"]
416 |
417 | return response.data.category
418 |
419 | def update_transaction(
420 | self, transaction_id: str, update_data: dict[str, Any]
421 | ) -> ynab.TransactionDetail:
422 | """Update a transaction with the provided data."""
423 | with ynab.ApiClient(self.configuration) as api_client:
424 | transactions_api = ynab.TransactionsApi(api_client)
425 |
426 | # Create the save transaction object
427 | existing_transaction = ynab.ExistingTransaction(**update_data)
428 | put_wrapper = ynab.PutTransactionWrapper(transaction=existing_transaction)
429 |
430 | response = transactions_api.update_transaction(
431 | self.budget_id, transaction_id, put_wrapper
432 | )
433 |
434 | # Invalidate transactions cache since transaction was modified
435 | with self._lock:
436 | if "transactions" in self._data:
437 | del self._data["transactions"]
438 | if "transactions" in self._server_knowledge:
439 | del self._server_knowledge["transactions"]
440 |
441 | return response.data.transaction
442 |
443 | def get_transaction_by_id(self, transaction_id: str) -> ynab.TransactionDetail:
444 | """Get a specific transaction by ID."""
445 | with ynab.ApiClient(self.configuration) as api_client:
446 | transactions_api = ynab.TransactionsApi(api_client)
447 | response = transactions_api.get_transaction_by_id(
448 | self.budget_id, transaction_id
449 | )
450 | return response.data.transaction
451 |
452 | def get_transactions_by_filters(
453 | self,
454 | account_id: str | None = None,
455 | category_id: str | None = None,
456 | payee_id: str | None = None,
457 | since_date: date | None = None,
458 | ) -> list[ynab.TransactionDetail | ynab.HybridTransaction]:
459 | """Get transactions using specific YNAB API endpoints for filtering."""
460 | with ynab.ApiClient(self.configuration) as api_client:
461 | transactions_api = ynab.TransactionsApi(api_client)
462 |
463 | if account_id:
464 | account_response = transactions_api.get_transactions_by_account(
465 | self.budget_id, account_id, since_date=since_date, type=None
466 | )
467 | return list(account_response.data.transactions)
468 | elif category_id:
469 | category_response = transactions_api.get_transactions_by_category(
470 | self.budget_id, category_id, since_date=since_date, type=None
471 | )
472 | return list(category_response.data.transactions)
473 | elif payee_id:
474 | payee_response = transactions_api.get_transactions_by_payee(
475 | self.budget_id, payee_id, since_date=since_date, type=None
476 | )
477 | return list(payee_response.data.transactions)
478 | else:
479 | # Use general transactions endpoint
480 | general_response = transactions_api.get_transactions(
481 | self.budget_id, since_date=since_date, type=None
482 | )
483 | return list(general_response.data.transactions)
484 |
485 | def get_scheduled_transactions(self) -> list[ynab.ScheduledTransactionDetail]:
486 | """Get scheduled transactions."""
487 | with ynab.ApiClient(self.configuration) as api_client:
488 | scheduled_transactions_api = ynab.ScheduledTransactionsApi(api_client)
489 | response = scheduled_transactions_api.get_scheduled_transactions(
490 | self.budget_id
491 | )
492 | return list(response.data.scheduled_transactions)
493 |
494 | def get_month_category_by_id(self, month: date, category_id: str) -> ynab.Category:
495 | """Get a specific category for a specific month."""
496 | with ynab.ApiClient(self.configuration) as api_client:
497 | categories_api = ynab.CategoriesApi(api_client)
498 | response = categories_api.get_month_category_by_id(
499 | self.budget_id, month, category_id
500 | )
501 | return response.data.category
502 |
503 | def get_budget_month(self, month: date) -> ynab.MonthDetail:
504 | """Get budget month data for a specific month."""
505 | with ynab.ApiClient(self.configuration) as api_client:
506 | months_api = ynab.MonthsApi(api_client)
507 | response = months_api.get_budget_month(self.budget_id, month)
508 | return response.data.month
509 |
```