diff --git a/pkg/schema/converter/k8stonorman.go b/pkg/schema/converter/k8stonorman.go
index 6e9097fd..039e594f 100644
--- a/pkg/schema/converter/k8stonorman.go
+++ b/pkg/schema/converter/k8stonorman.go
@@ -49,10 +49,12 @@ func GVRToPluralName(gvr schema.GroupVersionResource) string {
 	return fmt.Sprintf("%s.%s", gvr.Group, gvr.Resource)
 }
 
-// GetGVKForKind attempts to retrieve a GVK for a given Kind. Not all kind represent top level resources,
-// so this function may return nil if the kind did not have a gvk extension
-func GetGVKForKind(kind *proto.Kind) *schema.GroupVersionKind {
-	extensions, ok := kind.Extensions[gvkExtensionName].([]any)
+// GetGVKForProto attempts to retrieve a GVK for a given OpenAPI V2 schema
+// object.
+// The GVK is defined in an extension. It is possible that the protoSchema does
+// not have the GVK extension set - in that case, we return nil.
+func GetGVKForProtoSchema(protoSchema proto.Schema) *schema.GroupVersionKind {
+	extensions, ok := protoSchema.GetExtensions()[gvkExtensionName].([]any)
 	if !ok {
 		return nil
 	}
@@ -69,6 +71,12 @@ func GetGVKForKind(kind *proto.Kind) *schema.GroupVersionKind {
 	return nil
 }
 
+// GetGVKForKind attempts to retrieve a GVK for a given Kind. Not all kind represent top level resources,
+// so this function may return nil if the kind did not have a gvk extension
+func GetGVKForKind(kind *proto.Kind) *schema.GroupVersionKind {
+	return GetGVKForProtoSchema(kind)
+}
+
 // ToSchemas creates the schemas for a K8s server, using client to discover groups/resources, and crd to potentially
 // add additional information about new fields/resources. Mostly ties together addDiscovery and addCustomResources.
 func ToSchemas(crd v1.CustomResourceDefinitionClient, client discovery.DiscoveryInterface) (map[string]*types.APISchema, error) {
diff --git a/pkg/schema/definitions/converter.go b/pkg/schema/definitions/converter.go
new file mode 100644
index 00000000..57c4f8d4
--- /dev/null
+++ b/pkg/schema/definitions/converter.go
@@ -0,0 +1,246 @@
+package definitions
+
+import (
+	"fmt"
+
+	"github.com/rancher/apiserver/pkg/types"
+	wranglerDefinition "github.com/rancher/wrangler/v2/pkg/schemas/definition"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/kube-openapi/pkg/util/proto"
+)
+
+// crdToDefinition builds a schemaDefinition for a CustomResourceDefinition
+func crdToDefinition(jsonSchemaProps *apiextv1.JSONSchemaProps, modelName string) schemaDefinition {
+	path := proto.NewPath(modelName)
+
+	definitions := make(map[string]definition)
+	convertJSONSchemaPropsToDefinition(*jsonSchemaProps, path, definitions)
+
+	return schemaDefinition{
+		DefinitionType: modelName,
+		Definitions:    definitions,
+	}
+}
+
+// convertJSONSchemaPropsToDefinition recurses through the given schema props of
+// type object and adds each definition found to the map of definitions
+//
+// This supports all OpenAPI V3 types: boolean, number, integer, string, object and array
+// as defined here: https://swagger.io/specification/v3/
+func convertJSONSchemaPropsToDefinition(props apiextv1.JSONSchemaProps, path proto.Path, definitions map[string]definition) {
+	def := definition{
+		Type:           path.String(),
+		Description:    props.Description,
+		ResourceFields: map[string]definitionField{},
+	}
+
+	requiredSet := make(map[string]struct{})
+	for _, name := range props.Required {
+		requiredSet[name] = struct{}{}
+	}
+
+	for name, prop := range props.Properties {
+		_, required := requiredSet[name]
+		field := convertJSONSchemaPropsToDefinitionField(prop, path.FieldPath(name), required)
+		def.ResourceFields[name] = field
+
+		if prop.Type != "object" && prop.Type != "array" {
+			continue
+		}
+
+		schema := prop
+		// Get the properties of the items inside the array
+		if prop.Type == "array" {
+			items := getItemsSchema(prop)
+			if items == nil {
+				continue
+			}
+			schema = *items
+		}
+
+		if schema.Type == "object" {
+			convertJSONSchemaPropsToDefinition(schema, path.FieldPath(name), definitions)
+		}
+	}
+	definitions[path.String()] = def
+}
+
+func convertJSONSchemaPropsToDefinitionField(props apiextv1.JSONSchemaProps, path proto.Path, required bool) definitionField {
+	field := definitionField{
+		Description: props.Description,
+		Required:    required,
+		Type:        getPrimitiveType(props.Type),
+	}
+	switch props.Type {
+	case "array":
+		field.Type = "array"
+		if item := getItemsSchema(props); item != nil {
+			if item.Type == "object" || item.Type == "array" {
+				field.SubType = path.String()
+			} else {
+				field.SubType = getPrimitiveType(item.Type)
+			}
+		}
+	case "object":
+		field.Type = path.String()
+	}
+	return field
+}
+
+// typ is a OpenAPI V2 or V3 type
+func getPrimitiveType(typ string) string {
+	switch typ {
+	case "integer", "number":
+		return "int"
+	default:
+		return typ
+	}
+}
+
+func getItemsSchema(props apiextv1.JSONSchemaProps) *apiextv1.JSONSchemaProps {
+	if props.Items == nil {
+		return nil
+	}
+
+	if props.Items.Schema != nil {
+		return props.Items.Schema
+	} else if len(props.Items.JSONSchemas) > 0 {
+		return &props.Items.JSONSchemas[0]
+	}
+	return nil
+}
+
+var _ proto.Reference = (*openAPIV2Reference)(nil)
+var _ proto.Schema = (*openAPIV2Reference)(nil)
+
+type openAPIV2Reference struct {
+	proto.BaseSchema
+	reference string
+	subSchema proto.Schema
+}
+
+func (r *openAPIV2Reference) Accept(v proto.SchemaVisitor) {
+	v.VisitReference(r)
+}
+
+func (r *openAPIV2Reference) Reference() string {
+	return r.reference
+}
+
+func (r *openAPIV2Reference) SubSchema() proto.Schema {
+	return r.subSchema
+}
+
+func (r *openAPIV2Reference) GetName() string {
+	return fmt.Sprintf("Reference to %q", r.reference)
+}
+
+// newObjectMetaReference creates an OpenAPI V2 reference for the .metadata
+// field.
+// path is expected to be the path of the object without a .metadata suffix
+func newObjectMetaReference(path proto.Path, models proto.Models) *openAPIV2Reference {
+	objectMetaModel := models.LookupModel("io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta")
+	ref := &openAPIV2Reference{
+		BaseSchema: proto.BaseSchema{
+			Description: "Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata",
+			Path:        path.FieldPath("metadata"),
+		},
+		reference: "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+		subSchema: objectMetaModel,
+	}
+	return ref
+}
+
+// mapToKind converts a *proto.Map to a *proto.Kind by keeping the same
+// description, etc but also adding the 3 minimum fields - apiVersion, kind and
+// metadata.
+// This function assumes that the protoMap given is a top-level object (eg: a CRD).
+func mapToKind(protoMap *proto.Map, models proto.Models) *proto.Kind {
+	apiVersion := &proto.Primitive{
+		BaseSchema: proto.BaseSchema{
+			Description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
+			Path:        protoMap.Path.FieldPath("apiVersion"),
+		},
+		Type: "string",
+	}
+	kind := &proto.Primitive{
+		BaseSchema: proto.BaseSchema{
+			Description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
+			Path:        protoMap.Path.FieldPath("kind"),
+		},
+		Type: "string",
+	}
+	metadata := newObjectMetaReference(protoMap.Path, models)
+	return &proto.Kind{
+		BaseSchema: protoMap.BaseSchema,
+		Fields: map[string]proto.Schema{
+			"apiVersion": apiVersion,
+			"kind":       kind,
+			"metadata":   metadata,
+		},
+	}
+}
+
+// openAPIV2ToDefinition builds a schemaDefinition for the given schemaID based on
+// Resource information from OpenAPI v2 endpoint
+func openAPIV2ToDefinition(protoSchema proto.Schema, models proto.Models, modelName string) (schemaDefinition, error) {
+	switch m := protoSchema.(type) {
+	case *proto.Map:
+		// If the schema is a *proto.Map, it will not have any Fields associated with it
+		// even though all Kubernetes resources have at least apiVersion, kind and metadata.
+		//
+		// We transform this Map to a Kind and inject these fields
+		protoSchema = mapToKind(m, models)
+	case *proto.Kind:
+	default:
+		return schemaDefinition{}, fmt.Errorf("model for %s was type %T, not a *proto.Kind nor *proto.Map", modelName, protoSchema)
+	}
+	definitions := map[string]definition{}
+	visitor := schemaFieldVisitor{
+		definitions: definitions,
+		models:      models,
+	}
+	protoSchema.Accept(&visitor)
+
+	return schemaDefinition{
+		DefinitionType: modelName,
+		Definitions:    definitions,
+	}, nil
+}
+
+// baseSchemaToDefinition converts a given schema to the definition map. This should only be used with baseSchemas, whose definitions
+// are expected to be set by another application and may not be k8s resources.
+func baseSchemaToDefinition(schema types.APISchema) map[string]definition {
+	definitions := map[string]definition{}
+	def := definition{
+		Description:    schema.Description,
+		Type:           schema.ID,
+		ResourceFields: map[string]definitionField{},
+	}
+	for fieldName, field := range schema.ResourceFields {
+		fieldType, subType := parseFieldType(field.Type)
+		def.ResourceFields[fieldName] = definitionField{
+			Type:        fieldType,
+			SubType:     subType,
+			Description: field.Description,
+			Required:    field.Required,
+		}
+	}
+	definitions[schema.ID] = def
+	return definitions
+}
+
+// parseFieldType parses a schemas.Field's type to a type (first return) and subType (second return)
+func parseFieldType(fieldType string) (string, string) {
+	subType := wranglerDefinition.SubType(fieldType)
+	if wranglerDefinition.IsMapType(fieldType) {
+		return "map", subType
+	}
+	if wranglerDefinition.IsArrayType(fieldType) {
+		return "array", subType
+	}
+	if wranglerDefinition.IsReferenceType(fieldType) {
+		return "reference", subType
+	}
+	return fieldType, ""
+}
diff --git a/pkg/schema/definitions/converter_test.go b/pkg/schema/definitions/converter_test.go
new file mode 100644
index 00000000..0889206c
--- /dev/null
+++ b/pkg/schema/definitions/converter_test.go
@@ -0,0 +1,8 @@
+package definitions
+
+import (
+	"testing"
+)
+
+func TestMapToKind(t *testing.T) {
+}
diff --git a/pkg/schema/definitions/openapi_test.go b/pkg/schema/definitions/fixtures_test.go
similarity index 65%
rename from pkg/schema/definitions/openapi_test.go
rename to pkg/schema/definitions/fixtures_test.go
index b49d90e5..d701b24a 100644
--- a/pkg/schema/definitions/openapi_test.go
+++ b/pkg/schema/definitions/fixtures_test.go
@@ -1,5 +1,47 @@
 package definitions
 
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/rancher/wrangler/v2/pkg/yaml"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+)
+
+var (
+	rawCRDs = `apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: userattributes.management.cattle.io
+spec:
+  conversion:
+    strategy: None
+  group: management.cattle.io
+  names:
+    kind: UserAttribute
+    listKind: UserAttributeList
+    plural: userattributes
+    singular: userattribute
+  scope: Cluster
+  versions:
+  - name: v2
+    schema:
+      openAPIV3Schema:
+        type: object
+        x-kubernetes-preserve-unknown-fields: true
+    served: true
+    storage: true
+`
+)
+
+func getCRDs() ([]*apiextv1.CustomResourceDefinition, error) {
+	crds, err := yaml.UnmarshalWithJSONDecoder[*apiextv1.CustomResourceDefinition](bytes.NewBuffer([]byte(rawCRDs)))
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal CRD: %w", err)
+	}
+	return crds, err
+}
+
 const openapi_raw = `
 swagger: "2.0"
 info:
@@ -172,6 +214,23 @@ definitions:
     - group: "noversion.cattle.io"
       version: "v1"
       kind: "Resource"
+  io.cattle.management.v1.DeprecatedResource:
+    description: "A resource that is not present in v2"
+    type: "object"
+    properties:
+      apiVersion:
+        description: "The APIVersion of this resource"
+        type: "string"
+      kind:
+        description: "The kind"
+        type: "string"
+      metadata:
+        description: "The metadata"
+        $ref: "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta"
+    x-kubernetes-group-version-kind:
+    - group: "management.cattle.io"
+      version: "v1"
+      kind: "DeprecatedResource"
   io.cattle.missinggroup.v2.Resource:
     description: "A Missing Group V2 resource is for a group not listed by server groups"
     type: "object"
@@ -236,6 +295,12 @@ definitions:
   io.cattle.management.NotAKind:
     type: "string"
     description: "Some string which isn't a kind"
+  io.cattle.management.v2.UserAttribute:
+    type: "object"
+    x-kubernetes-group-version-kind:
+    - group: "management.cattle.io"
+      version: "v2"
+      kind: "UserAttribute"
   io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta:
     description: "Object Metadata"
     properties:
@@ -247,4 +312,35 @@ definitions:
       name:
         description: "name of the resource"
         type: "string"
+  io.k8s.api.core.v1.ConfigMap:
+    type: "object"
+    description: "ConfigMap holds configuration data for pods to consume."
+    properties:
+      apiVersion:
+        description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources"
+        type: "string"
+      kind:
+        description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds"
+        type: "string"
+      metadata:
+        description: "Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata"
+        $ref: "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta"
+      binaryData:
+        description: "BinaryData contains the binary data. Each key must consist of alphanumeric characters, '-', '_' or '.'. BinaryData can contain byte sequences that are not in the UTF-8 range. The keys stored in BinaryData must not overlap with the ones in the Data field, this is enforced during validation process. Using this field will require 1.10+ apiserver and kubelet."
+        type: "object"
+        additionalProperties:
+          type: "string"
+          format: "byte"
+      data:
+        description: "Data contains the configuration data. Each key must consist of alphanumeric characters, '-', '_' or '.'. Values with non-UTF-8 byte sequences must use the BinaryData field. The keys stored in Data must not overlap with the keys in the BinaryData field, this is enforced during validation process."
+        type: "object"
+        additionalProperties:
+          type: "string"
+      immutable:
+        description: "Immutable, if set to true, ensures that data stored in the ConfigMap cannot be updated (only object metadata can be modified). If not set to true, the field can be modified at any time. Defaulted to nil."
+        type: "boolean"
+    x-kubernetes-group-version-kind:
+    - group: ""
+      kind: "ConfigMap"
+      version: "v1"
 `
diff --git a/pkg/schema/definitions/handler.go b/pkg/schema/definitions/handler.go
index c0a869f7..b794bf0e 100644
--- a/pkg/schema/definitions/handler.go
+++ b/pkg/schema/definitions/handler.go
@@ -1,6 +1,7 @@
 package definitions
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"sync"
@@ -8,14 +9,21 @@ import (
 	"github.com/rancher/apiserver/pkg/apierror"
 	"github.com/rancher/apiserver/pkg/types"
 	"github.com/rancher/steve/pkg/schema/converter"
-	wranglerDefinition "github.com/rancher/wrangler/v2/pkg/schemas/definition"
+	wapiextv1 "github.com/rancher/wrangler/v2/pkg/generated/controllers/apiextensions.k8s.io/v1"
 	"github.com/rancher/wrangler/v2/pkg/schemas/validation"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/discovery"
 	"k8s.io/kube-openapi/pkg/util/proto"
 )
 
 var (
+	badRequestErrorCode = validation.ErrorCode{
+		Status: http.StatusBadRequest,
+		Code:   "BadRequest",
+	}
 	internalServerErrorCode = validation.ErrorCode{
 		Status: http.StatusInternalServerError,
 		Code:   "InternalServerError",
@@ -26,20 +34,60 @@ var (
 	}
 )
 
-// SchemaDefinitionHandler is a byID handler for a specific schema, which provides field definitions for all schemas.
-// Does not implement any method allowing a caller to list definitions for all schemas.
+type gvkModel struct {
+	// ModelName is the name of the OpenAPI V2 model.
+	// For example, the GVK Group=management.cattle.io/v2, Kind=UserAttribute will have
+	// the following model name: io.cattle.management.v2.UserAttribute.
+	ModelName string
+	// Schema is the OpenAPI V2 schema for this GVK
+	Schema proto.Schema
+	// CRD is the schema from the CRD for this GVK, if there exists a CRD
+	// for this group/kind
+	CRD *apiextv1.JSONSchemaProps
+}
+
+// SchemaDefinitionHandler provides a schema definition for the schema ID provided. The schema definition is built
+// using the following information, in order:
+//
+//  1. If the schema ID refers to a BaseSchema (a schema that doesn't exist in Kubernetes), then we use the
+//     schema to return the schema definition - we return early. Otherwise:
+//  2. We build a schema definition from the OpenAPI V2 info.
+//  3. If the schemaID refers to a CRD, then we also build a schema definition from the CRD.
+//  4. We merge both the OpenAPI V2 and the CRD schema definition. CRD will ALWAYS override whatever is
+//     in OpenAPI V2. This makes sense because CRD is defined by OpenAPI V3, so has more information. This
+//     merged schema definition is returned.
+//
+// Note: SchemaDefinitionHandler only implements a ByID handler. It does not implement any method allowing a caller
+// to list definitions for all schemas.
 type SchemaDefinitionHandler struct {
-	sync.RWMutex
+	// schemaDefinitions stores a schema definition for each OpenAPI V2
+	// models that has a GVK. The schemaDefinition is the result of merging
+	// OpenAPI V2 information with the CRD information (if any).
+	schemaDefinitions     map[string]schemaDefinition
+	schemaDefinitionsLock sync.RWMutex
 
 	// baseSchema are the schemas (which may not represent a real CRD) added to the server
 	baseSchema *types.APISchemas
-	// client is the discovery client used to get the groups/resources/fields from kubernetes.
+
+	// crdCache is used to add more information to a schema definition by getting information
+	// from the CRD of the resource being accessed (if said resource is a CRD)
+	crdCache wapiextv1.CustomResourceDefinitionCache
+
+	// client is the discovery client used to get the groups/resources/fields from kubernetes
 	client discovery.DiscoveryInterface
-	// models are the cached models from the last response from kubernetes.
-	models *proto.Models
-	// schemaToModel is a map of the schema name to the model for that schema. Can be used to load the
-	// top-level definition for a schema, which can then be processed by the schemaFieldVisitor.
-	schemaToModel map[string]string
+}
+
+func NewSchemaDefinitionHandler(
+	baseSchema *types.APISchemas,
+	crdCache wapiextv1.CustomResourceDefinitionCache,
+	client discovery.DiscoveryInterface,
+) *SchemaDefinitionHandler {
+	handler := &SchemaDefinitionHandler{
+		baseSchema: baseSchema,
+		crdCache:   crdCache,
+		client:     client,
+	}
+	return handler
 }
 
 // Refresh writeLocks and updates the cache with new schemaDefinitions. Will result in a call to kubernetes to retrieve
@@ -47,7 +95,7 @@ type SchemaDefinitionHandler struct {
 func (s *SchemaDefinitionHandler) Refresh() error {
 	openapi, err := s.client.OpenAPISchema()
 	if err != nil {
-		return fmt.Errorf("unable to fetch openapi definition: %w", err)
+		return fmt.Errorf("unable to fetch openapi v2 definition: %w", err)
 	}
 	models, err := proto.NewOpenAPIData(openapi)
 	if err != nil {
@@ -57,15 +105,59 @@ func (s *SchemaDefinitionHandler) Refresh() error {
 	if err != nil {
 		return fmt.Errorf("unable to retrieve groups: %w", err)
 	}
-	s.Lock()
-	defer s.Unlock()
-	nameIndex := s.indexSchemaNames(models, groups)
-	s.schemaToModel = nameIndex
-	s.models = &models
-	return nil
+
+	gvkModels, err := listGVKModels(models, groups, s.crdCache)
+	if err != nil {
+		return err
+	}
+
+	var errs []error
+
+	schemaDefinitions := map[string]schemaDefinition{}
+	for schemaID, gvkModel := range gvkModels {
+		schemaDef, err := s.buildSchemaDefinitionForModel(models, gvkModel)
+		if err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		schemaDefinitions[schemaID] = schemaDef
+	}
+
+	s.schemaDefinitionsLock.Lock()
+	defer s.schemaDefinitionsLock.Unlock()
+	s.schemaDefinitions = schemaDefinitions
+	return errors.Join(errs...)
+}
+
+func (s *SchemaDefinitionHandler) buildSchemaDefinitionForModel(models proto.Models, gvk gvkModel) (schemaDefinition, error) {
+	definitions, err := openAPIV2ToDefinition(gvk.Schema, models, gvk.ModelName)
+	if err != nil {
+		return schemaDefinition{}, fmt.Errorf("OpenAPI V2 to definition error: %w", err)
+	}
+
+	// CRDs don't always exists (eg: Pods, Deployments, etc)
+	if gvk.CRD != nil {
+		// CRD definitions generally has more information than the OpenAPI V2
+		// because it embeds an OpenAPI V3 document. However, these 3 fields
+		// are the exception where the Open API V2 endpoint has more
+		// information.
+		props := gvk.CRD.DeepCopy()
+		delete(props.Properties, "apiVersion")
+		delete(props.Properties, "kind")
+		delete(props.Properties, "metadata")
+
+		// We want to merge the OpenAPI V2 information with the CRD information
+		// whenever possible because the CRD is defined by OpenAPI V3 which
+		// _generally_ ends up with more information than OpenAPI V2
+		// (eg: Optional fields wrongly ends up as type string in V2)
+		crdDefinitions := crdToDefinition(props, gvk.ModelName)
+		definitions.Merge(crdDefinitions)
+	}
+
+	return definitions, nil
 }
 
-// byIDHandler is the Handler method for a request to get the schema definition for a specifc schema. Will use the
+// byIDHandler is the Handler method for a request to get the schema definition for a specific schema. Will use the
 // cached models found during the last refresh as part of this process.
 func (s *SchemaDefinitionHandler) byIDHandler(request *types.APIRequest) (types.APIObject, error) {
 	// pseudo-access check, designed to make sure that users have access to the schema for the definition that they
@@ -75,11 +167,6 @@ func (s *SchemaDefinitionHandler) byIDHandler(request *types.APIRequest) (types.
 		return types.APIObject{}, apierror.NewAPIError(validation.NotFound, "no such schema")
 	}
 
-	// lock only in read-mode so that we don't read while refresh writes. Only use a read-lock - using a write lock
-	// would make this endpoint only usable by one caller at a time
-	s.RLock()
-	defer s.RUnlock()
-
 	if baseSchema := s.baseSchema.LookupSchema(requestSchema.ID); baseSchema != nil {
 		// if this schema is a base schema it won't be in the model cache. In this case, and only this case, we process
 		// the fields independently
@@ -94,106 +181,87 @@ func (s *SchemaDefinitionHandler) byIDHandler(request *types.APIRequest) (types.
 		}, nil
 	}
 
-	if s.models == nil {
+	s.schemaDefinitionsLock.RLock()
+	schemaDefinitions := s.schemaDefinitions
+	s.schemaDefinitionsLock.RUnlock()
+
+	if schemaDefinitions == nil {
 		return types.APIObject{}, apierror.NewAPIError(notRefreshedErrorCode, "schema definitions not yet refreshed")
 	}
-	models := *s.models
-	modelName, ok := s.schemaToModel[requestSchema.ID]
+
+	schemaDef, ok := schemaDefinitions[requestSchema.ID]
 	if !ok {
 		return types.APIObject{}, apierror.NewAPIError(notRefreshedErrorCode, "no model found for schema, try again after refresh")
 	}
-	model := models.LookupModel(modelName)
-	protoKind, ok := model.(*proto.Kind)
-	if !ok {
-		errorMsg := fmt.Sprintf("model for %s was type %T, not a proto.Kind", modelName, model)
-		return types.APIObject{}, apierror.NewAPIError(internalServerErrorCode, errorMsg)
-	}
-	definitions := map[string]definition{}
-	visitor := schemaFieldVisitor{
-		definitions: definitions,
-		models:      models,
-	}
-	protoKind.Accept(&visitor)
 
 	return types.APIObject{
-		ID:   request.Name,
-		Type: "schemaDefinition",
-		Object: schemaDefinition{
-			DefinitionType: modelName,
-			Definitions:    definitions,
-		},
+		ID:     request.Name,
+		Type:   "schemaDefinition",
+		Object: schemaDef,
 	}, nil
 }
 
-// indexSchemaNames returns a map of schemaID to the modelName for a given schema. Will use the preferred version of a
-// resource if possible. Can return an error if unable to find groups.
-func (s *SchemaDefinitionHandler) indexSchemaNames(models proto.Models, groups *metav1.APIGroupList) map[string]string {
-	preferredResourceVersions := map[string]string{}
-	if groups != nil {
-		for _, group := range groups.Groups {
-			preferredResourceVersions[group.Name] = group.PreferredVersion.Version
+// listGVKModels returns a map of schemaID to the gvkModel. Will use the preferred version of a
+// resource if possible.
+func listGVKModels(models proto.Models, groups *metav1.APIGroupList, crdCache wapiextv1.CustomResourceDefinitionCache) (map[string]gvkModel, error) {
+	groupToPreferredVersion := make(map[string]string)
+	for _, group := range groups.Groups {
+		groupToPreferredVersion[group.Name] = group.PreferredVersion.Version
+	}
+
+	gvkToCRD := make(map[schema.GroupVersionKind]*apiextv1.JSONSchemaProps)
+	crds, err := crdCache.List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+	for _, crd := range crds {
+		for _, version := range crd.Spec.Versions {
+			gvk := schema.GroupVersionKind{
+				Group:   crd.Spec.Group,
+				Version: version.Name,
+				Kind:    crd.Spec.Names.Kind,
+			}
+			gvkToCRD[gvk] = version.Schema.OpenAPIV3Schema
 		}
 	}
-	schemaToModel := map[string]string{}
+
+	schemaToGVKModel := map[string]gvkModel{}
 	for _, modelName := range models.ListModels() {
-		protoKind, ok := models.LookupModel(modelName).(*proto.Kind)
-		if !ok {
-			// no need to process models that aren't kinds
+		protoSchema := models.LookupModel(modelName)
+		switch protoSchema.(type) {
+		// It is possible that a Kubernetes resources ends up being treated as
+		// a *proto.Map instead of *proto.Kind when it doesn't have any fields
+		// defined. (eg: management.cattle.io.v1.UserAttributes)
+		//
+		// For that reason, we accept both *proto.Kind and *proto.Map
+		// as long as they have a GVK assigned
+		case *proto.Kind, *proto.Map:
+		default:
+			// no need to process models that aren't kind or map
 			continue
 		}
-		gvk := converter.GetGVKForKind(protoKind)
+
+		// Makes sure the schema has a GVK (whether it's a Map or a Kind)
+		gvk := converter.GetGVKForProtoSchema(protoSchema)
 		if gvk == nil {
-			// not all kinds are for top-level resources, since these won't have a schema,
-			// we can safely continue
 			continue
 		}
+
 		schemaID := converter.GVKToSchemaID(*gvk)
-		prefVersion := preferredResourceVersions[gvk.Group]
-		_, ok = schemaToModel[schemaID]
+
+		prefVersion := groupToPreferredVersion[gvk.Group]
+		_, ok := schemaToGVKModel[schemaID]
 		// we always add the preferred version to the map. However, if this isn't the preferred version the preferred group could
 		// be missing this resource (e.x. v1alpha1 has a resource, it's removed in v1). In those cases, we add the model name
 		// only if we don't already have an entry. This way we always choose the preferred, if possible, but still have 1 version
 		// for everything
 		if !ok || prefVersion == gvk.Version {
-			schemaToModel[schemaID] = modelName
-		}
-	}
-	return schemaToModel
-}
-
-// baseSchemaToDefinition converts a given schema to the definition map. This should only be used with baseSchemas, whose definitions
-// are expected to be set by another application and may not be k8s resources.
-func baseSchemaToDefinition(schema types.APISchema) map[string]definition {
-	definitions := map[string]definition{}
-	def := definition{
-		Description:    schema.Description,
-		Type:           schema.ID,
-		ResourceFields: map[string]definitionField{},
-	}
-	for fieldName, field := range schema.ResourceFields {
-		fieldType, subType := parseFieldType(field.Type)
-		def.ResourceFields[fieldName] = definitionField{
-			Type:        fieldType,
-			SubType:     subType,
-			Description: field.Description,
-			Required:    field.Required,
+			schemaToGVKModel[schemaID] = gvkModel{
+				ModelName: modelName,
+				Schema:    protoSchema,
+				CRD:       gvkToCRD[*gvk],
+			}
 		}
 	}
-	definitions[schema.ID] = def
-	return definitions
-}
-
-// parseFieldType parses a schemas.Field's type to a type (first return) and subType (second return)
-func parseFieldType(fieldType string) (string, string) {
-	subType := wranglerDefinition.SubType(fieldType)
-	if wranglerDefinition.IsMapType(fieldType) {
-		return "map", subType
-	}
-	if wranglerDefinition.IsArrayType(fieldType) {
-		return "array", subType
-	}
-	if wranglerDefinition.IsReferenceType(fieldType) {
-		return "reference", subType
-	}
-	return fieldType, ""
+	return schemaToGVKModel, nil
 }
diff --git a/pkg/schema/definitions/handler_test.go b/pkg/schema/definitions/handler_test.go
index 9adb3693..089ca81e 100644
--- a/pkg/schema/definitions/handler_test.go
+++ b/pkg/schema/definitions/handler_test.go
@@ -4,113 +4,112 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/golang/mock/gomock"
 	openapi_v2 "github.com/google/gnostic-models/openapiv2"
 	"github.com/rancher/apiserver/pkg/apierror"
 	"github.com/rancher/apiserver/pkg/types"
+	"github.com/rancher/wrangler/v2/pkg/generic/fake"
 	wschemas "github.com/rancher/wrangler/v2/pkg/schemas"
 	"github.com/stretchr/testify/require"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/openapi"
 	restclient "k8s.io/client-go/rest"
-	"k8s.io/kube-openapi/pkg/util/proto"
 )
 
 func TestRefresh(t *testing.T) {
-	defaultDocument, err := openapi_v2.ParseDocument([]byte(openapi_raw))
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	discoveryClient, err := buildDefaultDiscovery()
 	require.NoError(t, err)
-	defaultModels, err := proto.NewOpenAPIData(defaultDocument)
+
+	crds, err := getCRDs()
 	require.NoError(t, err)
-	defaultSchemaToModel := map[string]string{
-		"management.cattle.io.globalrole":  "io.cattle.management.v2.GlobalRole",
-		"management.cattle.io.newresource": "io.cattle.management.v2.NewResource",
-		"noversion.cattle.io.resource":     "io.cattle.noversion.v1.Resource",
-		"missinggroup.cattle.io.resource":  "io.cattle.missinggroup.v1.Resource",
-	}
-	tests := []struct {
-		name              string
-		openapiError      error
-		serverGroupsErr   error
-		useBadOpenApiDoc  bool
-		nilGroups         bool
-		wantModels        *proto.Models
-		wantSchemaToModel map[string]string
-		wantError         bool
-	}{
-		{
-			name:              "success",
-			wantModels:        &defaultModels,
-			wantSchemaToModel: defaultSchemaToModel,
-		},
-		{
-			name:         "error - openapi doc unavailable",
-			openapiError: fmt.Errorf("server unavailable"),
-			wantError:    true,
-		},
-		{
-			name:             "error - unable to parse openapi doc",
-			useBadOpenApiDoc: true,
-			wantError:        true,
-		},
-		{
-			name:            "error - unable to retrieve groups and resources",
-			serverGroupsErr: fmt.Errorf("server not available"),
-			wantError:       true,
+
+	crdCache := fake.NewMockNonNamespacedCacheInterface[*apiextv1.CustomResourceDefinition](ctrl)
+	crdCache.EXPECT().Get(gomock.Any()).Return(nil, apierrors.NewNotFound(schema.GroupResource{}, "")).AnyTimes()
+	crdCache.EXPECT().List(gomock.Any()).Return(crds, nil).AnyTimes()
+
+	baseSchemas := types.EmptyAPISchemas()
+	handler := NewSchemaDefinitionHandler(baseSchemas, crdCache, discoveryClient)
+
+	schemas := types.EmptyAPISchemas()
+	schemas.MustAddSchema(types.APISchema{
+		Schema: &wschemas.Schema{
+			ID:                "management.cattle.io.globalrole",
+			CollectionMethods: []string{"get"},
+			ResourceMethods:   []string{"get"},
 		},
-		{
-			name:       "no groups or error from server",
-			nilGroups:  true,
-			wantModels: &defaultModels,
-			wantSchemaToModel: map[string]string{
-				"management.cattle.io.globalrole":  "io.cattle.management.v1.GlobalRole",
-				"management.cattle.io.newresource": "io.cattle.management.v2.NewResource",
-				"noversion.cattle.io.resource":     "io.cattle.noversion.v1.Resource",
-				"missinggroup.cattle.io.resource":  "io.cattle.missinggroup.v1.Resource",
+	})
+
+	request := types.APIRequest{
+		Schemas: schemas,
+		Name:    "management.cattle.io.globalrole",
+	}
+	_, err = handler.byIDHandler(&request)
+	require.True(t, apierror.IsAPIError(err))
+	apiErr, _ := err.(*apierror.APIError)
+	require.Equal(t, 503, apiErr.Code.Status)
+
+	err = handler.Refresh()
+	require.NoError(t, err)
+
+	response, err := handler.byIDHandler(&request)
+	require.NoError(t, err)
+
+	definition := response.Object.(schemaDefinition)
+	require.Equal(t, "io.cattle.management.v2.GlobalRole", definition.DefinitionType)
+
+	discoveryClient.Groups = &metav1.APIGroupList{
+		Groups: []metav1.APIGroup{
+			{
+				Name: "management.cattle.io",
+				PreferredVersion: metav1.GroupVersionForDiscovery{
+					GroupVersion: "management.cattle.io/v1",
+					Version:      "v1",
+				},
+				Versions: []metav1.GroupVersionForDiscovery{
+					{
+						GroupVersion: "management.cattle.io/v1",
+						Version:      "v1",
+					},
+					{
+						GroupVersion: "management.cattle.io/v2",
+						Version:      "v2",
+					},
+				},
 			},
 		},
 	}
-	for _, test := range tests {
-		test := test
-		t.Run(test.name, func(t *testing.T) {
-			t.Parallel()
-			client, err := buildDefaultDiscovery()
-			client.DocumentErr = test.openapiError
-			client.GroupsErr = test.serverGroupsErr
-			if test.useBadOpenApiDoc {
-				schema := client.Document.Definitions.AdditionalProperties[0]
-				schema.Value.Type = &openapi_v2.TypeItem{
-					Value: []string{"multiple", "entries"},
-				}
-			}
-			if test.nilGroups {
-				client.Groups = nil
-			}
-			require.Nil(t, err)
-			handler := SchemaDefinitionHandler{
-				client: client,
-			}
-			err = handler.Refresh()
-			if test.wantError {
-				require.Error(t, err)
-			} else {
-				require.NoError(t, err)
-			}
-			require.Equal(t, test.wantModels, handler.models)
-			require.Equal(t, test.wantSchemaToModel, handler.schemaToModel)
-		})
 
-	}
+	err = handler.Refresh()
+	require.NoError(t, err)
+
+	response, err = handler.byIDHandler(&request)
+	require.NoError(t, err)
+	definition = response.Object.(schemaDefinition)
+	require.Equal(t, "io.cattle.management.v1.GlobalRole", definition.DefinitionType)
+
+	discoveryClient.GroupsErr = fmt.Errorf("fake error")
+	err = handler.Refresh()
+	require.Error(t, err)
+
+	// Still able to query even if we have a refresh error
+	response, err = handler.byIDHandler(&request)
+	require.NoError(t, err)
+	definition = response.Object.(schemaDefinition)
+	require.Equal(t, "io.cattle.management.v1.GlobalRole", definition.DefinitionType)
 }
 
 func Test_byID(t *testing.T) {
-	defaultDocument, err := openapi_v2.ParseDocument([]byte(openapi_raw))
+	discoveryClient, err := buildDefaultDiscovery()
 	require.NoError(t, err)
-	defaultModels, err := proto.NewOpenAPIData(defaultDocument)
-	require.NoError(t, err)
-	defaultSchemaToModel := map[string]string{
-		"management.cattle.io.globalrole": "io.cattle.management.v2.GlobalRole",
-	}
+
 	schemas := types.EmptyAPISchemas()
 	addSchema := func(names ...string) {
 		for _, name := range names {
@@ -158,7 +157,14 @@ func Test_byID(t *testing.T) {
 			},
 		},
 	}
-	addSchema("management.cattle.io.globalrole", "management.cattle.io.missingfrommodel", "management.cattle.io.notakind")
+	addSchema(
+		"configmap",
+		"management.cattle.io.globalrole",
+		"management.cattle.io.missingfrommodel",
+		"management.cattle.io.notakind",
+		"management.cattle.io.userattribute",
+		"management.cattle.io.deprecatedresource",
+	)
 	baseSchemas := types.EmptyAPISchemas()
 	baseSchemas.MustAddSchema(builtinSchema)
 	schemas.MustAddSchema(builtinSchema)
@@ -166,17 +172,125 @@ func Test_byID(t *testing.T) {
 	tests := []struct {
 		name          string
 		schemaName    string
-		models        *proto.Models
-		schemaToModel map[string]string
 		wantObject    *types.APIObject
 		wantError     bool
 		wantErrorCode *int
 	}{
 		{
-			name:          "global role definition",
-			schemaName:    "management.cattle.io.globalrole",
-			models:        &defaultModels,
-			schemaToModel: defaultSchemaToModel,
+			// ConfigMaps is NOT a CRD but it is defined in OpenAPI V2
+			name:       "configmap",
+			schemaName: "configmap",
+			wantObject: &types.APIObject{
+				ID:   "configmap",
+				Type: "schemaDefinition",
+				Object: schemaDefinition{
+					DefinitionType: "io.k8s.api.core.v1.ConfigMap",
+					Definitions: map[string]definition{
+						"io.k8s.api.core.v1.ConfigMap": {
+							Type:        "io.k8s.api.core.v1.ConfigMap",
+							Description: "ConfigMap holds configuration data for pods to consume.",
+							ResourceFields: map[string]definitionField{
+								"apiVersion": {
+									Type:        "string",
+									Description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
+								},
+								"kind": {
+									Type:        "string",
+									Description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
+								},
+								"metadata": {
+									Type:        "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+									Description: "Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata",
+								},
+								"binaryData": {
+									Type:        "map",
+									SubType:     "string",
+									Description: "BinaryData contains the binary data. Each key must consist of alphanumeric characters, '-', '_' or '.'. BinaryData can contain byte sequences that are not in the UTF-8 range. The keys stored in BinaryData must not overlap with the ones in the Data field, this is enforced during validation process. Using this field will require 1.10+ apiserver and kubelet.",
+								},
+								"data": {
+									Type:        "map",
+									SubType:     "string",
+									Description: "Data contains the configuration data. Each key must consist of alphanumeric characters, '-', '_' or '.'. Values with non-UTF-8 byte sequences must use the BinaryData field. The keys stored in Data must not overlap with the keys in the BinaryData field, this is enforced during validation process.",
+								},
+								"immutable": {
+									Type:        "boolean",
+									Description: "Immutable, if set to true, ensures that data stored in the ConfigMap cannot be updated (only object metadata can be modified). If not set to true, the field can be modified at any time. Defaulted to nil.",
+								},
+							},
+						},
+						"io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta": {
+							ResourceFields: map[string]definitionField{
+								"annotations": {
+									Type:        "map",
+									SubType:     "string",
+									Description: "annotations of the resource",
+								},
+								"name": {
+									Type:        "string",
+									SubType:     "",
+									Description: "name of the resource",
+								},
+							},
+							Type:        "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+							Description: "Object Metadata",
+						},
+					},
+				},
+			},
+		},
+		{
+			// UserAttribute is a CRD, but is not a Kind because the CRD misses some
+			// fields for the object.
+			// We still want it to be defined correctly and have default values applied (apiVersion, kind, metadata)
+			name:       "user attribute",
+			schemaName: "management.cattle.io.userattribute",
+			wantObject: &types.APIObject{
+				ID:   "management.cattle.io.userattribute",
+				Type: "schemaDefinition",
+				Object: schemaDefinition{
+					DefinitionType: "io.cattle.management.v2.UserAttribute",
+					Definitions: map[string]definition{
+						"io.cattle.management.v2.UserAttribute": {
+							Type:        "io.cattle.management.v2.UserAttribute",
+							Description: "",
+							ResourceFields: map[string]definitionField{
+								"apiVersion": {
+									Type:        "string",
+									Description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
+								},
+								"kind": {
+									Type:        "string",
+									Description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
+								},
+								"metadata": {
+									Type:        "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+									Description: "Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata",
+								},
+							},
+						},
+						"io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta": {
+							ResourceFields: map[string]definitionField{
+								"annotations": {
+									Type:        "map",
+									SubType:     "string",
+									Description: "annotations of the resource",
+								},
+								"name": {
+									Type:        "string",
+									SubType:     "",
+									Description: "name of the resource",
+								},
+							},
+							Type:        "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+							Description: "Object Metadata",
+						},
+					},
+				},
+			},
+		},
+		{
+			name:       "global role definition",
+			schemaName: "management.cattle.io.globalrole",
 			wantObject: &types.APIObject{
 				ID:   "management.cattle.io.globalrole",
 				Type: "schemaDefinition",
@@ -249,10 +363,57 @@ func Test_byID(t *testing.T) {
 			},
 		},
 		{
-			name:          "baseSchema",
-			schemaName:    "builtin",
-			models:        &defaultModels,
-			schemaToModel: defaultSchemaToModel,
+			// The preferred group for management.cattle.io is V2, but DeprecatedResource doesn't
+			// exist in V2. Steve should be able to fallback to another version (V1).
+			name:       "deprecated resource",
+			schemaName: "management.cattle.io.deprecatedresource",
+			wantObject: &types.APIObject{
+				ID:   "management.cattle.io.deprecatedresource",
+				Type: "schemaDefinition",
+				Object: schemaDefinition{
+					DefinitionType: "io.cattle.management.v1.DeprecatedResource",
+					Definitions: map[string]definition{
+						"io.cattle.management.v1.DeprecatedResource": {
+							Type:        "io.cattle.management.v1.DeprecatedResource",
+							Description: "A resource that is not present in v2",
+							ResourceFields: map[string]definitionField{
+								"apiVersion": {
+									Type:        "string",
+									Description: "The APIVersion of this resource",
+								},
+								"kind": {
+									Type:        "string",
+									Description: "The kind",
+								},
+								"metadata": {
+									Type:        "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+									Description: "The metadata",
+								},
+							},
+						},
+						"io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta": {
+							ResourceFields: map[string]definitionField{
+								"annotations": {
+									Type:        "map",
+									SubType:     "string",
+									Description: "annotations of the resource",
+								},
+								"name": {
+									Type:        "string",
+									SubType:     "",
+									Description: "name of the resource",
+								},
+							},
+							Type:        "io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta",
+							Description: "Object Metadata",
+						},
+					},
+				},
+			},
+		},
+		{
+			name:       "baseSchema",
+			schemaName: "builtin",
 			wantObject: &types.APIObject{
 				ID:   "builtin",
 				Type: "schemaDefinition",
@@ -294,36 +455,16 @@ func Test_byID(t *testing.T) {
 			},
 		},
 		{
-			name:          "missing definition",
-			schemaName:    "management.cattle.io.cluster",
-			models:        &defaultModels,
-			schemaToModel: defaultSchemaToModel,
-			wantError:     true,
-			wantErrorCode: intPtr(404),
-		},
-		{
-			name:          "not refreshed",
-			schemaName:    "management.cattle.io.globalrole",
+			name:          "not a kind",
+			schemaName:    "management.cattle.io.notakind",
 			wantError:     true,
 			wantErrorCode: intPtr(503),
 		},
 		{
-			name:          "has schema, missing from model",
-			schemaName:    "management.cattle.io.missingfrommodel",
-			models:        &defaultModels,
-			schemaToModel: defaultSchemaToModel,
-			wantError:     true,
-			wantErrorCode: intPtr(503),
-		},
-		{
-			name:       "has schema, model is not a kind",
-			schemaName: "management.cattle.io.notakind",
-			models:     &defaultModels,
-			schemaToModel: map[string]string{
-				"management.cattle.io.notakind": "io.management.cattle.NotAKind",
-			},
+			name:          "missing definition",
+			schemaName:    "management.cattle.io.cluster",
 			wantError:     true,
-			wantErrorCode: intPtr(500),
+			wantErrorCode: intPtr(404),
 		},
 	}
 
@@ -331,11 +472,23 @@ func Test_byID(t *testing.T) {
 		test := test
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
-			handler := SchemaDefinitionHandler{
-				baseSchema:    baseSchemas,
-				models:        test.models,
-				schemaToModel: test.schemaToModel,
+
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+
+			crdCache := fake.NewMockNonNamespacedCacheInterface[*apiextv1.CustomResourceDefinition](ctrl)
+			crds, err := getCRDs()
+			require.NoError(t, err)
+
+			for _, crd := range crds {
+				crdCache.EXPECT().Get(crd.GetName()).Return(crd, nil).AnyTimes()
 			}
+			crdCache.EXPECT().Get(gomock.Any()).AnyTimes().Return(nil, apierrors.NewNotFound(schema.GroupResource{}, ""))
+			crdCache.EXPECT().List(gomock.Any()).Return(crds, nil).AnyTimes()
+
+			handler := NewSchemaDefinitionHandler(baseSchemas, crdCache, discoveryClient)
+			err = handler.Refresh()
+			require.NoError(t, err)
 			request := types.APIRequest{
 				Schemas: schemas,
 				Name:    test.schemaName,
@@ -362,6 +515,20 @@ func buildDefaultDiscovery() (*fakeDiscovery, error) {
 		return nil, fmt.Errorf("unable to parse openapi document %w", err)
 	}
 	groups := []metav1.APIGroup{
+		// The core groups (eg: Pods, ConfigMaps, etc)
+		{
+			Name: "",
+			PreferredVersion: metav1.GroupVersionForDiscovery{
+				GroupVersion: "v1",
+				Version:      "v1",
+			},
+			Versions: []metav1.GroupVersionForDiscovery{
+				{
+					GroupVersion: "v1",
+					Version:      "v1",
+				},
+			},
+		},
 		{
 			Name: "management.cattle.io",
 			PreferredVersion: metav1.GroupVersionForDiscovery{
diff --git a/pkg/schema/definitions/schema.go b/pkg/schema/definitions/schema.go
index a4ee622f..0c5679b7 100644
--- a/pkg/schema/definitions/schema.go
+++ b/pkg/schema/definitions/schema.go
@@ -2,6 +2,7 @@ package definitions
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"strconv"
 	"time"
@@ -40,7 +41,33 @@ type definitionField struct {
 	Type        string `json:"type"`
 	SubType     string `json:"subtype,omitempty"`
 	Description string `json:"description,omitempty"`
-	Required    bool   `json:"required,omitempty"`
+	// TODO: What do we mean by required? OpenAPI has both concept:
+	// - required: The field must be present, whatever its value is (eg: null would be a valid value)
+	// - nullable: The field can be null
+	Required bool `json:"required,omitempty"`
+}
+
+// mergeSchemaDefinitions the given schemaDefinitions in order, overriding values
+// when they are set.
+func (s *schemaDefinition) Merge(schema schemaDefinition) error {
+	if s.DefinitionType != schema.DefinitionType {
+		return fmt.Errorf("invalid definition type: %s != %s", s.DefinitionType, schema.DefinitionType)
+	}
+
+	for key, value := range schema.Definitions {
+		mergedDef := s.Definitions[key]
+		if mergedDef.ResourceFields == nil {
+			mergedDef.ResourceFields = make(map[string]definitionField)
+		}
+
+		mergedDef.Type = value.Type
+		mergedDef.Description = value.Description
+		for fieldKey, fieldValue := range value.ResourceFields {
+			mergedDef.ResourceFields[fieldKey] = fieldValue
+		}
+		s.Definitions[key] = mergedDef
+	}
+	return nil
 }
 
 // Register registers the schemaDefinition schema.
@@ -49,10 +76,7 @@ func Register(ctx context.Context,
 	client discovery.DiscoveryInterface,
 	crd apiextcontrollerv1.CustomResourceDefinitionController,
 	apiService v1.APIServiceController) {
-	handler := SchemaDefinitionHandler{
-		baseSchema: baseSchema,
-		client:     client,
-	}
+	handler := NewSchemaDefinitionHandler(baseSchema, crd.Cache(), client)
 	baseSchema.MustAddSchema(types.APISchema{
 		Schema: &schemas.Schema{
 			ID:              "schemaDefinition",
@@ -63,7 +87,7 @@ func Register(ctx context.Context,
 	})
 
 	debounce := debounce.DebounceableRefresher{
-		Refreshable: &handler,
+		Refreshable: handler,
 	}
 	crdDebounce := getDurationEnvVarOrDefault(delayEnvVar, defaultDelay, delayUnit)
 	refHandler := refreshHandler{
diff --git a/pkg/schema/definitions/schema_test.go b/pkg/schema/definitions/schema_test.go
index b47d6108..e5e554ab 100644
--- a/pkg/schema/definitions/schema_test.go
+++ b/pkg/schema/definitions/schema_test.go
@@ -22,6 +22,7 @@ func TestRegister(t *testing.T) {
 	apisvcController := fake.NewMockNonNamespacedControllerInterface[*apiregv1.APIService, *apiregv1.APIServiceList](ctrl)
 	ctx, cancel := context.WithCancel(context.Background())
 	crdController.EXPECT().OnChange(ctx, handlerKey, gomock.Any())
+	crdController.EXPECT().Cache().AnyTimes()
 	apisvcController.EXPECT().OnChange(ctx, handlerKey, gomock.Any())
 	Register(ctx, schemas, &client, crdController, apisvcController)
 	registeredSchema := schemas.LookupSchema("schemaDefinition")
@@ -74,3 +75,249 @@ func Test_getDurationEnvVarOrDefault(t *testing.T) {
 		})
 	}
 }
+
+func TestSchemaDefinitionMerge(t *testing.T) {
+	tests := []struct {
+		name     string
+		schemas  [2]schemaDefinition
+		wantErr  bool
+		expected schemaDefinition
+	}{
+		{
+			name: "merge top-level definitions",
+			schemas: [2]schemaDefinition{
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "foo",
+							Description: "Foo",
+						},
+					},
+				},
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"bar": {
+							Type:           "bar",
+							Description:    "Bar",
+							ResourceFields: map[string]definitionField{},
+						},
+					},
+				},
+			},
+			expected: schemaDefinition{
+				DefinitionType: "foo",
+				Definitions: map[string]definition{
+					"foo": {
+						Type:        "foo",
+						Description: "Foo",
+					},
+					"bar": {
+						Type:           "bar",
+						Description:    "Bar",
+						ResourceFields: map[string]definitionField{},
+					},
+				},
+			},
+		},
+		{
+			name: "merge resource fields",
+			schemas: [2]schemaDefinition{
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "foo",
+							Description: "Foo",
+							ResourceFields: map[string]definitionField{
+								"old": {
+									Type:        "string",
+									Description: "foo.old",
+								},
+								"inBoth": {
+									Type:        "string",
+									Description: "foo.inBoth",
+								},
+							},
+						},
+					},
+				},
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "foo",
+							Description: "Foo",
+							ResourceFields: map[string]definitionField{
+								"new": {
+									Type:        "string",
+									Description: "foo.new",
+								},
+								"inBoth": {
+									Type:        "array",
+									SubType:     "number",
+									Description: "foo.inBoth",
+									Required:    true,
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: schemaDefinition{
+				DefinitionType: "foo",
+				Definitions: map[string]definition{
+					"foo": {
+						Type:        "foo",
+						Description: "Foo",
+						ResourceFields: map[string]definitionField{
+							"new": {
+								Type:        "string",
+								Description: "foo.new",
+							},
+							"old": {
+								Type:        "string",
+								Description: "foo.old",
+							},
+							"inBoth": {
+								Type:        "array",
+								SubType:     "number",
+								Description: "foo.inBoth",
+								Required:    true,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "empty resource fields in old",
+			schemas: [2]schemaDefinition{
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "foo",
+							Description: "Foo",
+						},
+					},
+				},
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "foo",
+							Description: "Foo",
+							ResourceFields: map[string]definitionField{
+								"new": {
+									Type:        "string",
+									Description: "foo.new",
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: schemaDefinition{
+				DefinitionType: "foo",
+				Definitions: map[string]definition{
+					"foo": {
+						Type:        "foo",
+						Description: "Foo",
+						ResourceFields: map[string]definitionField{
+							"new": {
+								Type:        "string",
+								Description: "foo.new",
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "empty resource fields in new",
+			schemas: [2]schemaDefinition{
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "string",
+							Description: "Foo",
+							ResourceFields: map[string]definitionField{
+								"old": {
+									Type:        "string",
+									Description: "foo.old",
+								},
+							},
+						},
+					},
+				},
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "string",
+							Description: "Foo",
+						},
+					},
+				},
+			},
+			expected: schemaDefinition{
+				DefinitionType: "foo",
+				Definitions: map[string]definition{
+					"foo": {
+						Type:        "string",
+						Description: "Foo",
+						ResourceFields: map[string]definitionField{
+							"old": {
+								Type:        "string",
+								Description: "foo.old",
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "empty definition type",
+			schemas: [2]schemaDefinition{
+				{
+					DefinitionType: "foo",
+					Definitions: map[string]definition{
+						"foo": {
+							Type:        "foo",
+							Description: "Foo",
+							ResourceFields: map[string]definitionField{
+								"old": {
+									Type:        "string",
+									Description: "foo.old",
+								},
+							},
+						},
+					},
+				},
+				{
+					DefinitionType: "",
+					Definitions:    map[string]definition{},
+				},
+			},
+			wantErr: true,
+		},
+	}
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			err := test.schemas[0].Merge(test.schemas[1])
+			if test.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.expected, test.schemas[0])
+			}
+
+		})
+	}
+}
diff --git a/pkg/schema/definitions/visitor.go b/pkg/schema/definitions/visitor.go
index c112dbd0..eddff989 100644
--- a/pkg/schema/definitions/visitor.go
+++ b/pkg/schema/definitions/visitor.go
@@ -46,11 +46,7 @@ func (s *schemaFieldVisitor) VisitPrimitive(primitive *proto.Primitive) {
 	field := definitionField{
 		Description: primitive.GetDescription(),
 	}
-	if primitive.Type == "number" || primitive.Type == "integer" {
-		field.Type = "int"
-	} else {
-		field.Type = primitive.Type
-	}
+	field.Type = getPrimitiveType(primitive.Type)
 	s.field = field
 }
 
diff --git a/pkg/schema/definitions/visitor_test.go b/pkg/schema/definitions/visitor_test.go
index b2f14503..c9497d4f 100644
--- a/pkg/schema/definitions/visitor_test.go
+++ b/pkg/schema/definitions/visitor_test.go
@@ -1,7 +1,6 @@
 package definitions
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -55,13 +54,13 @@ var (
 			"missing",
 		},
 	}
-	protoRefNoSubSchema = testRef{
+	protoRefNoSubSchema = openAPIV2Reference{
 		BaseSchema: proto.BaseSchema{
 			Description: "testRef - no subSchema",
 		},
 		reference: "some-other-type",
 	}
-	protoRef = testRef{
+	protoRef = openAPIV2Reference{
 		BaseSchema: proto.BaseSchema{
 			Description: "testRef",
 		},
@@ -75,29 +74,6 @@ var (
 	}
 )
 
-// testRef implements proto.Reference to test VisitReference
-type testRef struct {
-	proto.BaseSchema
-	reference string
-	subSchema proto.Schema
-}
-
-func (t *testRef) Reference() string {
-	return t.reference
-}
-
-func (t *testRef) SubSchema() proto.Schema {
-	return t.subSchema
-}
-
-func (t *testRef) Accept(v proto.SchemaVisitor) {
-	v.VisitReference(t)
-}
-
-func (t *testRef) GetName() string {
-	return fmt.Sprintf("Reference to %q", t.reference)
-}
-
 func TestSchemaFieldVisitor(t *testing.T) {
 	protoKind.Fields["protoRef"] = &protoRef
 	tests := []struct {