YNAB MCP Server

Instructions

Implementation Reference

• src/ynab_mcp/server.py:141-173 (handler)

  MCP tool handler implementation for get_transactions. Registers the tool via @mcp.tool() decorator, validates inputs via type hints/docstring, delegates to YNABClient, and serializes response to JSON.
  @mcp.tool() async def get_transactions( budget_id: str, since_date: str = None, until_date: str = None, account_id: str = None, category_id: str = None, limit: int = None, page: int = None, ) -> str: """Get transactions with optional filtering and pagination. Args: budget_id: The ID of the budget (use 'last-used' for default budget) since_date: Only return transactions on or after this date (YYYY-MM-DD format) until_date: Only return transactions on or before this date (YYYY-MM-DD format) account_id: Filter by account ID (optional) category_id: Filter by category ID (optional) limit: Number of transactions per page (default: 100, max: 500) page: Page number for pagination (1-indexed, default: 1) Returns: JSON string with transactions array and pagination metadata Note: For large date ranges (>1 year), use get_category_spending_summary or compare_spending_by_year instead to avoid timeouts and reduce context usage. """ client = get_ynab_client() result = await client.get_transactions( budget_id, since_date, until_date, account_id, category_id, limit, page ) return json.dumps(result, indent=2)
• src/ynab_mcp/ynab_client.py:525-629 (helper)

  Supporting method in YNABClient class that performs the actual API request to YNAB, applies client-side filtering for category/until_date/pagination, formats transaction data, and handles errors.
  async def get_transactions( self, budget_id: str, since_date: str | None = None, until_date: str | None = None, account_id: str | None = None, category_id: str | None = None, limit: int | None = None, page: int | None = None, ) -> dict[str, Any]: """Get transactions with optional filtering and pagination. Args: budget_id: The budget ID or 'last-used' since_date: Only return transactions on or after this date (YYYY-MM-DD) until_date: Only return transactions on or before this date (YYYY-MM-DD) account_id: Filter by account ID category_id: Filter by category ID limit: Number of transactions per page (default: 100, max: 500) page: Page number for pagination (1-indexed, default: 1) Returns: Dictionary with transactions, pagination info, and total count Note: For large date ranges (>1 year), consider using get_category_spending_summary or compare_spending_by_year instead to avoid timeouts and reduce context usage. """ try: # Use direct API call for better filtering support url = f"{self.api_base_url}/budgets/{budget_id}/transactions" params = {} if since_date: params["since_date"] = since_date if account_id: url = f"{self.api_base_url}/budgets/{budget_id}/accounts/{account_id}/transactions" result = await self._make_request_with_retry("get", url, params=params) txn_data = result["data"]["transactions"] # Apply filters filtered_transactions = [] for txn in txn_data: # Filter by category_id if provided if category_id and txn.get("category_id") != category_id: continue # Filter by until_date if provided (client-side filtering) if until_date and txn["date"] > until_date: continue filtered_transactions.append(txn) # Pagination page_size = min(limit or 100, 500) # Default 100, max 500 page_num = max(page or 1, 1) # Default to page 1, minimum 1 total_count = len(filtered_transactions) total_pages = (total_count + page_size - 1) // page_size if total_count > 0 else 1 start_idx = (page_num - 1) * page_size end_idx = start_idx + page_size paginated_txns = filtered_transactions[start_idx:end_idx] transactions = [] for txn in paginated_txns: transactions.append( { "id": txn["id"], "date": txn["date"], "amount": txn["amount"] / 1000 if txn.get("amount") else 0, "memo": txn.get("memo"), "cleared": txn.get("cleared"), "approved": txn.get("approved"), "account_id": txn.get("account_id"), "account_name": txn.get("account_name"), "payee_id": txn.get("payee_id"), "payee_name": txn.get("payee_name"), "category_id": txn.get("category_id"), "category_name": txn.get("category_name"), "transfer_account_id": txn.get("transfer_account_id"), "deleted": txn.get("deleted"), } ) return { "transactions": transactions, "pagination": { "page": page_num, "per_page": page_size, "total_count": total_count, "total_pages": total_pages, "has_next_page": page_num < total_pages, "has_prev_page": page_num > 1, }, } except httpx.HTTPStatusError as e: raise Exception( f"Failed to get transactions: HTTP {e.response.status_code} - {e.response.text}" ) from e except Exception as e: raise Exception(f"Failed to get transactions: {type(e).__name__}: {e}") from e