MCP Server for MySQL

import { describe, it, expect, beforeAll, afterAll } from 'vitest'; import { Server } from '@modelcontextprotocol/sdk/server/index.js'; import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'; import * as mysql2 from 'mysql2/promise'; import * as dotenv from 'dotenv'; import { spawn, ChildProcess } from 'child_process'; import * as path from 'path'; // Load test environment variables dotenv.config({ path: '.env.test' }); // Helper function to create a test client function createTestClient() { // This would be a simplified version of an MCP client for testing return { async listTools() { // Implementation would communicate with the server return { tools: [ { name: 'mysql_query', description: 'Run a read-only MySQL query', inputSchema: { type: 'object', properties: { sql: { type: 'string' }, }, }, }, ], }; }, async callTool(name: string, args: any) { // Implementation would send the request to the server if (name !== 'mysql_query') { throw new Error(`Unknown tool: ${name}`); } // This is a mock response - in a real test, this would come from the server return { content: [ { type: 'text', text: JSON.stringify([{ result: 'test' }], null, 2), }, ], isError: false, }; }, async listResources() { // Implementation would communicate with the server return { resources: [ { uri: `mysql://127.0.0.1:3306/test_table/schema`, mimeType: 'application/json', name: '"test_table" database schema', }, ], }; }, async readResource(uri: string) { // Implementation would communicate with the server return { contents: [ { uri, mimeType: 'application/json', text: JSON.stringify([ { column_name: 'id', data_type: 'int' }, { column_name: 'name', data_type: 'varchar' }, { column_name: 'created_at', data_type: 'timestamp' }, ], null, 2), }, ], }; }, close() { // Clean up resources } }; } describe('Server', () => { let serverProcess: any; let pool: any; let client: ReturnType<typeof createTestClient>; beforeAll(async () => { // Set up test database pool = mysql2.createPool({ host: process.env.MYSQL_HOST || '127.0.0.1', port: Number(process.env.MYSQL_PORT || '3306'), user: process.env.MYSQL_USER || 'root', password: process.env.MYSQL_PASS || '', database: process.env.MYSQL_DB || 'mcp_test', connectionLimit: 5, }); // Create test client client = createTestClient(); }); afterAll(async () => { // Clean up if (serverProcess) { serverProcess.kill(); } if (pool) { await pool.end(); } if (client) { client.close(); } }); it('should list available tools', async () => { const result = await client.listTools(); expect(result.tools).toHaveLength(1); expect(result.tools[0].name).toBe('mysql_query'); }); it('should execute a query tool', async () => { const result = await client.callTool('mysql_query', { sql: 'SELECT * FROM test_table' }); expect(result.isError).toBe(false); expect(result.content).toHaveLength(1); expect(result.content[0].type).toBe('text'); }); it('should list available resources', async () => { const result = await client.listResources(); expect(result.resources).toHaveLength(1); expect(result.resources[0].name).toContain('test_table'); }); it('should read a resource', async () => { const uri = 'mysql://127.0.0.1:3306/test_table/schema'; const result = await client.readResource(uri); expect(result.contents).toHaveLength(1); const content = JSON.parse(result.contents[0].text); expect(Array.isArray(content)).toBe(true); expect(content.length).toBeGreaterThan(0); expect(content[0]).toHaveProperty('column_name'); expect(content[0]).toHaveProperty('data_type'); }); });