MariaDB MCP Server

Instructions

Implementation Reference

• src/mariadb_mcp/server.py:231-280 (handler)

  The main handler function for the 'get_table_schema' tool. It is registered via the @mcp.tool() decorator. Uses DESCRIBE query to fetch column details and SHOW TABLE STATUS for additional table metadata, formatting the output as a markdown table.

  @mcp.tool()
  async def get_table_schema(table_name: str, database: Optional[str] = None) -> str:
      """Get the schema/structure of a specific table."""
      try:
          if database:
              full_table_name = f"`{database}`.`{table_name}`"
          else:
              full_table_name = f"`{table_name}`"
          
          query = f"DESCRIBE {full_table_name}"
          results = await db_connection.execute_query(query)
          
          if not results:
              return f"Table '{table_name}' not found"
          
          schema_info = f"Schema for table '{table_name}':\n\n"
          schema_info += "| Field | Type | Null | Key | Default | Extra |\n"
          schema_info += "|-------|------|------|-----|---------|-------|\n"
          
          for row in results:
              field = row['Field']
              type_info = row['Type']
              null_info = row['Null']
              key_info = row['Key'] or ''
              default_info = row['Default'] or ''
              extra_info = row['Extra'] or ''
              
              schema_info += f"| {field} | {type_info} | {null_info} | {key_info} | {default_info} | {extra_info} |\n"
          
          # Also get table status for additional info
          status_query = f"SHOW TABLE STATUS LIKE '{table_name}'"
          if database:
              status_query = f"SHOW TABLE STATUS FROM `{database}` LIKE '{table_name}'"
          
          status_results = await db_connection.execute_query(status_query)
          if status_results:
              status = status_results[0]
              schema_info += f"\nTable Info:\n"
              schema_info += f"- Engine: {status.get('Engine', 'N/A')}\n"
              schema_info += f"- Rows: {status.get('Rows', 'N/A')}\n"
              schema_info += f"- Data Length: {status.get('Data_length', 'N/A')} bytes\n"
              schema_info += f"- Auto Increment: {status.get('Auto_increment', 'N/A')}\n"
              schema_info += f"- Create Time: {status.get('Create_time', 'N/A')}\n"
              schema_info += f"- Comment: {status.get('Comment', 'N/A')}\n"
          
          return schema_info
      
      except Exception as e:
          logger.error(f"Error getting table schema: {e}")
          return f"Error getting table schema: {str(e)}"