MCP SQL Agent

Instructions

Implementation Reference

• servers/mcp_server_mysql.py:86-90 (handler)

  Handler function for 'get_schema' tool registered via @mcp.tool() decorator. Delegates to get_database_schema() helper.
  @mcp.tool() def get_schema() -> str: """Get the complete database schema with table structures and sample data""" return get_database_schema()
• servers/mcp_server_sqlite.py:59-62 (handler)

  Handler function for 'get_schema' tool registered via @mcp.tool() decorator. Delegates to get_database_schema() helper.
  @mcp.tool() def get_schema() -> str: """Get the complete database schema with table structures and sample data""" return get_database_schema()
• servers/mcp_server_oracle.py:96-99 (handler)

  Handler function for 'get_schema' tool registered via @mcp.tool() decorator. Delegates to get_database_schema() helper.
  @mcp.tool() def get_schema() -> str: """Get the complete database schema with table structures and sample data""" return get_database_schema()
• servers/mcp_server_mysql.py:24-85 (helper)

  Core logic for fetching MySQL database schema, including tables, columns, and sample data.
  def get_database_schema() -> str: """Get the database schema information""" try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor() # Get all table names tables_query = """ SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s """ cursor.execute(tables_query, (db_config['database'],)) tables = cursor.fetchall() schema_info = f"Database Schema for '{db_config['database']}':\n\n" for (table_name,) in tables: schema_info += f"Table: {table_name}\n" # Get column information for each table columns_query = """ SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_KEY, EXTRA FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """ cursor.execute(columns_query, (db_config['database'], table_name)) columns = cursor.fetchall() for column in columns: col_name, data_type, is_nullable, default_value, column_key, extra = column pk_indicator = " (PRIMARY KEY)" if column_key == "PRI" else "" null_indicator = " NOT NULL" if is_nullable == "NO" else "" default_indicator = f" DEFAULT {default_value}" if default_value else "" auto_inc = f" {extra}" if extra else "" schema_info += f" - {col_name}: {data_type}{pk_indicator}{null_indicator}{default_indicator}{auto_inc}\n" # Get sample data (first 3 rows) sample_query = f"SELECT * FROM `{table_name}` LIMIT 3" try: cursor.execute(sample_query) sample_data = cursor.fetchall() if sample_data: schema_info += f" Sample data:\n" for row in sample_data: schema_info += f" {row}\n" except Exception as e: schema_info += f" Sample data: Error reading sample data - {e}\n" schema_info += "\n" return schema_info except Exception as e: return f"Error getting schema: {str(e)}" finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close()
• servers/mcp_server_sqlite.py:15-58 (helper)

  Core logic for fetching SQLite database schema using PRAGMA table_info.
  def get_database_schema() -> str: """Get the database schema information""" conn = sqlite3.connect(db_path) try: # Get all table names tables_query = "SELECT name FROM sqlite_master WHERE type='table';" tables = conn.execute(tables_query).fetchall() schema_info = "Database Schema:\n\n" for (table_name,) in tables: schema_info += f"Table: {table_name}\n" # Get column information for each table pragma_query = f"PRAGMA table_info({table_name});" columns = conn.execute(pragma_query).fetchall() for column in columns: cid, name, data_type, notnull, default_value, pk = column pk_indicator = " (PRIMARY KEY)" if pk else "" null_indicator = " NOT NULL" if notnull else "" default_indicator = f" DEFAULT {default_value}" if default_value else "" schema_info += f" - {name}: {data_type}{pk_indicator}{null_indicator}{default_indicator}\n" # Get sample data (first 3 rows) sample_query = f"SELECT * FROM {table_name} LIMIT 3;" try: sample_data = conn.execute(sample_query).fetchall() if sample_data: schema_info += f" Sample data:\n" for row in sample_data: schema_info += f" {row}\n" except Exception as e: schema_info += f" Sample data: Error reading sample data - {e}\n" schema_info += "\n" return schema_info except Exception as e: return f"Error getting schema: {str(e)}" finally: conn.close()