This is page 1 of 2. Use http://codebase.md/chrisguidry/you-need-an-mcp?lines=false&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:
--------------------------------------------------------------------------------
```
3.12.8
```
--------------------------------------------------------------------------------
/.envrc:
--------------------------------------------------------------------------------
```
if [ -f .venv/bin/activate ]; then
source .venv/bin/activate
fi
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
.venv
.coverage
.python-version
```
--------------------------------------------------------------------------------
/.pre-commit-config.yaml:
--------------------------------------------------------------------------------
```yaml
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-merge-conflict
- id: check-yaml
- id: check-toml
- id: check-added-large-files
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.12.0
hooks:
- id: ruff-format
- id: ruff
args: [--fix, --exit-non-zero-on-fix]
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.16.1
hooks:
- id: mypy
additional_dependencies:
- types-requests
- pytest
- fastmcp
- pydantic
- ynab
args: [--strict, --show-error-codes]
files: ^(server\.py|models\.py|tests/.+\.py)$
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# you-need-an-mcp
An MCP server providing LLMs access to a YNAB budget.
## Setup
### 1. Install Dependencies
```bash
uv sync
```
### 2. Get YNAB Access Token
To use this MCP server, you need a YNAB Personal Access Token:
1. Log into your YNAB account at https://app.youneedabudget.com
2. Go to **Account Settings** (click your email in the top right corner)
3. Click on **Developer Settings** in the left sidebar
4. Click **New Token**
5. Enter a token name (e.g., "MCP Server")
6. Click **Generate**
7. Copy the generated token (you won't be able to see it again)
### 3. Set Environment Variables
```bash
export YNAB_ACCESS_TOKEN=your_token_here
```
Optionally, set a default budget ID to avoid having to specify it in every call:
```bash
export YNAB_DEFAULT_BUDGET=your_budget_id_here
```
### 4. Run the Server
```bash
uv run python server.py
```
## Available Tools
- `list_budgets()` - Returns all your YNAB budgets
- `list_accounts(budget_id=None, limit=100, offset=0, include_closed=False)` - Returns accounts with pagination and filtering
- `list_categories(budget_id=None, limit=50, offset=0, include_hidden=False)` - Returns categories with pagination and filtering
- `list_category_groups(budget_id=None)` - Returns category groups with totals (lighter weight overview)
### Pagination
The `list_accounts` and `list_categories` tools support pagination. Use the `offset` parameter to get subsequent pages:
- First page: `list_categories(limit=50, offset=0)`
- Second page: `list_categories(limit=50, offset=50)`
- Check `pagination.has_more` to see if there are more results
## Security Note
Keep your YNAB access token secure and never commit it to version control. The token provides read access to all your budget data.
```
--------------------------------------------------------------------------------
/CLAUDE.md:
--------------------------------------------------------------------------------
```markdown
# CLAUDE.md
## Project Overview
MCP (Model Context Protocol) server for YNAB (You Need A Budget) using FastMCP. Provides structured access to YNAB financial data.
## Essential Commands
```bash
# Development
uv sync --group dev # Install all dependencies
pytest --cov=server --cov=models --cov-report=term-missing # Run tests with coverage
fastmcp run server.py:mcp # Run MCP server
# Authentication (required)
export YNAB_ACCESS_TOKEN=your_token_here
export YNAB_BUDGET=your_budget_id_here # Required
```
## Critical Design Principles
1. **100% test coverage is mandatory** - No exceptions. All new code must have complete coverage.
2. **All listing tools must implement pagination** - Use `limit` and `offset` parameters.
3. **Automatically filter out deleted/hidden/closed data** - Only show active, relevant data.
4. **Use real YNAB SDK models in tests** - Mock only the API calls, not the data structures.
5. **Handle milliunits properly** - 1000 milliunits = 1 currency unit.
6. **Try to avoid using abbreviations in names**
## Product Philosophy
- **Household finance assistant** - Help heads of household be more effective with YNAB
- **Insights and notifications** - Surface important financial patterns and alerts
- **Safe evolution** - Currently read-only, will add careful mutations (transactions, imports)
- **Natural language friendly** - Enable text-based transaction entry and import assistance
- **User-friendly defaults** - Sensible limits, current month defaults, active data only
- **Performance conscious** - Pagination prevents token overflow, efficient payee search
## Security & Privacy
- **Never log financial amounts or account numbers** - Use debug logs carefully
- **Sanitize error messages** - Don't expose internal IDs or sensitive details
- **Token safety** - Never commit or expose YNAB access tokens
- **Fail securely** - Return generic errors for auth failures
## Tool Documentation
**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.
## Testing Guidelines
- Always run the full test suite after changes: `uv run pytest`
- Use FastMCP's testing pattern with direct client-server testing
- Mock YNAB API calls, but use real YNAB model instances
- Verify pagination works correctly for all listing endpoints
- Prefer not to use test classes, just simple test functions. Test organization should come through new files
## Architecture
- `server.py` - MCP server implementation using `@mcp.tool()` decorators
- `models.py` - Pydantic models matching YNAB's data structures
- `DESIGN.md` - Detailed use cases and design philosophy
- Uses context managers for YNAB client lifecycle
- Returns structured JSON with consistent pagination format
- Handle YNAB API errors gracefully with user-friendly messages
```
--------------------------------------------------------------------------------
/tests/test_assertions.py:
--------------------------------------------------------------------------------
```python
"""Test assertion helpers."""
import pytest
from assertions import extract_response_data
def test_extract_response_data_invalid_type() -> None:
"""Test that extract_response_data raises TypeError for invalid input."""
with pytest.raises(TypeError, match="Expected CallToolResult with content"):
extract_response_data("invalid_input")
def test_extract_response_data_invalid_list() -> None:
"""Test that extract_response_data raises TypeError for old list format."""
with pytest.raises(TypeError, match="Expected CallToolResult with content"):
extract_response_data([])
```
--------------------------------------------------------------------------------
/tests/assertions.py:
--------------------------------------------------------------------------------
```python
"""
Test assertion helpers for YNAB MCP Server tests.
This module provides helper functions for common test assertions and response
parsing to reduce boilerplate in test files.
"""
import json
from typing import Any
from mcp.types import TextContent
def extract_response_data(result: Any) -> dict[str, Any]:
"""Extract JSON data from MCP client response."""
# Handle FastMCP CallToolResult format
if not hasattr(result, "content"):
raise TypeError(f"Expected CallToolResult with content, got {type(result)}")
content = result.content
assert len(content) == 1
response_data: dict[str, Any] | None = (
json.loads(content[0].text) if isinstance(content[0], TextContent) else None
)
assert response_data is not None
return response_data
def assert_pagination_info(
pagination: dict[str, Any],
*,
total_count: int,
limit: int,
offset: int = 0,
has_more: bool = False,
) -> None:
"""Assert pagination info matches expected values."""
assert pagination["total_count"] == total_count
assert pagination["limit"] == limit
assert pagination["offset"] == offset
assert pagination["has_more"] == has_more
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "you-need-an-mcp"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = ["fastmcp>=2.11.3", "pydantic>=2.11.5", "ynab>=1.4.0"]
[dependency-groups]
dev = [
"pytest>=8.4.0",
"pytest-asyncio>=1.0.0",
"pytest-cov>=6.2.1",
"pytest-xdist>=3.7.0",
"pytest-env>=1.1.5",
"mypy>=1.8.0",
"pre-commit>=3.6.0",
"ruff>=0.12.0",
]
[tool.pytest.ini_options]
addopts = [
"--cov=server",
"--cov=models",
"--cov=tests/",
"--cov-branch",
"--cov-report=term-missing",
"--cov-fail-under=100",
"--strict-markers",
"--strict-config",
"-Werror",
]
asyncio_mode = "auto"
testpaths = ["tests"]
filterwarnings = ["error"]
env = [
"YNAB_BUDGET=test_budget_id",
"YNAB_ACCESS_TOKEN=test_token_123",
]
[tool.mypy]
python_version = "3.12"
strict = true
[[tool.mypy.overrides]]
module = ["ynab", "ynab.*"]
implicit_reexport = true
[tool.ruff]
target-version = "py312"
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"B", # flake8-bugbear
"UP", # pyupgrade (enforce modern python syntax)
"RUF", # ruff-specific rules
]
ignore = []
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
"""
Test fixtures for YNAB MCP Server tests.
This module contains pytest fixtures for testing without calling the actual YNAB API.
"""
import sys
from collections.abc import AsyncGenerator, Generator
from datetime import date
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, Mock, patch
import fastmcp
import pytest
import ynab
from fastmcp.client import Client, FastMCPTransport
# Add parent directory to path to import server module
sys.path.insert(0, str(Path(__file__).parent.parent))
import server
@pytest.fixture
def mock_environment_variables(monkeypatch: pytest.MonkeyPatch) -> None:
"""Mock environment variables for testing."""
monkeypatch.setenv("YNAB_ACCESS_TOKEN", "test_token_123")
monkeypatch.setenv("YNAB_BUDGET", "test_budget_id")
@pytest.fixture
def ynab_client(mock_environment_variables: None) -> Generator[MagicMock, None, None]:
"""Mock YNAB client with proper autospec for testing."""
mock_client = MagicMock(spec=ynab.ApiClient)
mock_client.__enter__.return_value = mock_client
mock_client.__exit__.return_value = None
yield mock_client
@pytest.fixture
def mock_repository() -> Generator[MagicMock, None, None]:
"""Mock the repository to prevent API calls during testing."""
with patch("server._repository") as mock_repo:
yield mock_repo
@pytest.fixture
def categories_api(ynab_client: MagicMock) -> Generator[MagicMock, None, None]:
mock_api = Mock(spec=ynab.CategoriesApi)
with patch("ynab.CategoriesApi", return_value=mock_api):
yield mock_api
@pytest.fixture
async def mcp_client() -> AsyncGenerator[Client[FastMCPTransport], None]:
"""Mock MCP client with proper autospec for testing."""
async with fastmcp.Client(server.mcp) as client:
yield client
# Test data factories
def create_ynab_account(
*,
id: str = "acc-1",
name: str = "Test Account",
account_type: ynab.AccountType = ynab.AccountType.CHECKING,
on_budget: bool = True,
closed: bool = False,
balance: int = 100_000,
deleted: bool = False,
**kwargs: Any,
) -> ynab.Account:
"""Create a YNAB Account for testing with sensible defaults."""
return ynab.Account(
id=id,
name=name,
type=account_type,
on_budget=on_budget,
closed=closed,
note=kwargs.get("note"),
balance=balance,
cleared_balance=kwargs.get("cleared_balance", balance - 5_000),
uncleared_balance=kwargs.get("uncleared_balance", 5_000),
transfer_payee_id=kwargs.get("transfer_payee_id"),
direct_import_linked=kwargs.get("direct_import_linked", False),
direct_import_in_error=kwargs.get("direct_import_in_error", False),
last_reconciled_at=kwargs.get("last_reconciled_at"),
debt_original_balance=kwargs.get("debt_original_balance"),
debt_interest_rates=kwargs.get("debt_interest_rates"),
debt_minimum_payments=kwargs.get("debt_minimum_payments"),
debt_escrow_amounts=kwargs.get("debt_escrow_amounts"),
deleted=deleted,
)
def create_ynab_payee(
*,
id: str = "payee-1",
name: str = "Test Payee",
deleted: bool = False,
**kwargs: Any,
) -> ynab.Payee:
"""Create a YNAB Payee for testing with sensible defaults."""
return ynab.Payee(
id=id,
name=name,
transfer_account_id=kwargs.get("transfer_account_id"),
deleted=deleted,
)
def create_ynab_category(
*,
id: str = "cat-1",
name: str = "Test Category",
category_group_id: str = "group-1",
hidden: bool = False,
deleted: bool = False,
budgeted: int = 50_000,
activity: int = -30_000,
balance: int = 20_000,
**kwargs: Any,
) -> ynab.Category:
"""Create a YNAB Category for testing with sensible defaults."""
return ynab.Category(
id=id,
category_group_id=category_group_id,
category_group_name=kwargs.get("category_group_name"),
name=name,
hidden=hidden,
original_category_group_id=kwargs.get("original_category_group_id"),
note=kwargs.get("note"),
budgeted=budgeted,
activity=activity,
balance=balance,
goal_type=kwargs.get("goal_type"),
goal_needs_whole_amount=kwargs.get("goal_needs_whole_amount"),
goal_day=kwargs.get("goal_day"),
goal_cadence=kwargs.get("goal_cadence"),
goal_cadence_frequency=kwargs.get("goal_cadence_frequency"),
goal_creation_month=kwargs.get("goal_creation_month"),
goal_target=kwargs.get("goal_target"),
goal_target_month=kwargs.get("goal_target_month"),
goal_percentage_complete=kwargs.get("goal_percentage_complete"),
goal_months_to_budget=kwargs.get("goal_months_to_budget"),
goal_under_funded=kwargs.get("goal_under_funded"),
goal_overall_funded=kwargs.get("goal_overall_funded"),
goal_overall_left=kwargs.get("goal_overall_left"),
deleted=deleted,
)
def create_ynab_transaction(
*,
id: str = "txn-1",
transaction_date: date = date(2024, 1, 15),
amount: int = -50_000,
account_id: str = "acc-1",
deleted: bool = False,
**kwargs: Any,
) -> ynab.TransactionDetail:
"""Create a YNAB TransactionDetail for testing with sensible defaults."""
return ynab.TransactionDetail(
id=id,
date=transaction_date,
amount=amount,
memo=kwargs.get("memo"),
cleared=kwargs.get("cleared", ynab.TransactionClearedStatus.CLEARED),
approved=kwargs.get("approved", True),
flag_color=kwargs.get("flag_color"),
account_id=account_id,
account_name=kwargs.get("account_name", "Test Account"),
payee_id=kwargs.get("payee_id"),
payee_name=kwargs.get("payee_name"),
category_id=kwargs.get("category_id"),
category_name=kwargs.get("category_name"),
transfer_account_id=kwargs.get("transfer_account_id"),
transfer_transaction_id=kwargs.get("transfer_transaction_id"),
matched_transaction_id=kwargs.get("matched_transaction_id"),
import_id=kwargs.get("import_id"),
import_payee_name=kwargs.get("import_payee_name"),
import_payee_name_original=kwargs.get("import_payee_name_original"),
debt_transaction_type=kwargs.get("debt_transaction_type"),
deleted=deleted,
subtransactions=kwargs.get("subtransactions", []),
)
```
--------------------------------------------------------------------------------
/tests/test_categories.py:
--------------------------------------------------------------------------------
```python
"""
Test category-related MCP tools.
"""
from unittest.mock import MagicMock
import ynab
from assertions import assert_pagination_info, extract_response_data
from conftest import create_ynab_category
from fastmcp.client import Client, FastMCPTransport
async def test_list_categories_success(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test successful category listing."""
visible_category = create_ynab_category(
id="cat-1",
name="Groceries",
note="Food shopping",
goal_type="TB",
goal_target=100_000,
goal_percentage_complete=50,
)
hidden_category = create_ynab_category(
id="cat-hidden",
name="Hidden Category",
hidden=True, # Should be excluded
budgeted=10_000,
activity=0,
balance=10_000,
)
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Monthly Bills",
hidden=False,
deleted=False,
categories=[visible_category, hidden_category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
result = await mcp_client.call_tool("list_categories", {})
response_data = extract_response_data(result)
# Should only include visible category
categories = response_data["categories"]
assert len(categories) == 1
assert categories[0]["id"] == "cat-1"
assert categories[0]["name"] == "Groceries"
assert categories[0]["category_group_name"] == "Monthly Bills"
assert_pagination_info(
response_data["pagination"],
total_count=1,
limit=50,
has_more=False,
)
async def test_list_category_groups_success(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test successful category group listing."""
category = ynab.Category(
id="cat-1",
category_group_id="group-1",
category_group_name="Monthly Bills",
name="Test Category",
hidden=False,
original_category_group_id=None,
note=None,
budgeted=50000,
activity=-30000,
balance=20000,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Monthly Bills",
hidden=False,
deleted=False,
categories=[category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
result = await mcp_client.call_tool("list_category_groups", {})
groups_data = extract_response_data(result)
# Should return a list of category groups
assert isinstance(groups_data, list)
assert len(groups_data) == 1
group = groups_data[0]
assert group["id"] == "group-1"
assert group["name"] == "Monthly Bills"
async def test_list_categories_filters_deleted_and_hidden(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test that list_categories automatically filters out deleted and hidden."""
# Active category (should be included)
mock_active_category = ynab.Category(
id="cat-active",
name="Active Category",
category_group_id="group-1",
hidden=False,
deleted=False,
note="Active",
budgeted=10000,
activity=-5000,
balance=5000,
goal_type=None,
goal_target=None,
goal_percentage_complete=None,
goal_under_funded=None,
goal_creation_month=None,
goal_target_month=None,
goal_overall_funded=None,
goal_overall_left=None,
)
# Hidden category (should be excluded)
mock_hidden_category = ynab.Category(
id="cat-hidden",
name="Hidden Category",
category_group_id="group-1",
hidden=True,
deleted=False,
note="Hidden",
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_target=None,
goal_percentage_complete=None,
goal_under_funded=None,
goal_creation_month=None,
goal_target_month=None,
goal_overall_funded=None,
goal_overall_left=None,
)
# Deleted category (should be excluded)
mock_deleted_category = ynab.Category(
id="cat-deleted",
name="Deleted Category",
category_group_id="group-1",
hidden=False,
deleted=True,
note="Deleted",
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_target=None,
goal_percentage_complete=None,
goal_under_funded=None,
goal_creation_month=None,
goal_target_month=None,
goal_overall_funded=None,
goal_overall_left=None,
)
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Monthly Bills",
hidden=False,
deleted=False,
categories=[
mock_active_category,
mock_hidden_category,
mock_deleted_category,
],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
result = await mcp_client.call_tool("list_categories", {})
response_data = extract_response_data(result)
# Should only include the active category
assert len(response_data["categories"]) == 1
assert response_data["categories"][0]["id"] == "cat-active"
assert response_data["categories"][0]["name"] == "Active Category"
async def test_list_category_groups_filters_deleted(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test that list_category_groups automatically filters out deleted groups."""
# Active group (should be included)
active_group = ynab.CategoryGroupWithCategories(
id="group-active",
name="Active Group",
hidden=False,
deleted=False,
categories=[],
)
# Deleted group (should be excluded)
deleted_group = ynab.CategoryGroupWithCategories(
id="group-deleted",
name="Deleted Group",
hidden=False,
deleted=True,
categories=[],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [active_group, deleted_group]
result = await mcp_client.call_tool("list_category_groups", {})
response_data = extract_response_data(result)
# Should only include the active group
assert isinstance(response_data, list)
assert len(response_data) == 1
group = response_data[0]
assert group["id"] == "group-active"
assert group["name"] == "Active Group"
```
--------------------------------------------------------------------------------
/tests/test_utilities.py:
--------------------------------------------------------------------------------
```python
"""
Test utility functions in server module.
"""
from datetime import date, datetime
from decimal import Decimal
from unittest.mock import patch
import pytest
import server
def test_decimal_precision_milliunits_conversion() -> None:
"""Test that milliunits conversion maintains Decimal precision."""
# Test various milliunits values that could lose precision with floats
test_cases = [
(123456, Decimal("123.456")), # Regular amount
(1, Decimal("0.001")), # Smallest unit
(999, Decimal("0.999")), # Just under 1
(1000, Decimal("1")), # Exactly 1
(1001, Decimal("1.001")), # Just over 1
(999999999, Decimal("999999.999")), # Large amount
(-50000, Decimal("-50")), # Negative amount
(0, Decimal("0")), # Zero
]
for milliunits, expected in test_cases:
from models import milliunits_to_currency
result = milliunits_to_currency(milliunits)
assert result == expected, (
f"Failed for {milliunits}: got {result}, expected {expected}"
)
# Ensure result is actually a Decimal, not float
assert isinstance(result, Decimal), (
f"Result {result} is not a Decimal but {type(result)}"
)
def test_milliunits_to_currency_valid_input() -> None:
"""Test milliunits conversion with valid input."""
from models import milliunits_to_currency
result = milliunits_to_currency(123456)
assert result == Decimal("123.456")
def test_milliunits_to_currency_none_input() -> None:
"""Test milliunits conversion with None input raises TypeError."""
from typing import Any
with pytest.raises(TypeError):
from models import milliunits_to_currency
none_value: Any = None
milliunits_to_currency(none_value)
def test_milliunits_to_currency_zero() -> None:
"""Test milliunits conversion with zero."""
from models import milliunits_to_currency
result = milliunits_to_currency(0)
assert result == Decimal("0")
def test_milliunits_to_currency_negative() -> None:
"""Test milliunits conversion with negative value."""
from models import milliunits_to_currency
result = milliunits_to_currency(-50000)
assert result == Decimal("-50")
def test_convert_month_to_date_with_date_object() -> None:
"""Test convert_month_to_date with date object returns unchanged."""
test_date = date(2024, 3, 15)
result = server.convert_month_to_date(test_date)
assert result == test_date
def test_convert_month_to_date_with_current() -> None:
"""Test convert_month_to_date with 'current' returns current month date."""
with patch("server.datetime") as mock_datetime:
mock_datetime.now.return_value = datetime(2024, 9, 20, 16, 45, 0)
result = server.convert_month_to_date("current")
assert result == date(2024, 9, 1)
def test_convert_month_to_date_with_last_and_next() -> None:
"""Test convert_month_to_date with 'last' and 'next' literals."""
# Test normal month (June -> May and July)
with patch("server.datetime") as mock_datetime:
mock_datetime.now.return_value = datetime(2024, 6, 15, 10, 30, 0)
result_last = server.convert_month_to_date("last")
assert result_last == date(2024, 5, 1)
result_next = server.convert_month_to_date("next")
assert result_next == date(2024, 7, 1)
# Test January edge case (January -> December previous year)
with patch("server.datetime") as mock_datetime:
mock_datetime.now.return_value = datetime(2024, 1, 10, 14, 45, 0)
result_last = server.convert_month_to_date("last")
assert result_last == date(2023, 12, 1)
result_next = server.convert_month_to_date("next")
assert result_next == date(2024, 2, 1)
# Test December edge case (December -> January next year)
with patch("server.datetime") as mock_datetime:
mock_datetime.now.return_value = datetime(2024, 12, 25, 9, 15, 0)
result_last = server.convert_month_to_date("last")
assert result_last == date(2024, 11, 1)
result_next = server.convert_month_to_date("next")
assert result_next == date(2025, 1, 1)
def test_convert_month_to_date_invalid_value() -> None:
"""Test convert_month_to_date with invalid value raises error."""
from typing import Any
with pytest.raises(ValueError, match="Invalid month value: invalid"):
invalid_value: Any = "invalid"
server.convert_month_to_date(invalid_value)
def test_convert_transaction_to_model_basic() -> None:
"""Test Transaction.from_ynab with basic transaction."""
import ynab
from models import Transaction
txn = ynab.TransactionDetail(
id="txn-123",
date=date(2024, 6, 15),
amount=-50000,
memo="Test transaction",
cleared=ynab.TransactionClearedStatus.CLEARED,
approved=True,
flag_color=ynab.TransactionFlagColor.RED,
account_id="acc-1",
payee_id="payee-1",
category_id="cat-1",
transfer_account_id=None,
transfer_transaction_id=None,
matched_transaction_id=None,
import_id=None,
import_payee_name=None,
import_payee_name_original=None,
debt_transaction_type=None,
deleted=False,
account_name="Checking",
payee_name="Test Payee",
category_name="Test Category",
subtransactions=[],
)
result = Transaction.from_ynab(txn)
assert result.id == "txn-123"
assert result.date == date(2024, 6, 15)
assert result.amount == Decimal("-50")
assert result.account_name == "Checking"
assert result.payee_name == "Test Payee"
assert result.category_name == "Test Category"
assert result.subtransactions is None
def test_convert_transaction_to_model_without_optional_attributes() -> None:
"""Test Transaction.from_ynab with minimal TransactionDetail."""
import ynab
from models import Transaction
minimal_txn = ynab.TransactionDetail(
id="txn-456",
date=date(2024, 6, 16),
amount=-25000,
memo="Minimal transaction",
cleared=ynab.TransactionClearedStatus.UNCLEARED,
approved=True,
flag_color=None,
account_id="acc-2",
payee_id="payee-2",
category_id="cat-2",
transfer_account_id=None,
transfer_transaction_id=None,
matched_transaction_id=None,
import_id=None,
import_payee_name=None,
import_payee_name_original=None,
debt_transaction_type=None,
deleted=False,
account_name="Test Account 2",
payee_name="Test Payee 2",
category_name="Test Category 2",
subtransactions=[],
)
result = Transaction.from_ynab(minimal_txn)
assert result.id == "txn-456"
assert result.account_name == "Test Account 2"
assert result.payee_name == "Test Payee 2"
assert result.category_name == "Test Category 2"
def test_milliunits_to_currency_from_models() -> None:
"""Test milliunits_to_currency function from models module."""
from models import milliunits_to_currency
assert milliunits_to_currency(50000) == Decimal("50")
assert milliunits_to_currency(-25000) == Decimal("-25")
assert milliunits_to_currency(1000) == Decimal("1")
assert milliunits_to_currency(0) == Decimal("0")
```
--------------------------------------------------------------------------------
/tests/test_accounts.py:
--------------------------------------------------------------------------------
```python
"""
Test account-related MCP tools.
"""
from unittest.mock import MagicMock
import ynab
from assertions import assert_pagination_info, extract_response_data
from conftest import create_ynab_account
from fastmcp.client import Client, FastMCPTransport
async def test_list_accounts_success(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test successful account listing."""
open_account = create_ynab_account(
id="acc-1",
name="Checking",
account_type=ynab.AccountType.CHECKING,
note="Main account",
)
closed_account = create_ynab_account(
id="acc-2",
name="Savings",
account_type=ynab.AccountType.SAVINGS,
closed=True, # Should be excluded
balance=0,
)
# Mock repository to return test accounts
mock_repository.get_accounts.return_value = [open_account, closed_account]
result = await mcp_client.call_tool("list_accounts", {})
response_data = extract_response_data(result)
# Should only include open account
accounts = response_data["accounts"]
assert len(accounts) == 1
assert accounts[0]["id"] == "acc-1"
assert accounts[0]["name"] == "Checking"
assert_pagination_info(
response_data["pagination"],
total_count=1,
limit=100,
has_more=False,
)
async def test_list_accounts_filters_closed_accounts(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test that list_accounts automatically excludes closed accounts."""
open_checking = create_ynab_account(
id="acc-1",
name="Checking",
account_type=ynab.AccountType.CHECKING,
closed=False,
)
closed_savings = create_ynab_account(
id="acc-2",
name="Old Savings",
account_type=ynab.AccountType.SAVINGS,
closed=True,
)
open_credit = create_ynab_account(
id="acc-3",
name="Credit Card",
account_type=ynab.AccountType.CREDITCARD,
closed=False,
)
mock_repository.get_accounts.return_value = [
open_checking,
closed_savings,
open_credit,
]
result = await mcp_client.call_tool("list_accounts", {})
response_data = extract_response_data(result)
# Should only include open accounts
accounts = response_data["accounts"]
assert len(accounts) == 2
account_names = [acc["name"] for acc in accounts]
assert "Checking" in account_names
assert "Credit Card" in account_names
assert "Old Savings" not in account_names # Closed account excluded
async def test_list_accounts_pagination(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test account listing with pagination."""
accounts = []
for i in range(5):
accounts.append(
create_ynab_account(
id=f"acc-{i}",
name=f"Account {i}",
closed=False,
)
)
mock_repository.get_accounts.return_value = accounts
# Test first page
result = await mcp_client.call_tool("list_accounts", {"limit": 2, "offset": 0})
response_data = extract_response_data(result)
assert len(response_data["accounts"]) == 2
assert_pagination_info(
response_data["pagination"],
total_count=5,
limit=2,
has_more=True,
)
# Test second page
result = await mcp_client.call_tool("list_accounts", {"limit": 2, "offset": 2})
response_data = extract_response_data(result)
assert len(response_data["accounts"]) == 2
assert_pagination_info(
response_data["pagination"],
total_count=5,
limit=2,
offset=2,
has_more=True,
)
async def test_list_accounts_with_repository_sync(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test that list_accounts triggers repository sync when needed."""
account = create_ynab_account(id="acc-1", name="Test Account")
# Mock repository to return empty initially (not initialized)
mock_repository.get_accounts.return_value = [account]
result = await mcp_client.call_tool("list_accounts", {})
response_data = extract_response_data(result)
# Verify repository was called
mock_repository.get_accounts.assert_called_once()
# Verify account data was returned
assert len(response_data["accounts"]) == 1
assert response_data["accounts"][0]["id"] == "acc-1"
async def test_list_accounts_account_types(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test that different account types are handled correctly."""
checking_account = create_ynab_account(
id="acc-checking",
name="My Checking",
account_type=ynab.AccountType.CHECKING,
on_budget=True,
)
savings_account = create_ynab_account(
id="acc-savings",
name="Emergency Fund",
account_type=ynab.AccountType.SAVINGS,
on_budget=True,
)
credit_card = create_ynab_account(
id="acc-credit",
name="Visa Card",
account_type=ynab.AccountType.CREDITCARD,
on_budget=True,
)
investment_account = create_ynab_account(
id="acc-investment",
name="401k",
account_type=ynab.AccountType.OTHERASSET,
on_budget=False, # Typically off-budget
)
mock_repository.get_accounts.return_value = [
checking_account,
savings_account,
credit_card,
investment_account,
]
result = await mcp_client.call_tool("list_accounts", {})
response_data = extract_response_data(result)
# All account types should be included (none are closed)
accounts = response_data["accounts"]
assert len(accounts) == 4
# Verify account types are preserved
account_types = {acc["id"]: acc["type"] for acc in accounts}
assert account_types["acc-checking"] == "checking"
assert account_types["acc-savings"] == "savings"
assert account_types["acc-credit"] == "creditCard"
assert account_types["acc-investment"] == "otherAsset"
# Verify on_budget status
on_budget_status = {acc["id"]: acc["on_budget"] for acc in accounts}
assert on_budget_status["acc-checking"] is True
assert on_budget_status["acc-investment"] is False
async def test_list_accounts_with_debt_fields(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test that debt-related fields are properly included for debt accounts."""
# Create a mortgage account with debt fields
mortgage_account = create_ynab_account(
id="acc-mortgage",
name="Home Mortgage",
account_type=ynab.AccountType.MORTGAGE,
on_budget=False,
balance=-250_000_000, # -$250,000 in milliunits
debt_interest_rates={
"2024-01-01": 3375,
"2024-07-01": 3250,
}, # 3.375%, 3.25% in milliunits
debt_minimum_payments={
"2024-01-01": 1500_000,
"2024-07-01": 1450_000,
}, # $1500, $1450 in milliunits
debt_escrow_amounts={
"2024-01-01": 300_000,
"2024-07-01": 325_000,
}, # $300, $325 in milliunits
)
# Create a credit card with empty debt fields
credit_card = create_ynab_account(
id="acc-credit",
name="Visa Card",
account_type=ynab.AccountType.CREDITCARD,
on_budget=True,
balance=-2500_000, # -$2,500 in milliunits
debt_interest_rates={}, # Empty for credit cards
debt_minimum_payments={},
debt_escrow_amounts={},
)
# Create a regular checking account without debt fields
checking_account = create_ynab_account(
id="acc-checking",
name="Checking",
account_type=ynab.AccountType.CHECKING,
balance=5000_000, # $5,000 in milliunits
)
mock_repository.get_accounts.return_value = [
mortgage_account,
credit_card,
checking_account,
]
result = await mcp_client.call_tool("list_accounts", {})
response_data = extract_response_data(result)
accounts = response_data["accounts"]
assert len(accounts) == 3
# Find mortgage account and verify debt fields
mortgage = next(acc for acc in accounts if acc["id"] == "acc-mortgage")
assert mortgage["debt_interest_rates"] == {
"2024-01-01": "0.03375", # 3.375% as decimal
"2024-07-01": "0.0325", # 3.25% as decimal
}
assert mortgage["debt_minimum_payments"] == {
"2024-01-01": "1500",
"2024-07-01": "1450",
}
assert mortgage["debt_escrow_amounts"] == {
"2024-01-01": "300",
"2024-07-01": "325",
}
# Verify credit card has null debt fields (empty dicts become None)
credit = next(acc for acc in accounts if acc["id"] == "acc-credit")
assert credit["debt_interest_rates"] is None
assert credit["debt_minimum_payments"] is None
assert credit["debt_escrow_amounts"] is None
# Verify checking account has null debt fields
checking = next(acc for acc in accounts if acc["id"] == "acc-checking")
assert checking["debt_interest_rates"] is None
assert checking["debt_minimum_payments"] is None
assert checking["debt_escrow_amounts"] is None
```
--------------------------------------------------------------------------------
/tests/test_payees.py:
--------------------------------------------------------------------------------
```python
"""
Test suite for payee-related functionality in YNAB MCP Server.
"""
from unittest.mock import MagicMock
import ynab
from assertions import extract_response_data
from conftest import create_ynab_payee
from fastmcp.client import Client, FastMCPTransport
async def test_list_payees_success(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test successful payee listing."""
payee1 = create_ynab_payee(id="payee-1", name="Amazon")
payee2 = create_ynab_payee(id="payee-2", name="Whole Foods")
# Deleted payee should be excluded by default
payee_deleted = create_ynab_payee(
id="payee-deleted",
name="Closed Store",
deleted=True,
)
# Transfer payee
payee_transfer = create_ynab_payee(
id="payee-transfer",
name="Transfer : Savings",
transfer_account_id="acc-savings",
)
# Mock repository to return test payees
mock_repository.get_payees.return_value = [
payee2,
payee1,
payee_deleted,
payee_transfer,
]
result = await mcp_client.call_tool("list_payees", {})
response_data = extract_response_data(result)
# Should have 3 payees (deleted one excluded)
assert len(response_data["payees"]) == 3
# Should be sorted by name
assert response_data["payees"][0]["name"] == "Amazon"
assert response_data["payees"][1]["name"] == "Transfer : Savings"
assert response_data["payees"][2]["name"] == "Whole Foods"
# Check transfer payee details
transfer_payee = response_data["payees"][1]
assert transfer_payee["id"] == "payee-transfer"
# Check pagination
assert response_data["pagination"]["total_count"] == 3
assert response_data["pagination"]["has_more"] is False
async def test_list_payees_pagination(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test payee listing with pagination."""
# Create multiple payees
payees = []
for i in range(5):
payee = ynab.Payee(
id=f"payee-{i}",
name=f"Store {i:02d}", # Store 00, Store 01, etc. for predictable sorting
transfer_account_id=None,
deleted=False,
)
payees.append(payee)
mock_repository.get_payees.return_value = payees
# Test first page
result = await mcp_client.call_tool("list_payees", {"limit": 2, "offset": 0})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["payees"]) == 2
assert response_data["pagination"]["total_count"] == 5
assert response_data["pagination"]["has_more"] is True
# Should be sorted alphabetically
assert response_data["payees"][0]["name"] == "Store 00"
assert response_data["payees"][1]["name"] == "Store 01"
async def test_list_payees_filters_deleted(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test that list_payees automatically filters out deleted payees."""
# Active payee (should be included)
payee_active = ynab.Payee(
id="payee-active",
name="Active Store",
transfer_account_id=None,
deleted=False,
)
# Deleted payee (should be excluded)
payee_deleted = ynab.Payee(
id="payee-deleted",
name="Deleted Store",
transfer_account_id=None,
deleted=True,
)
mock_repository.get_payees.return_value = [payee_active, payee_deleted]
result = await mcp_client.call_tool("list_payees", {})
response_data = extract_response_data(result)
assert response_data is not None
# Should only include the active payee
assert len(response_data["payees"]) == 1
assert response_data["payees"][0]["name"] == "Active Store"
assert response_data["payees"][0]["id"] == "payee-active"
async def test_find_payee_filters_deleted(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test that find_payee automatically filters out deleted payees."""
# Both payees have "amazon" in name, but one is deleted
payee_active = ynab.Payee(
id="payee-active", name="Amazon", transfer_account_id=None, deleted=False
)
payee_deleted = ynab.Payee(
id="payee-deleted",
name="Amazon Prime",
transfer_account_id=None,
deleted=True,
)
mock_repository.get_payees.return_value = [payee_active, payee_deleted]
result = await mcp_client.call_tool("find_payee", {"name_search": "amazon"})
response_data = extract_response_data(result)
assert response_data is not None
# Should only find the active Amazon payee, not the deleted one
assert len(response_data["payees"]) == 1
assert response_data["payees"][0]["name"] == "Amazon"
assert response_data["payees"][0]["id"] == "payee-active"
async def test_find_payee_success(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test successful payee search by name."""
# Create payees with different names for searching
payees = [
ynab.Payee(
id="payee-amazon",
name="Amazon",
transfer_account_id=None,
deleted=False,
),
ynab.Payee(
id="payee-amazon-web",
name="Amazon Web Services",
transfer_account_id=None,
deleted=False,
),
ynab.Payee(
id="payee-starbucks",
name="Starbucks",
transfer_account_id=None,
deleted=False,
),
ynab.Payee(
id="payee-grocery",
name="Whole Foods Market",
transfer_account_id=None,
deleted=False,
),
ynab.Payee(
id="payee-deleted",
name="Amazon Prime",
transfer_account_id=None,
deleted=True,
),
]
mock_repository.get_payees.return_value = payees
# Test searching for "amazon" (case-insensitive)
result = await mcp_client.call_tool("find_payee", {"name_search": "amazon"})
response_data = extract_response_data(result)
assert response_data is not None
# Should find Amazon and Amazon Web Services, but not deleted Amazon Prime
assert len(response_data["payees"]) == 2
assert response_data["pagination"]["total_count"] == 2
assert response_data["pagination"]["has_more"] is False
# Should be sorted alphabetically
payee_names = [p["name"] for p in response_data["payees"]]
assert payee_names == ["Amazon", "Amazon Web Services"]
async def test_find_payee_case_insensitive(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test that payee search is case-insensitive."""
payees = [
ynab.Payee(
id="payee-1",
name="Starbucks Coffee",
transfer_account_id=None,
deleted=False,
)
]
mock_repository.get_payees.return_value = payees
# Test various case combinations
search_terms_matches = [
("STARBUCKS", 1),
("starbucks", 1),
("StArBuCkS", 1),
("coffee", 1),
("COFFEE", 1),
("nonexistent", 0), # This will test the else branch
]
for search_term, expected_count in search_terms_matches:
result = await mcp_client.call_tool("find_payee", {"name_search": search_term})
response_data = extract_response_data(result)
assert len(response_data["payees"]) == expected_count
if expected_count > 0:
assert response_data["payees"][0]["name"] == "Starbucks Coffee"
async def test_find_payee_limit(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test payee search with limit parameter."""
# Create multiple payees with "store" in the name
payees = []
for i in range(5):
payees.append(
ynab.Payee(
id=f"payee-{i}",
name=f"Store {i:02d}", # Store 00, Store 01, etc.
transfer_account_id=None,
deleted=False,
)
)
mock_repository.get_payees.return_value = payees
# Test with limit of 2
result = await mcp_client.call_tool(
"find_payee", {"name_search": "store", "limit": 2}
)
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["payees"]) == 2
assert response_data["pagination"]["total_count"] == 5
assert response_data["pagination"]["has_more"] is True
# Should be first 2 in alphabetical order
assert response_data["payees"][0]["name"] == "Store 00"
assert response_data["payees"][1]["name"] == "Store 01"
async def test_find_payee_no_matches(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test payee search with no matching results."""
payees = [
ynab.Payee(
id="payee-1", name="Starbucks", transfer_account_id=None, deleted=False
)
]
mock_repository.get_payees.return_value = payees
result = await mcp_client.call_tool("find_payee", {"name_search": "nonexistent"})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["payees"]) == 0
assert response_data["pagination"]["total_count"] == 0
assert response_data["pagination"]["has_more"] is False
```
--------------------------------------------------------------------------------
/plans/caching.md:
--------------------------------------------------------------------------------
```markdown
# YNAB Local Repository with Differential Sync
## Overview
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.
## Core Design Principles
1. **Local-First**: All reads from in-memory repository, zero API calls during normal operation
2. **Differential Sync**: Use YNAB's `server_knowledge` to fetch only changes, not full datasets
3. **Repository Pattern**: Speaks only YNAB SDK models, no MCP-specific types
4. **Single Budget**: Server operates on one budget specified by `YNAB_BUDGET` env var
5. **Background Sync**: Updates happen out-of-band, never blocking MCP tool calls
## Key Constraints
- **YNAB is source of truth**: Local repository is read-only mirror, never modifies data
- **Eventually consistent**: Repository converges to YNAB state within sync interval
- **Handle stale knowledge**: When server returns 409, fall back to full refresh
- **Thread-safe**: Multiple MCP tools can read concurrently during sync
- **Memory-only initially**: Start with dicts, defer persistence to later
## Repository Interface
```python
class YNABRepository:
"""Local repository for YNAB data with background differential sync."""
def __init__(self, budget_id: str, api_client_factory: Callable):
self.budget_id = budget_id # Set once from YNAB_BUDGET env var
self.api_client_factory = api_client_factory
# In-memory storage - simple dicts
self._data: dict[str, list] = {} # entity_type -> list of entities
self._server_knowledge: dict[str, int] = {} # entity_type -> server_knowledge
self._lock = threading.RLock()
self._last_sync: datetime | None = None
# Data access - returns YNAB SDK models directly
def get_accounts(self) -> list[ynab.Account]:
def get_categories(self) -> list[ynab.CategoryGroupWithCategories]:
def get_transactions(self, since_date: date | None = None) -> list[ynab.TransactionDetail]:
def get_payees(self) -> list[ynab.Payee]:
def get_budget_month(self, month: date) -> ynab.MonthDetail:
# Sync management
def sync(self) -> None: # Fetch deltas and update repository
def needs_sync(self) -> bool: # Check if sync is needed
def last_sync_time(self) -> datetime | None:
```
## How Differential Sync Works
### Initial Load
```python
# First call without last_knowledge_of_server
response = api.get_accounts(budget_id)
# Returns: all accounts + server_knowledge: 100
self._data["accounts"] = response.data.accounts
self._server_knowledge["accounts"] = response.data.server_knowledge
```
### Delta Sync
```python
# Subsequent calls with last_knowledge_of_server
response = api.get_accounts(budget_id, last_knowledge_of_server=100)
# Returns: only changed accounts + server_knowledge: 101
# Apply changes: add/update/remove based on response
```
### Applying Deltas
```python
def apply_deltas(current: list, deltas: list) -> list:
entity_map = {e.id: e for e in current}
for delta in deltas:
if delta.deleted:
entity_map.pop(delta.id, None)
else:
entity_map[delta.id] = delta # Add or update
return list(entity_map.values())
```
## Budget Configuration Change
### Environment Variables
- **OLD**: `YNAB_DEFAULT_BUDGET` (optional, with fallback logic)
- **NEW**: `YNAB_BUDGET` (required for server startup)
The server will fail to start if `YNAB_BUDGET` is not set, making configuration explicit and removing ambiguity.
### Tool Signature Simplification
Remove `budget_id` parameter from all MCP tools since the server operates on a single budget:
```python
# Before
@mcp.tool()
def list_accounts(budget_id: str | None = None, limit: int = 100, offset: int = 0):
budget_id = budget_id_or_default(budget_id)
...
# After
@mcp.tool()
def list_accounts(limit: int = 100, offset: int = 0):
# No budget_id needed - using server's configured budget
...
```
This change applies to all tools: `list_accounts`, `list_categories`, `list_transactions`, `list_payees`, `get_budget_month`, etc.
Additionally, the `list_budgets` tool becomes unnecessary and should be removed since the server operates on a single configured budget.
### MCP Instructions Update
Simplify the MCP server instructions to remove budget_id complexity:
```python
mcp = FastMCP[None](
name="YNAB",
instructions="""
Access to your YNAB budget data including accounts, categories, and transactions.
The server operates on the budget configured via YNAB_BUDGET environment variable.
All data is served from a local repository that syncs with YNAB in the background.
"""
)
```
## Integration with MCP Tools
### Current Pattern (Direct API with budget_id)
```python
@mcp.tool()
def list_accounts(budget_id: str | None = None):
budget_id = budget_id_or_default(budget_id)
with get_ynab_client() as api_client:
accounts_api = ynab.AccountsApi(api_client)
response = accounts_api.get_accounts(budget_id)
# Process and return
```
### New Pattern (Repository without budget_id)
```python
# Global repository instance for the configured budget
_repository: YNABRepository | None = None
def get_repository() -> YNABRepository:
global _repository
if _repository is None:
budget_id = os.environ["YNAB_BUDGET"] # Required at startup
_repository = YNABRepository(
budget_id=budget_id,
api_client_factory=get_ynab_client
)
# Initial sync to populate
_repository.sync()
return _repository
@mcp.tool()
def list_accounts(limit: int = 100, offset: int = 0): # No budget_id parameter
repo = get_repository()
# Trigger background sync if needed (non-blocking)
if repo.needs_sync():
threading.Thread(target=repo.sync).start()
# Return data instantly from repository
accounts = repo.get_accounts()
# Apply existing filtering/pagination
return process_accounts(accounts)
```
## Critical Implementation Details
### Entity Types to Sync
- `accounts` - All accounts in budget
- `categories` - Category groups with nested categories
- `transactions` - Transaction history (consider date limits)
- `payees` - All payees
- `scheduled_transactions` - Scheduled/recurring transactions
- `budget_months` - Month-specific budget data (current/last/next)
### Thread Safety
- Use `threading.RLock()` for all repository data access
- Sync updates entire entity list atomically
- Reads can happen during sync (old data until sync completes)
### Error Handling
- **Network failure**: Continue serving stale data, retry sync later
- **409 (stale knowledge)**: Clear entity type, fetch all without last_knowledge
- **429 (rate limit)**: YNAB allows 200 requests/hour per token. Use exponential backoff, track request count
- **Invalid token**: Fail gracefully, log error, serve cached data
### Memory Management
- Typical budget: ~1-5MB in memory
- Consider transaction date limits (e.g., last 2 years only)
- Clear old budget month data (keep current + last + next)
## Benefits
- **Performance**: Sub-millisecond reads vs 100-500ms API calls
- **Reliability**: Works offline, degrades gracefully
- **Efficiency**: 60-80% fewer API calls after initial sync
- **User Experience**: Instant responses in MCP tools
## Future Considerations
- **Persistence**: SQLite for data survival across restarts
- **Selective Sync**: Only sync entity types actually used
- **Smart Scheduling**: Sync more frequently during business hours
- **Multi-Budget**: Support switching between budgets efficiently
## Migration Notes
### ✅ Completed Breaking Changes
1. **Environment variable**: `YNAB_DEFAULT_BUDGET` → `YNAB_BUDGET` (now required) ✅
2. **Tool signatures**: Remove `budget_id` parameter from all tools ✅
3. **Tool removal**: Delete `list_budgets` tool entirely ✅
4. **Error handling**: Server fails to start without `YNAB_BUDGET` ✅
5. **Test infrastructure**: Updated with pytest-env for environment variable support ✅
### User Impact
- Users must set `YNAB_BUDGET` before starting the server ✅
- LLMs no longer need to handle budget selection logic ✅
- Simpler, cleaner tool interfaces without optional budget_id parameters ✅
### Implementation Status
- **Phase 0: Budget ID Removal** ✅ COMPLETED
- All 57 tests passing with 100% coverage
- Clean foundation ready for repository pattern implementation
- **Phase 1: Repository Pattern** ✅ COMPLETED
- ✅ YNABRepository class created with differential sync
- ✅ Thread-safe data access with RLock
- ✅ Delta application for add/update/delete operations
- ✅ Lazy initialization per entity type
- ✅ Server integration - all tools use repository
- ✅ Background sync (non-blocking, triggered when data is stale)
- ✅ needs_sync() method for staleness detection
- ✅ Proper error handling (ConflictException, 429 rate limiting, fallback)
- ✅ Initial population at server startup
- ✅ Python logging with structured error handling
- ✅ Test coverage migration (all 97 tests passing with 100% coverage)
- **Phase 2: Test Quality Improvements** ✅ COMPLETED
- ✅ Hoisted all inline imports to top of test files
- ✅ Removed unhelpful comments that just repeated code
- ✅ Fixed poor test patterns (replaced try/except: pass with pytest.raises)
- ✅ Consolidated duplicate test helper functions into conftest.py
- ✅ Eliminated code duplication across 8+ test files
- ✅ Maintained 100% test coverage throughout cleanup
## Success Criteria
1. MCP tools never wait for API calls during normal operation
2. Repository stays synchronized within 5 minutes of YNAB changes
3. All existing MCP tool functionality works unchanged (except budget_id removal)
4. Memory usage stays under 10MB for typical budgets
5. Graceful degradation when YNAB API is unavailable
```
--------------------------------------------------------------------------------
/tests/test_updates.py:
--------------------------------------------------------------------------------
```python
"""
Tests for update functionality in YNAB MCP Server.
Tests the update_category_budget and update_transaction tools.
"""
from datetime import date
from typing import Any
from unittest.mock import MagicMock
import ynab
from assertions import extract_response_data
from fastmcp.client import Client, FastMCPTransport
def create_ynab_category(
*,
id: str = "cat-1",
category_group_id: str = "group-1",
budgeted: int = 100_000, # $100.00
activity: int = -50_000, # -$50.00
balance: int = 50_000, # $50.00
**kwargs: Any,
) -> ynab.Category:
"""Create a YNAB Category for testing with sensible defaults."""
return ynab.Category(
id=id,
category_group_id=category_group_id,
category_group_name=kwargs.get("category_group_name", "Test Group"),
name=kwargs.get("name", "Test Category"),
hidden=kwargs.get("hidden", False),
original_category_group_id=kwargs.get("original_category_group_id"),
note=kwargs.get("note"),
budgeted=budgeted,
activity=activity,
balance=balance,
goal_type=kwargs.get("goal_type"),
goal_needs_whole_amount=kwargs.get("goal_needs_whole_amount"),
goal_day=kwargs.get("goal_day"),
goal_cadence=kwargs.get("goal_cadence"),
goal_cadence_frequency=kwargs.get("goal_cadence_frequency"),
goal_creation_month=kwargs.get("goal_creation_month"),
goal_target=kwargs.get("goal_target"),
goal_target_month=kwargs.get("goal_target_month"),
goal_percentage_complete=kwargs.get("goal_percentage_complete"),
goal_months_to_budget=kwargs.get("goal_months_to_budget"),
goal_under_funded=kwargs.get("goal_under_funded"),
goal_overall_funded=kwargs.get("goal_overall_funded"),
goal_overall_left=kwargs.get("goal_overall_left"),
deleted=kwargs.get("deleted", False),
)
def create_ynab_transaction_detail(
*,
id: str = "txn-1",
date: date = date(2024, 1, 15),
amount: int = -50_000, # -$50.00
account_id: str = "acc-1",
**kwargs: Any,
) -> ynab.TransactionDetail:
"""Create a YNAB TransactionDetail for testing with sensible defaults."""
return ynab.TransactionDetail(
id=id,
date=date,
amount=amount,
memo=kwargs.get("memo"),
cleared=kwargs.get("cleared", ynab.TransactionClearedStatus.CLEARED),
approved=kwargs.get("approved", True),
flag_color=kwargs.get("flag_color"),
account_id=account_id,
account_name=kwargs.get("account_name", "Test Account"),
payee_id=kwargs.get("payee_id"),
payee_name=kwargs.get("payee_name"),
category_id=kwargs.get("category_id"),
category_name=kwargs.get("category_name"),
transfer_account_id=kwargs.get("transfer_account_id"),
transfer_transaction_id=kwargs.get("transfer_transaction_id"),
matched_transaction_id=kwargs.get("matched_transaction_id"),
import_id=kwargs.get("import_id"),
import_payee_name=kwargs.get("import_payee_name"),
import_payee_name_original=kwargs.get("import_payee_name_original"),
debt_transaction_type=kwargs.get("debt_transaction_type"),
deleted=kwargs.get("deleted", False),
subtransactions=kwargs.get("subtransactions", []),
)
async def test_update_category_budget_success(
mock_environment_variables: None,
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test successful category budget update."""
# Create the updated category that will be returned
updated_category = create_ynab_category(
id="cat-groceries",
category_group_id="group-everyday",
name="Groceries",
budgeted=200_000, # $200.00 (new budgeted amount)
activity=-150_000, # -$150.00
balance=50_000, # $50.00
)
# Mock repository methods
mock_repository.update_month_category.return_value = updated_category
# Mock the categories response for group names
category_group = ynab.CategoryGroupWithCategories(
id="group-everyday",
name="Everyday Expenses",
hidden=False,
deleted=False,
categories=[updated_category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
# Execute the tool
result = await mcp_client.call_tool(
"update_category_budget",
{
"category_id": "cat-groceries",
"budgeted": "200.00",
"month": "current",
},
)
# Verify the response
category_data = extract_response_data(result)
assert category_data["id"] == "cat-groceries"
assert category_data["name"] == "Groceries"
assert category_data["category_group_name"] == "Everyday Expenses"
assert category_data["budgeted"] == "200" # $200.00
assert category_data["activity"] == "-150" # -$150.00
assert category_data["balance"] == "50" # $50.00
# Verify the repository was called correctly
mock_repository.update_month_category.assert_called_once()
call_args = mock_repository.update_month_category.call_args
assert call_args[0][0] == "cat-groceries" # category_id
assert call_args[0][1].year == 2025 # current month (from date.today())
assert call_args[0][2] == 200_000 # budgeted_milliunits
async def test_update_transaction_success(
mock_environment_variables: None,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test successful transaction update."""
# Create the updated transaction that will be returned
updated_transaction = create_ynab_transaction_detail(
id="txn-123",
date=date(2024, 1, 15),
amount=-75_000, # -$75.00
account_id="acc-checking",
account_name="Checking",
payee_id="payee-amazon",
payee_name="Amazon",
category_id="cat-household", # Updated category
category_name="Household Items", # Updated category name
memo="Amazon purchase - household items", # Updated memo
cleared=ynab.TransactionClearedStatus.CLEARED,
approved=True,
)
# Mock the existing transaction response (what we fetch before updating)
original_transaction = create_ynab_transaction_detail(
id="txn-123",
date=date(2024, 1, 15),
amount=-75_000, # -$75.00
account_id="acc-checking",
account_name="Checking",
payee_id="payee-amazon",
payee_name="Amazon",
category_id="cat-food", # Original category
category_name="Food", # Original category name
memo="Amazon purchase", # Original memo
cleared=ynab.TransactionClearedStatus.CLEARED,
approved=True,
)
# Mock the API to return the transaction directly via the repository
mock_repository.get_transaction_by_id.return_value = original_transaction
mock_repository.update_transaction.return_value = updated_transaction
# Execute the tool
result = await mcp_client.call_tool(
"update_transaction",
{
"transaction_id": "txn-123",
"category_id": "cat-household",
"memo": "Amazon purchase - household items",
},
)
# Verify the response
transaction_data = extract_response_data(result)
assert transaction_data["id"] == "txn-123"
assert transaction_data["amount"] == "-75" # -$75.00
assert transaction_data["category_id"] == "cat-household"
assert transaction_data["category_name"] == "Household Items"
assert transaction_data["memo"] == "Amazon purchase - household items"
assert transaction_data["cleared"] == "cleared"
# Verify the repository was called correctly
mock_repository.get_transaction_by_id.assert_called_once_with("txn-123")
mock_repository.update_transaction.assert_called_once()
async def test_update_category_budget_with_specific_month(
mock_environment_variables: None,
categories_api: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test category budget update for a specific month."""
updated_category = create_ynab_category(
id="cat-dining",
name="Dining Out",
budgeted=150_000, # $150.00
)
save_response = ynab.SaveCategoryResponse(
data=ynab.SaveCategoryResponseData(
category=updated_category, server_knowledge=0
)
)
categories_api.update_month_category.return_value = save_response
# Mock categories response for group names
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Fun Money",
hidden=False,
deleted=False,
categories=[updated_category],
)
categories_response = ynab.CategoriesResponse(
data=ynab.CategoriesResponseData(
category_groups=[category_group], server_knowledge=0
)
)
categories_api.get_categories.return_value = categories_response
# Execute with specific date
result = await mcp_client.call_tool(
"update_category_budget",
{
"category_id": "cat-dining",
"budgeted": "150.00",
"month": "2024-03-01", # Specific month
},
)
# Verify the response
category_data = extract_response_data(result)
assert category_data["budgeted"] == "150"
# Verify correct month was passed to API
call_args = categories_api.update_month_category.call_args[0]
month_arg = call_args[1]
assert month_arg == date(2024, 3, 1)
async def test_update_transaction_minimal_fields(
mock_environment_variables: None,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test transaction update with only category change."""
# Mock the existing transaction response (what we fetch before updating)
original_transaction = create_ynab_transaction_detail(
id="txn-456",
category_id="cat-food",
category_name="Food",
)
updated_transaction = create_ynab_transaction_detail(
id="txn-456",
category_id="cat-gas",
category_name="Gas & Fuel",
)
# Mock the repository methods
mock_repository.get_transaction_by_id.return_value = original_transaction
mock_repository.update_transaction.return_value = updated_transaction
# Execute with only category_id change
result = await mcp_client.call_tool(
"update_transaction",
{
"transaction_id": "txn-456",
"category_id": "cat-gas",
},
)
# Verify the response
transaction_data = extract_response_data(result)
assert transaction_data["category_id"] == "cat-gas"
# Verify the repository was called correctly
mock_repository.get_transaction_by_id.assert_called_once_with("txn-456")
mock_repository.update_transaction.assert_called_once()
async def test_update_transaction_with_payee(
mock_environment_variables: None,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test transaction update with payee_id to cover all branches."""
# Mock the existing transaction response (what we fetch before updating)
original_transaction = create_ynab_transaction_detail(
id="txn-789",
amount=-25_500, # -$25.50
payee_id="payee-generic",
payee_name="Generic Store",
memo="Store purchase",
)
updated_transaction = create_ynab_transaction_detail(
id="txn-789",
amount=-25_500, # -$25.50
payee_id="payee-starbucks",
payee_name="Starbucks",
memo="Coffee run",
)
# Mock the repository methods
mock_repository.get_transaction_by_id.return_value = original_transaction
mock_repository.update_transaction.return_value = updated_transaction
# Execute with payee_id
result = await mcp_client.call_tool(
"update_transaction",
{
"transaction_id": "txn-789",
"payee_id": "payee-starbucks",
"memo": "Coffee run",
},
)
# Verify the response
transaction_data = extract_response_data(result)
assert transaction_data["payee_id"] == "payee-starbucks"
assert transaction_data["memo"] == "Coffee run"
assert transaction_data["amount"] == "-25.5"
# Verify the repository was called correctly
mock_repository.get_transaction_by_id.assert_called_once_with("txn-789")
mock_repository.update_transaction.assert_called_once()
```
--------------------------------------------------------------------------------
/DESIGN.md:
--------------------------------------------------------------------------------
```markdown
# DESIGN.md
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.
## Division of Responsibilities
**MCP Server provides**:
- Structured access to YNAB data with consistent formatting
- Efficient filtering and search capabilities
- Pagination for large datasets
- Clean abstractions over YNAB's API complexity
**LLM provides**:
- Natural language understanding
- Pattern recognition and analysis
- Intelligent summarization
- Actionable recommendations
- Context-aware interpretation
## Use Case Format Guide
Each use case follows this structure for clarity:
- **User says**: Natural language examples
- **Why it matters**: User context and pain points
- **MCP Server Role**: Which tools to use and what data they provide
- **LLM Role**: Analysis and intelligence needed
- **Example Implementation Flow**: Step-by-step approach (when helpful)
- **Edge Cases**: YNAB-specific gotchas to handle (when applicable)
## Core Use Cases
### 1. Weekly Family Finance Check-ins
**User says**: "How are we doing on our budget this month? Which categories are we overspending?"
**Why it matters**: Busy parents need quick weekly snapshots without opening YNAB. They want conversational summaries that highlight what needs attention.
**MCP Server Role**:
- `get_budget_month()` - Provides current month's budgeted amounts, activity, and balances
- `list_categories()` - Gets category structure for organization
- Returns clean, structured data with proper currency formatting
**LLM Role**:
- Interprets which categories are overspent/underspent
- Prioritizes what needs attention
- Suggests fund reallocation
- Generates conversational summary
**Example Implementation Flow**:
1. Call `get_budget_month()` to get current state
2. Identify categories where `balance < 0` (overspent)
3. Find categories with available funds (`balance > 0`)
4. Prioritize by spending velocity and days left in month
5. Format as conversational response with specific suggestions
**Edge Cases**:
- Handle credit card payment categories differently
- Consider "Inflow: Ready to Assign" as special
- Account for scheduled transactions not yet posted
### 2. Kid-Related Expense Tracking
**User says**: "How much have we spent on soccer this year?" or "Show me all transactions for Emma's activities"
**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.
**MCP Server Role**:
- `find_payee()` - Searches for "soccer", "dance academy", etc.
- `list_transactions()` - Filters by payee_id, date ranges
- Handles pagination for large transaction sets
- Returns transaction details with amounts and dates
**LLM Role**:
- Identifies relevant search terms from natural language
- Aggregates totals across multiple payees/categories
- Groups related expenses
- Formats results for specific use (taxes, custody docs)
**Key Implementation Notes**:
- May need multiple payee searches (e.g., "Soccer Club", "Soccer Store", "Soccer Camp")
- Consider memo fields for additional context
- Date ranges should align with tax year or custody period
### 3. Subscription and Recurring Expense Audits
**User says**: "What subscriptions are we paying for?" or "List all our monthly recurring expenses"
**Why it matters**: Subscription creep affects every family. Parents need to identify forgotten subscriptions and understand their true monthly commitments.
**MCP Server Role**:
- `list_transactions()` - Provides transaction history with dates
- `list_payees()` - Gets payee details for merchant identification
- Efficient pagination for analyzing patterns over time
**LLM Role**:
- Pattern recognition to identify recurring transactions
- Frequency analysis (monthly, annual, etc.)
- Grouping by merchant
- Flagging unusual patterns or new subscriptions
### 4. Pre-Shopping Budget Checks
**User says**: "Can we afford to spend $300 at Costco today?" or "How much grocery money do we have left?"
**Why it matters**: Quick budget checks before shopping trips prevent overspending and the stress of moving money after the fact.
**MCP Server Role**:
- `get_budget_month()` - Current balances for relevant categories
- `list_categories()` - Category relationships and groupings
- Real-time accurate balance data
**LLM Role**:
- Maps "Costco shopping" to relevant categories (groceries, household, etc.)
- Calculates total available across multiple categories
- Suggests reallocation strategies
- Provides go/no-go recommendation
### 5. Financial Partnership Transparency
**User says**: "Give me a simple summary of our finances" or "Are we okay financially?" or "Did that Amazon return get credited?"
**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.
**MCP Server Role**:
- `get_budget_month()` - Overall budget health data
- `list_accounts()` - Account balances for net worth
- `list_transactions()` - Recent transaction verification
- `find_payee()` - Quick lookup for specific merchants
**LLM Role**:
- Translates budget complexity into simple terms
- Provides reassuring summaries ("Yes, you're on track")
- Answers specific concerns without overwhelming detail
- Bridges the knowledge gap between budget manager and partner
### 6. End-of-Month Category Sweep
**User says**: "Which categories have money left over?" or "Help me zero out my budget"
**Why it matters**: YNAB's zero-based budgeting requires monthly cleanup. Parents need quick identification of surplus funds and smart reallocation suggestions.
**MCP Server Role**:
- `get_budget_month()` - All category balances for current month
- `list_category_groups()` - Organized view of budget structure
- Accurate to-the-penny balance data
**LLM Role**:
- Identifies categories with positive balances
- Analyzes historical spending to suggest reallocations
- Prioritizes based on upcoming needs
- Future: Generates reallocation transactions
### 7. Emergency Fund Reality Checks
**User says**: "How many months could we survive on our emergency fund?" or "What's our true available emergency money?"
**Why it matters**: Provides peace of mind by calculating realistic burn rates based on essential expenses and actual family spending patterns.
**MCP Server Role**:
- `list_accounts()` - Gets emergency fund account balances
- `list_transactions()` - Historical spending data for analysis
- `list_categories()` - Category structure for expense classification
**LLM Role**:
- Categorizes expenses as essential vs. discretionary
- Calculates average monthly burn rate
- Projects survival duration
- Scenario modeling based on different assumptions
### 8. Financial Scenario Planning
**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?"
**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.
**MCP Server Role**:
- `get_budget_month()` - Current budget as baseline
- `list_transactions()` - Historical spending patterns
- `list_categories()` - Understanding fixed vs. variable expenses
- `list_accounts()` - Current financial position
**LLM Role**:
- Models income changes across categories
- Projects new expense impacts
- Identifies categories that would need adjustment
- Calculates how long until savings goals are met
- Suggests budget reallocations for new scenarios
**Future MCP Tools Needed**:
- `create_budget_scenario()` - Clone budget for what-if analysis
- `get_category_spending_history()` - Trends over time for better projections
### 9. AI-Assisted Payee Cleanup
**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?"
**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.
**MCP Server Role**:
- `list_payees()` - Get all payees with transaction counts
- `list_transactions()` - Analyze transaction patterns per payee
- Returns payee names, IDs, and usage statistics
**LLM Role**:
- Pattern recognition to identify similar payees (fuzzy matching)
- Groups variations of the same merchant (e.g., "Amazon", "AMZN", "Amazon Prime")
- Suggests canonical names based on clarity and frequency
- Identifies rarely-used payees that could be merged
- Prioritizes high-impact cleanups (most transactions affected)
- Generates step-by-step cleanup instructions for YNAB UI
**Example Implementation Flow**:
1. Call `list_payees()` to get all payees with pagination
2. Use NLP/fuzzy matching to identify potential duplicate groups
3. For each group, analyze transaction patterns to confirm similarity
4. Suggest a canonical name (prefer clear, human-readable versions)
5. Calculate impact (number of transactions that would be affected)
6. Present recommendations ranked by impact with manual cleanup steps
**Edge Cases**:
- Transfer payees (contain account names) need special handling
- Starting Balance and Manual Balance Adjustment are system payees
- Credit card payment payees follow specific patterns
- Venmo/PayPal transactions may need memo analysis for better grouping
- Consider frequency of use when suggesting merge direction
**SDK Limitations (as of 2025-06)**:
- ❌ No payee merging/combining operations in YNAB API
- ❌ No payee deletion capability
- ❌ No bulk payee operations
- ✅ Only individual payee renaming supported via `update_payee()`
**Current Implementation Approach**:
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.
## Design Principles for Use Cases
1. **Conversational First**: Every query should feel natural to speak or type
2. **Context Aware**: Understand "we", "our", "the kids" in the context of a family
3. **Action Oriented**: Don't just report data, suggest next steps
4. **Time Sensitive**: Respect that parents are asking between activities
5. **Trust Building**: Be transparent about calculations and assumptions
## Future Use Case Directions
### Receipt/Transaction Quick Entry
**User says**: "Add Costco $127.43 groceries and household"
**Future MCP Tools Needed**:
- `create_transaction()` - Add new transactions with splits
- `get_recent_payees()` - Smart payee matching
- `suggest_categories()` - Based on payee history
### Bill Reminders
**User says**: "What bills are due this week?"
**MCP Server Role**:
- `list_scheduled_transactions()` - Future tool for recurring transactions
- Current workaround: Analyze transaction history for patterns
### Import Assistance
**User says**: "Help me categorize these Venmo transactions"
**Future MCP Tools Needed**:
- `import_transactions()` - Bulk import capability
- `update_transaction()` - Modify imported transactions
- `match_payees()` - Fuzzy matching for payee cleanup
## Tool Implementation Status
### Currently Implemented (11 tools)
- ✅ `list_budgets()` - All use cases
- ✅ `list_accounts()` - Emergency fund calculations
- ✅ `list_categories()` - Budget structure understanding
- ✅ `list_category_groups()` - Efficient category overview
- ✅ `get_budget_month()` - Weekly check-ins, category sweep
- ✅ `get_month_category_by_id()` - Specific category details
- ✅ `list_transactions()` - Expense tracking, subscriptions, transparency
- ✅ `list_payees()` - Payee analysis
- ✅ `find_payee()` - Efficient payee search
- ✅ `list_scheduled_transactions()` - Bill reminders, recurring expenses
### Recently Implemented
- ✅ `list_scheduled_transactions()` - Bill reminders, recurring expenses
- Supports all major use cases: subscription audits, bill reminders, recurring expense analysis
- Comprehensive filtering: account, category, payee, frequency, upcoming days, amount range
- Full pagination support following existing patterns
- Consistent field naming with regular transactions using shared base model
- 100% test coverage with extensive edge case testing
### Planned Tools (SDK-supported)
- 🔄 `create_transaction()` - Quick entry
- 🔄 `update_transaction()` - Import assistance
- 🔄 `import_transactions()` - Bulk import
### Optimization Considerations
The SDK offers specialized transaction endpoints that could optimize specific use cases:
- `get_transactions_by_account()` - Direct account filtering
- `get_transactions_by_category()` - Direct category filtering
- `get_transactions_by_month()` - Month-specific queries
- `get_transactions_by_payee()` - Direct payee filtering
**Current approach**: Single `list_transactions()` with flexible filtering
**Trade-offs**:
- ✅ Simpler API surface for LLMs to learn
- ✅ One tool handles all filtering combinations
- ❌ Potentially less efficient for single-filter queries
- ❌ May miss SDK-specific optimizations
**Recommendation**: Keep the single `list_transactions()` approach because:
1. LLMs perform better with fewer, more flexible tools
2. Most use cases need multiple filters anyway (date + payee, category + amount)
3. The performance difference is negligible for household-scale data
4. Reduces tool discovery complexity for the LLM
### Creative Solutions Needed
- 💡 Smart payee matching - Build on existing tools
- 💡 Category suggestions - Analyze transaction history
- 💡 Fuzzy payee matching - Custom logic required
## Success Metrics
A use case is successful when:
- It saves the user time vs. using YNAB directly
- It provides insights not easily visible in YNAB's interface
- It helps prevent financial stress or surprises
- It works within the natural flow of family life
```
--------------------------------------------------------------------------------
/tests/test_budget_months.py:
--------------------------------------------------------------------------------
```python
"""
Test budget month and month category-related MCP tools.
"""
from collections.abc import Generator
from datetime import date
from unittest.mock import MagicMock, Mock, patch
import pytest
import ynab
from assertions import extract_response_data
from fastmcp.client import Client, FastMCPTransport
@pytest.fixture
def months_api(ynab_client: MagicMock) -> Generator[MagicMock, None, None]:
mock_api = Mock(spec=ynab.MonthsApi)
with patch("ynab.MonthsApi", return_value=mock_api):
yield mock_api
async def test_get_budget_month_success(
months_api: MagicMock,
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test successful budget month retrieval."""
category = ynab.Category(
id="cat-1",
category_group_id="group-1",
category_group_name="Monthly Bills",
name="Groceries",
hidden=False,
original_category_group_id=None,
note="Food",
budgeted=50000,
activity=-30000,
balance=20000,
goal_type="TB",
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=100000,
goal_target_month=None,
goal_percentage_complete=50,
goal_months_to_budget=None,
goal_under_funded=0,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
month = ynab.MonthDetail(
month=date(2024, 1, 1),
note="January budget",
income=400000,
budgeted=350000,
activity=-200000,
to_be_budgeted=50000,
age_of_money=15,
deleted=False,
categories=[category],
)
# Mock repository methods
mock_repository.get_budget_month.return_value = month
# Mock the categories API call for getting group names
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Monthly Bills",
hidden=False,
deleted=False,
categories=[category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
result = await mcp_client.call_tool("get_budget_month", {})
response_data = extract_response_data(result)
assert response_data["note"] == "January budget"
assert len(response_data["categories"]) == 1
assert response_data["categories"][0]["id"] == "cat-1"
assert response_data["categories"][0]["category_group_name"] == "Monthly Bills"
async def test_get_month_category_by_id_success(
months_api: MagicMock,
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test successful month category retrieval by ID."""
mock_category = ynab.Category(
id="cat-1",
category_group_id="group-1",
category_group_name="Monthly Bills",
name="Groceries",
hidden=False,
original_category_group_id=None,
note="Food",
budgeted=50000,
activity=-30000,
balance=20000,
goal_type="TB",
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=100000,
goal_target_month=None,
goal_percentage_complete=50,
goal_months_to_budget=None,
goal_under_funded=0,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
# Mock repository method
mock_repository.get_month_category_by_id.return_value = mock_category
# Mock the categories API call for getting group names
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Monthly Bills",
hidden=False,
deleted=False,
categories=[mock_category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
result = await mcp_client.call_tool(
"get_month_category_by_id",
{"category_id": "cat-1"},
)
response_data = extract_response_data(result)
assert response_data["id"] == "cat-1"
assert response_data["name"] == "Groceries"
assert response_data["category_group_name"] == "Monthly Bills"
async def test_get_month_category_by_id_default_budget(
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test month category retrieval using default budget."""
mock_category = ynab.Category(
id="cat-2",
category_group_id="group-2",
category_group_name="Fun Money",
name="Entertainment",
hidden=False,
original_category_group_id=None,
note="Fun stuff",
budgeted=25000,
activity=-15000,
balance=10000,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
# Mock repository method
mock_repository.get_month_category_by_id.return_value = mock_category
# Mock the categories API call for getting group names
category_group = ynab.CategoryGroupWithCategories(
id="group-2",
name="Fun Money",
hidden=False,
deleted=False,
categories=[mock_category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
# Call without budget_id to test default
result = await mcp_client.call_tool(
"get_month_category_by_id", {"category_id": "cat-2"}
)
response_data = extract_response_data(result)
assert response_data["id"] == "cat-2"
assert response_data["name"] == "Entertainment"
assert response_data["category_group_name"] == "Fun Money"
async def test_get_month_category_by_id_no_groups(
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test month category retrieval when no category groups exist."""
mock_category = ynab.Category(
id="cat-orphan",
category_group_id="group-missing",
category_group_name="Missing Group",
name="Orphan Category",
hidden=False,
original_category_group_id=None,
note="Category with no group",
budgeted=10000,
activity=-5000,
balance=5000,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
# Mock repository method
mock_repository.get_month_category_by_id.return_value = mock_category
# Mock empty category groups response
mock_repository.get_category_groups.return_value = []
result = await mcp_client.call_tool(
"get_month_category_by_id", {"category_id": "cat-orphan"}
)
response_data = extract_response_data(result)
assert response_data["id"] == "cat-orphan"
assert response_data["category_group_name"] is None
async def test_get_month_category_by_id_category_not_in_groups(
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test month category retrieval when category is not found in any group."""
mock_category = ynab.Category(
id="cat-notfound",
category_group_id="group-old",
category_group_name="Old Group",
name="Not Found Category",
hidden=False,
original_category_group_id=None,
note="Category not in groups",
budgeted=5000,
activity=-2000,
balance=3000,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
# Create some other categories that don't match
other_category1 = ynab.Category(
id="cat-other1",
category_group_id="group-1",
category_group_name="Group 1",
name="Other Category 1",
hidden=False,
original_category_group_id=None,
note=None,
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
other_category2 = ynab.Category(
id="cat-other2",
category_group_id="group-2",
category_group_name="Group 2",
name="Other Category 2",
hidden=False,
original_category_group_id=None,
note=None,
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
# Mock repository method
mock_repository.get_month_category_by_id.return_value = mock_category
# Mock category groups with categories that don't include our target
category_group1 = ynab.CategoryGroupWithCategories(
id="group-1",
name="Group 1",
hidden=False,
deleted=False,
categories=[other_category1],
)
category_group2 = ynab.CategoryGroupWithCategories(
id="group-2",
name="Group 2",
hidden=False,
deleted=False,
categories=[other_category2],
)
# Add an empty category group to test the empty categories branch
empty_group = ynab.CategoryGroupWithCategories(
id="group-empty",
name="Empty Group",
hidden=False,
deleted=False,
categories=[],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [
category_group1,
empty_group,
category_group2,
]
result = await mcp_client.call_tool(
"get_month_category_by_id", {"category_id": "cat-notfound"}
)
response_data = extract_response_data(result)
assert response_data["id"] == "cat-notfound"
assert response_data["category_group_name"] is None
async def test_get_budget_month_with_default_budget(
months_api: MagicMock,
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test budget month retrieval with default budget."""
category = ynab.Category(
id="cat-default",
category_group_id="group-default",
category_group_name="Default Group",
name="Default Category",
hidden=False,
original_category_group_id=None,
note=None,
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
month = ynab.MonthDetail(
month=date(2024, 2, 1),
note=None,
income=0,
budgeted=0,
activity=0,
to_be_budgeted=0,
age_of_money=None,
deleted=False,
categories=[category],
)
# Mock repository method
mock_repository.get_budget_month.return_value = month
# Mock the categories API call for getting group names
category_group = ynab.CategoryGroupWithCategories(
id="group-default",
name="Default Group",
hidden=False,
deleted=False,
categories=[category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
# Call without budget_id to test default
result = await mcp_client.call_tool("get_budget_month", {})
response_data = extract_response_data(result)
assert len(response_data["categories"]) == 1
assert response_data["categories"][0]["id"] == "cat-default"
async def test_get_budget_month_filters_deleted_and_hidden(
months_api: MagicMock,
categories_api: MagicMock,
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test that get_budget_month filters out deleted and hidden categories."""
# Create active category
active_category = ynab.Category(
id="cat-active",
category_group_id="group-1",
category_group_name="Group 1",
name="Active Category",
hidden=False,
original_category_group_id=None,
note=None,
budgeted=10000,
activity=-5000,
balance=5000,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
# Create deleted category (should be filtered out)
deleted_category = ynab.Category(
id="cat-deleted",
category_group_id="group-1",
category_group_name="Group 1",
name="Deleted Category",
hidden=False,
original_category_group_id=None,
note=None,
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=True,
)
# Create hidden category (should be filtered out)
hidden_category = ynab.Category(
id="cat-hidden",
category_group_id="group-1",
category_group_name="Group 1",
name="Hidden Category",
hidden=True,
original_category_group_id=None,
note=None,
budgeted=0,
activity=0,
balance=0,
goal_type=None,
goal_needs_whole_amount=None,
goal_day=None,
goal_cadence=None,
goal_cadence_frequency=None,
goal_creation_month=None,
goal_target=None,
goal_target_month=None,
goal_percentage_complete=None,
goal_months_to_budget=None,
goal_under_funded=None,
goal_overall_funded=None,
goal_overall_left=None,
deleted=False,
)
month = ynab.MonthDetail(
month=date(2024, 1, 1),
note=None,
income=100000,
budgeted=10000,
activity=-5000,
to_be_budgeted=95000,
age_of_money=10,
deleted=False,
categories=[active_category, deleted_category, hidden_category],
)
# Mock repository method
mock_repository.get_budget_month.return_value = month
# Mock the categories API call for getting group names
category_group = ynab.CategoryGroupWithCategories(
id="group-1",
name="Group 1",
hidden=False,
deleted=False,
categories=[active_category, deleted_category, hidden_category],
)
# Mock repository to return category groups
mock_repository.get_category_groups.return_value = [category_group]
result = await mcp_client.call_tool("get_budget_month", {})
response_data = extract_response_data(result)
# Should only include the active category
assert len(response_data["categories"]) == 1
assert response_data["categories"][0]["id"] == "cat-active"
```
--------------------------------------------------------------------------------
/models.py:
--------------------------------------------------------------------------------
```python
"""
Pydantic models for YNAB MCP Server responses.
These models provide structured, well-documented data types for all YNAB API responses,
including detailed explanations of YNAB's data model subtleties and conventions.
"""
from __future__ import annotations
import datetime
from decimal import Decimal
from typing import TYPE_CHECKING
import ynab
from pydantic import BaseModel, Field
if TYPE_CHECKING: # pragma: no cover
from repository import YNABRepository
def milliunits_to_currency(milliunits: int, decimal_digits: int = 2) -> Decimal:
"""Convert YNAB milliunits to currency amount.
YNAB uses milliunits where 1000 milliunits = 1 currency unit.
"""
return Decimal(milliunits) / Decimal("1000")
class PaginationInfo(BaseModel):
"""Pagination metadata for listing endpoints."""
total_count: int = Field(..., description="Total number of items available")
limit: int = Field(..., description="Maximum items per page")
offset: int = Field(..., description="Number of items skipped")
has_more: bool = Field(..., description="Whether more items are available")
class Account(BaseModel):
"""A YNAB account with balance information.
All amounts are in currency units with Decimal precision.
"""
id: str = Field(..., description="Unique account identifier")
name: str = Field(..., description="User-defined account name")
type: str = Field(
...,
description="Account type. Common values: 'checking', 'savings', 'creditCard', "
"'cash', 'lineOfCredit', 'otherAsset', 'otherLiability', 'mortgage', "
"'autoLoan', 'studentLoan'",
)
on_budget: bool = Field(
..., description="Whether this account is included in budget calculations"
)
closed: bool = Field(..., description="Whether this account has been closed")
note: str | None = Field(None, description="User-defined account notes")
balance: Decimal | None = Field(
None, description="Current account balance in currency units"
)
cleared_balance: Decimal | None = Field(
None, description="Balance of cleared transactions in currency units"
)
debt_interest_rates: dict[datetime.date, Decimal] | None = Field(
None,
description="Interest rates by date for debt accounts. Keys are dates, "
"values are interest rates as decimals (e.g., 0.03375 for 3.375%)",
)
debt_minimum_payments: dict[datetime.date, Decimal] | None = Field(
None,
description="Minimum payment amounts by date for debt accounts. "
"Keys are dates, values are payment amounts in currency units",
)
debt_escrow_amounts: dict[datetime.date, Decimal] | None = Field(
None,
description="Escrow amounts by date for mortgage accounts. Keys are dates, "
"values are escrow amounts in currency units",
)
@classmethod
def from_ynab(cls, account: ynab.Account) -> Account:
"""Convert YNAB Account object to our Account model."""
# Convert debt interest rates from milliunits to decimal (e.g., 3375 -> 0.03375)
debt_interest_rates = None
if hasattr(account, "debt_interest_rates") and account.debt_interest_rates:
debt_interest_rates = {
datetime.date.fromisoformat(date_str): milliunits_to_currency(rate)
/ 100
for date_str, rate in account.debt_interest_rates.items()
}
# Convert debt minimum payments from milliunits to currency
debt_minimum_payments = None
if hasattr(account, "debt_minimum_payments") and account.debt_minimum_payments:
debt_minimum_payments = {
datetime.date.fromisoformat(date_str): milliunits_to_currency(amount)
for date_str, amount in account.debt_minimum_payments.items()
}
# Convert debt escrow amounts from milliunits to currency
debt_escrow_amounts = None
if hasattr(account, "debt_escrow_amounts") and account.debt_escrow_amounts:
debt_escrow_amounts = {
datetime.date.fromisoformat(date_str): milliunits_to_currency(amount)
for date_str, amount in account.debt_escrow_amounts.items()
}
return cls(
id=account.id,
name=account.name,
type=account.type,
on_budget=account.on_budget,
closed=account.closed,
note=account.note,
balance=milliunits_to_currency(account.balance)
if account.balance is not None
else None,
cleared_balance=milliunits_to_currency(account.cleared_balance)
if account.cleared_balance is not None
else None,
debt_interest_rates=debt_interest_rates,
debt_minimum_payments=debt_minimum_payments,
debt_escrow_amounts=debt_escrow_amounts,
)
class Category(BaseModel):
"""A YNAB category with budget and goal information."""
id: str = Field(..., description="Unique category identifier")
name: str = Field(..., description="Category name")
category_group_id: str = Field(..., description="Category group ID")
category_group_name: str | None = Field(None, description="Category group name")
note: str | None = Field(None, description="Category notes")
budgeted: Decimal | None = Field(None, description="Amount budgeted")
activity: Decimal | None = Field(
None,
description="Spending activity (negative = spending)",
)
balance: Decimal | None = Field(None, description="Available balance")
goal_type: str | None = Field(
None,
description="Goal type: NEED (refill up to X monthly - budget full target), "
"TB (target balance by date), TBD (target by specific date), MF (funding)",
)
goal_target: Decimal | None = Field(None, description="Goal target amount")
goal_percentage_complete: int | None = Field(
None, description="Goal percentage complete"
)
goal_under_funded: Decimal | None = Field(
None, description="Amount under-funded for goal"
)
@classmethod
def from_ynab(
cls, category: ynab.Category, category_group_name: str | None = None
) -> Category:
"""Convert YNAB Category object to our Category model.
Args:
category: The YNAB category object
category_group_name: Optional category group name to include
"""
return cls(
id=category.id,
name=category.name,
category_group_id=category.category_group_id,
category_group_name=category_group_name,
note=category.note,
budgeted=milliunits_to_currency(category.budgeted)
if category.budgeted is not None
else None,
activity=milliunits_to_currency(category.activity)
if category.activity is not None
else None,
balance=milliunits_to_currency(category.balance)
if category.balance is not None
else None,
goal_type=category.goal_type,
goal_target=milliunits_to_currency(category.goal_target)
if category.goal_target is not None
else None,
goal_percentage_complete=category.goal_percentage_complete,
goal_under_funded=milliunits_to_currency(category.goal_under_funded)
if category.goal_under_funded is not None
else None,
)
class CategoryGroup(BaseModel):
"""A YNAB category group with summary totals."""
id: str = Field(..., description="Unique category group identifier")
name: str = Field(..., description="Category group name")
hidden: bool = Field(..., description="Whether hidden from budget view")
category_count: int = Field(..., description="Number of categories in group")
total_budgeted: Decimal | None = Field(None, description="Total budgeted amount")
total_activity: Decimal | None = Field(None, description="Total activity")
total_balance: Decimal | None = Field(None, description="Total balance")
@classmethod
def from_ynab(
cls, category_group: ynab.CategoryGroupWithCategories
) -> CategoryGroup:
"""Convert YNAB CategoryGroup object to our CategoryGroup model.
Calculates aggregated totals from active (non-deleted, non-hidden) categories.
"""
# Calculate totals for the group (exclude deleted and hidden categories)
active_categories = [
cat
for cat in category_group.categories
if not cat.deleted and not cat.hidden
]
total_budgeted = sum(cat.budgeted or 0 for cat in active_categories)
total_activity = sum(cat.activity or 0 for cat in active_categories)
total_balance = sum(cat.balance or 0 for cat in active_categories)
return cls(
id=category_group.id,
name=category_group.name,
hidden=category_group.hidden,
category_count=len(active_categories),
total_budgeted=milliunits_to_currency(total_budgeted),
total_activity=milliunits_to_currency(total_activity),
total_balance=milliunits_to_currency(total_balance),
)
class BudgetMonth(BaseModel):
"""Monthly budget summary with category details.
Includes income, budgeted amounts, spending activity, and category breakdowns.
"""
month: datetime.date | None = Field(None, description="Budget month date")
note: str | None = Field(
None, description="User-defined notes for this budget month"
)
income: Decimal | None = Field(
None, description="Total income for the month in currency units"
)
budgeted: Decimal | None = Field(
None, description="Total amount budgeted across all categories"
)
activity: Decimal | None = Field(
None, description="Total spending activity for the month"
)
to_be_budgeted: Decimal | None = Field(
None, description="Amount remaining to be budgeted (can be negative)"
)
age_of_money: int | None = Field(
None,
description="Age of money in days (how long money sits before being spent)",
)
categories: list[Category] = Field(
..., description="Categories with monthly budget data"
)
pagination: PaginationInfo | None = Field(
None, description="Pagination information"
)
# Response models for tools that need pagination
class AccountsResponse(BaseModel):
"""Response for list_accounts tool."""
accounts: list[Account] = Field(..., description="List of accounts")
pagination: PaginationInfo = Field(..., description="Pagination information")
class CategoriesResponse(BaseModel):
"""Response for list_categories tool."""
categories: list[Category] = Field(..., description="List of categories")
pagination: PaginationInfo = Field(..., description="Pagination information")
def format_flag(flag_color: str | None, flag_name: str | None) -> str | None:
"""Format flag as 'Name (Color)' or just color if no name."""
if not flag_color:
return None
if flag_name:
return f"{flag_name} ({flag_color.title()})"
return flag_color.title()
class BaseTransaction(BaseModel):
"""Base fields shared between Transaction and ScheduledTransaction models."""
id: str = Field(..., description="Unique identifier")
amount: Decimal | None = Field(
None,
description="Amount in currency units (negative = spending, positive = income)",
)
memo: str | None = Field(None, description="User-entered memo")
flag: str | None = Field(
None,
description="Flag as 'Name (Color)' format",
)
account_id: str = Field(..., description="Account ID")
account_name: str | None = Field(None, description="Account name")
payee_id: str | None = Field(None, description="Payee ID")
payee_name: str | None = Field(None, description="Payee name")
category_id: str | None = Field(None, description="Category ID")
category_name: str | None = Field(None, description="Category name")
class Subtransaction(BaseModel):
"""A subtransaction within a split transaction."""
id: str = Field(..., description="Unique subtransaction identifier")
amount: Decimal | None = Field(None, description="Amount in currency units")
memo: str | None = Field(None, description="Memo")
payee_id: str | None = Field(None, description="Payee ID")
payee_name: str | None = Field(None, description="Payee name")
category_id: str | None = Field(None, description="Category ID")
category_name: str | None = Field(None, description="Category name")
class ScheduledSubtransaction(BaseModel):
"""A scheduled subtransaction within a split scheduled transaction."""
id: str = Field(..., description="Unique scheduled subtransaction identifier")
amount: Decimal | None = Field(None, description="Amount in currency units")
memo: str | None = Field(None, description="Memo")
payee_id: str | None = Field(None, description="Payee ID")
payee_name: str | None = Field(None, description="Payee name")
category_id: str | None = Field(None, description="Category ID")
category_name: str | None = Field(None, description="Category name")
class Transaction(BaseTransaction):
"""A YNAB transaction with full details."""
date: datetime.date = Field(..., description="Transaction date")
cleared: str = Field(..., description="Cleared status")
approved: bool = Field(
...,
description="Whether transaction is approved",
)
parent_transaction_id: str | None = Field(
None, description="Parent transaction ID if this is a subtransaction"
)
subtransactions: list[Subtransaction] | None = Field(
None, description="Subtransactions for splits"
)
@classmethod
def from_ynab(
cls,
txn: ynab.TransactionDetail | ynab.HybridTransaction,
repository: YNABRepository | None = None,
) -> Transaction:
"""Convert YNAB transaction object to our Transaction model.
Args:
txn: The YNAB transaction object
repository: Optional repository to resolve parent transaction info
"""
# Convert amount from milliunits
amount = milliunits_to_currency(txn.amount)
# Handle HybridTransaction subtransactions that need parent payee resolution
payee_id = txn.payee_id
payee_name = getattr(txn, "payee_name", None)
# Check if this is a subtransaction that needs parent payee info
if (
hasattr(txn, "type")
and txn.type == "subtransaction"
and not payee_id
and not payee_name
and hasattr(txn, "parent_transaction_id")
and txn.parent_transaction_id
and repository
):
parent_txn = repository.get_transaction_by_id(txn.parent_transaction_id)
parent_payee_id = parent_txn.payee_id
parent_payee_name = getattr(parent_txn, "payee_name", None)
if parent_payee_id or parent_payee_name:
payee_id = parent_payee_id
payee_name = parent_payee_name
# Handle subtransactions if present and available
subtransactions = None
if hasattr(txn, "subtransactions") and txn.subtransactions:
subtransactions = []
for sub in txn.subtransactions:
if not sub.deleted:
# Inherit parent payee info if subtransaction payee is null
sub_payee_id = sub.payee_id if sub.payee_id else payee_id
sub_payee_name = sub.payee_name if sub.payee_name else payee_name
subtransactions.append(
Subtransaction(
id=sub.id,
amount=milliunits_to_currency(sub.amount),
memo=sub.memo,
payee_id=sub_payee_id,
payee_name=sub_payee_name,
category_id=sub.category_id,
category_name=sub.category_name,
)
)
return cls(
id=txn.id,
date=txn.var_date,
amount=amount,
memo=txn.memo,
cleared=txn.cleared,
approved=txn.approved,
flag=format_flag(txn.flag_color, getattr(txn, "flag_name", None)),
account_id=txn.account_id,
account_name=getattr(txn, "account_name", None),
payee_id=payee_id,
payee_name=payee_name,
category_id=txn.category_id,
category_name=getattr(txn, "category_name", None),
parent_transaction_id=getattr(txn, "parent_transaction_id", None),
subtransactions=subtransactions,
)
class ScheduledTransaction(BaseTransaction):
"""A YNAB scheduled transaction with frequency and timing details."""
date_first: datetime.date = Field(..., description="First occurrence date")
date_next: datetime.date = Field(..., description="Next occurrence date")
frequency: str = Field(
...,
description="Recurrence frequency",
)
subtransactions: list[ScheduledSubtransaction] | None = Field(
None, description="Scheduled subtransactions for splits"
)
@classmethod
def from_ynab(cls, st: ynab.ScheduledTransactionDetail) -> ScheduledTransaction:
"""Convert YNAB scheduled transaction to ScheduledTransaction model."""
# Convert amount from milliunits
amount = milliunits_to_currency(st.amount)
# Handle scheduled subtransactions if present and available
subtransactions = None
if hasattr(st, "subtransactions") and st.subtransactions:
subtransactions = []
for sub in st.subtransactions:
if not sub.deleted:
subtransactions.append(
ScheduledSubtransaction(
id=sub.id,
amount=milliunits_to_currency(sub.amount),
memo=sub.memo,
payee_id=sub.payee_id,
payee_name=sub.payee_name,
category_id=sub.category_id,
category_name=sub.category_name,
)
)
return cls(
id=st.id,
date_first=st.date_first,
date_next=st.date_next,
frequency=st.frequency,
amount=amount,
memo=st.memo,
flag=format_flag(st.flag_color, getattr(st, "flag_name", None)),
account_id=st.account_id,
account_name=getattr(st, "account_name", None),
payee_id=st.payee_id,
payee_name=getattr(st, "payee_name", None),
category_id=st.category_id,
category_name=getattr(st, "category_name", None),
subtransactions=subtransactions,
)
class TransactionsResponse(BaseModel):
"""Response for list_transactions tool."""
transactions: list[Transaction] = Field(..., description="List of transactions")
pagination: PaginationInfo = Field(..., description="Pagination information")
class Payee(BaseModel):
"""A YNAB payee (person, company, or entity that receives payments)."""
id: str = Field(..., description="Unique payee identifier")
name: str = Field(..., description="Payee name")
@classmethod
def from_ynab(cls, payee: ynab.Payee) -> Payee:
"""Convert YNAB Payee object to our Payee model."""
return cls(
id=payee.id,
name=payee.name,
)
class PayeesResponse(BaseModel):
"""Response for list_payees tool."""
payees: list[Payee] = Field(..., description="List of payees")
pagination: PaginationInfo = Field(..., description="Pagination information")
class ScheduledTransactionsResponse(BaseModel):
"""Response for list_scheduled_transactions tool."""
scheduled_transactions: list[ScheduledTransaction] = Field(
..., description="List of scheduled transactions"
)
pagination: PaginationInfo = Field(..., description="Pagination information")
```
--------------------------------------------------------------------------------
/repository.py:
--------------------------------------------------------------------------------
```python
"""
YNAB Repository with differential sync.
Provides local-first access to YNAB data with background synchronization.
"""
import logging
import threading
import time
from collections.abc import Callable
from datetime import date, datetime
from typing import Any
import ynab
from ynab.exceptions import ApiException, ConflictException
logger = logging.getLogger(__name__)
class YNABRepository:
"""Local repository for YNAB data with background differential sync."""
def __init__(self, budget_id: str, access_token: str):
self.budget_id = budget_id
self.configuration = ynab.Configuration(access_token=access_token)
# In-memory storage - generic dict for different entity types
self._data: dict[str, list[Any]] = {}
self._server_knowledge: dict[str, int] = {}
self._lock = threading.RLock()
self._last_sync: datetime | None = None
# Testing flag to disable background sync
self._background_sync_enabled = True
def get_accounts(self) -> list[ynab.Account]:
"""Get all accounts from local repository."""
with self._lock:
# If no data exists, do synchronous sync (first time)
if "accounts" not in self._data:
logger.info("No accounts data - performing initial sync")
self.sync_accounts()
# If data exists but is stale, trigger background sync
elif self.needs_sync():
logger.info("Accounts data is stale - triggering background sync")
self._trigger_background_sync("accounts")
return self._data.get("accounts", [])
def get_payees(self) -> list[ynab.Payee]:
"""Get all payees from local repository."""
with self._lock:
# If no data exists, do synchronous sync (first time)
if "payees" not in self._data:
logger.info("No payees data - performing initial sync")
self.sync_payees()
# If data exists but is stale, trigger background sync
elif self.needs_sync():
logger.info("Payees data is stale - triggering background sync")
self._trigger_background_sync("payees")
return self._data.get("payees", [])
def get_category_groups(self) -> list[ynab.CategoryGroupWithCategories]:
"""Get all category groups from local repository."""
with self._lock:
# If no data exists, do synchronous sync (first time)
if "category_groups" not in self._data:
logger.info("No category groups data - performing initial sync")
self.sync_category_groups()
# If data exists but is stale, trigger background sync
elif self.needs_sync():
logger.info(
"Category groups data is stale - triggering background sync"
)
self._trigger_background_sync("category_groups")
return self._data.get("category_groups", [])
def get_transactions(self) -> list[ynab.TransactionDetail]:
"""Get all transactions from local repository."""
with self._lock:
# If no data exists, do synchronous sync (first time)
if "transactions" not in self._data:
logger.info("No transactions data - performing initial sync")
self.sync_transactions()
# If data exists but is stale, trigger background sync
elif self.needs_sync():
logger.info("Transactions data is stale - triggering background sync")
self._trigger_background_sync("transactions")
return self._data.get("transactions", [])
def sync_accounts(self) -> None:
"""Sync accounts with YNAB API using differential sync."""
self._sync_entity("accounts", self._sync_accounts_from_api)
def sync_payees(self) -> None:
"""Sync payees with YNAB API using differential sync."""
self._sync_entity("payees", self._sync_payees_from_api)
def sync_category_groups(self) -> None:
"""Sync category groups with YNAB API using differential sync."""
self._sync_entity("category_groups", self._sync_category_groups_from_api)
def sync_transactions(self) -> None:
"""Sync transactions with YNAB API using differential sync."""
self._sync_entity("transactions", self._sync_transactions_from_api)
def _sync_accounts_from_api(
self, last_knowledge: int | None
) -> tuple[list[ynab.Account], int]:
"""Fetch accounts from YNAB API with optional server knowledge."""
with ynab.ApiClient(self.configuration) as api_client:
accounts_api = ynab.AccountsApi(api_client)
if last_knowledge is not None:
try:
# Try delta sync first
response = self._handle_api_call_with_retry(
lambda: accounts_api.get_accounts(
self.budget_id, last_knowledge_of_server=last_knowledge
)
)
except ConflictException as e:
# Fall back to full sync on stale knowledge
logger.info(
f"Falling back to full accounts sync due to conflict: {e}"
)
response = self._handle_api_call_with_retry(
lambda: accounts_api.get_accounts(self.budget_id)
)
except ApiException as e:
# Log API error and fall back to full sync
logger.warning(f"API error during accounts delta sync: {e}")
response = self._handle_api_call_with_retry(
lambda: accounts_api.get_accounts(self.budget_id)
)
except Exception as e:
# Log unexpected error and re-raise
logger.error(f"Unexpected error during accounts delta sync: {e}")
raise
else:
response = self._handle_api_call_with_retry(
lambda: accounts_api.get_accounts(self.budget_id)
)
return list(response.data.accounts), response.data.server_knowledge
def _sync_payees_from_api(
self, last_knowledge: int | None
) -> tuple[list[ynab.Payee], int]:
"""Fetch payees from YNAB API with optional server knowledge."""
with ynab.ApiClient(self.configuration) as api_client:
payees_api = ynab.PayeesApi(api_client)
if last_knowledge is not None:
try:
# Try delta sync first
response = self._handle_api_call_with_retry(
lambda: payees_api.get_payees(
self.budget_id, last_knowledge_of_server=last_knowledge
)
)
except ConflictException as e:
# Fall back to full sync on stale knowledge
logger.info(
f"Falling back to full payees sync due to conflict: {e}"
)
response = self._handle_api_call_with_retry(
lambda: payees_api.get_payees(self.budget_id)
)
except ApiException as e:
# Log API error and fall back to full sync
logger.warning(f"API error during payees delta sync: {e}")
response = self._handle_api_call_with_retry(
lambda: payees_api.get_payees(self.budget_id)
)
except Exception as e:
# Log unexpected error and re-raise
logger.error(f"Unexpected error during payees delta sync: {e}")
raise
else:
response = self._handle_api_call_with_retry(
lambda: payees_api.get_payees(self.budget_id)
)
return list(response.data.payees), response.data.server_knowledge
def _sync_category_groups_from_api(
self, last_knowledge: int | None
) -> tuple[list[ynab.CategoryGroupWithCategories], int]:
"""Fetch category groups from YNAB API with optional server knowledge."""
with ynab.ApiClient(self.configuration) as api_client:
categories_api = ynab.CategoriesApi(api_client)
if last_knowledge is not None:
try:
# Try delta sync first
response = self._handle_api_call_with_retry(
lambda: categories_api.get_categories(
self.budget_id, last_knowledge_of_server=last_knowledge
)
)
except ConflictException as e:
# Fall back to full sync on stale knowledge
logger.info(
f"Category groups conflict, falling back to full sync: {e}"
)
response = self._handle_api_call_with_retry(
lambda: categories_api.get_categories(self.budget_id)
)
except ApiException as e:
# Log API error and fall back to full sync
logger.warning(f"API error during category groups delta sync: {e}")
response = self._handle_api_call_with_retry(
lambda: categories_api.get_categories(self.budget_id)
)
except Exception as e:
# Log unexpected error and re-raise
logger.error(
f"Unexpected error during category groups delta sync: {e}"
)
raise
else:
response = self._handle_api_call_with_retry(
lambda: categories_api.get_categories(self.budget_id)
)
return list(response.data.category_groups), response.data.server_knowledge
def _sync_transactions_from_api(
self, last_knowledge: int | None
) -> tuple[list[ynab.TransactionDetail], int]:
"""Fetch transactions from YNAB API with optional server knowledge."""
with ynab.ApiClient(self.configuration) as api_client:
transactions_api = ynab.TransactionsApi(api_client)
if last_knowledge is not None:
try:
# Try delta sync first
response = self._handle_api_call_with_retry(
lambda: transactions_api.get_transactions(
self.budget_id, last_knowledge_of_server=last_knowledge
)
)
except ConflictException as e:
# Fall back to full sync on stale knowledge
logger.info(
f"Falling back to full transactions sync due to conflict: {e}"
)
response = self._handle_api_call_with_retry(
lambda: transactions_api.get_transactions(self.budget_id)
)
except ApiException as e:
# Log API error and fall back to full sync
logger.warning(f"API error during transactions delta sync: {e}")
response = self._handle_api_call_with_retry(
lambda: transactions_api.get_transactions(self.budget_id)
)
except Exception as e:
# Log unexpected error and re-raise
logger.error(
f"Unexpected error during transactions delta sync: {e}"
)
raise
else:
response = self._handle_api_call_with_retry(
lambda: transactions_api.get_transactions(self.budget_id)
)
return list(response.data.transactions), response.data.server_knowledge
def _sync_entity(
self, entity_type: str, sync_func: Callable[[int | None], tuple[list[Any], int]]
) -> None:
"""Generic sync method for any entity type."""
with self._lock:
current_knowledge = self._server_knowledge.get(entity_type, 0)
last_knowledge = current_knowledge if current_knowledge > 0 else None
# Fetch from API
entities, new_knowledge = sync_func(last_knowledge)
# Apply changes
if last_knowledge is not None and entity_type in self._data:
# Apply delta changes
self._apply_deltas(entity_type, entities)
else:
# Full refresh
self._data[entity_type] = entities
# Update metadata
self._server_knowledge[entity_type] = new_knowledge
self._last_sync = datetime.now()
def _apply_deltas(self, entity_type: str, delta_entities: list[Any]) -> None:
"""Apply delta changes to an entity list."""
current_entities = self._data.get(entity_type, [])
entity_map = {entity.id: entity for entity in current_entities}
for delta_entity in delta_entities:
if hasattr(delta_entity, "deleted") and delta_entity.deleted:
# Remove deleted entity
entity_map.pop(delta_entity.id, None)
else:
# Add new or update existing entity
entity_map[delta_entity.id] = delta_entity
# Update the entity list
self._data[entity_type] = list(entity_map.values())
def is_initialized(self) -> bool:
"""Check if repository has been initially populated."""
with self._lock:
return len(self._data) > 0 or self._last_sync is not None
def last_sync_time(self) -> datetime | None:
"""Get the last sync time."""
with self._lock:
return self._last_sync
def needs_sync(self, max_age_minutes: int = 5) -> bool:
"""Check if repository needs to be synced based on staleness."""
with self._lock:
if self._last_sync is None:
return True
age_minutes = (datetime.now() - self._last_sync).total_seconds() / 60
return age_minutes > max_age_minutes
def _trigger_background_sync(self, entity_type: str) -> None:
"""Trigger background sync for a specific entity type."""
if not self._background_sync_enabled:
return
sync_method = {
"accounts": self.sync_accounts,
"payees": self.sync_payees,
"category_groups": self.sync_category_groups,
"transactions": self.sync_transactions,
}.get(entity_type)
if sync_method:
sync_thread = threading.Thread(
target=self._background_sync_entity,
args=(entity_type, sync_method),
daemon=True,
name=f"ynab-sync-{entity_type}",
)
sync_thread.start()
def _background_sync_entity(
self, entity_type: str, sync_method: Callable[[], None]
) -> None:
"""Background sync for a specific entity type with error handling."""
try:
logger.info(f"Starting background sync for {entity_type}")
sync_method()
logger.info(f"Completed background sync for {entity_type}")
except Exception as e:
logger.error(f"Background sync failed for {entity_type}: {e}")
# Continue serving stale data on error
def _handle_api_call_with_retry(
self, api_call: Callable[[], Any], max_retries: int = 3
) -> Any:
"""Handle API call with exponential backoff for rate limiting."""
for attempt in range(max_retries):
try:
return api_call()
except ConflictException:
# Let the calling method handle ConflictException for fallback logic
raise
except ApiException as e:
if e.status == 429:
# Rate limited - YNAB allows 200 requests/hour
wait_time = 2**attempt
logger.warning(
f"Rate limited - waiting {wait_time}s (retry {attempt + 1})"
)
if attempt < max_retries - 1:
time.sleep(wait_time)
continue
else:
logger.error("Max retries exceeded for rate limiting")
raise
else:
# Other API error - don't retry, let caller handle
logger.error(f"API error {e.status}: {e}")
raise
except Exception as e:
logger.error(f"Unexpected error during API call: {e}")
raise
def update_month_category(
self, category_id: str, month: date, budgeted_milliunits: int
) -> ynab.Category:
"""Update a category's budget for a specific month."""
with ynab.ApiClient(self.configuration) as api_client:
categories_api = ynab.CategoriesApi(api_client)
save_month_category = ynab.SaveMonthCategory(budgeted=budgeted_milliunits)
patch_wrapper = ynab.PatchMonthCategoryWrapper(category=save_month_category)
response = categories_api.update_month_category(
self.budget_id, month, category_id, patch_wrapper
)
# Invalidate category groups cache since budget amounts changed
with self._lock:
if "category_groups" in self._data:
del self._data["category_groups"]
if "category_groups" in self._server_knowledge:
del self._server_knowledge["category_groups"]
return response.data.category
def update_transaction(
self, transaction_id: str, update_data: dict[str, Any]
) -> ynab.TransactionDetail:
"""Update a transaction with the provided data."""
with ynab.ApiClient(self.configuration) as api_client:
transactions_api = ynab.TransactionsApi(api_client)
# Create the save transaction object
existing_transaction = ynab.ExistingTransaction(**update_data)
put_wrapper = ynab.PutTransactionWrapper(transaction=existing_transaction)
response = transactions_api.update_transaction(
self.budget_id, transaction_id, put_wrapper
)
# Invalidate transactions cache since transaction was modified
with self._lock:
if "transactions" in self._data:
del self._data["transactions"]
if "transactions" in self._server_knowledge:
del self._server_knowledge["transactions"]
return response.data.transaction
def get_transaction_by_id(self, transaction_id: str) -> ynab.TransactionDetail:
"""Get a specific transaction by ID."""
with ynab.ApiClient(self.configuration) as api_client:
transactions_api = ynab.TransactionsApi(api_client)
response = transactions_api.get_transaction_by_id(
self.budget_id, transaction_id
)
return response.data.transaction
def get_transactions_by_filters(
self,
account_id: str | None = None,
category_id: str | None = None,
payee_id: str | None = None,
since_date: date | None = None,
) -> list[ynab.TransactionDetail | ynab.HybridTransaction]:
"""Get transactions using specific YNAB API endpoints for filtering."""
with ynab.ApiClient(self.configuration) as api_client:
transactions_api = ynab.TransactionsApi(api_client)
if account_id:
account_response = transactions_api.get_transactions_by_account(
self.budget_id, account_id, since_date=since_date, type=None
)
return list(account_response.data.transactions)
elif category_id:
category_response = transactions_api.get_transactions_by_category(
self.budget_id, category_id, since_date=since_date, type=None
)
return list(category_response.data.transactions)
elif payee_id:
payee_response = transactions_api.get_transactions_by_payee(
self.budget_id, payee_id, since_date=since_date, type=None
)
return list(payee_response.data.transactions)
else:
# Use general transactions endpoint
general_response = transactions_api.get_transactions(
self.budget_id, since_date=since_date, type=None
)
return list(general_response.data.transactions)
def get_scheduled_transactions(self) -> list[ynab.ScheduledTransactionDetail]:
"""Get scheduled transactions."""
with ynab.ApiClient(self.configuration) as api_client:
scheduled_transactions_api = ynab.ScheduledTransactionsApi(api_client)
response = scheduled_transactions_api.get_scheduled_transactions(
self.budget_id
)
return list(response.data.scheduled_transactions)
def get_month_category_by_id(self, month: date, category_id: str) -> ynab.Category:
"""Get a specific category for a specific month."""
with ynab.ApiClient(self.configuration) as api_client:
categories_api = ynab.CategoriesApi(api_client)
response = categories_api.get_month_category_by_id(
self.budget_id, month, category_id
)
return response.data.category
def get_budget_month(self, month: date) -> ynab.MonthDetail:
"""Get budget month data for a specific month."""
with ynab.ApiClient(self.configuration) as api_client:
months_api = ynab.MonthsApi(api_client)
response = months_api.get_budget_month(self.budget_id, month)
return response.data.month
```
--------------------------------------------------------------------------------
/tests/test_transactions.py:
--------------------------------------------------------------------------------
```python
"""
Tests for transaction-related functionality in YNAB MCP Server.
"""
from datetime import date
from unittest.mock import MagicMock
import pytest
import ynab
from assertions import extract_response_data
from conftest import create_ynab_transaction
from fastmcp.client import Client, FastMCPTransport
from fastmcp.exceptions import ToolError
async def test_list_transactions_basic(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test basic transaction listing without filters."""
txn1 = create_ynab_transaction(
id="txn-1",
transaction_date=date(2024, 1, 15),
amount=-50_000, # -$50.00 outflow
memo="Grocery shopping",
flag_color=ynab.TransactionFlagColor.RED,
account_name="Checking",
payee_id="payee-1",
payee_name="Whole Foods",
category_id="cat-1",
category_name="Groceries",
)
txn2 = create_ynab_transaction(
id="txn-2",
transaction_date=date(2024, 1, 20),
amount=-75_000, # -$75.00 outflow
memo="Dinner",
cleared=ynab.TransactionClearedStatus.UNCLEARED,
account_name="Checking",
payee_id="payee-2",
payee_name="Restaurant XYZ",
category_id="cat-2",
category_name="Dining Out",
)
# Add a deleted transaction that should be filtered out
txn_deleted = create_ynab_transaction(
id="txn-deleted",
transaction_date=date(2024, 1, 10),
amount=-25_000,
memo="Deleted transaction",
account_name="Checking",
payee_id="payee-3",
payee_name="Store ABC",
category_id="cat-1",
category_name="Groceries",
deleted=True, # Should be excluded
)
# Mock repository to return transactions
mock_repository.get_transactions.return_value = [txn2, txn1, txn_deleted]
result = await mcp_client.call_tool("list_transactions", {})
response_data = extract_response_data(result)
# Should have 2 transactions (deleted one excluded)
assert len(response_data["transactions"]) == 2
# Should be sorted by date descending
assert response_data["transactions"][0]["id"] == "txn-2"
assert response_data["transactions"][0]["date"] == "2024-01-20"
assert response_data["transactions"][0]["amount"] == "-75"
assert response_data["transactions"][0]["payee_name"] == "Restaurant XYZ"
assert response_data["transactions"][0]["category_name"] == "Dining Out"
assert response_data["transactions"][1]["id"] == "txn-1"
assert response_data["transactions"][1]["date"] == "2024-01-15"
assert response_data["transactions"][1]["amount"] == "-50"
assert response_data["transactions"][1]["flag"] == "Red"
# Check pagination
assert response_data["pagination"]["total_count"] == 2
assert response_data["pagination"]["has_more"] is False
async def test_list_transactions_with_account_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test transaction listing filtered by account."""
# Create transaction
txn = create_ynab_transaction(
id="txn-acc-1",
transaction_date=date(2024, 2, 1),
amount=-30_000,
memo="Account filtered",
account_id="acc-checking",
account_name="Main Checking",
payee_id="payee-1",
payee_name="Store",
category_id="cat-1",
category_name="Shopping",
)
# Mock repository to return filtered transactions
mock_repository.get_transactions_by_filters.return_value = [txn]
result = await mcp_client.call_tool(
"list_transactions", {"account_id": "acc-checking"}
)
response_data = extract_response_data(result)
assert len(response_data["transactions"]) == 1
assert response_data["transactions"][0]["account_id"] == "acc-checking"
# Verify correct repository method was called
mock_repository.get_transactions_by_filters.assert_called_once_with(
account_id="acc-checking",
category_id=None,
payee_id=None,
since_date=None,
)
async def test_list_transactions_with_amount_filters(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test transaction listing with amount range filters."""
# Create transactions with different amounts
txn_small = create_ynab_transaction(
id="txn-small",
transaction_date=date(2024, 3, 1),
amount=-25_000, # -$25
memo="Small purchase",
payee_id="payee-1",
payee_name="Coffee Shop",
category_id="cat-1",
category_name="Dining Out",
)
txn_medium = create_ynab_transaction(
id="txn-medium",
transaction_date=date(2024, 3, 2),
amount=-60_000, # -$60
memo="Medium purchase",
payee_id="payee-2",
payee_name="Restaurant",
category_id="cat-1",
category_name="Dining Out",
)
txn_large = create_ynab_transaction(
id="txn-large",
transaction_date=date(2024, 3, 3),
amount=-120_000, # -$120
memo="Large purchase",
payee_id="payee-3",
payee_name="Electronics Store",
category_id="cat-2",
category_name="Shopping",
)
# Mock repository to return all transactions for filtering
mock_repository.get_transactions.return_value = [txn_small, txn_medium, txn_large]
# Test with min_amount filter (transactions >= -$50)
result = await mcp_client.call_tool(
"list_transactions",
{
"min_amount": -50.0 # -$50
},
)
response_data = extract_response_data(result)
assert response_data is not None
# Should only include small transaction (-$25 > -$50)
assert len(response_data["transactions"]) == 1
assert response_data["transactions"][0]["id"] == "txn-small"
# Test with max_amount filter (transactions <= -$100)
result = await mcp_client.call_tool(
"list_transactions",
{
"max_amount": -100.0 # -$100
},
)
response_data = extract_response_data(result)
assert response_data is not None
# Should only include large transaction (-$120 < -$100)
assert len(response_data["transactions"]) == 1
assert response_data["transactions"][0]["id"] == "txn-large"
# Test with both min and max filters
result = await mcp_client.call_tool(
"list_transactions",
{
"min_amount": -80.0, # >= -$80
"max_amount": -40.0, # <= -$40
},
)
response_data = extract_response_data(result)
assert response_data is not None
# Should only include medium transaction (-$60)
assert len(response_data["transactions"]) == 1
assert response_data["transactions"][0]["id"] == "txn-medium"
async def test_list_transactions_with_subtransactions(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test transaction listing with split transactions (subtransactions)."""
sub1 = ynab.SubTransaction(
id="sub-1",
transaction_id="txn-split",
amount=-30_000, # -$30
memo="Groceries portion",
payee_id=None,
payee_name=None,
category_id="cat-groceries",
category_name="Groceries",
transfer_account_id=None,
transfer_transaction_id=None,
deleted=False,
)
sub2 = ynab.SubTransaction(
id="sub-2",
transaction_id="txn-split",
amount=-20_000, # -$20
memo="Household items",
payee_id=None,
payee_name=None,
category_id="cat-household",
category_name="Household",
transfer_account_id=None,
transfer_transaction_id=None,
deleted=False,
)
# Deleted subtransaction should be filtered out
sub_deleted = ynab.SubTransaction(
id="sub-deleted",
transaction_id="txn-split",
amount=-10_000,
memo="Deleted sub",
payee_id=None,
payee_name=None,
category_id="cat-other",
category_name="Other",
transfer_account_id=None,
transfer_transaction_id=None,
deleted=True,
)
# Create split transaction
txn_split = create_ynab_transaction(
id="txn-split",
transaction_date=date(2024, 4, 1),
amount=-50_000, # -$50 total
memo="Split transaction at Target",
payee_id="payee-target",
payee_name="Target",
category_id=None, # Split transactions don't have a single category
category_name=None,
subtransactions=[sub1, sub2, sub_deleted],
)
# Mock repository to return split transaction
mock_repository.get_transactions.return_value = [txn_split]
result = await mcp_client.call_tool("list_transactions", {})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["transactions"]) == 1
txn = response_data["transactions"][0]
assert txn["id"] == "txn-split"
assert txn["amount"] == "-50"
# Should have 2 subtransactions (deleted one excluded)
assert len(txn["subtransactions"]) == 2
assert txn["subtransactions"][0]["id"] == "sub-1"
assert txn["subtransactions"][0]["amount"] == "-30"
assert txn["subtransactions"][0]["category_name"] == "Groceries"
assert txn["subtransactions"][1]["id"] == "sub-2"
assert txn["subtransactions"][1]["amount"] == "-20"
assert txn["subtransactions"][1]["category_name"] == "Household"
async def test_list_transactions_pagination(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test transaction listing with pagination."""
# Create many transactions to test pagination
transactions = []
for i in range(5):
txn = create_ynab_transaction(
id=f"txn-{i}",
transaction_date=date(2024, 1, i + 1),
amount=-10_000 * (i + 1),
memo=f"Transaction {i}",
payee_id=f"payee-{i}",
payee_name=f"Store {i}",
category_id="cat-1",
category_name="Shopping",
)
transactions.append(txn)
# Mock repository to return all transactions
mock_repository.get_transactions.return_value = transactions
# Test first page
result = await mcp_client.call_tool("list_transactions", {"limit": 2, "offset": 0})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["transactions"]) == 2
assert response_data["pagination"]["total_count"] == 5
assert response_data["pagination"]["has_more"] is True
# Transactions should be sorted by date descending
assert response_data["transactions"][0]["id"] == "txn-4"
assert response_data["transactions"][1]["id"] == "txn-3"
# Test second page
result = await mcp_client.call_tool("list_transactions", {"limit": 2, "offset": 2})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["transactions"]) == 2
assert response_data["transactions"][0]["id"] == "txn-2"
assert response_data["transactions"][1]["id"] == "txn-1"
async def test_list_transactions_with_category_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test transaction listing filtered by category."""
# Create transaction
txn = create_ynab_transaction(
id="txn-cat-1",
transaction_date=date(2024, 2, 1),
amount=-40_000,
memo="Category filtered",
payee_id="payee-1",
payee_name="Store",
category_id="cat-dining",
category_name="Dining Out",
)
# Mock repository to return filtered transactions
mock_repository.get_transactions_by_filters.return_value = [txn]
result = await mcp_client.call_tool(
"list_transactions", {"category_id": "cat-dining"}
)
response_data = extract_response_data(result)
assert len(response_data["transactions"]) == 1
assert response_data["transactions"][0]["category_id"] == "cat-dining"
# Verify correct repository method was called
mock_repository.get_transactions_by_filters.assert_called_once_with(
account_id=None,
category_id="cat-dining",
payee_id=None,
since_date=None,
)
async def test_list_transactions_with_payee_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test transaction listing filtered by payee."""
# Create transaction
txn = create_ynab_transaction(
id="txn-payee-1",
transaction_date=date(2024, 3, 1),
amount=-80_000,
memo="Payee filtered",
payee_id="payee-amazon",
payee_name="Amazon",
category_id="cat-shopping",
category_name="Shopping",
)
# Mock repository to return filtered transactions
mock_repository.get_transactions_by_filters.return_value = [txn]
result = await mcp_client.call_tool(
"list_transactions", {"payee_id": "payee-amazon"}
)
response_data = extract_response_data(result)
assert len(response_data["transactions"]) == 1
assert response_data["transactions"][0]["payee_id"] == "payee-amazon"
# Verify correct repository method was called
mock_repository.get_transactions_by_filters.assert_called_once_with(
account_id=None,
category_id=None,
payee_id="payee-amazon",
since_date=None,
)
async def test_split_transaction_payee_inheritance(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test that subtransactions inherit parent payee when their payee is null."""
# Create subtransactions where payee is null (simulating API response issue)
sub1 = ynab.SubTransaction(
id="sub-1",
transaction_id="txn-split",
amount=-30_000,
memo="Groceries portion",
payee_id=None, # Null payee_id
payee_name=None, # Null payee_name (the bug we're fixing)
category_id="cat-groceries",
category_name="Groceries",
transfer_account_id=None,
transfer_transaction_id=None,
deleted=False,
)
sub2 = ynab.SubTransaction(
id="sub-2",
transaction_id="txn-split",
amount=-20_000,
memo="Household items",
payee_id=None, # Null payee_id
payee_name=None, # Null payee_name (the bug we're fixing)
category_id="cat-household",
category_name="Household",
transfer_account_id=None,
transfer_transaction_id=None,
deleted=False,
)
# Create parent transaction with valid payee (what user sees in YNAB interface)
txn_split = create_ynab_transaction(
id="txn-split",
transaction_date=date(2024, 8, 11),
amount=-50_000,
memo="Split transaction at Walmart",
payee_id="payee-walmart",
payee_name="Walmart", # Parent has payee name
category_id=None, # Split transactions don't have single category
category_name=None,
subtransactions=[sub1, sub2],
)
# Mock repository to return split transaction
mock_repository.get_transactions.return_value = [txn_split]
result = await mcp_client.call_tool("list_transactions", {})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["transactions"]) == 1
txn = response_data["transactions"][0]
assert txn["id"] == "txn-split"
assert txn["payee_name"] == "Walmart" # Parent should have payee name
assert txn["payee_id"] == "payee-walmart"
# Both subtransactions should inherit parent payee info
assert len(txn["subtransactions"]) == 2
assert txn["subtransactions"][0]["id"] == "sub-1"
assert txn["subtransactions"][0]["payee_name"] == "Walmart" # Inherited!
assert txn["subtransactions"][0]["payee_id"] == "payee-walmart" # Inherited!
assert txn["subtransactions"][1]["id"] == "sub-2"
assert txn["subtransactions"][1]["payee_name"] == "Walmart" # Inherited!
assert txn["subtransactions"][1]["payee_id"] == "payee-walmart" # Inherited!
async def test_hybrid_transaction_subtransaction_payee_resolution(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test HybridTransaction subtransactions that need parent payee resolution."""
# Create a HybridTransaction subtransaction (like from filtered API)
from datetime import date
from ynab.models.hybrid_transaction import HybridTransaction
# This simulates what we get from get_transactions_by_filters()
hybrid_subtxn = HybridTransaction(
id="28a0ce46-a33b-4c3b-bcfc-633a05d9f9ec",
date=date(2025, 8, 11),
amount=-239660, # $239.66 in milliunits
memo=None,
cleared=ynab.TransactionClearedStatus.RECONCILED,
approved=True,
flag_color=None,
flag_name=None,
account_id="914dcb14-13da-49d2-86de-ba241c48f047",
account_name="American Express",
payee_id=None, # Missing payee info (the bug)
payee_name=None, # Missing payee info (the bug)
category_id="cd7c0b0e-7895-4f9f-aa1e-b6e0a22020cd",
category_name="Groceries",
transfer_account_id=None,
transfer_transaction_id=None,
matched_transaction_id=None,
import_id="YNAB:-339660:2025-08-11:1",
import_payee_name=None,
import_payee_name_original=None,
debt_transaction_type=None,
deleted=False,
type="subtransaction", # This is key!
parent_transaction_id="5db47639-1867-41df-a807-23cc23b0ffe9",
)
# Create the parent transaction that has the payee info
parent_txn = create_ynab_transaction(
id="5db47639-1867-41df-a807-23cc23b0ffe9",
transaction_date=date(2025, 8, 11),
amount=-339660, # Total amount
memo=None,
payee_id="payee-walmart",
payee_name="Walmart", # Parent has the payee name
category_id=None,
category_name="Split",
)
# Mock repository to return both transactions
mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
mock_repository.get_transaction_by_id.return_value = parent_txn # For parent lookup
result = await mcp_client.call_tool(
"list_transactions", {"category_id": "cd7c0b0e-7895-4f9f-aa1e-b6e0a22020cd"}
)
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["transactions"]) == 1
txn = response_data["transactions"][0]
assert txn["id"] == "28a0ce46-a33b-4c3b-bcfc-633a05d9f9ec"
assert txn["amount"] == "-239.66"
# The key test: should have resolved parent payee info
assert txn["payee_name"] == "Walmart" # Resolved from parent!
assert txn["payee_id"] == "payee-walmart" # Resolved from parent!
assert (
txn["parent_transaction_id"] == "5db47639-1867-41df-a807-23cc23b0ffe9"
) # Should surface parent ID
async def test_hybrid_transaction_with_missing_parent(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test HybridTransaction subtransaction when parent is not found."""
from datetime import date
from ynab.models.hybrid_transaction import HybridTransaction
# Create a HybridTransaction subtransaction with non-existent parent
hybrid_subtxn = HybridTransaction(
id="orphan-subtxn",
date=date(2025, 8, 11),
amount=-50000,
memo=None,
cleared=ynab.TransactionClearedStatus.CLEARED,
approved=True,
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Test Account",
payee_id=None,
payee_name=None,
category_id="cat-1",
category_name="Test Category",
transfer_account_id=None,
transfer_transaction_id=None,
matched_transaction_id=None,
import_id=None,
import_payee_name=None,
import_payee_name_original=None,
debt_transaction_type=None,
deleted=False,
type="subtransaction",
parent_transaction_id="nonexistent-parent-id", # Parent doesn't exist
)
# Mock repository - parent transaction not found
mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
# Mock get_transaction_by_id to raise an exception (transaction not found)
mock_repository.get_transaction_by_id.side_effect = Exception(
"Transaction not found"
)
# Should raise an exception when parent lookup fails
with pytest.raises(ToolError, match="Transaction not found"):
await mcp_client.call_tool("list_transactions", {"category_id": "cat-1"})
async def test_hybrid_transaction_parent_resolver_exception(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test HybridTransaction when parent resolver throws exception."""
from datetime import date
from ynab.models.hybrid_transaction import HybridTransaction
hybrid_subtxn = HybridTransaction(
id="exception-subtxn",
date=date(2025, 8, 11),
amount=-75000,
memo=None,
cleared=ynab.TransactionClearedStatus.CLEARED,
approved=True,
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Test Account",
payee_id=None,
payee_name=None,
category_id="cat-1",
category_name="Test Category",
transfer_account_id=None,
transfer_transaction_id=None,
matched_transaction_id=None,
import_id=None,
import_payee_name=None,
import_payee_name_original=None,
debt_transaction_type=None,
deleted=False,
type="subtransaction",
parent_transaction_id="exception-parent-id",
)
# Mock repository to return the subtransaction
mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
# Make get_transaction_by_id raise an exception to test exception handling
mock_repository.get_transaction_by_id.side_effect = Exception("Database error")
# Should raise an exception when parent lookup fails
with pytest.raises(ToolError, match="Database error"):
await mcp_client.call_tool("list_transactions", {"category_id": "cat-1"})
async def test_hybrid_transaction_parent_with_null_payee(
mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
) -> None:
"""Test HybridTransaction when parent transaction also has null payee."""
from datetime import date
from ynab.models.hybrid_transaction import HybridTransaction
hybrid_subtxn = HybridTransaction(
id="null-payee-subtxn",
date=date(2025, 8, 11),
amount=-80000,
memo=None,
cleared=ynab.TransactionClearedStatus.CLEARED,
approved=True,
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Test Account",
payee_id=None,
payee_name=None,
category_id="cat-1",
category_name="Test Category",
transfer_account_id=None,
transfer_transaction_id=None,
matched_transaction_id=None,
import_id=None,
import_payee_name=None,
import_payee_name_original=None,
debt_transaction_type=None,
deleted=False,
type="subtransaction",
parent_transaction_id="null-payee-parent-id",
)
# Create parent transaction that also has null payee info
parent_txn = create_ynab_transaction(
id="null-payee-parent-id",
transaction_date=date(2025, 8, 11),
amount=-80000,
memo=None,
payee_id=None, # Parent also has null payee
payee_name=None, # Parent also has null payee
category_id=None,
category_name="Split",
)
# Mock repository to return both
mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
mock_repository.get_transaction_by_id.return_value = (
parent_txn # Parent found but has null payee
)
result = await mcp_client.call_tool("list_transactions", {"category_id": "cat-1"})
response_data = extract_response_data(result)
assert response_data is not None
assert len(response_data["transactions"]) == 1
txn = response_data["transactions"][0]
assert txn["id"] == "null-payee-subtxn"
# Should remain null when parent also has null payee
assert txn["payee_name"] is None
assert txn["payee_id"] is None
assert txn["parent_transaction_id"] == "null-payee-parent-id"
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
import logging
import os
from collections.abc import Sequence
from datetime import date, datetime
from decimal import Decimal
from typing import Literal
import ynab
from fastmcp import FastMCP
from models import (
Account,
AccountsResponse,
BudgetMonth,
CategoriesResponse,
Category,
CategoryGroup,
PaginationInfo,
Payee,
PayeesResponse,
ScheduledTransaction,
ScheduledTransactionsResponse,
Transaction,
TransactionsResponse,
milliunits_to_currency,
)
from repository import YNABRepository
# Configure logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
mcp = FastMCP[None](
name="YNAB",
instructions="""
Gives you access to a user's YNAB budget, including accounts, categories, and
transactions. If a user is ever asking about budgeting, their personal finances,
banking, saving, or investing, their YNAB budget is very relevant to them.
When the user asks about budget categories and "how much is left", they are
talking about the current month.
Budget categories are grouped into category groups, which are important groupings
to the user and should be displayed in a hierarchical manner. Categories will have
the category_group_name and category_group_id available.
The server operates on a single budget configured via the YNAB_BUDGET environment
variable. All tools work with this budget automatically.
""",
)
# Load configuration at module import - fail fast if not configured
BUDGET_ID = os.environ["YNAB_BUDGET"]
ACCESS_TOKEN = os.environ["YNAB_ACCESS_TOKEN"]
ynab_api_configuration = ynab.Configuration(access_token=ACCESS_TOKEN)
# Initialize repository at module level
_repository = YNABRepository(budget_id=BUDGET_ID, access_token=ACCESS_TOKEN)
def _paginate_items[T](
items: list[T], limit: int, offset: int
) -> tuple[list[T], PaginationInfo]:
"""Apply pagination to a list of items and return the page with pagination info."""
total_count = len(items)
start_index = offset
end_index = min(offset + limit, total_count)
items_page = items[start_index:end_index]
has_more = end_index < total_count
pagination = PaginationInfo(
total_count=total_count,
limit=limit,
offset=offset,
has_more=has_more,
)
return items_page, pagination
def _filter_active_items[T](
items: list[T],
*,
exclude_deleted: bool = True,
exclude_hidden: bool = False,
exclude_closed: bool = False,
) -> list[T]:
"""Filter items to exclude deleted/hidden/closed based on flags."""
filtered = []
for item in items:
if exclude_deleted and getattr(item, "deleted", False):
continue
if exclude_hidden and getattr(item, "hidden", False):
continue
if exclude_closed and getattr(item, "closed", False):
continue
filtered.append(item)
return filtered
def _build_category_group_map(
category_groups: list[ynab.CategoryGroupWithCategories],
) -> dict[str, str]:
"""Build a mapping of category_id to category_group_name."""
mapping = {}
for category_group in category_groups:
for category in category_group.categories:
mapping[category.id] = category_group.name
return mapping
def convert_month_to_date(
month: date | Literal["current", "last", "next"],
) -> date:
"""Convert month parameter to appropriate date object for YNAB API.
Args:
month: Month in ISO format (date object), or "current", "last", "next" literals
Returns:
date object representing the first day of the specified month:
- "current": first day of current month
- "last": first day of previous month
- "next": first day of next month
- date object unchanged if already a date
"""
if isinstance(month, date):
return month
today = datetime.now().date()
year, month_num = today.year, today.month
match month:
case "current":
return date(year, month_num, 1)
case "last":
return (
date(year - 1, 12, 1)
if month_num == 1
else date(year, month_num - 1, 1)
)
case "next":
return (
date(year + 1, 1, 1)
if month_num == 12
else date(year, month_num + 1, 1)
)
case _:
raise ValueError(f"Invalid month value: {month}")
@mcp.tool()
def list_accounts(
limit: int = 100,
offset: int = 0,
) -> AccountsResponse:
"""List accounts with pagination.
Only returns open/active accounts. Closed accounts are excluded automatically.
Args:
limit: Maximum number of accounts to return per page (default: 100)
offset: Number of accounts to skip for pagination (default: 0)
Returns:
AccountsResponse with accounts list and pagination information
"""
# Get accounts from repository (handles sync automatically if needed)
accounts = _repository.get_accounts()
# Apply existing filtering and pagination logic
active_accounts = _filter_active_items(accounts, exclude_closed=True)
all_accounts = [Account.from_ynab(account) for account in active_accounts]
accounts_page, pagination = _paginate_items(all_accounts, limit, offset)
return AccountsResponse(accounts=accounts_page, pagination=pagination)
@mcp.tool()
def list_categories(
limit: int = 50,
offset: int = 0,
) -> CategoriesResponse:
"""List categories with pagination.
Only returns active/visible categories. Hidden and deleted categories are excluded
automatically.
Args:
limit: Maximum number of categories to return per page (default: 50)
offset: Number of categories to skip for pagination (default: 0)
Returns:
CategoriesResponse with categories list and pagination information
"""
category_groups = _repository.get_category_groups()
all_categories = []
for category_group in category_groups:
active_categories = _filter_active_items(
category_group.categories, exclude_hidden=True
)
for category in active_categories:
all_categories.append(
Category.from_ynab(category, category_group.name).model_dump()
)
categories_page, pagination = _paginate_items(all_categories, limit, offset)
# Convert dict categories back to Category objects
category_objects = [Category(**cat_dict) for cat_dict in categories_page]
return CategoriesResponse(categories=category_objects, pagination=pagination)
@mcp.tool()
def list_category_groups() -> list[CategoryGroup]:
"""List category groups (lighter weight than full categories).
Returns:
List of category groups
"""
category_groups = _repository.get_category_groups()
active_groups = _filter_active_items(category_groups)
groups = [
CategoryGroup.from_ynab(category_group) for category_group in active_groups
]
return groups
@mcp.tool()
def get_budget_month(
month: date | Literal["current", "last", "next"] = "current",
limit: int = 50,
offset: int = 0,
) -> BudgetMonth:
"""Get budget data for a specific month including category budgets, activity, and
balances with pagination.
Only returns active/visible categories. Hidden and deleted categories are excluded
automatically.
Args:
month: Specifies which budget month to retrieve:
• "current": Current calendar month
• "last": Previous calendar month
• "next": Next calendar month
• date object: Specific month (uses first day of month)
Examples: "current", date(2024, 3, 1) for March 2024 (default: "current")
limit: Maximum number of categories to return per page (default: 50)
offset: Number of categories to skip for pagination (default: 0)
Returns:
BudgetMonth with month info, categories, and pagination
"""
converted_month = convert_month_to_date(month)
month_data = _repository.get_budget_month(converted_month)
# Map category IDs to group names
category_groups = _repository.get_category_groups()
category_group_map = _build_category_group_map(category_groups)
all_categories = []
active_categories = _filter_active_items(month_data.categories, exclude_hidden=True)
for category in active_categories:
group_name = category_group_map.get(category.id)
all_categories.append(Category.from_ynab(category, group_name))
categories_page, pagination = _paginate_items(all_categories, limit, offset)
return BudgetMonth(
month=month_data.month,
note=month_data.note,
income=milliunits_to_currency(month_data.income),
budgeted=milliunits_to_currency(month_data.budgeted),
activity=milliunits_to_currency(month_data.activity),
to_be_budgeted=milliunits_to_currency(month_data.to_be_budgeted),
age_of_money=month_data.age_of_money,
categories=categories_page,
pagination=pagination,
)
@mcp.tool()
def get_month_category_by_id(
category_id: str,
month: date | Literal["current", "last", "next"] = "current",
) -> Category:
"""Get a specific category's data for a specific month.
Args:
category_id: Unique identifier for the category (required)
month: Specifies which budget month to retrieve:
• "current": Current calendar month
• "last": Previous calendar month
• "next": Next calendar month
• date object: Specific month (uses first day of month)
Examples: "current", date(2024, 3, 1) for March 2024 (default: "current")
Returns:
Category with budget data for the specified month
"""
converted_month = convert_month_to_date(month)
category = _repository.get_month_category_by_id(converted_month, category_id)
# Fetch category groups to get group name
category_groups = _repository.get_category_groups()
category_group_map = _build_category_group_map(category_groups)
group_name = category_group_map.get(category_id)
return Category.from_ynab(category, group_name)
@mcp.tool()
def list_transactions(
account_id: str | None = None,
category_id: str | None = None,
payee_id: str | None = None,
since_date: date | None = None,
min_amount: Decimal | None = None,
max_amount: Decimal | None = None,
limit: int = 25,
offset: int = 0,
) -> TransactionsResponse:
"""List transactions with powerful filtering options for financial analysis.
This tool supports various filters that can be combined:
- Filter by account to see transactions for a specific account
- Filter by category to analyze spending in a category (e.g., "Dining Out")
- Filter by payee to see all transactions with a specific merchant (e.g., "Amazon")
- Filter by date range using since_date
- Filter by amount range using min_amount and/or max_amount
Example queries this tool can answer:
- "Show me all transactions over $50 in Dining Out this year"
→ Use: category_id="cat_dining_out_id", min_amount=50.00,
since_date=date(2024, 1, 1)
- "How much have I spent at Amazon this month"
→ Use: payee_id="payee_amazon_id", since_date=date(2024, 12, 1)
- "List recent transactions in my checking account"
→ Use: account_id="acc_checking_id"
Args:
account_id: Filter by specific account (optional)
category_id: Filter by specific category (optional)
payee_id: Filter by specific payee (optional)
since_date: Only show transactions on or after this date. Accepts date objects
in YYYY-MM-DD format (e.g., date(2024, 1, 1)) (optional)
min_amount: Only show transactions with amount >= this value in currency units.
Use negative values for outflows/expenses
(e.g., -50.00 for $50+ expenses) (optional)
max_amount: Only show transactions with amount <= this value in currency units.
Use negative values for outflows/expenses
(e.g., -10.00 for under $10 expenses) (optional)
limit: Maximum number of transactions to return per page (default: 25)
offset: Number of transactions to skip for pagination (default: 0)
Returns:
TransactionsResponse with filtered transactions and pagination info
"""
# Use repository to get transactions with appropriate filtering
transactions_data: Sequence[ynab.TransactionDetail | ynab.HybridTransaction]
if account_id or category_id or payee_id or since_date:
# Use filtered endpoint for specific filters
transactions_data = _repository.get_transactions_by_filters(
account_id=account_id,
category_id=category_id,
payee_id=payee_id,
since_date=since_date,
)
else:
# Use cached transactions for general queries
transactions_data = _repository.get_transactions()
active_transactions = _filter_active_items(list(transactions_data))
all_transactions = []
for txn in active_transactions:
# Apply amount filters (check milliunits directly for efficiency)
if (
min_amount is not None
and txn.amount is not None
and txn.amount < (min_amount * 1000)
):
continue
if (
max_amount is not None
and txn.amount is not None
and txn.amount > (max_amount * 1000)
):
continue
all_transactions.append(Transaction.from_ynab(txn, _repository))
# Sort by date descending (most recent first)
all_transactions.sort(key=lambda t: t.date, reverse=True)
transactions_page, pagination = _paginate_items(all_transactions, limit, offset)
return TransactionsResponse(transactions=transactions_page, pagination=pagination)
@mcp.tool()
def list_payees(
limit: int = 50,
offset: int = 0,
) -> PayeesResponse:
"""List payees for a specific budget with pagination.
Payees are the entities you pay money to (merchants, people, companies, etc.).
This tool helps you find payee IDs for filtering transactions or analyzing spending
patterns. Only returns active payees. Deleted payees are excluded automatically.
Example queries this tool can answer:
- "List all my payees"
- "Find the payee ID for Amazon"
- "Show me all merchants I've paid"
Args:
limit: Maximum number of payees to return per page (default: 50)
offset: Number of payees to skip for pagination (default: 0)
Returns:
PayeesResponse with payees list and pagination information
"""
# Get payees from repository (syncs automatically if needed)
payees = _repository.get_payees()
active_payees = _filter_active_items(payees)
all_payees = [Payee.from_ynab(payee) for payee in active_payees]
# Sort by name for easier browsing
all_payees.sort(key=lambda p: p.name.lower())
payees_page, pagination = _paginate_items(all_payees, limit, offset)
return PayeesResponse(payees=payees_page, pagination=pagination)
@mcp.tool()
def find_payee(
name_search: str,
limit: int = 10,
) -> PayeesResponse:
"""Find payees by searching for name substrings (case-insensitive).
This tool is perfect for finding specific payees when you know part of their name.
Much more efficient than paginating through all payees with list_payees.
Only returns active payees. Deleted payees are excluded automatically.
Example queries this tool can answer:
- "Find Amazon payee ID" (use name_search="amazon")
- "Show me all Starbucks locations" (use name_search="starbucks")
- "Find payees with 'grocery' in the name" (use name_search="grocery")
Args:
name_search: Search term to match against payee names (case-insensitive
substring match). Examples: "amazon", "starbucks", "grocery"
limit: Maximum number of matching payees to return (default: 10)
Returns:
PayeesResponse with matching payees and pagination information
"""
# Get payees from repository (syncs automatically if needed)
payees = _repository.get_payees()
active_payees = _filter_active_items(payees)
search_term = name_search.lower().strip()
matching_payees = [
Payee.from_ynab(payee)
for payee in active_payees
if search_term in payee.name.lower()
]
# Sort by name for easier browsing
matching_payees.sort(key=lambda p: p.name.lower())
# Apply limit (no offset since this is a search, not pagination)
limited_payees = matching_payees[:limit]
# Create pagination info showing search results
total_count = len(matching_payees)
has_more = len(matching_payees) > limit
pagination = PaginationInfo(
total_count=total_count,
limit=limit,
offset=0,
has_more=has_more,
)
return PayeesResponse(payees=limited_payees, pagination=pagination)
@mcp.tool()
def list_scheduled_transactions(
account_id: str | None = None,
category_id: str | None = None,
payee_id: str | None = None,
frequency: str | None = None,
upcoming_days: int | None = None,
min_amount: Decimal | None = None,
max_amount: Decimal | None = None,
limit: int = 25,
offset: int = 0,
) -> ScheduledTransactionsResponse:
"""List scheduled transactions with powerful filtering options for analysis.
This tool supports various filters that can be combined:
- Filter by account to see scheduled transactions for a specific account
- Filter by category to analyze recurring spending (e.g., "Monthly Bills")
- Filter by payee to see scheduled transactions (e.g., "Netflix")
- Filter by frequency to find daily, weekly, monthly, etc. recurring transactions
- Filter by upcoming_days to see what's scheduled in the next N days
- Filter by amount range using min_amount and/or max_amount
Example queries this tool can answer:
- "Show me all monthly recurring expenses" (use frequency="monthly")
- "What bills are due in the next 7 days?" (use upcoming_days=7)
- "List all Netflix subscriptions" (use payee search first, then filter by payee_id)
- "Show scheduled transactions over $100" (use min_amount=100)
Args:
account_id: Filter by specific account (optional)
category_id: Filter by specific category (optional)
payee_id: Filter by specific payee (optional)
frequency: Filter by recurrence frequency. Valid values:
• never, daily, weekly
• everyOtherWeek, twiceAMonth, every4Weeks
• monthly, everyOtherMonth, every3Months, every4Months
• twiceAYear, yearly, everyOtherYear
(optional)
upcoming_days: Only show scheduled transactions with next occurrence
within this many days (optional)
min_amount: Only show scheduled transactions with amount >= this value
in currency units. Use negative values for outflows/expenses
(optional)
max_amount: Only show scheduled transactions with amount <= this value
in currency units. Use negative values for outflows/expenses
(optional)
limit: Maximum number of scheduled transactions to return per page (default: 25)
offset: Number of scheduled transactions to skip for pagination (default: 0)
Returns:
ScheduledTransactionsResponse with filtered scheduled transactions and
pagination info
"""
scheduled_transactions_data = _repository.get_scheduled_transactions()
active_scheduled_transactions = _filter_active_items(scheduled_transactions_data)
all_scheduled_transactions = []
for st in active_scheduled_transactions:
# Apply filters
if account_id and st.account_id != account_id:
continue
if category_id and st.category_id != category_id:
continue
if payee_id and st.payee_id != payee_id:
continue
if frequency and st.frequency != frequency:
continue
# Apply upcoming_days filter
if upcoming_days is not None:
days_until_next = (st.date_next - datetime.now().date()).days
if days_until_next > upcoming_days:
continue
# Apply amount filters (check milliunits directly for efficiency)
if min_amount is not None and st.amount < (min_amount * 1000):
continue
if max_amount is not None and st.amount > (max_amount * 1000):
continue
all_scheduled_transactions.append(ScheduledTransaction.from_ynab(st))
# Sort by next date ascending (earliest scheduled first)
all_scheduled_transactions.sort(key=lambda st: st.date_next)
scheduled_transactions_page, pagination = _paginate_items(
all_scheduled_transactions, limit, offset
)
return ScheduledTransactionsResponse(
scheduled_transactions=scheduled_transactions_page, pagination=pagination
)
@mcp.tool()
def update_category_budget(
category_id: str,
budgeted: Decimal,
month: date | Literal["current", "last", "next"] = "current",
) -> Category:
"""Update the budgeted amount for a category in a specific month.
This tool allows you to assign money to budget categories, which is essential
for monthly budget maintenance and reallocation.
IMPORTANT: For categories with NEED goals (refill up to X monthly), budget the
full goal_target amount regardless of current balance. These goals expect the
full target to be budgeted each month.
Args:
category_id: Unique identifier for the category to update (required)
budgeted: Amount to budget for this category in currency units (required)
month: Budget month to update:
• "current": Current calendar month
• "last": Previous calendar month
• "next": Next calendar month
• date object: Specific month (uses first day of month)
(default: "current")
Returns:
Category with updated budget information
"""
converted_month = convert_month_to_date(month)
# Convert currency units to milliunits
budgeted_milliunits = int(budgeted * 1000)
# Use repository update method with cache invalidation
updated_category = _repository.update_month_category(
category_id, converted_month, budgeted_milliunits
)
# Get category group name for the response
category_groups = _repository.get_category_groups()
category_group_map = _build_category_group_map(category_groups)
group_name = category_group_map.get(category_id)
return Category.from_ynab(updated_category, group_name)
@mcp.tool()
def update_transaction(
transaction_id: str,
category_id: str | None = None,
payee_id: str | None = None,
memo: str | None = None,
) -> Transaction:
"""Update an existing transaction's details.
This tool allows you to modify transaction properties, most commonly
to assign the correct category to imported or uncategorized transactions.
Args:
transaction_id: Unique identifier for the transaction to update (required)
category_id: Category ID to assign (optional)
payee_id: Payee ID to assign (optional)
memo: Transaction memo (optional)
Returns:
Transaction with updated information
"""
# First, get the existing transaction to preserve its current values
existing_txn = _repository.get_transaction_by_id(transaction_id)
# Build the update data starting with existing transaction values
update_data = {
"account_id": existing_txn.account_id,
"date": existing_txn.var_date, # ExistingTransaction uses 'date'
"amount": existing_txn.amount,
"payee_id": existing_txn.payee_id,
"payee_name": existing_txn.payee_name,
"category_id": existing_txn.category_id,
"memo": existing_txn.memo,
"cleared": existing_txn.cleared,
"approved": existing_txn.approved,
"flag_color": existing_txn.flag_color,
"subtransactions": existing_txn.subtransactions,
}
# Apply only the fields we want to change
if category_id is not None:
update_data["category_id"] = category_id
if payee_id is not None:
update_data["payee_id"] = payee_id
if memo is not None:
update_data["memo"] = memo
# Use repository update method with cache invalidation
updated_transaction = _repository.update_transaction(transaction_id, update_data)
return Transaction.from_ynab(updated_transaction, _repository)
```
--------------------------------------------------------------------------------
/tests/test_scheduled_transactions.py:
--------------------------------------------------------------------------------
```python
"""
Tests for scheduled transaction functionality in YNAB MCP Server.
Tests the list_scheduled_transactions tool with various filters and scenarios.
"""
from datetime import date
from unittest.mock import MagicMock
import ynab
from assertions import extract_response_data
from fastmcp.client import Client, FastMCPTransport
async def test_list_scheduled_transactions_basic(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test basic scheduled transaction listing without filters."""
st1 = ynab.ScheduledTransactionDetail(
id="st-1",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-120_000, # -$120.00 outflow
memo="Netflix subscription",
flag_color=ynab.TransactionFlagColor.RED,
flag_name="Entertainment",
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Netflix",
category_id="cat-1",
category_name="Entertainment",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st2 = ynab.ScheduledTransactionDetail(
id="st-2",
date_first=date(2024, 1, 15),
date_next=date(2024, 1, 29),
frequency="weekly",
amount=-5_000, # -$5.00 outflow
memo="Weekly coffee",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-2",
payee_name="Coffee Shop",
category_id="cat-2",
category_name="Dining Out",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Add a deleted scheduled transaction that should be filtered out
st_deleted = ynab.ScheduledTransactionDetail(
id="st-deleted",
date_first=date(2024, 1, 1),
date_next=date(2024, 3, 1),
frequency="monthly",
amount=-50_000,
memo="Deleted subscription",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-3",
payee_name="Old Service",
category_id="cat-1",
category_name="Entertainment",
transfer_account_id=None,
deleted=True, # Should be excluded
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [
st2,
st1,
st_deleted,
]
result = await mcp_client.call_tool("list_scheduled_transactions", {})
response_data = extract_response_data(result)
# Should have 2 scheduled transactions (deleted one excluded)
assert len(response_data["scheduled_transactions"]) == 2
# Should be sorted by next date ascending (earliest scheduled first)
assert response_data["scheduled_transactions"][0]["id"] == "st-2"
assert response_data["scheduled_transactions"][0]["date_next"] == "2024-01-29"
assert response_data["scheduled_transactions"][0]["frequency"] == "weekly"
assert response_data["scheduled_transactions"][0]["amount"] == "-5"
assert response_data["scheduled_transactions"][0]["payee_name"] == "Coffee Shop"
assert response_data["scheduled_transactions"][1]["id"] == "st-1"
assert response_data["scheduled_transactions"][1]["date_next"] == "2024-02-01"
assert response_data["scheduled_transactions"][1]["frequency"] == "monthly"
assert response_data["scheduled_transactions"][1]["amount"] == "-120"
assert response_data["scheduled_transactions"][1]["flag"] == "Entertainment (Red)"
# Check pagination
assert response_data["pagination"]["total_count"] == 2
assert response_data["pagination"]["has_more"] is False
async def test_list_scheduled_transactions_with_frequency_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by frequency."""
st_monthly = ynab.ScheduledTransactionDetail(
id="st-monthly",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-100_000,
memo="Monthly bill",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Electric Company",
category_id="cat-1",
category_name="Utilities",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st_weekly = ynab.ScheduledTransactionDetail(
id="st-weekly",
date_first=date(2024, 1, 8),
date_next=date(2024, 1, 15),
frequency="weekly",
amount=-2_500,
memo="Weekly groceries",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-2",
payee_name="Grocery Store",
category_id="cat-2",
category_name="Groceries",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_monthly, st_weekly]
# Test filtering by monthly frequency
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"frequency": "monthly"}
)
response_data = extract_response_data(result)
# Should only have the monthly scheduled transaction
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-monthly"
assert response_data["scheduled_transactions"][0]["frequency"] == "monthly"
assert (
response_data["scheduled_transactions"][0]["payee_name"] == "Electric Company"
)
# Check pagination
assert response_data["pagination"]["total_count"] == 1
assert response_data["pagination"]["has_more"] is False
async def test_list_scheduled_transactions_with_upcoming_days_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by upcoming days."""
# Scheduled for 5 days from now
st_soon = ynab.ScheduledTransactionDetail(
id="st-soon",
date_first=date(2024, 1, 1),
date_next=date(2024, 1, 20), # 5 days from "today" (2024-01-15)
frequency="monthly",
amount=-50_000,
memo="Due soon",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Due Soon Co",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Scheduled for 15 days from now
st_later = ynab.ScheduledTransactionDetail(
id="st-later",
date_first=date(2024, 1, 1),
date_next=date(2024, 1, 30), # 15 days from "today" (2024-01-15)
frequency="monthly",
amount=-75_000,
memo="Due later",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-2",
payee_name="Due Later Co",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_soon, st_later]
# Mock datetime.now() to return a fixed date for testing
from unittest.mock import patch
import server
with patch.object(server, "datetime") as mock_datetime:
mock_datetime.now.return_value.date.return_value = date(2024, 1, 15)
# Test filtering by upcoming 7 days
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"upcoming_days": 7}
)
response_data = extract_response_data(result)
# Should only have the transaction due within 7 days
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-soon"
assert response_data["scheduled_transactions"][0]["payee_name"] == "Due Soon Co"
async def test_list_scheduled_transactions_with_amount_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by amount range."""
st_small = ynab.ScheduledTransactionDetail(
id="st-small",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-1_000, # -$1.00
memo="Small expense",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Small Store",
category_id="cat-1",
category_name="Misc",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st_large = ynab.ScheduledTransactionDetail(
id="st-large",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-500_000, # -$500.00
memo="Large expense",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-2",
payee_name="Large Store",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_small, st_large]
# Test filtering by minimum amount (expenses <= -$10, i.e., larger expenses)
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"max_amount": -10}
)
response_data = extract_response_data(result)
# Should only have the large transaction (<= -$10)
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-large"
assert response_data["scheduled_transactions"][0]["amount"] == "-500"
async def test_list_scheduled_transactions_with_account_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by account."""
st_checking = ynab.ScheduledTransactionDetail(
id="st-checking",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-100_000,
memo="Checking account expense",
flag_color=None,
flag_name=None,
account_id="acc-checking",
account_name="Checking",
payee_id="payee-1",
payee_name="Merchant A",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st_savings = ynab.ScheduledTransactionDetail(
id="st-savings",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-50_000,
memo="Savings account expense",
flag_color=None,
flag_name=None,
account_id="acc-savings",
account_name="Savings",
payee_id="payee-2",
payee_name="Merchant B",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_checking, st_savings]
# Test filtering by checking account
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"account_id": "acc-checking"}
)
response_data = extract_response_data(result)
# Should only have the checking account scheduled transaction
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-checking"
assert response_data["scheduled_transactions"][0]["account_name"] == "Checking"
async def test_list_scheduled_transactions_with_category_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by category."""
st_bills = ynab.ScheduledTransactionDetail(
id="st-bills",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-100_000,
memo="Monthly bill",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Utility Co",
category_id="cat-bills",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st_entertainment = ynab.ScheduledTransactionDetail(
id="st-entertainment",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-1_500,
memo="Entertainment subscription",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-2",
payee_name="Streaming Service",
category_id="cat-entertainment",
category_name="Entertainment",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [
st_bills,
st_entertainment,
]
# Test filtering by bills category
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"category_id": "cat-bills"}
)
response_data = extract_response_data(result)
# Should only have the bills category scheduled transaction
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-bills"
assert response_data["scheduled_transactions"][0]["category_name"] == "Bills"
async def test_list_scheduled_transactions_with_min_amount_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by minimum amount."""
st_small = ynab.ScheduledTransactionDetail(
id="st-small",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-1_000, # -$1.00
memo="Small expense",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Small Store",
category_id="cat-1",
category_name="Misc",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st_large = ynab.ScheduledTransactionDetail(
id="st-large",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-500_000, # -$500.00
memo="Large expense",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-2",
payee_name="Large Store",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_small, st_large]
# Test filtering by minimum amount (only expenses >= -$5, excludes -$500)
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"min_amount": -5}
)
response_data = extract_response_data(result)
# Should only have the small transaction (>= -$5)
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-small"
assert response_data["scheduled_transactions"][0]["amount"] == "-1"
async def test_list_scheduled_transactions_with_payee_filter(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing filtered by payee."""
st_netflix = ynab.ScheduledTransactionDetail(
id="st-netflix",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-1_500,
memo="Netflix subscription",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-netflix",
payee_name="Netflix",
category_id="cat-1",
category_name="Entertainment",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
st_spotify = ynab.ScheduledTransactionDetail(
id="st-spotify",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-1_000,
memo="Spotify subscription",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-spotify",
payee_name="Spotify",
category_id="cat-1",
category_name="Entertainment",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_netflix, st_spotify]
# Test filtering by Netflix payee
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"payee_id": "payee-netflix"}
)
response_data = extract_response_data(result)
# Should only have the Netflix scheduled transaction
assert len(response_data["scheduled_transactions"]) == 1
assert response_data["scheduled_transactions"][0]["id"] == "st-netflix"
assert response_data["scheduled_transactions"][0]["payee_name"] == "Netflix"
async def test_list_scheduled_transactions_pagination(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing with pagination."""
# Create multiple scheduled transactions
scheduled_transactions = []
for i in range(15):
st = ynab.ScheduledTransactionDetail(
id=f"st-{i}",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, i + 1), # Different next dates for sorting
frequency="monthly",
amount=-10_000 * (i + 1),
memo=f"Transaction {i}",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id=f"payee-{i}",
payee_name=f"Payee {i}",
category_id="cat-1",
category_name="Bills",
transfer_account_id=None,
deleted=False,
subtransactions=[],
)
scheduled_transactions.append(st)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = scheduled_transactions
# Test first page with limit
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"limit": 5, "offset": 0}
)
response_data = extract_response_data(result)
# Should have 5 scheduled transactions
assert len(response_data["scheduled_transactions"]) == 5
assert response_data["pagination"]["total_count"] == 15
assert response_data["pagination"]["has_more"] is True
# Test second page
result = await mcp_client.call_tool(
"list_scheduled_transactions", {"limit": 5, "offset": 5}
)
response_data = extract_response_data(result)
# Should have next 5 scheduled transactions
assert len(response_data["scheduled_transactions"]) == 5
assert response_data["pagination"]["total_count"] == 15
assert response_data["pagination"]["has_more"] is True
async def test_list_scheduled_transactions_with_subtransactions(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing with split transactions (subtransactions)."""
# Create scheduled subtransactions
sub1 = ynab.ScheduledSubTransaction(
id="sub-1",
scheduled_transaction_id="st-split",
amount=-30_000, # -$30.00 for groceries
memo="Groceries portion",
payee_id="payee-1",
payee_name="Grocery Store",
category_id="cat-groceries",
category_name="Groceries",
transfer_account_id=None,
deleted=False,
)
sub2 = ynab.ScheduledSubTransaction(
id="sub-2",
scheduled_transaction_id="st-split",
amount=-20_000, # -$20.00 for household
memo="Household portion",
payee_id="payee-1",
payee_name="Grocery Store",
category_id="cat-household",
category_name="Household",
transfer_account_id=None,
deleted=False,
)
st_split = ynab.ScheduledTransactionDetail(
id="st-split",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-50_000, # -$50.00 total (should equal sum of subtransactions)
memo="Split transaction",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Grocery Store",
category_id=None, # Split transactions don't have a main category
category_name=None,
transfer_account_id=None,
deleted=False,
subtransactions=[sub1, sub2],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_split]
result = await mcp_client.call_tool("list_scheduled_transactions", {})
response_data = extract_response_data(result)
# Should have 1 scheduled transaction with subtransactions
assert len(response_data["scheduled_transactions"]) == 1
st = response_data["scheduled_transactions"][0]
assert st["id"] == "st-split"
assert st["amount"] == "-50"
assert st["memo"] == "Split transaction"
# Check subtransactions
assert len(st["subtransactions"]) == 2
assert st["subtransactions"][0]["id"] == "sub-1"
assert st["subtransactions"][0]["amount"] == "-30"
assert st["subtransactions"][0]["category_name"] == "Groceries"
assert st["subtransactions"][1]["id"] == "sub-2"
assert st["subtransactions"][1]["amount"] == "-20"
assert st["subtransactions"][1]["category_name"] == "Household"
async def test_list_scheduled_transactions_with_deleted_subtransactions(
mock_repository: MagicMock,
mcp_client: Client[FastMCPTransport],
) -> None:
"""Test scheduled transaction listing excludes deleted subtransactions."""
# Create active and deleted scheduled subtransactions
sub_active = ynab.ScheduledSubTransaction(
id="sub-active",
scheduled_transaction_id="st-mixed",
amount=-30_000, # -$30.00
memo="Active subtransaction",
payee_id="payee-1",
payee_name="Store",
category_id="cat-1",
category_name="Active Category",
transfer_account_id=None,
deleted=False,
)
sub_deleted = ynab.ScheduledSubTransaction(
id="sub-deleted",
scheduled_transaction_id="st-mixed",
amount=-20_000, # -$20.00
memo="Deleted subtransaction",
payee_id="payee-1",
payee_name="Store",
category_id="cat-2",
category_name="Deleted Category",
transfer_account_id=None,
deleted=True, # Should be excluded
)
st_mixed = ynab.ScheduledTransactionDetail(
id="st-mixed",
date_first=date(2024, 1, 1),
date_next=date(2024, 2, 1),
frequency="monthly",
amount=-50_000, # -$50.00 total
memo="Mixed subtransactions",
flag_color=None,
flag_name=None,
account_id="acc-1",
account_name="Checking",
payee_id="payee-1",
payee_name="Store",
category_id=None,
category_name=None,
transfer_account_id=None,
deleted=False,
subtransactions=[sub_active, sub_deleted],
)
# Mock repository to return scheduled transactions
mock_repository.get_scheduled_transactions.return_value = [st_mixed]
result = await mcp_client.call_tool("list_scheduled_transactions", {})
response_data = extract_response_data(result)
# Should have 1 scheduled transaction with only active subtransactions
assert len(response_data["scheduled_transactions"]) == 1
st = response_data["scheduled_transactions"][0]
assert st["id"] == "st-mixed"
assert st["amount"] == "-50"
# Should only have the active subtransaction (deleted one excluded)
assert len(st["subtransactions"]) == 1
assert st["subtransactions"][0]["id"] == "sub-active"
assert st["subtransactions"][0]["category_name"] == "Active Category"
```