first

.gitignore

@@ -0,0 +1,3 @@
+.env
+__pycache__
+.gradio

db_work.py

@@ -0,0 +1,105 @@
+import os
+from typing import Dict, List, Any, Optional
+from dotenv import load_dotenv
+import psycopg2
+from psycopg2.extras import RealDictCursor
+import pandas as pd
+from mcp.server.fastmcp import FastMCP
+from pathlib import Path
+
+# Load environment variables
+load_dotenv()
+
+LIST_SCHEMA=os.getenv('LIST_SCHEMA')
+LIST_DATABASE_INFOS=os.getenv('LIST_DATABASE_INFOS')
+TABLE_IN_SCHEMA=os.getenv('TABLE_IN_SCHEMA')
+COLUMN_IN_TABLE=os.getenv('COLUMN_IN_TABLE')
+
+class DatabaseInterface:
+	def __init__(self):
+		# Initialize FastMCP server
+		self.mcp = FastMCP("ecommerce-mcp-server")
+		
+		self.db_config = {
+			'host': os.getenv('DB_HOST'),
+			'port': os.getenv('DB_PORT'),
+			'database': os.getenv('DB_NAME'),
+			'user': os.getenv('DB_USER'),
+			'password': os.getenv('DB_PASSWORD')
+		}
+		print('=============>',self.db_config)
+		
+	def get_db_connection(self):
+		"""Create database connection"""
+		return psycopg2.connect(**self.db_config)
+
+	def list_schemas(self):
+		print("=======>", LIST_SCHEMA)
+		sql_path = Path(LIST_SCHEMA)
+		with sql_path.open("r", encoding="utf-8") as f:
+			query = f.read()
+
+		conn = self.get_db_connection()
+		try:
+			with conn.cursor() as cur:
+				cur.execute(query)
+				result = cur.fetchone()[0]  # JSON object
+				return result
+		finally:
+			conn.close()
+
+	def list_database_info(self):
+		sql_path = Path(LIST_DATABASE_INFOS)
+		with sql_path.open("r", encoding="utf-8") as f:
+			query = f.read()
+
+		conn = self.get_db_connection()
+		try:
+			with conn.cursor() as cur:
+				cur.execute(query)
+				result = cur.fetchone()[0]  # JSON object
+				return result
+		finally:
+			conn.close()
+	
+	def list_tables_in_schema(self, schema_name: str):
+		sql_path = Path(TABLE_IN_SCHEMA)
+		with sql_path.open("r", encoding="utf-8") as f:
+			query = f.read()
+
+		conn = self.get_db_connection()
+		try:
+			with conn.cursor() as cur:
+				cur.execute(query, {'schema_name': schema_name})
+				result = cur.fetchone()[0]  # JSON object
+				return result
+		finally:
+			conn.close()
+	
+	def list_columns_in_table(self, schema_name: str, table_name: str):
+		sql_path = Path(COLUMN_IN_TABLE)
+		with sql_path.open("r", encoding="utf-8") as f:
+			query = f.read()
+
+		conn = self.get_db_connection()
+		try:
+			with conn.cursor() as cur:
+				cur.execute(query, {
+					'schema_name': schema_name,
+					'table_name': table_name
+				})
+				result = cur.fetchone()[0]  # JSON object
+				return result
+		finally:
+			conn.close()
+
+	def read_only_query(self, query):
+		try:
+			conn = self.get_db_connection()
+			with conn.cursor() as cur:
+				cur.execute("SET TRANSACTION READ ONLY")
+				cur.execute(query)
+				result = cur.fetchall()  # JSON object
+				return result
+		finally:
+			conn.close()

gradio_mcp.py

