MySQL MCP Server

Instructions

Implementation Reference

• src/mysql_mcp_server/server.py:127-174 (handler)

  The call_tool handler function that implements the execute_sql tool. It validates the tool name, extracts the SQL query, connects to MySQL, executes the query, handles results for SELECT/SHOW queries or reports rowcount for modifications, with special case for SHOW TABLES.
  @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Execute SQL commands.""" config = get_db_config() logger.info(f"Calling tool: {name} with arguments: {arguments}") if name != "execute_sql": raise ValueError(f"Unknown tool: {name}") query = arguments.get("query") if not query: raise ValueError("Query is required") try: logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}") with connect(**config) as conn: logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}") with conn.cursor() as cursor: cursor.execute(query) # Special handling for SHOW TABLES if query.strip().upper().startswith("SHOW TABLES"): tables = cursor.fetchall() result = ["Tables_in_" + config["database"]] # Header result.extend([table[0] for table in tables]) return [TextContent(type="text", text="\n".join(result))] # Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.) elif cursor.description is not None: columns = [desc[0] for desc in cursor.description] try: rows = cursor.fetchall() result = [",".join(map(str, row)) for row in rows] return [TextContent(type="text", text="\n".join([",".join(columns)] + result))] except Error as e: logger.warning(f"Error fetching results: {str(e)}") return [TextContent(type="text", text=f"Query executed but error fetching results: {str(e)}")] # Non-SELECT queries else: conn.commit() return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")] except Error as e: logger.error(f"Error executing SQL '{query}': {e}") logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}") return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
• src/mysql_mcp_server/server.py:114-123 (schema)

  Input schema definition for the execute_sql tool, specifying an object with a required 'query' property of type string.
  inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] }
• src/mysql_mcp_server/server.py:106-126 (registration)

  Registration of the execute_sql tool via the list_tools decorator, defining its name, description, and input schema.
  @app.list_tools() async def list_tools() -> list[Tool]: """List available MySQL tools.""" logger.info("Listing tools...") return [ Tool( name="execute_sql", description="Execute an SQL query on the MySQL server", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] } ) ]
• src/mysql_mcp_server/server.py:17-43 (helper)

  Helper function get_db_config() used by the handler to obtain MySQL connection parameters from environment variables.
  def get_db_config(): """Get database configuration from environment variables.""" config = { "host": os.getenv("MYSQL_HOST", "localhost"), "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER"), "password": os.getenv("MYSQL_PASSWORD"), "database": os.getenv("MYSQL_DATABASE"), # Add charset and collation to avoid utf8mb4_0900_ai_ci issues with older MySQL versions # These can be overridden via environment variables for specific MySQL versions "charset": os.getenv("MYSQL_CHARSET", "utf8mb4"), "collation": os.getenv("MYSQL_COLLATION", "utf8mb4_unicode_ci"), # Disable autocommit for better transaction control "autocommit": True, # Set SQL mode for better compatibility - can be overridden "sql_mode": os.getenv("MYSQL_SQL_MODE", "TRADITIONAL") } # Remove None values to let MySQL connector use defaults if not specified config = {k: v for k, v in config.items() if v is not None} if not all([config.get("user"), config.get("password"), config.get("database")]): logger.error("Missing required database configuration. Please check environment variables:") logger.error("MYSQL_USER, MYSQL_PASSWORD, and MYSQL_DATABASE are required") raise ValueError("Missing required database configuration") return config