diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/engines/components/tables/utils.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/engines/components/tables/utils.ts
index d9f41f8558b78..13cad349d75b0 100644
--- a/x-pack/plugins/enterprise_search/public/applications/app_search/components/engines/components/tables/utils.ts
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/engines/components/tables/utils.ts
@@ -10,7 +10,7 @@ import { EngineDetails } from '../../../engine/types';
 
 export const getConflictingEnginesFromConflictingField = (
   conflictingField: SchemaConflictFieldTypes
-): string[] => Object.values(conflictingField).flat();
+) => Object.values(conflictingField).flat() as string[];
 
 export const getConflictingEnginesFromSchemaConflicts = (
   schemaConflicts: SchemaConflicts
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/constants.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/constants.ts
index b9c4e2c0454d5..0ee428e87873e 100644
--- a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/constants.ts
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/constants.ts
@@ -10,3 +10,18 @@ import { i18n } from '@kbn/i18n';
 export const SCHEMA_TITLE = i18n.translate('xpack.enterpriseSearch.appSearch.engine.schema.title', {
   defaultMessage: 'Schema',
 });
+
+export const ADD_SCHEMA_ERROR = (fieldName: string) =>
+  i18n.translate('xpack.enterpriseSearch.appSearch.engine.schema.addSchemaErrorMessage', {
+    defaultMessage: 'Field name already exists: {fieldName}',
+    values: { fieldName },
+  });
+export const ADD_SCHEMA_SUCCESS = (fieldName: string) =>
+  i18n.translate('xpack.enterpriseSearch.appSearch.engine.schema.addSchemaSuccessMessage', {
+    defaultMessage: 'New field added: {fieldName}',
+    values: { fieldName },
+  });
+export const UPDATE_SCHEMA_SUCCESS = i18n.translate(
+  'xpack.enterpriseSearch.appSearch.engine.schema.updateSchemaSuccessMessage',
+  { defaultMessage: 'Schema updated' }
+);
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_base_logic.test.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_base_logic.test.ts
new file mode 100644
index 0000000000000..e5dbf97b971d9
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_base_logic.test.ts
@@ -0,0 +1,98 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { LogicMounter, mockFlashMessageHelpers, mockHttpValues } from '../../../__mocks__';
+import '../../__mocks__/engine_logic.mock';
+
+import { nextTick } from '@kbn/test/jest';
+
+import { SchemaType } from '../../../shared/schema/types';
+
+import { SchemaBaseLogic } from './schema_base_logic';
+
+describe('SchemaBaseLogic', () => {
+  const { mount } = new LogicMounter(SchemaBaseLogic);
+  const { http } = mockHttpValues;
+  const { flashAPIErrors } = mockFlashMessageHelpers;
+
+  const MOCK_SCHEMA = {
+    some_text_field: SchemaType.Text,
+    some_number_field: SchemaType.Number,
+  };
+  const MOCK_RESPONSE = {
+    schema: MOCK_SCHEMA,
+  } as any;
+
+  const DEFAULT_VALUES = {
+    dataLoading: true,
+    schema: {},
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('has expected default values', () => {
+    mount();
+    expect(SchemaBaseLogic.values).toEqual(DEFAULT_VALUES);
+  });
+
+  describe('actions', () => {
+    describe('onSchemaLoad', () => {
+      it('stores schema state and sets dataLoading to false', () => {
+        mount({ schema: {}, dataLoading: true });
+
+        SchemaBaseLogic.actions.onSchemaLoad(MOCK_RESPONSE);
+
+        expect(SchemaBaseLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          dataLoading: false,
+          schema: MOCK_SCHEMA,
+        });
+      });
+    });
+
+    describe('setSchema', () => {
+      it('updates schema state', () => {
+        mount({ schema: {} });
+
+        SchemaBaseLogic.actions.setSchema(MOCK_SCHEMA);
+
+        expect(SchemaBaseLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          schema: MOCK_SCHEMA,
+        });
+      });
+    });
+  });
+
+  describe('listeners', () => {
+    describe('loadSchema', () => {
+      it('should make an API call and then set schema state', async () => {
+        http.get.mockReturnValueOnce(Promise.resolve(MOCK_RESPONSE));
+        mount();
+        jest.spyOn(SchemaBaseLogic.actions, 'onSchemaLoad');
+
+        SchemaBaseLogic.actions.loadSchema();
+        await nextTick();
+
+        expect(http.get).toHaveBeenCalledWith('/api/app_search/engines/some-engine/schema');
+        expect(SchemaBaseLogic.actions.onSchemaLoad).toHaveBeenCalledWith(MOCK_RESPONSE);
+      });
+
+      it('handles errors', async () => {
+        http.get.mockReturnValueOnce(Promise.reject('error'));
+        mount();
+
+        SchemaBaseLogic.actions.loadSchema();
+        await nextTick();
+
+        expect(flashAPIErrors).toHaveBeenCalledWith('error');
+      });
+    });
+  });
+});
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_base_logic.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_base_logic.ts
new file mode 100644
index 0000000000000..c2196c01d402b
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_base_logic.ts
@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { kea, MakeLogicType } from 'kea';
+
+import { flashAPIErrors } from '../../../shared/flash_messages';
+import { HttpLogic } from '../../../shared/http';
+import { Schema } from '../../../shared/schema/types';
+import { EngineLogic } from '../engine';
+
+import { SchemaApiResponse, MetaEngineSchemaApiResponse } from './types';
+
+export interface SchemaBaseValues {
+  dataLoading: boolean;
+  schema: Schema;
+}
+
+export interface SchemaBaseActions {
+  loadSchema(): void;
+  onSchemaLoad(
+    response: SchemaApiResponse | MetaEngineSchemaApiResponse
+  ): SchemaApiResponse | MetaEngineSchemaApiResponse;
+  setSchema(schema: Schema): { schema: Schema };
+}
+
+export const SchemaBaseLogic = kea<MakeLogicType<SchemaBaseValues, SchemaBaseActions>>({
+  path: ['enterprise_search', 'app_search', 'schema_base_logic'],
+  actions: {
+    loadSchema: true,
+    onSchemaLoad: (response) => response,
+    setSchema: (schema) => ({ schema }),
+  },
+  reducers: {
+    dataLoading: [
+      true,
+      {
+        loadSchema: () => true,
+        onSchemaLoad: () => false,
+      },
+    ],
+    schema: [
+      {},
+      {
+        onSchemaLoad: (_, { schema }) => schema,
+        setSchema: (_, { schema }) => schema,
+      },
+    ],
+  },
+  listeners: ({ actions }) => ({
+    loadSchema: async () => {
+      const { http } = HttpLogic.values;
+      const { engineName } = EngineLogic.values;
+
+      try {
+        const response = await http.get(`/api/app_search/engines/${engineName}/schema`);
+        actions.onSchemaLoad(response);
+      } catch (e) {
+        flashAPIErrors(e);
+      }
+    },
+  }),
+});
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_logic.test.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_logic.test.ts
new file mode 100644
index 0000000000000..123f62af4eeba
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_logic.test.ts
@@ -0,0 +1,291 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { LogicMounter, mockFlashMessageHelpers, mockHttpValues } from '../../../__mocks__';
+import { mockEngineActions } from '../../__mocks__/engine_logic.mock';
+
+import { nextTick } from '@kbn/test/jest';
+
+import { SchemaType, Schema } from '../../../shared/schema/types';
+
+import { SchemaLogic } from './schema_logic';
+
+describe('SchemaLogic', () => {
+  const { mount } = new LogicMounter(SchemaLogic);
+  const { http } = mockHttpValues;
+  const { flashAPIErrors, flashSuccessToast, setErrorMessage } = mockFlashMessageHelpers;
+
+  const MOCK_RESPONSE = {
+    schema: {
+      some_text_field: SchemaType.Text,
+      some_number_field: SchemaType.Number,
+    },
+    mostRecentIndexJob: {
+      percentageComplete: 100,
+      numDocumentsWithErrors: 10,
+      activeReindexJobId: 'some-id',
+      isActive: false,
+      hasErrors: true,
+    },
+    unconfirmedFields: ['some_field'],
+    unsearchedUnconfirmedFields: true,
+  };
+
+  const DEFAULT_VALUES = {
+    dataLoading: true,
+    schema: {},
+    isUpdating: false,
+    hasSchema: false,
+    hasSchemaChanged: false,
+    cachedSchema: {},
+    mostRecentIndexJob: {},
+    unconfirmedFields: [],
+    hasUnconfirmedFields: false,
+    hasNewUnsearchedFields: false,
+    isModalOpen: false,
+  };
+
+  /*
+   * Unfortunately, we can't mount({ schema: ... }) & have to use an action to set schema
+   * because of the separate connected logic file - our LogicMounter test helper sets context
+   * for only the currently tested file
+   */
+  const mountAndSetSchema = ({ schema, ...values }: { schema: Schema; [key: string]: unknown }) => {
+    mount(values);
+    SchemaLogic.actions.setSchema(schema);
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('has expected default values', () => {
+    mount();
+    expect(SchemaLogic.values).toEqual(DEFAULT_VALUES);
+  });
+
+  describe('actions', () => {
+    describe('onSchemaLoad', () => {
+      it('stores the API response in state and sets isUpdating & isModalOpen to false', () => {
+        mount({ isModalOpen: true });
+
+        SchemaLogic.actions.onSchemaLoad(MOCK_RESPONSE);
+
+        expect(SchemaLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          // SchemaBaseLogic
+          dataLoading: false,
+          schema: MOCK_RESPONSE.schema,
+          // SchemaLogic
+          isUpdating: false,
+          isModalOpen: false,
+          cachedSchema: MOCK_RESPONSE.schema,
+          hasSchema: true,
+          mostRecentIndexJob: MOCK_RESPONSE.mostRecentIndexJob,
+          unconfirmedFields: MOCK_RESPONSE.unconfirmedFields,
+          hasUnconfirmedFields: true,
+          hasNewUnsearchedFields: MOCK_RESPONSE.unsearchedUnconfirmedFields,
+        });
+      });
+    });
+
+    describe('onSchemaUpdateError', () => {
+      it('sets isUpdating & isModalOpen to false', () => {
+        mount({ isUpdating: true, isModalOpen: true });
+
+        SchemaLogic.actions.onSchemaUpdateError();
+
+        expect(SchemaLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          isUpdating: false,
+          isModalOpen: false,
+        });
+      });
+    });
+
+    describe('openModal', () => {
+      it('sets isModalOpen to true', () => {
+        mount({ isModalOpen: false });
+
+        SchemaLogic.actions.openModal();
+
+        expect(SchemaLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          isModalOpen: true,
+        });
+      });
+    });
+
+    describe('closeModal', () => {
+      it('sets isModalOpen to false', () => {
+        mount({ isModalOpen: true });
+
+        SchemaLogic.actions.closeModal();
+
+        expect(SchemaLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          isModalOpen: false,
+        });
+      });
+    });
+  });
+
+  describe('selectors', () => {
+    describe('hasSchema', () => {
+      it('returns true when the schema obj has items', () => {
+        mountAndSetSchema({ schema: { test: SchemaType.Text } });
+        expect(SchemaLogic.values.hasSchema).toEqual(true);
+      });
+
+      it('returns false when the schema obj is empty', () => {
+        mountAndSetSchema({ schema: {} });
+        expect(SchemaLogic.values.hasSchema).toEqual(false);
+      });
+    });
+
+    describe('hasSchemaChanged', () => {
+      it('returns true when the schema state is different from the cachedSchema state', () => {
+        mountAndSetSchema({
+          schema: { test: SchemaType.Text },
+          cachedSchema: { test: SchemaType.Number },
+        });
+
+        expect(SchemaLogic.values.hasSchemaChanged).toEqual(true);
+      });
+
+      it('returns false when the stored schema is the same as cachedSchema', () => {
+        mountAndSetSchema({
+          schema: { test: SchemaType.Text },
+          cachedSchema: { test: SchemaType.Text },
+        });
+
+        expect(SchemaLogic.values.hasSchemaChanged).toEqual(false);
+      });
+    });
+
+    describe('hasUnconfirmedFields', () => {
+      it('returns true when the unconfirmedFields array has items', () => {
+        mount({ unconfirmedFields: ['hello_world'] });
+        expect(SchemaLogic.values.hasUnconfirmedFields).toEqual(true);
+      });
+
+      it('returns false when the unconfirmedFields array is empty', () => {
+        mount({ unconfirmedFields: [] });
+        expect(SchemaLogic.values.hasUnconfirmedFields).toEqual(false);
+      });
+    });
+  });
+
+  describe('listeners', () => {
+    describe('addSchemaField', () => {
+      describe('if the schema field already exists', () => {
+        it('flashes an error and closes the modal', () => {
+          mountAndSetSchema({ schema: { existing_field: SchemaType.Text } });
+          jest.spyOn(SchemaLogic.actions, 'closeModal');
+
+          SchemaLogic.actions.addSchemaField('existing_field', SchemaType.Text);
+
+          expect(setErrorMessage).toHaveBeenCalledWith('Field name already exists: existing_field');
+          expect(SchemaLogic.actions.closeModal).toHaveBeenCalled();
+        });
+      });
+
+      describe('if the schema field does not already exist', () => {
+        it('updates the schema state and calls updateSchema with a custom success message', () => {
+          mount();
+          jest.spyOn(SchemaLogic.actions, 'setSchema');
+          jest.spyOn(SchemaLogic.actions, 'updateSchema');
+
+          SchemaLogic.actions.addSchemaField('new_field', SchemaType.Date);
+
+          expect(SchemaLogic.actions.setSchema).toHaveBeenCalledWith({
+            new_field: SchemaType.Date,
+          });
+          expect(SchemaLogic.actions.updateSchema).toHaveBeenCalledWith(
+            'New field added: new_field'
+          );
+        });
+      });
+    });
+
+    describe('updateSchemaFieldType', () => {
+      it("updates an existing schema key's field type value", async () => {
+        mountAndSetSchema({ schema: { existing_field: SchemaType.Text } });
+        jest.spyOn(SchemaLogic.actions, 'setSchema');
+
+        SchemaLogic.actions.updateSchemaFieldType('existing_field', SchemaType.Geolocation);
+
+        expect(SchemaLogic.actions.setSchema).toHaveBeenCalledWith({
+          existing_field: SchemaType.Geolocation,
+        });
+      });
+    });
+
+    describe('updateSchema', () => {
+      it('sets isUpdating to true', () => {
+        mount({ isUpdating: false });
+
+        SchemaLogic.actions.updateSchema();
+
+        expect(SchemaLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          isUpdating: true,
+        });
+      });
+
+      it('should make an API call and then set schema state', async () => {
+        http.post.mockReturnValueOnce(Promise.resolve(MOCK_RESPONSE));
+        mount();
+        jest.spyOn(SchemaLogic.actions, 'onSchemaLoad');
+
+        SchemaLogic.actions.updateSchema();
+        await nextTick();
+
+        expect(http.post).toHaveBeenCalledWith('/api/app_search/engines/some-engine/schema', {
+          body: '{}',
+        });
+        expect(SchemaLogic.actions.onSchemaLoad).toHaveBeenCalledWith(MOCK_RESPONSE);
+      });
+
+      it('should call flashSuccessToast with a custom success message if passed', async () => {
+        http.post.mockReturnValueOnce(Promise.resolve(MOCK_RESPONSE));
+        mount();
+
+        SchemaLogic.actions.updateSchema('wow it worked!!');
+        await nextTick();
+
+        expect(flashSuccessToast).toHaveBeenCalledWith('wow it worked!!');
+      });
+
+      it('should always call EngineLogic.actions.initializeEngine to refresh engine-wide state', async () => {
+        mount();
+
+        SchemaLogic.actions.updateSchema();
+        await nextTick();
+
+        expect(mockEngineActions.initializeEngine).toHaveBeenCalled();
+      });
+
+      it('handles errors and resets bad schema state back to cached/server values', async () => {
+        const MOCK_ERROR = 'Fields cannot contain more than 64 characters';
+        const MOCK_CACHED_SCHEMA = { ok_field: SchemaType.Text };
+
+        http.post.mockReturnValueOnce(Promise.reject(MOCK_ERROR));
+        mount({ cachedSchema: MOCK_CACHED_SCHEMA });
+        jest.spyOn(SchemaLogic.actions, 'onSchemaUpdateError');
+        jest.spyOn(SchemaLogic.actions, 'setSchema');
+
+        SchemaLogic.actions.updateSchema();
+        await nextTick();
+
+        expect(flashAPIErrors).toHaveBeenCalledWith(MOCK_ERROR);
+        expect(SchemaLogic.actions.onSchemaUpdateError).toHaveBeenCalled();
+        expect(SchemaLogic.actions.setSchema).toHaveBeenCalledWith(MOCK_CACHED_SCHEMA);
+      });
+    });
+  });
+});
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_logic.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_logic.ts
new file mode 100644
index 0000000000000..3215a46c8e299
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_logic.ts
@@ -0,0 +1,164 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { kea, MakeLogicType } from 'kea';
+import { isEqual } from 'lodash';
+
+import {
+  flashAPIErrors,
+  setErrorMessage,
+  flashSuccessToast,
+  clearFlashMessages,
+} from '../../../shared/flash_messages';
+import { HttpLogic } from '../../../shared/http';
+import { Schema, SchemaType, IndexJob } from '../../../shared/schema/types';
+import { EngineLogic } from '../engine';
+
+import { ADD_SCHEMA_ERROR, ADD_SCHEMA_SUCCESS, UPDATE_SCHEMA_SUCCESS } from './constants';
+import { SchemaBaseLogic, SchemaBaseValues, SchemaBaseActions } from './schema_base_logic';
+import { SchemaApiResponse } from './types';
+
+interface SchemaValues extends SchemaBaseValues {
+  isUpdating: boolean;
+  hasSchema: boolean;
+  hasSchemaChanged: boolean;
+  cachedSchema: Schema;
+  mostRecentIndexJob: Partial<IndexJob>;
+  unconfirmedFields: string[];
+  hasUnconfirmedFields: boolean;
+  hasNewUnsearchedFields: boolean;
+  isModalOpen: boolean;
+}
+
+interface SchemaActions extends SchemaBaseActions {
+  onSchemaLoad(response: SchemaApiResponse): SchemaApiResponse;
+  addSchemaField(
+    fieldName: string,
+    fieldType: SchemaType
+  ): { fieldName: string; fieldType: SchemaType };
+  updateSchemaFieldType(
+    fieldName: string,
+    fieldType: SchemaType
+  ): { fieldName: string; fieldType: SchemaType };
+  updateSchema(successMessage?: string): string | undefined;
+  onSchemaUpdateError(): void;
+  openModal(): void;
+  closeModal(): void;
+}
+
+export const SchemaLogic = kea<MakeLogicType<SchemaValues, SchemaActions>>({
+  path: ['enterprise_search', 'app_search', 'schema_logic'],
+  connect: {
+    values: [SchemaBaseLogic, ['dataLoading', 'schema']],
+    actions: [SchemaBaseLogic, ['loadSchema', 'onSchemaLoad', 'setSchema']],
+  },
+  actions: {
+    addSchemaField: (fieldName, fieldType) => ({ fieldName, fieldType }),
+    updateSchemaFieldType: (fieldName, fieldType) => ({ fieldName, fieldType }),
+    updateSchema: (successMessage) => successMessage,
+    onSchemaUpdateError: true,
+    openModal: true,
+    closeModal: true,
+  },
+  reducers: {
+    isUpdating: [
+      false,
+      {
+        updateSchema: () => true,
+        onSchemaLoad: () => false,
+        onSchemaUpdateError: () => false,
+      },
+    ],
+    cachedSchema: [
+      {},
+      {
+        onSchemaLoad: (_, { schema }) => schema,
+      },
+    ],
+    mostRecentIndexJob: [
+      {},
+      {
+        onSchemaLoad: (_, { mostRecentIndexJob }) => mostRecentIndexJob,
+      },
+    ],
+    unconfirmedFields: [
+      [],
+      {
+        onSchemaLoad: (_, { unconfirmedFields }) => unconfirmedFields,
+      },
+    ],
+    hasNewUnsearchedFields: [
+      false,
+      {
+        onSchemaLoad: (_, { unsearchedUnconfirmedFields }) => unsearchedUnconfirmedFields,
+      },
+    ],
+    isModalOpen: [
+      false,
+      {
+        openModal: () => true,
+        closeModal: () => false,
+        onSchemaLoad: () => false,
+        onSchemaUpdateError: () => false,
+      },
+    ],
+  },
+  selectors: {
+    hasSchema: [(selectors) => [selectors.schema], (schema) => Object.keys(schema).length > 0],
+    hasSchemaChanged: [
+      (selectors) => [selectors.schema, selectors.cachedSchema],
+      (schema, cachedSchema) => !isEqual(schema, cachedSchema),
+    ],
+    hasUnconfirmedFields: [
+      (selectors) => [selectors.unconfirmedFields],
+      (unconfirmedFields) => unconfirmedFields.length > 0,
+    ],
+  },
+  listeners: ({ actions, values }) => ({
+    addSchemaField: ({ fieldName, fieldType }) => {
+      if (values.schema.hasOwnProperty(fieldName)) {
+        setErrorMessage(ADD_SCHEMA_ERROR(fieldName));
+        actions.closeModal();
+      } else {
+        const updatedSchema = { ...values.schema };
+        updatedSchema[fieldName] = fieldType;
+        actions.setSchema(updatedSchema);
+        actions.updateSchema(ADD_SCHEMA_SUCCESS(fieldName));
+      }
+    },
+    updateSchemaFieldType: ({ fieldName, fieldType }) => {
+      const updatedSchema = { ...values.schema };
+      updatedSchema[fieldName] = fieldType;
+      actions.setSchema(updatedSchema);
+    },
+    updateSchema: async (successMessage) => {
+      const { schema } = values;
+      const { http } = HttpLogic.values;
+      const { engineName } = EngineLogic.values;
+
+      clearFlashMessages();
+
+      try {
+        const response = await http.post(`/api/app_search/engines/${engineName}/schema`, {
+          body: JSON.stringify(schema),
+        });
+        actions.onSchemaLoad(response);
+        flashSuccessToast(successMessage || UPDATE_SCHEMA_SUCCESS);
+      } catch (e) {
+        flashAPIErrors(e);
+        actions.onSchemaUpdateError();
+        // Restore updated schema back to server/cached schema, so we don't keep
+        // erroneous or bad fields in-state
+        actions.setSchema(values.cachedSchema);
+      } finally {
+        // Re-fetch engine data so that other views also dynamically update
+        // (e.g. Documents results, nav icons for invalid boosts or unconfirmed flags)
+        EngineLogic.actions.initializeEngine();
+      }
+    },
+  }),
+});
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_meta_engine_logic.test.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_meta_engine_logic.test.ts
new file mode 100644
index 0000000000000..f265fb2d74113
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_meta_engine_logic.test.ts
@@ -0,0 +1,95 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { LogicMounter } from '../../../__mocks__';
+
+import { SchemaType } from '../../../shared/schema/types';
+
+import { MetaEngineSchemaLogic } from './schema_meta_engine_logic';
+
+describe('MetaEngineSchemaLogic', () => {
+  const { mount } = new LogicMounter(MetaEngineSchemaLogic);
+
+  const MOCK_RESPONSE = {
+    schema: {
+      some_text_field: SchemaType.Text,
+      some_number_field: SchemaType.Number,
+    },
+    fields: {
+      some_text_field: {
+        text: ['source-engine-a', 'source-engine-b'],
+      },
+    },
+    conflictingFields: {
+      some_number_field: {
+        number: ['source-engine-a'],
+        text: ['source-engine-b'],
+      },
+    },
+  };
+
+  const DEFAULT_VALUES = {
+    dataLoading: true,
+    schema: {},
+    fields: {},
+    conflictingFields: {},
+    conflictingFieldsCount: 0,
+    hasConflicts: false,
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('has expected default values', () => {
+    mount();
+    expect(MetaEngineSchemaLogic.values).toEqual(DEFAULT_VALUES);
+  });
+
+  describe('actions', () => {
+    describe('onSchemaLoad', () => {
+      it('stores the API response in state', () => {
+        mount();
+
+        MetaEngineSchemaLogic.actions.onSchemaLoad(MOCK_RESPONSE);
+
+        expect(MetaEngineSchemaLogic.values).toEqual({
+          ...DEFAULT_VALUES,
+          // SchemaBaseLogic
+          dataLoading: false,
+          schema: MOCK_RESPONSE.schema,
+          // MetaEngineSchemaLogic
+          fields: MOCK_RESPONSE.fields,
+          conflictingFields: MOCK_RESPONSE.conflictingFields,
+          hasConflicts: true,
+          conflictingFieldsCount: 1,
+        });
+      });
+    });
+  });
+
+  describe('selectors', () => {
+    describe('conflictingFieldsCount', () => {
+      it('returns the number of conflicting fields', () => {
+        mount({ conflictingFields: { field_a: {}, field_b: {} } });
+        expect(MetaEngineSchemaLogic.values.conflictingFieldsCount).toEqual(2);
+      });
+    });
+
+    describe('hasConflictingFields', () => {
+      it('returns true when the conflictingFields obj has items', () => {
+        mount({ conflictingFields: { field_c: {} } });
+        expect(MetaEngineSchemaLogic.values.hasConflicts).toEqual(true);
+      });
+
+      it('returns false when the conflictingFields obj is empty', () => {
+        mount({ conflictingFields: {} });
+        expect(MetaEngineSchemaLogic.values.hasConflicts).toEqual(false);
+      });
+    });
+  });
+});
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_meta_engine_logic.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_meta_engine_logic.ts
new file mode 100644
index 0000000000000..5c8ab0f4662e2
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/schema_meta_engine_logic.ts
@@ -0,0 +1,56 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { kea, MakeLogicType } from 'kea';
+
+import { SchemaBaseLogic, SchemaBaseValues, SchemaBaseActions } from './schema_base_logic';
+import { MetaEngineSchemaApiResponse } from './types';
+
+interface MetaEngineSchemaValues extends SchemaBaseValues {
+  fields: MetaEngineSchemaApiResponse['fields'];
+  conflictingFields: MetaEngineSchemaApiResponse['conflictingFields'];
+  conflictingFieldsCount: number;
+  hasConflicts: boolean;
+}
+
+interface MetaEngineSchemaActions extends SchemaBaseActions {
+  onSchemaLoad(response: MetaEngineSchemaApiResponse): MetaEngineSchemaApiResponse;
+}
+
+export const MetaEngineSchemaLogic = kea<
+  MakeLogicType<MetaEngineSchemaValues, MetaEngineSchemaActions>
+>({
+  path: ['enterprise_search', 'app_search', 'meta_engine_schema_logic'],
+  connect: {
+    values: [SchemaBaseLogic, ['dataLoading', 'schema']],
+    actions: [SchemaBaseLogic, ['loadSchema', 'onSchemaLoad']],
+  },
+  reducers: {
+    fields: [
+      {},
+      {
+        onSchemaLoad: (_, { fields }) => fields,
+      },
+    ],
+    conflictingFields: [
+      {},
+      {
+        onSchemaLoad: (_, { conflictingFields }) => conflictingFields,
+      },
+    ],
+  },
+  selectors: {
+    conflictingFieldsCount: [
+      (selectors) => [selectors.conflictingFields],
+      (conflictingFields) => Object.keys(conflictingFields).length,
+    ],
+    hasConflicts: [
+      (selectors) => [selectors.conflictingFieldsCount],
+      (conflictingFieldsCount) => conflictingFieldsCount > 0,
+    ],
+  },
+});
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/types.ts b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/types.ts
new file mode 100644
index 0000000000000..9d79909e5549b
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/types.ts
@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  Schema,
+  IndexJob,
+  SchemaConflicts,
+  FieldCoercionErrors,
+} from '../../../shared/schema/types';
+
+export interface SchemaApiResponse {
+  schema: Schema;
+  mostRecentIndexJob: IndexJob;
+  unconfirmedFields: string[];
+  unsearchedUnconfirmedFields: boolean;
+}
+
+export interface MetaEngineSchemaApiResponse {
+  schema: Schema;
+  fields: SchemaConflicts;
+  conflictingFields: SchemaConflicts;
+}
+
+export interface ReindexJobApiResponse {
+  fieldCoercionErrors: FieldCoercionErrors;
+}
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.test.tsx b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.test.tsx
index 8412af6455285..a6e9eef8efa70 100644
--- a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.test.tsx
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.test.tsx
@@ -5,15 +5,29 @@
  * 2.0.
  */
 