@@ -0,0 +1,109 @@
+import gradio as gr
+from db_work import DatabaseInterface
+import os
+from PIL import Image
+
+db_interface = DatabaseInterface()
+# Define the functions
+def get_schemas():
+    '''
+    this function allow you to acknowledge the database schema in order for 
+    you to know which schema to query to get the relevant informations
+    '''
+    return db_interface.list_schemas()
+
+def get_db_infos():
+    '''
+    this function allow you to acknowledge the relevant database information for you to better understand what is it about
+    '''
+    return db_interface.list_database_info()
+
+def get_list_of_tables_in_schema(schema):
+    """
+        this function allows you to get the list of tables (associated with their description if exist) 
+        of all the tables that exist in a schema
+    """
+    return db_interface.list_tables_in_schema(schema)
+
+def get_list_of_column_in_table(schema, table):
+    """
+        this function allows you to get the list of columns of a specific table of a specific schema.
+        each column is associated with its datatype and its description if exist 
+    """
+    return db_interface.list_columns_in_table(schema, table)
+
+def run_read_only_query(query:str):
+    """
+        based on what you know about the database properties, you can use this function to run read-only query
+        in order to make analysis
+        the output is of shape:
+        List(Tuple()) where each entry if the list is a row and each entry of the tuple is a column value
+    """
+    return db_interface.read_only_query(query)
+
+def create_sample_image():
+    img_path = "./EuijP.png"
+    if not os.path.exists(img_path):
+        img = Image.new("RGB", (300, 150), color="lightgreen")
+        img.save(img_path)
+    return img_path
+
+def serve_image_from_path():
+    """
+        get a scatter plot of 2 variables.
+        input type: [[list_x], [list_y]]
+        it is up to you to determine if the variable need to be standardize or not
+    """
+    return create_sample_image()
+
+# Create the Gradio Blocks interface
+with gr.Blocks() as interface:
+    with gr.Row():
+        with gr.Column(scale=1):
+            # Get info on the schema
+            discover_input = gr.Textbox(label="Get info on schemas of the database")
+            discover_btn = gr.Button("run get infos on the schemas of the database")
+            
+            # Get info on the database
+            database_info = gr.Textbox(label="Get info on the database")
+            database_info_btn = gr.Button("Run Get info on the database")
+            
+            # Get table in schema
+            table_in_schema_input = gr.Textbox(label="What schema you want table name for")
+            table_in_schema_btn = gr.Button("Run Get list of table in schema")
+
+            # Get Columns in Table
+            gr.Markdown("### Get Columns in Table\nRetrieve the columns of a table in a schema.")
+            schema_input = gr.Textbox(label="Schema Name")
+            table_input = gr.Textbox(label="Table Name")
+            column_btn = gr.Button("Get Columns")
+
+            gr.Markdown("### Enter a read-only query")
+            query_input = gr.Textbox(label="read-only query")
+            query_btn = gr.Button("Get Columns")
+
+            gr.Markdown("### generate a scatter-plot")
+            input_text = gr.Textbox(label="Prompt")
+            generate_button = gr.Button("Generate")
+        
+        with gr.Column(scale=2):
+            schema_info = gr.Textbox(label="Discover DB Output")
+            db_info = gr.Textbox(label="Query DB Output")
+            table_in_schema = gr.Textbox(label="what table are in the selected schema")
+            column_output = gr.Textbox(label="Table Columns Output")
+            query_output = gr.Textbox(label="your query output")
+            output_image = gr.Image(label="Generated Image", type="filepath")
+
+
+
+    # Bind functions to buttons
+    discover_btn.click(get_schemas, outputs=schema_info)
+    database_info_btn.click(get_db_infos, outputs=db_info)
+    table_in_schema_btn.click(get_list_of_tables_in_schema, inputs=table_in_schema_input, outputs=table_in_schema)
+    column_btn.click(get_list_of_column_in_table, inputs=[schema_input, table_input], outputs=column_output)
+    query_btn.click(run_read_only_query, inputs=query_input, outputs=query_output)
+    generate_button.click(fn=serve_image_from_path, outputs=output_image)
+
+# Launch the app
+interface.launch(mcp_server=True, share=True)
+

list_columns_in_table.sql

@@ -0,0 +1,25 @@
+-- server/resources/sql/list_columns_in_table.sql
+-- Returns column metadata for a specific table as a JSON object
+-- Uses parameters: :schema_name, :table_name
+
+WITH columns AS (
+    SELECT
+        cols.column_name,
+        cols.data_type,
+        col_description(('"' || cols.table_schema || '"."' || cols.table_name || '"')::regclass, cols.ordinal_position) AS description
+    FROM information_schema.columns cols
+    WHERE cols.table_schema = %(schema_name)s
+      AND cols.table_name = %(table_name)s
+    ORDER BY cols.ordinal_position
+)
+SELECT jsonb_build_object(
+    'columns',
+    jsonb_agg(
+        jsonb_build_object(
+            'name', column_name,
+            'type', data_type,
+            'description', description
+        )
+    )
+) AS column_list
+FROM columns;

list_database_infos.sql

@@ -0,0 +1,16 @@
+-- server/resources/sql/list_database_info.sql
+-- Returns basic information about the current database in JSON format
+
+SELECT jsonb_build_object(
+    'database',
+    jsonb_build_object(
+        'name', current_database(),
+        'description', (
+            SELECT description
+            FROM pg_shdescription
+            JOIN pg_database ON pg_database.oid = pg_shdescription.objoid
+            WHERE pg_database.datname = current_database()
+            LIMIT 1
+        )
+    )
+) AS database_info;

list_schema.sql

@@ -0,0 +1,25 @@
+-- server/resources/sql/list_schemas.sql
+-- List all non-system schemas in the database
+-- Returns a JSON array of schema objects
+
+WITH schemas AS (
+    SELECT
+        schema_name,
+        obj_description(pg_namespace.oid) as description
+    FROM information_schema.schemata
+    JOIN pg_namespace ON pg_namespace.nspname = schema_name
+    WHERE
+        schema_name NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
+        AND schema_name NOT LIKE 'pg_%'
+    ORDER BY schema_name
+)
+SELECT jsonb_build_object(
+    'schemas',
+    jsonb_agg(
+        jsonb_build_object(
+            'name', schema_name,
+            'description', description
+        )
+    )
+) AS schema_list
+FROM schemas;

list_tables_in_schema.sql

@@ -0,0 +1,23 @@
+-- server/resources/sql/list_tables_in_schema.sql
+-- Returns all user-defined tables in a given schema with their descriptions as JSON
+-- Uses a parameter :schema_name
+
+WITH tables AS (
+    SELECT
+        table_name,
+        obj_description(('"' || table_schema || '"."' || table_name || '"')::regclass) AS description
+    FROM information_schema.tables
+    WHERE table_schema = %(schema_name)s
+      AND table_type = 'BASE TABLE'
+    ORDER BY table_name
+)
+SELECT jsonb_build_object(
+    'tables',
+    jsonb_agg(
+        jsonb_build_object(
+            'name', table_name,
+            'description', description
+        )
+    )
+) AS table_list
+FROM tables;

requirements.txt

@@ -0,0 +1,13 @@
+mcp>=1.0.0
+gradio>=4.0.0
+gradio[mcp]
+psycopg2-binary>=2.9.0
+pandas>=2.0.0
+sqlalchemy>=2.0.0
+python-dotenv>=1.0.0
+numpy>=1.24.0
+scikit-learn>=1.3.0
+plotly>=5.0.0  # For interactive visualizations
+seaborn>=0.12.0  # For statistical plots
+asyncio-throttle>=1.0.0  # For rate limiting
+cachetools>=5.0.0  # For caching optimization