This is page 1 of 2. Use http://codebase.md/klauern/mcp-ynab?lines=true&page={x} to view the full context.
# Directory Structure
```
├── .cursor
│ └── rules
│ ├── mcp.mdc
│ └── project-setup.mdc
├── .env.example
├── .gitignore
├── CLAUDE.md
├── docs
│ ├── llms-full.txt
│ └── mcp-py-sdk.md
├── mise.toml
├── package-lock.json
├── pyproject.toml
├── README.md
├── src
│ └── mcp_ynab
│ ├── __init__.py
│ ├── __main__.py
│ └── server.py
├── Taskfile.yml
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_environment.py
│ └── test_server.py
├── todo.txt
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
1 | YNAB_API_KEY=your_ynab_api_key
2 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 |
12 | # local preference files
13 | .config/mcp-ynab/
14 | preferred_budget_id.json
15 | budget_category_cache.json
16 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # MCP YNAB Server
2 |
3 | An MCP server implementation that provides access to YNAB (You Need A Budget) functionality through the Model Context Protocol.
4 |
5 | ## Features
6 |
7 | - View account balances and transactions
8 | - Create new transactions
9 | - Access YNAB data through standardized MCP resources
10 |
11 | ## Installation
12 |
13 | ```bash
14 | uv pip install -e .
15 | ```
16 |
17 | ## Configuration
18 |
19 | The server requires a YNAB API key to function. You can obtain one from your [YNAB Developer Settings](https://app.ynab.com/settings/developer).
20 |
21 | The API key can be provided through:
22 |
23 | 1. Environment variable: `YNAB_API_KEY=your_api_key`
24 | 2. MCP secret management system
25 | 3. `.env` file in project root
26 |
27 | ## Usage
28 |
29 | ### Running the Server
30 |
31 | ```bash
32 | # Development mode with hot reload and browser launch
33 | task dev
34 |
35 | # Production install for Claude Desktop, Goose, or any other MCP-supported environment
36 | task install
37 | ```
38 |
39 | ### Available Resources
40 |
41 | - `ynab://accounts` - List all YNAB accounts
42 | - `ynab://transactions/{account_id}` - Get recent transactions for a specific account
43 |
44 | ### Available Tools
45 |
46 | - `create_transaction` - Create a new transaction
47 | - `get_account_balance` - Get the current balance of an account
48 |
49 | ## Example Usage
50 |
51 | ```python
52 | # Create a new transaction
53 | result = await create_transaction(
54 | account_id="your_account_id",
55 | amount=42.50, # in dollars
56 | payee_name="Coffee Shop",
57 | category_name="Dining Out",
58 | memo="Morning coffee"
59 | )
60 |
61 | # Get account balance
62 | balance = await get_account_balance("your_account_id")
63 |
64 | # List accounts
65 | accounts = await ctx.read_resource("ynab://accounts")
66 |
67 | # Get recent transactions
68 | transactions = await ctx.read_resource(f"ynab://transactions/{account_id}")
69 | ```
70 |
71 | ## Development
72 |
73 | ```bash
74 | # Install dependencies (uses uv)
75 | task deps
76 |
77 | # Run all tests including integration tests (you will need a YNAB API key for this)
78 | task test:all
79 |
80 | # Generate coverage report
81 | task coverage
82 |
83 | # Format and lint code
84 | task fmt # Should add this to Taskfile
85 | ```
86 |
87 | ## Project Tasks
88 |
89 | This project uses a Taskfile for common operations. Key commands:
90 |
91 | ```bash
92 | task dev # Start dev server with auto-reload
93 | task test # Run unit tests
94 | task coverage # Generate test coverage report
95 | task install # Install production build
96 | task deps # Synchronize dependencies
97 | ```
98 |
99 | See [Taskfile.yml](Taskfile.yml) for all available tasks.
100 |
```
--------------------------------------------------------------------------------
/CLAUDE.md:
--------------------------------------------------------------------------------
```markdown
1 | # MCP-YNAB Project Guide
2 |
3 | ## Commands
4 | - Build: `task install` (production) or `task dev` (development with browser)
5 | - Lint/Format: `task fmt` (runs ruff format and ruff check with --fix)
6 | - Tests:
7 | - All excluding integration: `pytest` or `task test`
8 | - Single test: `pytest tests/test_server.py::test_name`
9 | - Integration tests: `pytest -m "integration"` or `task test:integration`
10 | - With coverage: `task coverage`
11 | - Dependencies: `task deps` (uses uv sync)
12 |
13 | ## Code Style
14 | - Python 3.12+
15 | - Line length: 100 characters
16 | - Formatting: ruff format (Black-compatible)
17 | - Linting: ruff check
18 | - Imports: standard library first, third-party second, local modules last
19 | - Types: Use type hints consistently with modern Python typing
20 | - Testing: pytest with pytest-asyncio for async tests
21 | - Error handling: Use proper exception handling with specific exceptions
22 |
23 | ## Project Structure
24 | - src/mcp_ynab/ - Main package source code
25 | - tests/ - Test directory with pytest fixtures in conftest.py
26 | - Task definitions in Taskfile.yml (use `uv` for Python package management)
27 | - MCP server implementation following modelcontextprotocol.io guidelines
```
--------------------------------------------------------------------------------
/mise.toml:
--------------------------------------------------------------------------------
```toml
1 | [tools]
2 | node = "lts"
3 |
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Test package for mcp-ynab."""
2 |
```
--------------------------------------------------------------------------------
/src/mcp_ynab/__main__.py:
--------------------------------------------------------------------------------
```python
1 | from mcp_ynab import main
2 |
3 | main()
4 |
```
--------------------------------------------------------------------------------
/tests/test_environment.py:
--------------------------------------------------------------------------------
```python
1 | """Test environment setup and configuration."""
2 |
3 | import os
4 |
5 | import pytest
6 | from ynab.api.budgets_api import BudgetsApi
7 |
8 |
9 | def test_environment_variables():
10 | """Test that required environment variables are set."""
11 | assert "YNAB_API_KEY" in os.environ, "YNAB_API_KEY must be set in environment"
12 |
13 |
14 | @pytest.mark.integration
15 | def test_ynab_api_connection(ynab_client):
16 | """Test that we can connect to the YNAB API."""
17 | budgets_api = BudgetsApi(ynab_client)
18 | budgets_response = budgets_api.get_budgets()
19 | assert budgets_response.data.budgets is not None
20 | assert len(budgets_response.data.budgets) > 0
21 |
22 |
23 | def test_preferences_files_exist():
24 | """Test that the preference file is loaded, and if not, returns None."""
25 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "mcp-ynab"
3 | version = "0.1.0"
4 | description = "MCP server for YNAB API integration"
5 | readme = "README.md"
6 | requires-python = ">=3.12"
7 | dependencies = [
8 | "mcp[cli]>=0.5.0",
9 | "httpx>=0.26.0",
10 | "pydantic>=2.0.0",
11 | "ynab>=1.0.1",
12 | "python-dotenv>=1.0.0",
13 | "xdg>=6.0.0",
14 | ]
15 |
16 | [tool.ruff]
17 | line-length = 100
18 | target-version = "py312"
19 |
20 | [tool.black]
21 | line-length = 100
22 | target-version = ["py312"]
23 |
24 | [project.scripts]
25 | mcp-ynab = "mcp_ynab:main"
26 |
27 | [build-system]
28 | requires = ["hatchling"]
29 | build-backend = "hatchling.build"
30 |
31 | [dependency-groups]
32 | dev = [
33 | "pytest>=8.3.4",
34 | "pytest-asyncio>=0.25.3",
35 | "pytest-cov>=6.0.0",
36 | "black>=24.0.0",
37 | "ruff>=0.9.4",
38 | "mypy>=1.15.0",
39 | ]
40 |
41 | [tool.pytest.ini_options]
42 | pythonpath = ["src"]
43 | testpaths = ["tests"]
44 | markers = [
45 | "integration: marks tests as integration tests that require YNAB API access",
46 | "asyncio: mark tests as async tests",
47 | ]
48 | addopts = "-v -ra --strict-markers -m 'not integration'"
49 |
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
1 | """Pytest configuration and shared fixtures."""
2 |
3 | import os
4 | from typing import Generator
5 |
6 | import pytest
7 | from dotenv import load_dotenv
8 | from ynab.api_client import ApiClient
9 | from ynab.configuration import Configuration
10 |
11 |
12 | def pytest_configure(config):
13 | """Configure custom markers."""
14 | config.addinivalue_line(
15 | "markers",
16 | "integration: mark test as an integration test that requires YNAB API access",
17 | )
18 |
19 |
20 | @pytest.fixture(scope="session")
21 | def env_setup() -> None:
22 | """Load environment variables for tests."""
23 | load_dotenv(verbose=True)
24 | if not os.getenv("YNAB_API_KEY"):
25 | pytest.skip("YNAB_API_KEY not set in environment")
26 |
27 |
28 | @pytest.fixture
29 | def ynab_client(env_setup) -> Generator:
30 | """Create a YNAB API client for testing."""
31 | if not os.getenv("YNAB_API_KEY"):
32 | pytest.skip("YNAB_API_KEY not set in environment")
33 |
34 | configuration = Configuration(access_token=os.getenv("YNAB_API_KEY"))
35 | with ApiClient(configuration) as client:
36 | yield client
37 |
```
--------------------------------------------------------------------------------
/src/mcp_ynab/__init__.py:
--------------------------------------------------------------------------------
```python
1 | import argparse
2 | import signal
3 | import sys
4 | from typing import NoReturn
5 |
6 | from dotenv import load_dotenv
7 |
8 | from .server import mcp
9 |
10 |
11 | def handle_sigint(signum, frame):
12 | """Handle SIGINT (Ctrl+C) gracefully."""
13 | print("\nReceived SIGINT. Shutting down...", file=sys.stderr)
14 | sys.exit(0)
15 |
16 |
17 | def main() -> NoReturn:
18 | """Entry point for the YNAB MCP server."""
19 | parser = argparse.ArgumentParser(description="YNAB (You Need A Budget) API integration for MCP")
20 | parser.add_argument("--debug", action="store_true", help="Enable debug logging")
21 | args = parser.parse_args()
22 |
23 | # Load environment variables from .env file
24 | load_dotenv()
25 |
26 | # Set up signal handling
27 | signal.signal(signal.SIGINT, handle_sigint)
28 |
29 | # Run the MCP server
30 | try:
31 | if args.debug:
32 | print("Starting YNAB MCP server in debug mode...", file=sys.stderr)
33 | mcp.run()
34 | sys.exit(0) # This line will never be reached due to mcp.run() being blocking
35 | except KeyboardInterrupt:
36 | print("\nShutting down...", file=sys.stderr)
37 | sys.exit(0)
38 | except Exception as e:
39 | print(f"Error: {e}", file=sys.stderr)
40 | sys.exit(1)
41 |
42 |
43 | if __name__ == "__main__":
44 | main()
45 |
```
--------------------------------------------------------------------------------
/Taskfile.yml:
--------------------------------------------------------------------------------
```yaml
1 | # https://taskfile.dev
2 |
3 | version: "3"
4 |
5 | vars:
6 | GREETING: Hello, World!
7 |
8 | tasks:
9 | mcp-dev:
10 | desc: "Run the MCP server in development mode"
11 | cmds:
12 | - uv run mcp dev src/mcp_ynab/server.py
13 |
14 | open-browser:
15 | desc: "Open the browser"
16 | cmds:
17 | - sleep 2 && open http://localhost:5173
18 |
19 | dev:
20 | desc: "Run the MCP server in development mode and open the browser"
21 | deps:
22 | - mcp-dev
23 | - open-browser
24 |
25 | deps:
26 | desc: "Synchronize dependencies"
27 | cmds:
28 | - uv sync
29 | - npm install --global @modelcontextprotocol/inspector
30 |
31 | install:
32 | desc: "Install the package locally"
33 | cmds:
34 | - uv sync
35 | - uv pip install .
36 | - echo "installed mcp-ynab at $(which mcp-ynab)"
37 |
38 | test:
39 | desc: "Run the tests"
40 | cmds:
41 | - pytest
42 |
43 | test:integration:
44 | desc: "Run the integration tests"
45 | cmds:
46 | - pytest -m "integration"
47 |
48 | test:all:
49 | desc: "Run all tests including integration tests"
50 | cmds:
51 | - pytest -m ""
52 |
53 | coverage:
54 | desc: "Run tests with coverage reporting"
55 | cmds:
56 | - pytest --cov=src/mcp_ynab --cov-report=term-missing --cov-report=html -m ""
57 |
58 | fmt:
59 | desc: "Format and lint code"
60 | cmds:
61 | - ruff format src/ tests/
62 | - ruff check src/ tests/ --fix
63 |
```
--------------------------------------------------------------------------------
/todo.txt:
--------------------------------------------------------------------------------
```
1 | # MCP-YNAB Project To-Do List
2 |
3 | ## Code Organization
4 | - [ ] Split server.py into smaller modules (client, resources, utils, tools)
5 | - [ ] Create proper package structure with submodules
6 | - [ ] Extract formatting utilities to a dedicated module
7 | - [ ] Move YNAB client code to a dedicated client module
8 | - [ ] Separate persistence logic from business logic
9 |
10 | ## Performance Improvements
11 | - [ ] Implement proper caching with TTL for budget and category data
12 | - [ ] Add pagination support for transaction queries
13 | - [ ] Optimize category lookup by implementing indexed search
14 | - [ ] Implement batch operations for transaction updates
15 | - [ ] Reduce redundant API calls in tool implementations
16 |
17 | ## Error Handling & Robustness
18 | - [ ] Implement consistent error handling pattern across all API calls
19 | - [ ] Add input validation for all tool parameters
20 | - [ ] Implement retry logic for API failures
21 | - [ ] Add proper error reporting and logging infrastructure
22 | - [ ] Handle rate limiting for YNAB API
23 |
24 | ## Testing
25 | - [ ] Implement unit tests for all helper functions
26 | - [ ] Add integration tests with API simulation
27 | - [ ] Test edge cases and error conditions
28 | - [ ] Add test coverage for resource endpoints
29 | - [ ] Improve test fixtures and mocking patterns
30 |
31 | ## Documentation
32 | - [ ] Add docstrings to all public functions and classes
33 | - [ ] Create API documentation for resources and tools
34 | - [ ] Document data models and field definitions
35 | - [ ] Add usage examples for common operations
36 | - [ ] Document environment setup and configuration options
37 |
38 | ## Feature Enhancements
39 | - [ ] Add support for multiple budget switching
40 | - [ ] Implement transaction search functionality
41 | - [ ] Add budget adjustment capabilities
42 | - [ ] Implement transaction approval workflow
43 | - [ ] Add reporting and visualization features
44 |
45 | ## Development Experience
46 | - [ ] Add development environment setup documentation
47 | - [ ] Implement pre-commit hooks for code quality checks
48 | - [ ] Set up CI/CD workflow for automated testing
49 | - [ ] Create a development quick start guide
50 | - [ ] Improve debugging support
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
1 | from datetime import date, datetime, timedelta
2 | from unittest.mock import AsyncMock, MagicMock, patch
3 |
4 | import pytest
5 | from ynab.api.accounts_api import AccountsApi
6 | from ynab.api.budgets_api import BudgetsApi
7 | from ynab.api.categories_api import CategoriesApi
8 | from ynab.api.transactions_api import TransactionsApi
9 | from ynab.api_client import ApiClient
10 | from ynab.models.account import Account
11 | from ynab.models.budget_summary import BudgetSummary
12 | from ynab.models.category import Category
13 | from ynab.models.category_group_with_categories import CategoryGroupWithCategories
14 | from ynab.models.transaction_detail import TransactionDetail
15 |
16 | from mcp_ynab.server import YNABResources
17 |
18 | # Test constants
19 | TEST_BUDGET_ID = "test-budget-123"
20 | TEST_ACCOUNT_ID = "test-account-456"
21 | TEST_CATEGORY_ID = "test-category-789"
22 | TEST_TRANSACTION_ID = "test-transaction-012"
23 |
24 | # Common test data
25 | SAMPLE_ACCOUNT = {
26 | "id": TEST_ACCOUNT_ID,
27 | "name": "Test Account",
28 | "type": "checking",
29 | "balance": 100000, # $100 in milliunits
30 | "closed": False,
31 | "deleted": False
32 | }
33 |
34 | SAMPLE_TRANSACTION = {
35 | "id": TEST_TRANSACTION_ID,
36 | "date": date.today().isoformat(),
37 | "amount": -50000, # -$50 in milliunits
38 | "payee_name": "Test Payee",
39 | "category_id": TEST_CATEGORY_ID,
40 | "memo": "Test transaction",
41 | "cleared": True,
42 | "approved": True,
43 | "account_id": TEST_ACCOUNT_ID
44 | }
45 |
46 | SAMPLE_CATEGORY = {
47 | "id": TEST_CATEGORY_ID,
48 | "name": "Test Category",
49 | "budgeted": 200000, # $200 in milliunits
50 | "activity": -50000, # -$50 in milliunits
51 | "balance": 150000 # $150 in milliunits
52 | }
53 |
54 | @pytest.fixture
55 | def mock_ynab_client():
56 | """Mock YNAB API client."""
57 | with patch("mcp_ynab.server._get_client") as mock_get_client:
58 | client = AsyncMock(spec=ApiClient)
59 | mock_get_client.return_value = client
60 | yield client
61 |
62 | @pytest.fixture
63 | def mock_budgets_api():
64 | """Mock YNAB Budgets API."""
65 | with patch("ynab.api.budgets_api.BudgetsApi") as mock_api:
66 | api = MagicMock(spec=BudgetsApi)
67 | mock_api.return_value = api
68 | yield api
69 |
70 | @pytest.fixture
71 | def mock_accounts_api():
72 | """Mock YNAB Accounts API."""
73 | with patch("ynab.api.accounts_api.AccountsApi") as mock_api:
74 | api = MagicMock(spec=AccountsApi)
75 | mock_api.return_value = api
76 | yield api
77 |
78 | @pytest.fixture
79 | def mock_categories_api():
80 | """Mock YNAB Categories API."""
81 | with patch("ynab.api.categories_api.CategoriesApi") as mock_api:
82 | api = MagicMock(spec=CategoriesApi)
83 | mock_api.return_value = api
84 | yield api
85 |
86 | @pytest.fixture
87 | def mock_transactions_api():
88 | """Mock YNAB Transactions API."""
89 | with patch("ynab.api.transactions_api.TransactionsApi") as mock_api:
90 | api = MagicMock(spec=TransactionsApi)
91 | mock_api.return_value = api
92 | yield api
93 |
94 | @pytest.fixture
95 | def mock_xdg_config_home(tmp_path):
96 | """Mock XDG_CONFIG_HOME directory."""
97 | config_dir = tmp_path / "config"
98 | config_dir.mkdir()
99 | with patch("mcp_ynab.server.XDG_CONFIG_HOME", str(config_dir)):
100 | yield config_dir
101 |
102 | @pytest.fixture
103 | def ynab_resources(mock_xdg_config_home):
104 | """Create a YNABResources instance with mocked config directory."""
105 | return YNABResources()
106 |
107 | @pytest.fixture
108 | def sample_budget_summary():
109 | """Create a sample BudgetSummary."""
110 | return BudgetSummary(
111 | id=TEST_BUDGET_ID,
112 | name="Test Budget",
113 | last_modified_on=datetime.now()
114 | )
115 |
116 | @pytest.fixture
117 | def sample_account():
118 | """Create a sample Account."""
119 | return Account(**SAMPLE_ACCOUNT)
120 |
121 | @pytest.fixture
122 | def sample_transaction():
123 | """Create a sample TransactionDetail."""
124 | return TransactionDetail(**SAMPLE_TRANSACTION)
125 |
126 | @pytest.fixture
127 | def sample_category():
128 | """Create a sample Category."""
129 | return Category(**SAMPLE_CATEGORY)
130 |
131 | @pytest.fixture
132 | def sample_category_group():
133 | """Create a sample CategoryGroupWithCategories."""
134 | return CategoryGroupWithCategories(
135 | id="test-group-123",
136 | name="Test Group",
137 | categories=[sample_category()]
138 | )
139 |
140 | # Test helper functions
141 | class TestHelperFunctions:
142 | def test_build_markdown_table(self):
143 | """Test _build_markdown_table function."""
144 | headers = ["Name", "Value"]
145 | rows = [
146 | ["Test1", "100"],
147 | ["Test2", "200"]
148 | ]
149 | alignments = ["left", "right"]
150 | # TODO: Test table generation with various inputs
151 | # TODO: Test empty rows
152 | # TODO: Test different alignments
153 | # TODO: Test edge cases with special characters
154 |
155 | def test_format_accounts_output(self):
156 | """Test _format_accounts_output function."""
157 | # TODO: Test formatting different account types
158 | # TODO: Test closed/deleted accounts
159 | # TODO: Test negative balances
160 | # TODO: Test grouping by account type
161 | # TODO: Test summary calculations
162 |
163 | def test_load_save_json_file(self, tmp_path):
164 | """Test _load_json_file and _save_json_file functions."""
165 | # TODO: Test saving and loading valid JSON
166 | # TODO: Test loading non-existent file
167 | # TODO: Test saving to non-existent directory
168 | # TODO: Test with invalid JSON data
169 |
170 | # Test YNAB Resources
171 | class TestYNABResources:
172 | def test_init_loads_data(self, ynab_resources, mock_xdg_config_home):
173 | """Test YNABResources initialization loads data correctly."""
174 | # TODO: Test initialization with existing files
175 | # TODO: Test initialization with missing files
176 |
177 | def test_get_set_preferred_budget_id(self, ynab_resources):
178 | """Test getting and setting preferred budget ID."""
179 | # TODO: Test setting new budget ID
180 | # TODO: Test getting existing budget ID
181 | # TODO: Test persistence across instances
182 |
183 | def test_get_cached_categories(self, ynab_resources):
184 | """Test retrieving cached categories."""
185 | # TODO: Test with existing cached categories
186 | # TODO: Test with empty cache
187 | # TODO: Test with invalid cache data
188 |
189 | def test_cache_categories(self, ynab_resources):
190 | """Test caching categories."""
191 | # TODO: Test caching new categories
192 | # TODO: Test updating existing cache
193 | # TODO: Test with invalid category data
194 |
195 | # Test MCP Tools
196 | @pytest.mark.asyncio
197 | class TestMCPTools:
198 | async def test_create_transaction(self, mock_ynab_client, mock_transactions_api, sample_transaction):
199 | """Test create_transaction tool."""
200 | # TODO: Test creating with minimum required fields
201 | # TODO: Test with optional fields
202 | # TODO: Test with category
203 | # TODO: Test with invalid data
204 | pass
205 |
206 | async def test_get_account_balance(self, mock_ynab_client, mock_accounts_api, sample_account):
207 | """Test get_account_balance tool."""
208 | # TODO: Test getting balance for valid account
209 | # TODO: Test with non-existent account
210 | # TODO: Test with closed account
211 | # TODO: Test with various balance formats
212 |
213 | async def test_get_budgets(self, mock_ynab_client, mock_budgets_api, sample_budget_summary):
214 | """Test get_budgets tool."""
215 | # TODO: Test listing multiple budgets
216 | # TODO: Test with no budgets
217 | # TODO: Test markdown formatting
218 | # TODO: Test error handling
219 |
220 | async def test_get_accounts(self, mock_ynab_client, mock_accounts_api, sample_account):
221 | """Test get_accounts tool."""
222 | # TODO: Test listing different account types
223 | # TODO: Test with closed accounts
224 | # TODO: Test markdown formatting
225 | # TODO: Test summary calculations
226 |
227 | async def test_get_transactions(
228 | self, mock_ynab_client, mock_transactions_api, sample_transaction
229 | ):
230 | """Test get_transactions tool."""
231 | # TODO: Test with date range
232 | # TODO: Test with specific account
233 | # TODO: Test markdown formatting
234 | # TODO: Test pagination handling
235 |
236 | async def test_get_transactions_needing_attention(
237 | self, mock_ynab_client, mock_transactions_api, sample_transaction
238 | ):
239 | """Test get_transactions_needing_attention tool."""
240 | # TODO: Test uncategorized filter
241 | # TODO: Test unapproved filter
242 | # TODO: Test both filters
243 | # TODO: Test with different date ranges
244 | # TODO: Test markdown output formatting
245 |
246 | async def test_categorize_transaction(
247 | self, mock_ynab_client, mock_transactions_api, sample_transaction
248 | ):
249 | """Test categorize_transaction tool."""
250 | # TODO: Test with valid transaction and category
251 | # TODO: Test with different ID types
252 | # TODO: Test with non-existent transaction
253 | # TODO: Test with invalid category
254 |
255 | async def test_get_categories(
256 | self, mock_ynab_client, mock_categories_api, sample_category_group
257 | ):
258 | """Test get_categories tool."""
259 | # TODO: Test listing all categories
260 | # TODO: Test nested category groups
261 | # TODO: Test markdown formatting
262 | # TODO: Test budget/activity calculations
263 |
264 | async def test_set_preferred_budget_id(self, ynab_resources):
265 | """Test set_preferred_budget_id tool."""
266 | # TODO: Test setting new budget ID
267 | # TODO: Test persistence
268 | # TODO: Test validation
269 | # TODO: Test error cases
270 |
271 | async def test_cache_categories(
272 | self, mock_ynab_client, mock_categories_api, ynab_resources, sample_category_group
273 | ):
274 | """Test cache_categories tool."""
275 | # TODO: Test caching new categories
276 | # TODO: Test updating existing cache
277 | # TODO: Test cache format
278 | # TODO: Test error handling
279 |
280 | # Test API Client
281 | @pytest.mark.asyncio
282 | class TestAPIClient:
283 | async def test_get_client(self):
284 | """Test _get_client function."""
285 | # TODO: Test with valid API key
286 | # TODO: Test without API key
287 | # TODO: Test configuration options
288 | # TODO: Test error handling
289 |
290 | async def test_client_context_manager(self, mock_ynab_client):
291 | """Test AsyncYNABClient context manager."""
292 | # TODO: Test normal usage
293 | # TODO: Test error handling
294 | # TODO: Test resource cleanup
295 | # TODO: Test multiple context manager usage
296 |
```
--------------------------------------------------------------------------------
/docs/mcp-py-sdk.md:
--------------------------------------------------------------------------------
```markdown
1 | # MCP Python SDK
2 |
3 | <div align="center">
4 |
5 | <strong>Python implementation of the Model Context Protocol (MCP)</strong>
6 |
7 | [![PyPI][pypi-badge]][pypi-url]
8 | [![MIT licensed][mit-badge]][mit-url]
9 | [![Python Version][python-badge]][python-url]
10 | [![Documentation][docs-badge]][docs-url]
11 | [![Specification][spec-badge]][spec-url]
12 | [![GitHub Discussions][discussions-badge]][discussions-url]
13 |
14 | </div>
15 |
16 | <!-- omit in toc -->
17 | ## Table of Contents
18 |
19 | - [Overview](#overview)
20 | - [Installation](#installation)
21 | - [Quickstart](#quickstart)
22 | - [What is MCP?](#what-is-mcp)
23 | - [Core Concepts](#core-concepts)
24 | - [Server](#server)
25 | - [Resources](#resources)
26 | - [Tools](#tools)
27 | - [Prompts](#prompts)
28 | - [Images](#images)
29 | - [Context](#context)
30 | - [Running Your Server](#running-your-server)
31 | - [Development Mode](#development-mode)
32 | - [Claude Desktop Integration](#claude-desktop-integration)
33 | - [Direct Execution](#direct-execution)
34 | - [Examples](#examples)
35 | - [Echo Server](#echo-server)
36 | - [SQLite Explorer](#sqlite-explorer)
37 | - [Advanced Usage](#advanced-usage)
38 | - [Low-Level Server](#low-level-server)
39 | - [Writing MCP Clients](#writing-mcp-clients)
40 | - [MCP Primitives](#mcp-primitives)
41 | - [Server Capabilities](#server-capabilities)
42 | - [Documentation](#documentation)
43 | - [Contributing](#contributing)
44 | - [License](#license)
45 |
46 | [pypi-badge]: https://img.shields.io/pypi/v/mcp.svg
47 | [pypi-url]: https://pypi.org/project/mcp/
48 | [mit-badge]: https://img.shields.io/pypi/l/mcp.svg
49 | [mit-url]: https://github.com/modelcontextprotocol/python-sdk/blob/main/LICENSE
50 | [python-badge]: https://img.shields.io/pypi/pyversions/mcp.svg
51 | [python-url]: https://www.python.org/downloads/
52 | [docs-badge]: https://img.shields.io/badge/docs-modelcontextprotocol.io-blue.svg
53 | [docs-url]: https://modelcontextprotocol.io
54 | [spec-badge]: https://img.shields.io/badge/spec-spec.modelcontextprotocol.io-blue.svg
55 | [spec-url]: https://spec.modelcontextprotocol.io
56 | [discussions-badge]: https://img.shields.io/github/discussions/modelcontextprotocol/python-sdk
57 | [discussions-url]: https://github.com/modelcontextprotocol/python-sdk/discussions
58 |
59 | ## Overview
60 |
61 | The Model Context Protocol allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction. This Python SDK implements the full MCP specification, making it easy to:
62 |
63 | - Build MCP clients that can connect to any MCP server
64 | - Create MCP servers that expose resources, prompts and tools
65 | - Use standard transports like stdio and SSE
66 | - Handle all MCP protocol messages and lifecycle events
67 |
68 | ## Installation
69 |
70 | We recommend using [uv](https://docs.astral.sh/uv/) to manage your Python projects:
71 |
72 | ```bash
73 | uv add "mcp[cli]"
74 | ```
75 |
76 | Alternatively:
77 | ```bash
78 | pip install mcp
79 | ```
80 |
81 | ## Quickstart
82 |
83 | Let's create a simple MCP server that exposes a calculator tool and some data:
84 |
85 | ```python
86 | # server.py
87 | from mcp.server.fastmcp import FastMCP
88 |
89 | # Create an MCP server
90 | mcp = FastMCP("Demo")
91 |
92 | # Add an addition tool
93 | @mcp.tool()
94 | def add(a: int, b: int) -> int:
95 | """Add two numbers"""
96 | return a + b
97 |
98 | # Add a dynamic greeting resource
99 | @mcp.resource("greeting://{name}")
100 | def get_greeting(name: str) -> str:
101 | """Get a personalized greeting"""
102 | return f"Hello, {name}!"
103 | ```
104 |
105 | You can install this server in [Claude Desktop](https://claude.ai/download) and interact with it right away by running:
106 | ```bash
107 | mcp install server.py
108 | ```
109 |
110 | Alternatively, you can test it with the MCP Inspector:
111 | ```bash
112 | mcp dev server.py
113 | ```
114 |
115 | ## What is MCP?
116 |
117 | The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) lets you build servers that expose data and functionality to LLM applications in a secure, standardized way. Think of it like a web API, but specifically designed for LLM interactions. MCP servers can:
118 |
119 | - Expose data through **Resources** (think of these sort of like GET endpoints; they are used to load information into the LLM's context)
120 | - Provide functionality through **Tools** (sort of like POST endpoints; they are used to execute code or otherwise produce a side effect)
121 | - Define interaction patterns through **Prompts** (reusable templates for LLM interactions)
122 | - And more!
123 |
124 | ## Core Concepts
125 |
126 | ### Server
127 |
128 | The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:
129 |
130 | ```python
131 | from mcp.server.fastmcp import FastMCP
132 |
133 | # Create a named server
134 | mcp = FastMCP("My App")
135 |
136 | # Specify dependencies for deployment and development
137 | mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
138 | ```
139 |
140 | ### Resources
141 |
142 | Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects:
143 |
144 | ```python
145 | @mcp.resource("config://app")
146 | def get_config() -> str:
147 | """Static configuration data"""
148 | return "App configuration here"
149 |
150 | @mcp.resource("users://{user_id}/profile")
151 | def get_user_profile(user_id: str) -> str:
152 | """Dynamic user data"""
153 | return f"Profile data for user {user_id}"
154 | ```
155 |
156 | ### Tools
157 |
158 | Tools let LLMs take actions through your server. Unlike resources, tools are expected to perform computation and have side effects:
159 |
160 | ```python
161 | @mcp.tool()
162 | def calculate_bmi(weight_kg: float, height_m: float) -> float:
163 | """Calculate BMI given weight in kg and height in meters"""
164 | return weight_kg / (height_m ** 2)
165 |
166 | @mcp.tool()
167 | async def fetch_weather(city: str) -> str:
168 | """Fetch current weather for a city"""
169 | async with httpx.AsyncClient() as client:
170 | response = await client.get(f"https://api.weather.com/{city}")
171 | return response.text
172 | ```
173 |
174 | ### Prompts
175 |
176 | Prompts are reusable templates that help LLMs interact with your server effectively:
177 |
178 | ```python
179 | @mcp.prompt()
180 | def review_code(code: str) -> str:
181 | return f"Please review this code:\n\n{code}"
182 |
183 | @mcp.prompt()
184 | def debug_error(error: str) -> list[Message]:
185 | return [
186 | UserMessage("I'm seeing this error:"),
187 | UserMessage(error),
188 | AssistantMessage("I'll help debug that. What have you tried so far?")
189 | ]
190 | ```
191 |
192 | ### Images
193 |
194 | FastMCP provides an `Image` class that automatically handles image data:
195 |
196 | ```python
197 | from mcp.server.fastmcp import FastMCP, Image
198 | from PIL import Image as PILImage
199 |
200 | @mcp.tool()
201 | def create_thumbnail(image_path: str) -> Image:
202 | """Create a thumbnail from an image"""
203 | img = PILImage.open(image_path)
204 | img.thumbnail((100, 100))
205 | return Image(data=img.tobytes(), format="png")
206 | ```
207 |
208 | ### Context
209 |
210 | The Context object gives your tools and resources access to MCP capabilities:
211 |
212 | ```python
213 | from mcp.server.fastmcp import FastMCP, Context
214 |
215 | @mcp.tool()
216 | async def long_task(files: list[str], ctx: Context) -> str:
217 | """Process multiple files with progress tracking"""
218 | for i, file in enumerate(files):
219 | ctx.info(f"Processing {file}")
220 | await ctx.report_progress(i, len(files))
221 | data, mime_type = await ctx.read_resource(f"file://{file}")
222 | return "Processing complete"
223 | ```
224 |
225 | ## Running Your Server
226 |
227 | ### Development Mode
228 |
229 | The fastest way to test and debug your server is with the MCP Inspector:
230 |
231 | ```bash
232 | mcp dev server.py
233 |
234 | # Add dependencies
235 | mcp dev server.py --with pandas --with numpy
236 |
237 | # Mount local code
238 | mcp dev server.py --with-editable .
239 | ```
240 |
241 | ### Claude Desktop Integration
242 |
243 | Once your server is ready, install it in Claude Desktop:
244 |
245 | ```bash
246 | mcp install server.py
247 |
248 | # Custom name
249 | mcp install server.py --name "My Analytics Server"
250 |
251 | # Environment variables
252 | mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
253 | mcp install server.py -f .env
254 | ```
255 |
256 | ### Direct Execution
257 |
258 | For advanced scenarios like custom deployments:
259 |
260 | ```python
261 | from mcp.server.fastmcp import FastMCP
262 |
263 | mcp = FastMCP("My App")
264 |
265 | if __name__ == "__main__":
266 | mcp.run()
267 | ```
268 |
269 | Run it with:
270 | ```bash
271 | python server.py
272 | # or
273 | mcp run server.py
274 | ```
275 |
276 | ## Examples
277 |
278 | ### Echo Server
279 |
280 | A simple server demonstrating resources, tools, and prompts:
281 |
282 | ```python
283 | from mcp.server.fastmcp import FastMCP
284 |
285 | mcp = FastMCP("Echo")
286 |
287 | @mcp.resource("echo://{message}")
288 | def echo_resource(message: str) -> str:
289 | """Echo a message as a resource"""
290 | return f"Resource echo: {message}"
291 |
292 | @mcp.tool()
293 | def echo_tool(message: str) -> str:
294 | """Echo a message as a tool"""
295 | return f"Tool echo: {message}"
296 |
297 | @mcp.prompt()
298 | def echo_prompt(message: str) -> str:
299 | """Create an echo prompt"""
300 | return f"Please process this message: {message}"
301 | ```
302 |
303 | ### SQLite Explorer
304 |
305 | A more complex example showing database integration:
306 |
307 | ```python
308 | from mcp.server.fastmcp import FastMCP
309 | import sqlite3
310 |
311 | mcp = FastMCP("SQLite Explorer")
312 |
313 | @mcp.resource("schema://main")
314 | def get_schema() -> str:
315 | """Provide the database schema as a resource"""
316 | conn = sqlite3.connect("database.db")
317 | schema = conn.execute(
318 | "SELECT sql FROM sqlite_master WHERE type='table'"
319 | ).fetchall()
320 | return "\n".join(sql[0] for sql in schema if sql[0])
321 |
322 | @mcp.tool()
323 | def query_data(sql: str) -> str:
324 | """Execute SQL queries safely"""
325 | conn = sqlite3.connect("database.db")
326 | try:
327 | result = conn.execute(sql).fetchall()
328 | return "\n".join(str(row) for row in result)
329 | except Exception as e:
330 | return f"Error: {str(e)}"
331 | ```
332 |
333 | ## Advanced Usage
334 |
335 | ### Low-Level Server
336 |
337 | For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server:
338 |
339 | ```python
340 | from mcp.server.lowlevel import Server, NotificationOptions
341 | from mcp.server.models import InitializationOptions
342 | import mcp.server.stdio
343 | import mcp.types as types
344 |
345 | # Create a server instance
346 | server = Server("example-server")
347 |
348 | @server.list_prompts()
349 | async def handle_list_prompts() -> list[types.Prompt]:
350 | return [
351 | types.Prompt(
352 | name="example-prompt",
353 | description="An example prompt template",
354 | arguments=[
355 | types.PromptArgument(
356 | name="arg1",
357 | description="Example argument",
358 | required=True
359 | )
360 | ]
361 | )
362 | ]
363 |
364 | @server.get_prompt()
365 | async def handle_get_prompt(
366 | name: str,
367 | arguments: dict[str, str] | None
368 | ) -> types.GetPromptResult:
369 | if name != "example-prompt":
370 | raise ValueError(f"Unknown prompt: {name}")
371 |
372 | return types.GetPromptResult(
373 | description="Example prompt",
374 | messages=[
375 | types.PromptMessage(
376 | role="user",
377 | content=types.TextContent(
378 | type="text",
379 | text="Example prompt text"
380 | )
381 | )
382 | ]
383 | )
384 |
385 | async def run():
386 | async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
387 | await server.run(
388 | read_stream,
389 | write_stream,
390 | InitializationOptions(
391 | server_name="example",
392 | server_version="0.1.0",
393 | capabilities=server.get_capabilities(
394 | notification_options=NotificationOptions(),
395 | experimental_capabilities={},
396 | )
397 | )
398 | )
399 |
400 | if __name__ == "__main__":
401 | import asyncio
402 | asyncio.run(run())
403 | ```
404 |
405 | ### Writing MCP Clients
406 |
407 | The SDK provides a high-level client interface for connecting to MCP servers:
408 |
409 | ```python
410 | from mcp import ClientSession, StdioServerParameters
411 | from mcp.client.stdio import stdio_client
412 |
413 | # Create server parameters for stdio connection
414 | server_params = StdioServerParameters(
415 | command="python", # Executable
416 | args=["example_server.py"], # Optional command line arguments
417 | env=None # Optional environment variables
418 | )
419 |
420 | async def run():
421 | async with stdio_client(server_params) as (read, write):
422 | async with ClientSession(read, write) as session:
423 | # Initialize the connection
424 | await session.initialize()
425 |
426 | # List available prompts
427 | prompts = await session.list_prompts()
428 |
429 | # Get a prompt
430 | prompt = await session.get_prompt("example-prompt", arguments={"arg1": "value"})
431 |
432 | # List available resources
433 | resources = await session.list_resources()
434 |
435 | # List available tools
436 | tools = await session.list_tools()
437 |
438 | # Read a resource
439 | content, mime_type = await session.read_resource("file://some/path")
440 |
441 | # Call a tool
442 | result = await session.call_tool("tool-name", arguments={"arg1": "value"})
443 |
444 | if __name__ == "__main__":
445 | import asyncio
446 | asyncio.run(run())
447 | ```
448 |
449 | ### MCP Primitives
450 |
451 | The MCP protocol defines three core primitives that servers can implement:
452 |
453 | | Primitive | Control | Description | Example Use |
454 | |-----------|-----------------------|-----------------------------------------------------|------------------------------|
455 | | Prompts | User-controlled | Interactive templates invoked by user choice | Slash commands, menu options |
456 | | Resources | Application-controlled| Contextual data managed by the client application | File contents, API responses |
457 | | Tools | Model-controlled | Functions exposed to the LLM to take actions | API calls, data updates |
458 |
459 | ### Server Capabilities
460 |
461 | MCP servers declare capabilities during initialization:
462 |
463 | | Capability | Feature Flag | Description |
464 | |-------------|------------------------------|------------------------------------|
465 | | `prompts` | `listChanged` | Prompt template management |
466 | | `resources` | `subscribe`<br/>`listChanged`| Resource exposure and updates |
467 | | `tools` | `listChanged` | Tool discovery and execution |
468 | | `logging` | - | Server logging configuration |
469 | | `completion`| - | Argument completion suggestions |
470 |
471 | ## Documentation
472 |
473 | - [Model Context Protocol documentation](https://modelcontextprotocol.io)
474 | - [Model Context Protocol specification](https://spec.modelcontextprotocol.io)
475 | - [Officially supported servers](https://github.com/modelcontextprotocol/servers)
476 |
477 | ## Contributing
478 |
479 | We are passionate about supporting contributors of all levels of experience and would love to see you get involved in the project. See the [contributing guide](CONTRIBUTING.md) to get started.
480 |
481 | ## License
482 |
483 | This project is licensed under the MIT License - see the LICENSE file for details.
484 |
```
--------------------------------------------------------------------------------
/src/mcp_ynab/server.py:
--------------------------------------------------------------------------------
```python
1 | import json
2 | import os
3 | from datetime import date, datetime, timedelta
4 | from pathlib import Path
5 | from typing import Annotated, Any, Dict, List, Optional, cast
6 |
7 | import mcp.types as types # Import MCP types
8 | from dotenv import load_dotenv
9 | from mcp.server.fastmcp import FastMCP
10 | from pydantic import Field
11 | from xdg import XDG_CONFIG_HOME
12 | from ynab.api.accounts_api import AccountsApi
13 | from ynab.api.budgets_api import BudgetsApi
14 | from ynab.api.categories_api import CategoriesApi
15 | from ynab.api.transactions_api import TransactionsApi
16 | from ynab.api_client import ApiClient
17 | from ynab.configuration import Configuration
18 | from ynab.models.account import Account
19 | from ynab.models.category import Category
20 | from ynab.models.category_group_with_categories import CategoryGroupWithCategories
21 | from ynab.models.existing_transaction import ExistingTransaction
22 | from ynab.models.new_transaction import NewTransaction
23 | from ynab.models.post_transactions_wrapper import PostTransactionsWrapper
24 | from ynab.models.put_transaction_wrapper import PutTransactionWrapper
25 | from ynab.models.transaction_detail import TransactionDetail
26 |
27 | # 1. Load environment variables
28 | load_dotenv(verbose=True)
29 |
30 | # 2. Globals / configuration
31 | ynab_api_key = os.environ.get("YNAB_API_KEY")
32 |
33 | # Set up XDG config directory
34 | CONFIG_DIR = Path(XDG_CONFIG_HOME) / "mcp-ynab"
35 | CONFIG_DIR.mkdir(parents=True, exist_ok=True)
36 |
37 | PREFERRED_BUDGET_ID_FILE = CONFIG_DIR / "preferred_budget_id.json"
38 | BUDGET_CATEGORY_CACHE_FILE = CONFIG_DIR / "budget_category_cache.json"
39 |
40 | # 3. Private helper functions
41 |
42 |
43 | async def _get_client() -> ApiClient:
44 | """Get a configured YNAB API client. Reads API key from environment variables."""
45 | if not ynab_api_key:
46 | raise ValueError("YNAB_API_KEY not found in environment variables")
47 | configuration = Configuration(access_token=ynab_api_key)
48 | return ApiClient(configuration)
49 |
50 |
51 | class AsyncYNABClient:
52 | """Async context manager for YNAB API client."""
53 |
54 | def __init__(self):
55 | self.client: Optional[ApiClient] = None
56 |
57 | async def __aenter__(self) -> ApiClient:
58 | self.client = await _get_client()
59 | return self.client
60 |
61 | async def __aexit__(self, exc_type, exc_val, exc_tb):
62 | if self.client:
63 | # ApiClient doesn't have a close method, but we'll keep the context manager pattern
64 | pass
65 |
66 |
67 | async def get_ynab_client() -> AsyncYNABClient:
68 | """Get an async YNAB client context manager."""
69 | return AsyncYNABClient()
70 |
71 |
72 | def _get_empty_table(headers: List[str]) -> str:
73 | """Create an empty markdown table with just headers."""
74 | widths = [len(h) + 2 for h in headers]
75 | header_line = "| " + " | ".join(f"{headers[i]:<{widths[i]}}" for i in range(len(headers))) + " |\n"
76 | sep_line = "|" + "|".join("-" * (widths[i] + 2) for i in range(len(headers))) + "|\n"
77 | return header_line + sep_line + "\n"
78 |
79 |
80 | def _get_column_widths(headers: List[str], rows: List[List[str]], col_count: int) -> List[int]:
81 | """Calculate column widths based on content."""
82 | widths = [len(h) for h in headers]
83 | for row in rows:
84 | for i in range(col_count):
85 | widths[i] = max(widths[i], len(row[i]))
86 | return [w + 2 for w in widths]
87 |
88 |
89 | def _format_table_line(items: List[str], widths: List[int], alignments: List[str]) -> str:
90 | """Format a single line of the markdown table."""
91 | line = "| "
92 | for i, item in enumerate(items):
93 | if alignments[i] == "right":
94 | line += f"{item:>{widths[i]}} | "
95 | else:
96 | line += f"{item:<{widths[i]}} | "
97 | return line.rstrip() + "\n"
98 |
99 |
100 | def _build_markdown_table(
101 | rows: List[List[str]], headers: List[str], alignments: Optional[List[str]] = None
102 | ) -> str:
103 | """Build a markdown table from rows and headers."""
104 | if not rows:
105 | return _get_empty_table(headers)
106 |
107 | alignments = alignments if alignments is not None else ["left"] * len(headers)
108 | col_count = len(headers)
109 | widths = _get_column_widths(headers, rows, col_count)
110 |
111 | header_line = _format_table_line(headers, widths, alignments)
112 | sep_line = "|" + "|".join("-" * (w + 1) for w in widths) + "|\n"
113 |
114 | row_lines = "".join(_format_table_line(row, widths, alignments) for row in rows)
115 | return header_line + sep_line + row_lines
116 |
117 |
118 | def _format_accounts_output(accounts: List[Dict[str, Any]]) -> Dict[str, Any]:
119 | """Format account data into a user-friendly structure."""
120 | account_groups: Dict[str, List[Dict[str, Any]]] = {}
121 | type_order = [
122 | "checking",
123 | "savings",
124 | "creditCard",
125 | "mortgage",
126 | "autoLoan",
127 | "studentLoan",
128 | "otherAsset",
129 | "otherLiability",
130 | ]
131 |
132 | type_display_names = {
133 | "checking": "Checking Accounts",
134 | "savings": "Savings Accounts",
135 | "creditCard": "Credit Cards",
136 | "mortgage": "Mortgages",
137 | "autoLoan": "Auto Loans",
138 | "studentLoan": "Student Loans",
139 | "otherAsset": "Other Assets",
140 | "otherLiability": "Other Liabilities",
141 | }
142 |
143 | for account in accounts:
144 | if account.get("closed", False) or account.get("deleted", False):
145 | continue
146 |
147 | acct_type = account["type"]
148 | if acct_type not in account_groups:
149 | account_groups[acct_type] = []
150 |
151 | balance = float(account["balance"]) / 1000
152 | account_groups[acct_type].append(
153 | {
154 | "name": account["name"],
155 | "balance": f"${balance:,.2f}",
156 | "balance_raw": balance,
157 | "id": account["id"],
158 | }
159 | )
160 |
161 | for group in account_groups.values():
162 | group.sort(key=lambda x: abs(x["balance_raw"]), reverse=True)
163 |
164 | output: Dict[str, Any] = {
165 | "accounts": [],
166 | "summary": {
167 | "total_assets": 0.0,
168 | "total_liabilities": 0.0,
169 | "net_worth": 0.0,
170 | },
171 | }
172 |
173 | for acct_type in type_order:
174 | if acct_type in account_groups and account_groups[acct_type]:
175 | group_data = {
176 | "type": type_display_names.get(acct_type, acct_type),
177 | "accounts": account_groups[acct_type],
178 | }
179 | group_total = sum(acct["balance_raw"] for acct in account_groups[acct_type])
180 | group_data["total"] = f"${group_total:,.2f}"
181 |
182 | if acct_type in ["checking", "savings", "otherAsset"]:
183 | output["summary"]["total_assets"] += group_total
184 | elif acct_type in [
185 | "creditCard",
186 | "mortgage",
187 | "autoLoan",
188 | "studentLoan",
189 | "otherLiability",
190 | ]:
191 | output["summary"]["total_liabilities"] += abs(group_total)
192 |
193 | output["accounts"].append(group_data)
194 |
195 | output["summary"]["net_worth_raw"] = (
196 | output["summary"]["total_assets"] - output["summary"]["total_liabilities"]
197 | )
198 | output["summary"]["total_assets"] = f"${output['summary']['total_assets']:,.2f}"
199 | output["summary"]["total_liabilities"] = f"${output['summary']['total_liabilities']:,.2f}"
200 | output["summary"]["net_worth"] = f"${output['summary']['net_worth_raw']:,.2f}"
201 |
202 | return output
203 |
204 |
205 | def _load_json_file(filename: str | Path) -> Dict[str, Any]:
206 | """Load JSON data from a file."""
207 | try:
208 | with open(filename, "r") as f:
209 | return json.load(f)
210 | except FileNotFoundError:
211 | return {}
212 |
213 |
214 | def _save_json_file(filename: str | Path, data: Dict[str, Any]) -> None:
215 | """Save JSON data to a file."""
216 | with open(filename, "w") as f:
217 | json.dump(data, f, indent=2)
218 |
219 |
220 | # 4. Create the MCP server instance
221 | mcp = FastMCP("YNAB")
222 |
223 |
224 | # Define resources
225 | class YNABResources:
226 | def __init__(self):
227 | self._preferred_budget_id: Optional[str] = None
228 | self._category_cache: Dict[str, List[Dict[str, Any]]] = {}
229 | self._load_data()
230 |
231 | def _load_data(self) -> None:
232 | """Load data from files."""
233 | try:
234 | with open(PREFERRED_BUDGET_ID_FILE, "r") as f:
235 | self._preferred_budget_id = f.read().strip() or None
236 | except FileNotFoundError:
237 | self._preferred_budget_id = None
238 |
239 | try:
240 | self._category_cache = _load_json_file(BUDGET_CATEGORY_CACHE_FILE)
241 | except FileNotFoundError:
242 | self._category_cache = {}
243 |
244 | def get_preferred_budget_id(self) -> Optional[str]:
245 | """Get the preferred budget ID."""
246 | return self._preferred_budget_id
247 |
248 | def set_preferred_budget_id(self, budget_id: str) -> None:
249 | """Set the preferred budget ID."""
250 | self._preferred_budget_id = budget_id
251 | with open(PREFERRED_BUDGET_ID_FILE, "w") as f:
252 | f.write(budget_id)
253 |
254 | def get_cached_categories(self, budget_id: str) -> list[types.TextContent]:
255 | """Get categories from the cache formatted for MCP resources."""
256 | cached_categories = self._category_cache.get(budget_id, [])
257 | return [
258 | types.TextContent(
259 | type="text", text=f"{cat.get('name', 'Unnamed')} (ID: {cat.get('id', 'N/A')})"
260 | )
261 | for cat in cached_categories
262 | ]
263 |
264 | def cache_categories(self, budget_id: str, categories: List[Dict[str, Any]]) -> None:
265 | """Cache categories for a budget ID."""
266 | self._category_cache[budget_id] = [
267 | {
268 | "id": cat.get("id"),
269 | "name": cat.get("name"),
270 | "group": cat.get("category_group_name"),
271 | }
272 | for cat in categories
273 | ]
274 | _save_json_file(BUDGET_CATEGORY_CACHE_FILE, self._category_cache)
275 |
276 |
277 | # Instantiate the resources
278 | ynab_resources = YNABResources()
279 |
280 |
281 | # Define resources using decorators
282 | @mcp.resource("ynab://preferences/budget_id")
283 | def get_preferred_budget_id() -> Optional[str]:
284 | """Get the preferred YNAB budget ID."""
285 | return ynab_resources.get_preferred_budget_id()
286 |
287 |
288 | @mcp.resource("ynab://categories/{budget_id}")
289 | def get_cached_categories(budget_id: str) -> list[types.TextContent]:
290 | """Get cached categories for a budget ID."""
291 | return ynab_resources.get_cached_categories(budget_id)
292 |
293 |
294 | # 5. Public tool functions
295 |
296 |
297 | async def _find_category_id(client: ApiClient, budget_id: str, category_name: str) -> Optional[str]:
298 | """Find a category ID by name."""
299 | categories_api = CategoriesApi(client)
300 | categories_response = categories_api.get_categories(budget_id)
301 | categories = categories_response.data.category_groups
302 | for group in categories:
303 | for cat in group.categories:
304 | if cat.name.lower() == category_name.lower():
305 | return cat.id
306 | return None
307 |
308 |
309 | @mcp.tool()
310 | async def create_transaction(
311 | account_id: str,
312 | amount: Annotated[float, Field(description="Amount in dollars")],
313 | payee_name: str,
314 | category_name: Optional[str] = None,
315 | memo: Optional[str] = None,
316 | ) -> Dict[str, Any]:
317 | """Create a new transaction in YNAB."""
318 | async with await get_ynab_client() as client:
319 | transactions_api = TransactionsApi(client)
320 | budgets_api = BudgetsApi(client)
321 |
322 | amount_milliunits = int(amount * 1000)
323 |
324 | # Use preferred budget ID if available, otherwise fetch a list of budgets
325 | budget_id = ynab_resources.get_preferred_budget_id()
326 | if not budget_id:
327 | budgets_response = budgets_api.get_budgets()
328 | budget_id = budgets_response.data.budgets[0].id
329 |
330 | category_id = None
331 | if category_name:
332 | category_id = await _find_category_id(client, budget_id, category_name)
333 |
334 | # Create transaction data
335 | transaction = NewTransaction(
336 | account_id=account_id,
337 | date=date.today(),
338 | amount=amount_milliunits,
339 | payee_name=payee_name,
340 | memo=memo,
341 | category_id=category_id,
342 | )
343 |
344 | wrapper = PostTransactionsWrapper(transaction=transaction)
345 | response = transactions_api.create_transaction(budget_id, wrapper)
346 | if response.data and response.data.transaction:
347 | return response.data.transaction.to_dict()
348 | return {}
349 |
350 |
351 | @mcp.tool()
352 | async def get_account_balance(account_id: str) -> float:
353 | """Get the current balance of a YNAB account (in dollars)."""
354 | async with await get_ynab_client() as client:
355 | accounts_api = AccountsApi(client)
356 | budgets_api = BudgetsApi(client)
357 | budgets_response = budgets_api.get_budgets()
358 | budget_id = budgets_response.data.budgets[0].id
359 |
360 | response = accounts_api.get_account_by_id(budget_id, account_id)
361 | return float(response.data.account.balance) / 1000
362 |
363 |
364 | @mcp.tool()
365 | async def get_budgets() -> str:
366 | """List all YNAB budgets in Markdown format."""
367 | async with await get_ynab_client() as client:
368 | budgets_api = BudgetsApi(client)
369 | budgets_response = budgets_api.get_budgets()
370 | budgets_list = budgets_response.data.budgets
371 |
372 | markdown = "# YNAB Budgets\n\n"
373 | if not budgets_list:
374 | markdown += "_No budgets found._"
375 | else:
376 | for budget in budgets_list:
377 | b = budget.to_dict()
378 | markdown += f"- **{b.get('name', 'Unnamed Budget')}** (ID: {b.get('id')})\n"
379 | return markdown
380 |
381 |
382 | @mcp.tool()
383 | async def get_accounts(budget_id: str) -> str:
384 | """List all YNAB accounts in a specific budget in Markdown format."""
385 | async with await get_ynab_client() as client:
386 | accounts_api = AccountsApi(client)
387 | all_accounts: List[Dict[str, Any]] = []
388 | response = accounts_api.get_accounts(budget_id)
389 | for account in response.data.accounts:
390 | if isinstance(account, Account):
391 | all_accounts.append(account.to_dict())
392 |
393 | formatted = _format_accounts_output(all_accounts)
394 |
395 | markdown = "# YNAB Account Summary\n\n"
396 | markdown += "## Summary\n"
397 | markdown += f"- **Total Assets:** {formatted['summary']['total_assets']}\n"
398 | markdown += f"- **Total Liabilities:** {formatted['summary']['total_liabilities']}\n"
399 | markdown += f"- **Net Worth:** {formatted['summary']['net_worth']}\n\n"
400 |
401 | for group in formatted["accounts"]:
402 | markdown += f"## {group['type']}\n"
403 | markdown += f"**Group Total:** {group['total']}\n\n"
404 |
405 | rows = []
406 | for acct in group["accounts"]:
407 | rows.append([acct["name"], acct["balance"], acct["id"]])
408 |
409 | markdown += _build_markdown_table(
410 | rows, ["Account Name", "Balance", "ID"], ["left", "right", "left"]
411 | )
412 | markdown += "\n"
413 |
414 | return markdown
415 |
416 |
417 | @mcp.tool()
418 | async def get_transactions(budget_id: str, account_id: str) -> str:
419 | """Get recent transactions for a specific account in a specific budget."""
420 | async with await get_ynab_client() as client:
421 | transactions_api = TransactionsApi(client)
422 | all_transactions: List[TransactionDetail] = []
423 | since_date = datetime.now().replace(day=1).date()
424 | response = transactions_api.get_transactions_by_account(
425 | budget_id, account_id, since_date=since_date
426 | )
427 | all_transactions.extend(response.data.transactions)
428 |
429 | markdown = "# Recent Transactions\n\n"
430 | if not all_transactions:
431 | return markdown + "_No recent transactions found._\n"
432 |
433 | headers = ["ID", "Date", "Amount", "Payee Name", "Category Name", "Memo"]
434 | align = ["left", "left", "right", "left", "left", "left"]
435 | rows = []
436 |
437 | for txn in all_transactions:
438 | amount_str = f"${txn.amount / 1000:,.2f}"
439 | rows.append(
440 | [
441 | txn.id,
442 | txn.var_date.strftime("%Y-%m-%d"),
443 | amount_str,
444 | txn.payee_name or "N/A",
445 | txn.category_name or "N/A",
446 | txn.memo or "",
447 | ]
448 | )
449 |
450 | markdown += _build_markdown_table(rows, headers, align)
451 | return markdown
452 |
453 |
454 | def _get_transaction_row(
455 | txn: TransactionDetail, account_map: Dict[str, str], filter_type: str
456 | ) -> List[str]:
457 | """Format a transaction into a row for the markdown table."""
458 | amount_dollars = float(txn.amount) / 1000
459 | amount_str = f"${abs(amount_dollars):,.2f}"
460 | if amount_dollars < 0:
461 | amount_str = f"-{amount_str}"
462 |
463 | status = []
464 | if not txn.category_id:
465 | status.append("Uncategorized")
466 | if not txn.approved:
467 | status.append("Unapproved")
468 |
469 | return [
470 | txn.id,
471 | txn.var_date.strftime("%Y-%m-%d"),
472 | account_map.get(txn.account_id, "Unknown"),
473 | amount_str,
474 | txn.payee_name or "N/A",
475 | ", ".join(status),
476 | txn.memo or "",
477 | ]
478 |
479 |
480 | def _filter_transactions(
481 | transactions: List[TransactionDetail], filter_type: str
482 | ) -> List[TransactionDetail]:
483 | """Filter transactions based on the filter type."""
484 | needs_attention = []
485 | for txn in transactions:
486 | if isinstance(txn, TransactionDetail):
487 | needs_category = filter_type in ["uncategorized", "both"] and not txn.category_id
488 | needs_approval = filter_type in ["unapproved", "both"] and not txn.approved
489 | if needs_category or needs_approval:
490 | needs_attention.append(txn)
491 | return needs_attention
492 |
493 |
494 | @mcp.tool()
495 | async def get_transactions_needing_attention(
496 | budget_id: str,
497 | filter_type: Annotated[
498 | str,
499 | Field(
500 | description="Type of transactions to show. One of: 'uncategorized', 'unapproved', 'both'"
501 | ),
502 | ] = "both",
503 | days_back: Annotated[
504 | Optional[int], Field(description="Number of days to look back (default 30, None for all)")
505 | ] = 30,
506 | ) -> str:
507 | """List transactions that need attention based on specified filter type in a YNAB budget."""
508 | filter_type = filter_type.lower()
509 | if filter_type not in ["uncategorized", "unapproved", "both"]:
510 | return "Error: Invalid filter_type. Must be 'uncategorized', 'unapproved', or 'both'"
511 |
512 | async with await get_ynab_client() as client:
513 | transactions_api = TransactionsApi(client)
514 | accounts_api = AccountsApi(client)
515 |
516 | accounts_response = accounts_api.get_accounts(budget_id)
517 | account_map = {
518 | account.id: account.name
519 | for account in accounts_response.data.accounts
520 | if not account.closed and not account.deleted
521 | }
522 |
523 | since_date = (datetime.now() - timedelta(days=days_back)).date() if days_back else None
524 | response = transactions_api.get_transactions(budget_id, since_date=since_date)
525 | needs_attention = _filter_transactions(response.data.transactions, filter_type)
526 |
527 | markdown = f"# Transactions Needing Attention ({filter_type.title()})\n\n"
528 | if not needs_attention:
529 | return markdown + "_No transactions need attention._"
530 |
531 | markdown += "**Filters Applied:**\n"
532 | markdown += f"- Filter type: {filter_type}\n"
533 | if days_back:
534 | markdown += f"- Looking back {days_back} days\n"
535 | markdown += "\n"
536 |
537 | headers = ["ID", "Date", "Account", "Amount", "Payee", "Status", "Memo"]
538 | align = ["left", "left", "left", "right", "left", "left", "left"]
539 | rows = [_get_transaction_row(txn, account_map, filter_type) for txn in needs_attention]
540 |
541 | markdown += _build_markdown_table(rows, headers, align)
542 | return markdown
543 |
544 |
545 | @mcp.tool()
546 | def _find_transaction_by_id(
547 | transactions: List[TransactionDetail], transaction_id: str, id_type: str
548 | ) -> Optional[TransactionDetail]:
549 | """Find a transaction by its ID and ID type."""
550 | for txn in transactions:
551 | if (
552 | (id_type == "id" and txn.id == transaction_id)
553 | or (id_type == "import_id" and txn.import_id == transaction_id)
554 | or (
555 | id_type == "transfer_transaction_id"
556 | and txn.transfer_transaction_id == transaction_id
557 | )
558 | or (
559 | id_type == "matched_transaction_id" and txn.matched_transaction_id == transaction_id
560 | )
561 | ):
562 | return txn
563 | return None
564 |
565 |
566 | async def categorize_transaction(
567 | budget_id: str,
568 | transaction_id: str,
569 | category_id: str,
570 | id_type: str = "id", # One of: "id", "import_id", "transfer_transaction_id", "matched_transaction_id"
571 | ) -> str:
572 | """Categorize a transaction for a given YNAB budget with the provided category ID.
573 |
574 | Args:
575 | budget_id: The YNAB budget ID
576 | transaction_id: The transaction identifier
577 | category_id: The category ID to assign
578 | id_type: The type of transaction ID being provided. One of:
579 | - "id": Direct transaction ID (default)
580 | - "import_id": YNAB import ID format (YNAB:[milliunit_amount]:[iso_date]:[occurrence])
581 | - "transfer_transaction_id": ID of a transfer transaction
582 | - "matched_transaction_id": ID of a matched transaction
583 | """
584 | async with await get_ynab_client() as client:
585 | transactions_api = TransactionsApi(client)
586 |
587 | # Get since_date for import_id type
588 | since_date = None
589 | if id_type == "import_id" and ":" in transaction_id:
590 | try:
591 | since_date = datetime.strptime(transaction_id.split(":")[2], "%Y-%m-%d").date()
592 | except (ValueError, IndexError):
593 | pass
594 |
595 | response = transactions_api.get_transactions(budget_id, since_date=since_date)
596 | target_transaction = _find_transaction_by_id(
597 | response.data.transactions, transaction_id, id_type
598 | )
599 |
600 | if target_transaction:
601 | wrapper = PutTransactionWrapper(
602 | transaction=ExistingTransaction(
603 | account_id=target_transaction.account_id,
604 | amount=target_transaction.amount,
605 | category_id=category_id,
606 | )
607 | )
608 | transactions_api.update_transaction(
609 | budget_id=budget_id,
610 | transaction_id=target_transaction.id,
611 | data=wrapper,
612 | )
613 | return f"Transaction {transaction_id} (type: {id_type}) categorized as {category_id}."
614 |
615 | return f"Transaction {transaction_id} (type: {id_type}) not found."
616 |
617 |
618 | def _process_category_data(category: Category | Dict[str, Any]) -> tuple[str, str, float, float]:
619 | """Process category data and return tuple of (id, name, budgeted, activity)."""
620 | if isinstance(category, Category):
621 | return category.id, category.name, category.budgeted, category.activity
622 | cat_dict = cast(Dict[str, Any], category)
623 | return cat_dict["id"], cat_dict["name"], cat_dict["budgeted"], cat_dict["activity"]
624 |
625 |
626 | def _format_dollar_amount(amount: float) -> str:
627 | """Format a dollar amount with proper sign and formatting."""
628 | amount_str = f"${abs(amount):,.2f}"
629 | return f"-{amount_str}" if amount < 0 else amount_str
630 |
631 |
632 | @mcp.tool()
633 | async def get_categories(budget_id: str) -> str:
634 | """List all transaction categories for a given YNAB budget in Markdown format."""
635 | async with await get_ynab_client() as client:
636 | categories_api = CategoriesApi(client)
637 | response = categories_api.get_categories(budget_id)
638 | groups = response.data.category_groups
639 |
640 | markdown = "# YNAB Categories\n\n"
641 | headers = ["Category ID", "Category Name", "Budgeted", "Activity"]
642 | align = ["left", "left", "right", "right"]
643 |
644 | for group in groups:
645 | if isinstance(group, CategoryGroupWithCategories):
646 | categories_list = group.categories
647 | group_name = group.name
648 | else:
649 | group_dict = cast(Dict[str, Any], group.to_dict())
650 | categories_list = group_dict["categories"]
651 | group_name = group_dict["name"]
652 |
653 | if not categories_list:
654 | continue
655 |
656 | markdown += f"## {group_name}\n\n"
657 | rows = []
658 |
659 | for category in categories_list:
660 | cat_id, name, budgeted, activity = _process_category_data(category)
661 | budgeted_dollars = float(budgeted) / 1000 if budgeted else 0
662 | activity_dollars = float(activity) / 1000 if activity else 0
663 |
664 | rows.append(
665 | [
666 | cat_id,
667 | name,
668 | _format_dollar_amount(budgeted_dollars),
669 | _format_dollar_amount(activity_dollars),
670 | ]
671 | )
672 |
673 | table_md = _build_markdown_table(rows, headers, align)
674 | markdown += table_md + "\n"
675 | return markdown
676 |
677 |
678 | @mcp.tool()
679 | async def set_preferred_budget_id(budget_id: str) -> str:
680 | """Set the preferred YNAB budget ID."""
681 | ynab_resources.set_preferred_budget_id(budget_id)
682 | return f"Preferred budget ID set to {budget_id}"
683 |
684 |
685 | @mcp.tool()
686 | async def cache_categories(budget_id: str) -> str:
687 | """Cache all categories for a given YNAB budget ID."""
688 | async with await get_ynab_client() as client:
689 | categories_api = CategoriesApi(client)
690 | response = categories_api.get_categories(budget_id)
691 | groups = response.data.category_groups
692 | categories = []
693 | for group in groups:
694 | if isinstance(group, CategoryGroupWithCategories):
695 | categories.extend(group.categories)
696 |
697 | ynab_resources.cache_categories(budget_id, [cat.to_dict() for cat in categories])
698 | return f"Categories cached for budget ID {budget_id}"
699 |
```