MCP SQL Agent

Instructions

Implementation Reference

• servers/mcp_server_mysql.py:219-271 (handler)

  MySQL implementation of the search_tables MCP tool handler. Searches database tables and columns for a given keyword using information_schema queries.

  @mcp.tool()
  def search_tables(keyword: str) -> str:
      """Search for tables or columns containing a specific keyword"""
      try:
          conn = mysql.connector.connect(**db_config)
          cursor = conn.cursor()
          
          # Get all tables
          tables_query = """
          SELECT TABLE_NAME 
          FROM information_schema.TABLES 
          WHERE TABLE_SCHEMA = %s
          """
          cursor.execute(tables_query, (db_config['database'],))
          tables = cursor.fetchall()
          
          matches = []
          
          for (table_name,) in tables:
              # Check if table name contains keyword
              if keyword.lower() in table_name.lower():
                  matches.append(f"Table: {table_name}")
              
              # Check columns
              columns_query = """
              SELECT COLUMN_NAME 
              FROM information_schema.COLUMNS 
              WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
              """
              cursor.execute(columns_query, (db_config['database'], table_name))
              columns = cursor.fetchall()
              
              matching_columns = []
              for (column_name,) in columns:
                  if keyword.lower() in column_name.lower():
                      matching_columns.append(column_name)
              
              if matching_columns:
                  matches.append(f"Table '{table_name}' has columns: {', '.join(matching_columns)}")
          
          if matches:
              return f"Found matches for '{keyword}':\n" + "\n".join(matches)
          else:
              return f"No tables or columns found containing '{keyword}'"
              
      except Exception as e:
          return f"Error: {str(e)}"
      finally:
          if 'cursor' in locals():
              cursor.close()
          if 'conn' in locals():
              conn.close()

• servers/mcp_server_oracle.py:235-286 (handler)

  Oracle implementation of the search_tables MCP tool handler. Searches user tables and columns for a given keyword using Oracle system views.

  @mcp.tool()
  def search_tables(keyword: str) -> str:
      """Search for tables or columns containing a specific keyword"""
      try:
          conn = cx_Oracle.connect(**db_config)
          cursor = conn.cursor()
          
          # Get all tables
          tables_query = """
          SELECT table_name 
          FROM user_tables 
          ORDER BY table_name
          """
          cursor.execute(tables_query)
          tables = cursor.fetchall()
          
          matches = []
          
          for (table_name,) in tables:
              # Check if table name contains keyword
              if keyword.upper() in table_name.upper():
                  matches.append(f"Table: {table_name}")
              
              # Check columns
              columns_query = """
              SELECT column_name 
              FROM user_tab_columns 
              WHERE table_name = :1
              """
              cursor.execute(columns_query, (table_name,))
              columns = cursor.fetchall()
              
              matching_columns = []
              for (column_name,) in columns:
                  if keyword.upper() in column_name.upper():
                      matching_columns.append(column_name)
              
              if matching_columns:
                  matches.append(f"Table '{table_name}' has columns: {', '.join(matching_columns)}")
          
          if matches:
              return f"Found matches for '{keyword}':\n" + "\n".join(matches)
          else:
              return f"No tables or columns found containing '{keyword}'"
              
      except Exception as e:
          return f"Error: {str(e)}"
      finally:
          if 'cursor' in locals():
              cursor.close()
          if 'conn' in locals():
              conn.close()

• servers/mcp_server_sqlite.py:151-188 (handler)

  SQLite implementation of the search_tables MCP tool handler. Searches tables and columns for a given keyword using sqlite_master and PRAGMA.

  @mcp.tool()
  def search_tables(keyword: str) -> str:
      """Search for tables or columns containing a specific keyword"""
      conn = sqlite3.connect(db_path)
      try:
          # Get all tables
          tables_query = "SELECT name FROM sqlite_master WHERE type='table';"
          tables = conn.execute(tables_query).fetchall()
          
          matches = []
          
          for (table_name,) in tables:
              # Check if table name contains keyword
              if keyword.lower() in table_name.lower():
                  matches.append(f"Table: {table_name}")
              
              # Check columns
              pragma_query = f"PRAGMA table_info({table_name});"
              columns = conn.execute(pragma_query).fetchall()
              
              matching_columns = []
              for column in columns:
                  column_name = column[1]
                  if keyword.lower() in column_name.lower():
                      matching_columns.append(column_name)
              
              if matching_columns:
                  matches.append(f"Table '{table_name}' has columns: {', '.join(matching_columns)}")
          
          if matches:
              return f"Found matches for '{keyword}':\n" + "\n".join(matches)
          else:
              return f"No tables or columns found containing '{keyword}'"
              
      except Exception as e:
          return f"Error: {str(e)}"
      finally:
          conn.close()