+import { setMockValues, setMockActions } from '../../../../__mocks__';
+import '../../../../__mocks__/shallow_useeffect.mock';
+
 import React from 'react';
 
 import { shallow } from 'enzyme';
 
+import { Loading } from '../../../../shared/loading';
+
 import { MetaEngineSchema } from './';
 
 describe('MetaEngineSchema', () => {
+  const values = {
+    dataLoading: false,
+  };
+  const actions = {
+    loadSchema: jest.fn(),
+  };
+
   beforeEach(() => {
     jest.clearAllMocks();
+    setMockValues(values);
+    setMockActions(actions);
   });
 
   it('renders', () => {
@@ -22,4 +36,17 @@ describe('MetaEngineSchema', () => {
     expect(wrapper.isEmptyRender()).toBe(false);
     // TODO: Check for schema components
   });
+
+  it('calls loadSchema on mount', () => {
+    shallow(<MetaEngineSchema />);
+
+    expect(actions.loadSchema).toHaveBeenCalled();
+  });
+
+  it('renders a loading state', () => {
+    setMockValues({ ...values, dataLoading: true });
+    const wrapper = shallow(<MetaEngineSchema />);
+
+    expect(wrapper.find(Loading)).toHaveLength(1);
+  });
 });
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.tsx b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.tsx
index d79ddae3d9b78..234fcdb5a5a50 100644
--- a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.tsx
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/meta_engine_schema.tsx
@@ -5,14 +5,28 @@
  * 2.0.
  */
 
-import React from 'react';
+import React, { useEffect } from 'react';
+
+import { useValues, useActions } from 'kea';
 
 import { EuiPageHeader, EuiPageContentBody } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
 
 import { FlashMessages } from '../../../../shared/flash_messages';
+import { Loading } from '../../../../shared/loading';
+
+import { MetaEngineSchemaLogic } from '../schema_meta_engine_logic';
 
 export const MetaEngineSchema: React.FC = () => {
+  const { loadSchema } = useActions(MetaEngineSchemaLogic);
+  const { dataLoading } = useValues(MetaEngineSchemaLogic);
+
+  useEffect(() => {
+    loadSchema();
+  }, []);
+
+  if (dataLoading) return <Loading />;
+
   return (
     <>
       <EuiPageHeader
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.test.tsx b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.test.tsx
index 5b6367d9ce668..fd6a742d00cda 100644
--- a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.test.tsx
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.test.tsx
@@ -5,17 +5,31 @@
  * 2.0.
  */
 
+import { setMockValues, setMockActions } from '../../../../__mocks__';
+import '../../../../__mocks__/shallow_useeffect.mock';
+
 import React from 'react';
 
 import { shallow } from 'enzyme';
 
 import { EuiPageHeader, EuiButton } from '@elastic/eui';
 
+import { Loading } from '../../../../shared/loading';
+
 import { Schema } from './';
 
 describe('Schema', () => {
+  const values = {
+    dataLoading: false,
+  };
+  const actions = {
+    loadSchema: jest.fn(),
+  };
+
   beforeEach(() => {
     jest.clearAllMocks();
+    setMockValues(values);
+    setMockActions(actions);
   });
 
   it('renders', () => {
@@ -25,6 +39,19 @@ describe('Schema', () => {
     // TODO: Check for schema components
   });
 
+  it('calls loadSchema on mount', () => {
+    shallow(<Schema />);
+
+    expect(actions.loadSchema).toHaveBeenCalled();
+  });
+
+  it('renders a loading state', () => {
+    setMockValues({ ...values, dataLoading: true });
+    const wrapper = shallow(<Schema />);
+
+    expect(wrapper.find(Loading)).toHaveLength(1);
+  });
+
   it('renders page action buttons', () => {
     const wrapper = shallow(<Schema />)
       .find(EuiPageHeader)
diff --git a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.tsx b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.tsx
index ad53fd2c718b2..21dd52b04f4a7 100644
--- a/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.tsx
+++ b/x-pack/plugins/enterprise_search/public/applications/app_search/components/schema/views/schema.tsx
@@ -5,14 +5,28 @@
  * 2.0.
  */
 
-import React from 'react';
+import React, { useEffect } from 'react';
+
+import { useValues, useActions } from 'kea';
 
 import { EuiPageHeader, EuiButton, EuiPageContentBody } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
 
 import { FlashMessages } from '../../../../shared/flash_messages';
+import { Loading } from '../../../../shared/loading';
+
+import { SchemaLogic } from '../schema_logic';
 
 export const Schema: React.FC = () => {
+  const { loadSchema } = useActions(SchemaLogic);
+  const { dataLoading } = useValues(SchemaLogic);
+
+  useEffect(() => {
+    loadSchema();
+  }, []);
+
+  if (dataLoading) return <Loading />;
+
   return (
     <>
       <EuiPageHeader
diff --git a/x-pack/plugins/enterprise_search/public/applications/shared/schema/types.ts b/x-pack/plugins/enterprise_search/public/applications/shared/schema/types.ts
index 916478a0d9ccf..5a423ad510af1 100644
--- a/x-pack/plugins/enterprise_search/public/applications/shared/schema/types.ts
+++ b/x-pack/plugins/enterprise_search/public/applications/shared/schema/types.ts
@@ -31,7 +31,7 @@ export type Schema = Record<string, SchemaType>;
 
 // This is a mapping of schema field types ("text", "number", "geolocation", "date")
 // to the names of source engines which utilize that type
-export type SchemaConflictFieldTypes = Record<SchemaType, string[]>;
+export type SchemaConflictFieldTypes = Partial<Record<SchemaType, string[]>>;
 
 export interface SchemaConflict {
   fieldTypes: SchemaConflictFieldTypes;
diff --git a/x-pack/plugins/enterprise_search/server/routes/app_search/index.ts b/x-pack/plugins/enterprise_search/server/routes/app_search/index.ts
index 6ecdb8d8857c6..99aaaeeec38b3 100644
--- a/x-pack/plugins/enterprise_search/server/routes/app_search/index.ts
+++ b/x-pack/plugins/enterprise_search/server/routes/app_search/index.ts
@@ -16,6 +16,7 @@ import { registerEnginesRoutes } from './engines';
 import { registerOnboardingRoutes } from './onboarding';
 import { registerResultSettingsRoutes } from './result_settings';
 import { registerRoleMappingsRoutes } from './role_mappings';
+import { registerSchemaRoutes } from './schema';
 import { registerSearchSettingsRoutes } from './search_settings';
 import { registerSearchUIRoutes } from './search_ui';
 import { registerSettingsRoutes } from './settings';
@@ -28,6 +29,7 @@ export const registerAppSearchRoutes = (dependencies: RouteDependencies) => {
   registerAnalyticsRoutes(dependencies);
   registerDocumentsRoutes(dependencies);
   registerDocumentRoutes(dependencies);
+  registerSchemaRoutes(dependencies);
   registerCurationsRoutes(dependencies);
   registerSynonymsRoutes(dependencies);
   registerSearchSettingsRoutes(dependencies);
diff --git a/x-pack/plugins/enterprise_search/server/routes/app_search/schema.test.ts b/x-pack/plugins/enterprise_search/server/routes/app_search/schema.test.ts
new file mode 100644
index 0000000000000..408838a4de31b
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/server/routes/app_search/schema.test.ts
@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { MockRouter, mockRequestHandler, mockDependencies } from '../../__mocks__';
+
+import { registerSchemaRoutes } from './schema';
+
+describe('schema routes', () => {
+  describe('GET /api/app_search/engines/{engineName}/schema', () => {
+    let mockRouter: MockRouter;
+
+    beforeEach(() => {
+      jest.clearAllMocks();
+      mockRouter = new MockRouter({
+        method: 'get',
+        path: '/api/app_search/engines/{engineName}/schema',
+      });
+
+      registerSchemaRoutes({
+        ...mockDependencies,
+        router: mockRouter.router,
+      });
+    });
+
+    it('creates a request handler', () => {
+      expect(mockRequestHandler.createRequest).toHaveBeenCalledWith({
+        path: '/as/engines/:engineName/schema',
+      });
+    });
+  });
+
+  describe('POST /api/app_search/engines/{engineName}/schema', () => {
+    let mockRouter: MockRouter;
+
+    beforeEach(() => {
+      jest.clearAllMocks();
+      mockRouter = new MockRouter({
+        method: 'post',
+        path: '/api/app_search/engines/{engineName}/schema',
+      });
+
+      registerSchemaRoutes({
+        ...mockDependencies,
+        router: mockRouter.router,
+      });
+    });
+
+    it('creates a request handler', () => {
+      expect(mockRequestHandler.createRequest).toHaveBeenCalledWith({
+        path: '/as/engines/:engineName/schema',
+      });
+    });
+  });
+
+  describe('GET /api/app_search/engines/{engineName}/reindex_job/{reindexJobId}', () => {
+    let mockRouter: MockRouter;
+
+    beforeEach(() => {
+      jest.clearAllMocks();
+      mockRouter = new MockRouter({
+        method: 'get',
+        path: '/api/app_search/engines/{engineName}/reindex_job/{reindexJobId}',
+      });
+
+      registerSchemaRoutes({
+        ...mockDependencies,
+        router: mockRouter.router,
+      });
+    });
+
+    it('creates a request handler', () => {
+      expect(mockRequestHandler.createRequest).toHaveBeenCalledWith({
+        path: '/as/engines/:engineName/reindex_job/:reindexJobId',
+      });
+    });
+  });
+});
diff --git a/x-pack/plugins/enterprise_search/server/routes/app_search/schema.ts b/x-pack/plugins/enterprise_search/server/routes/app_search/schema.ts
new file mode 100644
index 0000000000000..74d07bd2bf75d
--- /dev/null
+++ b/x-pack/plugins/enterprise_search/server/routes/app_search/schema.ts
@@ -0,0 +1,59 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { schema } from '@kbn/config-schema';
+
+import { skipBodyValidation } from '../../lib/route_config_helpers';
+import { RouteDependencies } from '../../plugin';
+
+export function registerSchemaRoutes({
+  router,
+  enterpriseSearchRequestHandler,
+}: RouteDependencies) {
+  router.get(
+    {
+      path: '/api/app_search/engines/{engineName}/schema',
+      validate: {
+        params: schema.object({
+          engineName: schema.string(),
+        }),
+      },
+    },
+    enterpriseSearchRequestHandler.createRequest({
+      path: '/as/engines/:engineName/schema',
+    })
+  );
+
+  router.post(
+    skipBodyValidation({
+      path: '/api/app_search/engines/{engineName}/schema',
+      validate: {
+        params: schema.object({
+          engineName: schema.string(),
+        }),
+      },
+    }),
+    enterpriseSearchRequestHandler.createRequest({
+      path: '/as/engines/:engineName/schema',
+    })
+  );
+
+  router.get(
+    {
+      path: '/api/app_search/engines/{engineName}/reindex_job/{reindexJobId}',
+      validate: {
+        params: schema.object({
+          engineName: schema.string(),
+          reindexJobId: schema.string(),
+        }),
+      },
+    },
+    enterpriseSearchRequestHandler.createRequest({
+      path: '/as/engines/:engineName/reindex_job/:reindexJobId',
+    })
+  );
+}