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()