diff --git a/changelog/unreleased/kong/wasm-filter-plugins.yml b/changelog/unreleased/kong/wasm-filter-plugins.yml
new file mode 100644
index 000000000000..60b5dec2a281
--- /dev/null
+++ b/changelog/unreleased/kong/wasm-filter-plugins.yml
@@ -0,0 +1,3 @@
+message: "**proxy-wasm**: Added support for Wasm filters to be configured via the /plugins admin API"
+type: feature
+scope: Core
diff --git a/kong-3.9.0-0.rockspec b/kong-3.9.0-0.rockspec
index 375e500cc358..9ef6099d3589 100644
--- a/kong-3.9.0-0.rockspec
+++ b/kong-3.9.0-0.rockspec
@@ -238,6 +238,7 @@ build = {
     ["kong.runloop.plugin_servers.rpc.mp_rpc"] = "kong/runloop/plugin_servers/rpc/mp_rpc.lua",
     ["kong.runloop.plugin_servers.rpc.pb_rpc"] = "kong/runloop/plugin_servers/rpc/pb_rpc.lua",
     ["kong.runloop.wasm"] = "kong/runloop/wasm.lua",
+    ["kong.runloop.wasm.plugins"] = "kong/runloop/wasm/plugins.lua",
     ["kong.runloop.wasm.properties"] = "kong/runloop/wasm/properties.lua",
 
     ["kong.workspaces"] = "kong/workspaces/init.lua",
@@ -284,6 +285,7 @@ build = {
     ["kong.db.schema.json"] = "kong/db/schema/json.lua",
     ["kong.db.schema.others.migrations"] = "kong/db/schema/others/migrations.lua",
     ["kong.db.schema.others.declarative_config"] = "kong/db/schema/others/declarative_config.lua",
+    ["kong.db.schema.others.wasm_filter"] = "kong/db/schema/others/wasm_filter.lua",
     ["kong.db.schema.entity"] = "kong/db/schema/entity.lua",
     ["kong.db.schema.metaschema"] = "kong/db/schema/metaschema.lua",
     ["kong.db.schema.typedefs"] = "kong/db/schema/typedefs.lua",
diff --git a/kong/conf_loader/init.lua b/kong/conf_loader/init.lua
index 291bd9f058e7..b809821f167a 100644
--- a/kong/conf_loader/init.lua
+++ b/kong/conf_loader/init.lua
@@ -755,6 +755,14 @@ local function load(path, custom_conf, opts)
       end
     end
 
+    if conf.wasm_modules_parsed then
+      for _, filter in ipairs(conf.wasm_modules_parsed) do
+        assert(plugins[filter.name] == nil,
+               "duplicate plugin/wasm filter name: " .. filter.name)
+        plugins[filter.name] = true
+      end
+    end
+
     conf.loaded_plugins = setmetatable(plugins, conf_constants._NOP_TOSTRING_MT)
   end
 
diff --git a/kong/db/dao/plugins.lua b/kong/db/dao/plugins.lua
index e1b198b51d38..34199eaeaa46 100644
--- a/kong/db/dao/plugins.lua
+++ b/kong/db/dao/plugins.lua
@@ -3,6 +3,7 @@ local DAO = require "kong.db.dao"
 local plugin_loader = require "kong.db.schema.plugin_loader"
 local reports = require "kong.reports"
 local plugin_servers = require "kong.runloop.plugin_servers"
+local wasm_plugins = require "kong.runloop.wasm.plugins"
 local version = require "version"
 local load_module_if_exists = require "kong.tools.module".load_module_if_exists
 
@@ -167,6 +168,13 @@ local load_plugin_handler do
       end
     end
 
+    if not ok then
+      ok, handler = wasm_plugins.load_plugin(plugin)
+      if type(handler) == "table" then
+        handler._wasm = true
+      end
+    end
+
     if not ok then
       return nil, plugin .. " plugin is enabled but not installed;\n" .. handler
     end
diff --git a/kong/db/schema/entities/filter_chains.lua b/kong/db/schema/entities/filter_chains.lua
index 89ab188f4a9a..b933ce317b7a 100644
--- a/kong/db/schema/entities/filter_chains.lua
+++ b/kong/db/schema/entities/filter_chains.lua
@@ -1,7 +1,6 @@
 local typedefs = require "kong.db.schema.typedefs"
+local filter = require "kong.db.schema.others.wasm_filter"
 local wasm = require "kong.runloop.wasm"
-local constants = require "kong.constants"
-local json_schema = require "kong.db.schema.json"
 
 
 ---@class kong.db.schema.entities.filter_chain : table
@@ -9,66 +8,14 @@ local json_schema = require "kong.db.schema.json"
 ---@field id         string
 ---@field name       string|nil
 ---@field enabled    boolean
----@field route      table|nil
----@field service    table|nil
+---@field route      { id: string }|nil
+---@field service    { id: string }|nil
 ---@field created_at number
 ---@field updated_at number
 ---@field tags       string[]
 ---@field filters    kong.db.schema.entities.wasm_filter[]
 
 
----@class kong.db.schema.entities.wasm_filter : table
----
----@field name        string
----@field enabled     boolean
----@field config      any|nil
-
-
-local filter_config_schema = {
-  parent_subschema_key = "name",
-  namespace = constants.SCHEMA_NAMESPACES.PROXY_WASM_FILTERS,
-  optional = true,
-  default = {
-    ["$schema"] = json_schema.DRAFT_4,
-    -- filters with no user-defined JSON schema may accept an optional
-    -- config, but only as a string
-    type = { "string", "null" },
-  },
-}
-
-
-if kong and kong.configuration and kong.configuration.role == "data_plane" then
-  -- data plane nodes are not guaranteed to have access to filter metadata, so
-  -- they will use a JSON schema that permits all data types
-  --
-  -- this branch can be removed if we decide to turn off entity validation in
-  -- the data plane altogether
-  filter_config_schema = {
-    inline = {
-      ["$schema"] = json_schema.DRAFT_4,
-      type = { "array", "boolean", "integer", "null", "number", "object", "string" },
-    },
-  }
-end
-
-
-local filter = {
-  type = "record",
-  fields = {
-    { name       = { type = "string", required = true, one_of = wasm.filter_names,
-                     err = "no such filter", }, },
-    { enabled    = { type = "boolean", default = true, required = true, }, },
-
-    { config = {
-        type = "json",
-        required = false,
-        json_schema = filter_config_schema,
-      },
-    },
-
-  },
-}
-
 return {
   name = "filter_chains",
   primary_key = { "id" },
diff --git a/kong/db/schema/init.lua b/kong/db/schema/init.lua
index b6b1532416a8..f0bf9b3b881b 100644
--- a/kong/db/schema/init.lua
+++ b/kong/db/schema/init.lua
@@ -51,6 +51,7 @@ local random_string = require("kong.tools.rand").random_string
 local uuid = require("kong.tools.uuid").uuid
 local json_validate = json.validate
 
+local EMPTY = {}
 
 local Schema       = {}
 Schema.__index     = Schema
@@ -1076,6 +1077,9 @@ end
 -- @return true if compatible, false otherwise.
 local function compatible_fields(f1, f2)
   local t1, t2 = f1.type, f2.type
+  if t1 == "record" and t2 == "json" then
+    return true
+  end
   if t1 ~= t2 then
     return false
   end
@@ -1128,6 +1132,59 @@ local function resolve_field(self, k, field, subschema)
 end
 
 
+---@param field table
+---@param field_name string
+---@param input table
+---@return kong.db.schema.json.schema_doc? schema
+---@return string? error
+local function get_json_schema(field, field_name, input)
+  local json_schema = field.json_schema
+
+  local schema = json_schema.inline
+  if schema then
+    return schema
+  end
+
+  local parent_key = json_schema.parent_subschema_key
+  local subschema_key = input[parent_key]
+
+  if subschema_key then
+    local schema_name = json_schema.namespace .. "/" .. subschema_key
+    schema = json.get_schema(schema_name) or json_schema.default
+
+    if schema then
+      return schema
+
+    elseif not json_schema.optional then
+      return nil, validation_errors.JSON_SCHEMA_NOT_FOUND:format(schema_name)
+    end
+
+  elseif not json_schema.optional then
+    return nil, validation_errors.JSON_PARENT_KEY_MISSING:format(field_name, parent_key)
+  end
+
+  -- no error: schema is optional
+end
+
+
+---@param  field table # Lua schema definition for this field
+---@param  field_name string
+---@param  input table # full input table that this field appears in
+---@return boolean? ok
+---@return string? error
+local function validate_json_field(field, field_name, input)
+  local schema, err = get_json_schema(field, field_name, input)
+  if schema then
+    return json_validate(input[field_name], schema)
+
+  elseif err then
+    return nil, err
+  end
+
+  return true
+end
+
+
 --- Validate fields of a table, individually, against the schema.
 -- @param self The schema
 -- @param input The input table.
@@ -1141,37 +1198,17 @@ validate_fields = function(self, input)
   local errors, _ = {}
 
   local subschema = get_subschema(self, input)
+  local subschema_fields = subschema and subschema.fields or EMPTY
 
   for k, v in pairs(input) do
     local err
     local field = self.fields[tostring(k)]
+    local subschema_field = subschema_fields[tostring(k)]
 
-    if field and field.type == "json" then
-      local json_schema = field.json_schema
-      local inline_schema = json_schema.inline
-
-      if inline_schema then
-        _, errors[k] = json_validate(v, inline_schema)
-
-      else
-        local parent_key = json_schema.parent_subschema_key
-        local json_subschema_key = input[parent_key]
-
-        if json_subschema_key then
-          local schema_name = json_schema.namespace .. "/" .. json_subschema_key
-          inline_schema = json.get_schema(schema_name) or json_schema.default
-
-          if inline_schema then
-            _, errors[k] = json_validate(v, inline_schema)
-
-          elseif not json_schema.optional then
-            errors[k] = validation_errors.JSON_SCHEMA_NOT_FOUND:format(schema_name)
-          end
-
-        elseif not json_schema.optional then
-          errors[k] = validation_errors.JSON_PARENT_KEY_MISSING:format(k, parent_key)
-        end
-      end
+    if field and field.type == "json"
+      or (subschema_field and subschema_field.type == "json")
+    then
+      _, errors[k] = validate_json_field(subschema_field or field, k, input)
 
     elseif field and field.type == "self" then
       local pok
diff --git a/kong/db/schema/json.lua b/kong/db/schema/json.lua
index fb001775d897..e235ce225778 100644
--- a/kong/db/schema/json.lua
+++ b/kong/db/schema/json.lua
@@ -36,7 +36,7 @@ local DRAFT_4 = DRAFT_4_NO_FRAGMENT .. "#"
 _M.DRAFT_4 = DRAFT_4
 
 
----@type table<string, table>
+---@type table<string, kong.db.schema.json.schema_doc>
 local schemas = {}
 
 
@@ -165,7 +165,7 @@ end
 -- Retrieve a schema from local storage by name.
 --
 ---@param name string
----@return table|nil schema
+---@return kong.db.schema.json.schema_doc? schema
 function _M.get_schema(name)
   return schemas[name]
 end
@@ -175,7 +175,6 @@ end
 -- Remove a schema from local storage by name (if it exists).
 --
 ---@param name string
----@return table|nil schema
 function _M.remove_schema(name)
   schemas[name] = nil
 end
diff --git a/kong/db/schema/others/wasm_filter.lua b/kong/db/schema/others/wasm_filter.lua
new file mode 100644
index 000000000000..55ceb1591282
--- /dev/null
+++ b/kong/db/schema/others/wasm_filter.lua
@@ -0,0 +1,59 @@
+local constants = require "kong.constants"
+local json_schema = require "kong.db.schema.json"
+local wasm = require "kong.runloop.wasm"
+
+
+---@class kong.db.schema.entities.wasm_filter : table
+---
+---@field name        string
+---@field enabled     boolean
+---@field config      any|nil
+
+
+local filter_config_schema = {
+  parent_subschema_key = "name",
+  namespace = constants.SCHEMA_NAMESPACES.PROXY_WASM_FILTERS,
+  optional = true,
+  default = {
+    ["$schema"] = json_schema.DRAFT_4,
+    -- filters with no user-defined JSON schema may accept an optional
+    -- config, but only as a string
+    type = { "string", "null" },
+  },
+}
+
+
+-- FIXME: this is clunky and error-prone because a harmless refactor might
+-- affect whether this file is require()-ed before or after `kong.configuration`
+-- is initialized
+if kong and kong.configuration and kong.configuration.role == "data_plane" then
+  -- data plane nodes are not guaranteed to have access to filter metadata, so
+  -- they will use a JSON schema that permits all data types
+  --
+  -- this branch can be removed if we decide to turn off entity validation in
+  -- the data plane altogether
+  filter_config_schema = {
+    inline = {
+      ["$schema"] = json_schema.DRAFT_4,
+      type = { "array", "boolean", "integer", "null", "number", "object", "string" },
+    },
+  }
+end
+
+
+return {
+  type = "record",
+  fields = {
+    { name       = { type = "string", required = true, one_of = wasm.filter_names,
+                     err = "no such filter", }, },
+    { enabled    = { type = "boolean", default = true, required = true, }, },
+
+    { config = {
+        type = "json",
+        required = false,
+        json_schema = filter_config_schema,
+      },
+    },
+
+  },
+}
diff --git a/kong/db/schema/plugin_loader.lua b/kong/db/schema/plugin_loader.lua
index 7ae7d856e4aa..ec1964c09ddf 100644
--- a/kong/db/schema/plugin_loader.lua
+++ b/kong/db/schema/plugin_loader.lua
@@ -1,6 +1,7 @@
 local MetaSchema = require "kong.db.schema.metaschema"
 local Entity = require "kong.db.schema.entity"
 local plugin_servers = require "kong.runloop.plugin_servers"
+local wasm_plugins = require "kong.runloop.wasm.plugins"
 local is_array = require "kong.tools.table".is_array
 local load_module_if_exists = require "kong.tools.module".load_module_if_exists
 
@@ -18,6 +19,9 @@ function plugin_loader.load_subschema(parent_schema, plugin, errors)
   if not ok then
     ok, schema = plugin_servers.load_schema(plugin)
   end
+  if not ok then
+    ok, schema = wasm_plugins.load_schema(plugin)
+  end
 
   if not ok then
     return nil, "no configuration schema found for plugin: " .. plugin
diff --git a/kong/runloop/events.lua b/kong/runloop/events.lua
index dfc9718af3c4..3135cc4c53b5 100644
--- a/kong/runloop/events.lua
+++ b/kong/runloop/events.lua
@@ -312,15 +312,26 @@ local function crud_wasm_handler(data, schema_name)
     return
   end
 
-  -- cache is invalidated on service/route deletion to ensure we don't
-  -- have oprhaned filter chain data cached
-  local is_delete = data.operation == "delete"
-               and (schema_name == "services"
-                    or schema_name == "routes")
+  local invalidate = false
 
-  local updated = schema_name == "filter_chains" or is_delete
+  -- always invalidate for filter_chain entity changes
+  if schema_name == "filter_chains" then
+    invalidate = true
 
-  if updated then
+  -- invalidate on service/route deletion to ensure we don't have any orphaned
+  -- filter chain data cached
+  elseif schema_name == "services" or schema_name == "routes" then
+    invalidate = data.operation == "delete"
+
+  -- invalidate for changes to wasm filter plugin entities
+  elseif schema_name == "plugins" then
+    local new_name = data.entity.name
+    local old_name = data.old_entity and data.old_entity.name
+    invalidate = (new_name and wasm.filters_by_name[new_name]) or
+                 (old_name and wasm.filters_by_name[old_name])
+  end
+
+  if invalidate then
     log(DEBUG, "[events] wasm filter chains updated, invalidating cache")
     core_cache:invalidate("filter_chains:version")
   end
@@ -396,6 +407,7 @@ local LOCAL_HANDLERS = {
   { "crud"    , "filter_chains"  , crud_wasm_handler },
   { "crud"    , "services"       , crud_wasm_handler },
   { "crud"    , "routes"         , crud_wasm_handler },
+  { "crud"    , "plugins"        , crud_wasm_handler },
 
   -- ca certificate store caches invalidations
   { "crud"    , "ca_certificates" , crud_ca_certificates_handler },
diff --git a/kong/runloop/plugins_iterator.lua b/kong/runloop/plugins_iterator.lua
index ea0325b02d5b..fdeb0bed837a 100644
--- a/kong/runloop/plugins_iterator.lua
+++ b/kong/runloop/plugins_iterator.lua
@@ -2,6 +2,7 @@ local workspaces = require "kong.workspaces"
 local constants = require "kong.constants"
 local tablepool = require "tablepool"
 local req_dyn_hook = require "kong.dynamic_hook"
+local wasm = require "kong.runloop.wasm"
 
 
 local kong = kong
@@ -138,6 +139,10 @@ end
 
 
 local function should_process_plugin(plugin)
+  if wasm.filters_by_name[plugin.name] then
+    return false
+  end
+
   if plugin.enabled then
     local c = constants.PROTOCOLS_WITH_SUBSYSTEM
     for _, protocol in ipairs(plugin.protocols) do
diff --git a/kong/runloop/wasm.lua b/kong/runloop/wasm.lua
index 351499ad4f7e..d2efc2d59d0c 100644
--- a/kong/runloop/wasm.lua
+++ b/kong/runloop/wasm.lua
@@ -26,10 +26,6 @@ local _M = {
 ---@field name string
 ---@field path string
 
----@class kong.configuration.wasm_filter.meta
----
----@field config_schema kong.db.schema.json.schema_doc|nil
-
 
 local uuid = require "kong.tools.uuid"
 local reports = require "kong.reports"
@@ -47,6 +43,7 @@ local proxy_wasm
 
 local kong = _G.kong
 local ngx = ngx
+local null = ngx.null
 local log = ngx.log
 local DEBUG = ngx.DEBUG
 local ERR = ngx.ERR
@@ -63,10 +60,11 @@ local fmt = string.format
 
 local VERSION_KEY = "filter_chains:version"
 local TTL_ZERO = { ttl = 0 }
+local GLOBAL_QUERY_OPTS = { workspace = null, show_ws_id = true }
 
 ---@class kong.runloop.wasm.filter_meta
 ---
----@field config_schema table|nil
+---@field config_schema kong.db.schema.json.schema_doc|nil
 
 local FILTER_META_SCHEMA = {
   type = "object",
@@ -361,6 +359,60 @@ local function build_filter_list(service_chain, route_chain)
 end
 
 
+---@param config any
+---@return string|nil
+local function serialize_configuration(config)
+  -- Serialize all JSON configurations up front
+  --
+  -- NOTE: there is a subtle difference between a raw, non-JSON filter
+  -- configuration which requires no encoding (e.g. `my config bytes`)
+  -- and a JSON filter configuration of type=string, which should be
+  -- JSON-encoded (e.g. `"my config string"`).
+  --
+  -- Properly disambiguating between the two cases requires an
+  -- inspection of the filter metadata, which is not guaranteed to be
+  -- present on data-plane/proxy nodes.
+  if config ~= nil and type(config) ~= "string" then
+    return cjson_encode(config)
+  end
+
+  return config
+end
+
+
+---@param chain kong.db.schema.entities.filter_chain
+local function get_or_insert_chain(chains, chain)
+  local route_id = chain.route and chain.route.id
+  local service_id = chain.service and chain.service.id
+
+  local chain_type = service_id and TYPE_SERVICE or TYPE_ROUTE
+  local id = service_id or route_id
+
+  -- already exists
+  if chains.by_id[chain_type][id] then
+    return chains.by_id[chain_type][id]
+  end
+
+  chains.by_id[chain_type][id] = chain
+
+  if chain_type == TYPE_ROUTE then
+    insert(chains.route_chains, chain)
+  end
+
+  insert(chains.all_chain_refs, {
+    type           = chain_type,
+
+    service_chain  = (chain_type == TYPE_SERVICE and chain) or nil,
+    service_id     = service_id,
+
+    route_chain    = (chain_type == TYPE_ROUTE and chain) or nil,
+    route_id       = route_id,
+  })
+
+  return chain
+end
+
+
 ---
 -- Unconditionally rebuild and return a new wasm state table from the db.
 --
@@ -387,6 +439,15 @@ local function rebuild_state(db, version, old_state)
   ---@type kong.runloop.wasm.filter_chain_reference[]
   local all_chain_refs = {}
 
+  local chains = {
+    all_chain_refs = all_chain_refs,
+    by_id = {
+      [TYPE_SERVICE] = service_chains_by_id,
+      [TYPE_ROUTE] = {},
+    },
+    route_chains = route_chains,
+  }
+
   local page_size = db.filter_chains.max_page_size
 
   for chain, err in db.filter_chains:each(page_size) do
@@ -395,45 +456,37 @@ local function rebuild_state(db, version, old_state)
     end
 
     if chain.enabled then
-      local route_id = chain.route and chain.route.id
-      local service_id = chain.service and chain.service.id
-
-      local chain_type = service_id and TYPE_SERVICE or TYPE_ROUTE
-
       for _, filter in ipairs(chain.filters) do
         if filter.enabled then
-          -- Serialize all JSON configurations up front
-          --
-          -- NOTE: there is a subtle difference between a raw, non-JSON filter
-          -- configuration which requires no encoding (e.g. `my config bytes`)
-          -- and a JSON filter configuration of type=string, which should be
-          -- JSON-encoded (e.g. `"my config string"`).
-          --
-          -- Properly disambiguating between the two cases requires an
-          -- inspection of the filter metadata, which is not guaranteed to be
-          -- present on data-plane/proxy nodes.
-          if filter.config ~= nil and type(filter.config) ~= "string" then
-            filter.config = cjson_encode(filter.config)
-          end
+          filter.config = serialize_configuration(filter.config)
         end
       end
 
-      insert(all_chain_refs, {
-        type           = chain_type,
+      get_or_insert_chain(chains, chain)
+    end
+  end
 
-        service_chain  = (chain_type == TYPE_SERVICE and chain) or nil,
-        service_id     = service_id,
+  local plugin_pagesize = db.plugins.pagination.max_page_size
 
-        route_chain    = (chain_type == TYPE_ROUTE and chain) or nil,
-        route_id       = route_id,
-      })
+  for plugin, err in db.plugins:each(plugin_pagesize, GLOBAL_QUERY_OPTS) do
+    if err then
+      return nil, "failed iterating plugins: " .. tostring(err)
+    end
 
-      if chain_type == TYPE_SERVICE then
-        service_chains_by_id[service_id] = chain
+    if _M.filters_by_name[plugin.name] and plugin.enabled then
+      local chain = get_or_insert_chain(chains, {
+        id = uuid.uuid(),
+        enabled = true,
+        route = plugin.route,
+        service = plugin.service,
+        filters = {},
+      })
 
-      else
-        insert(route_chains, chain)
-      end
+      insert(chain.filters, {
+        name = plugin.name,
+        enabled = true,
+        config = serialize_configuration(plugin.config),
+      })
     end
   end
 
@@ -447,28 +500,28 @@ local function rebuild_state(db, version, old_state)
 
   -- locate matching route/service chain entities to build combined
   -- filter chain references
-  for _, rchain in ipairs(route_chains) do
-    local cache_key = routes:cache_key(rchain.route.id)
+  for _, route_chain in ipairs(route_chains) do
+    local cache_key = routes:cache_key(route_chain.route.id)
 
     local route, err = cache:get(cache_key, nil,
-                                 select_route, routes, rchain.route)
+                                 select_route, routes, route_chain.route)
 
     if err then
       return nil, "failed to load route for filter chain " ..
-                  rchain.id .. ": " .. tostring(err)
+                  route_chain.id .. ": " .. tostring(err)
     end
 
     local service_id = route and route.service and route.service.id
-    local schain = service_id and service_chains_by_id[service_id]
+    local service_chain = service_id and service_chains_by_id[service_id]
 
-    if schain then
+    if service_chain then
       insert(all_chain_refs, {
         type           = TYPE_COMBINED,
 
-        service_chain  = schain,
+        service_chain  = service_chain,
         service_id     = service_id,
 
-        route_chain    = rchain,
+        route_chain    = route_chain,
         route_id       = route.id,
       })
     end
diff --git a/kong/runloop/wasm/plugins.lua b/kong/runloop/wasm/plugins.lua
new file mode 100644
index 000000000000..3629953858f6
--- /dev/null
+++ b/kong/runloop/wasm/plugins.lua
@@ -0,0 +1,72 @@
+--------------------------------------------------------------------------------
+-- Kong plugin interface for Wasm filters
+--------------------------------------------------------------------------------
+local typedefs = require "kong.db.schema.typedefs"
+local wasm = require "kong.runloop.wasm"
+
+
+local wasm_filter_config
+
+-- lazily load the filter schema as late as possible because it may-or-may-not
+-- branch based on the contents of `kong.configuration`
+local function load_filter_config_schema()
+  if not wasm_filter_config then
+    local wasm_filter = require "kong.db.schema.others.wasm_filter"
+
+    for i = 1, #wasm_filter.fields do
+      local field = wasm_filter.fields[i]
+      local k, v = next(field)
+      if k == "config" then
+        wasm_filter_config = v
+        break
+      end
+    end
+    assert(wasm_filter_config)
+  end
+
+  return wasm_filter_config
+end
+
+
+local plugins = {}
+
+
+function plugins.load_plugin(name)
+  if not wasm.filters_by_name[name] then
+    return false, "no such Wasm plugin"
+  end
+
+  -- XXX: in the future these values may be sourced from filter metadata
+  local handler = {
+    PRIORITY = 0,
+    VERSION = "0.1.0",
+  }
+
+  return true, handler
+end
+
+
+function plugins.load_schema(name)
+  if not wasm.filters_by_name[name] then
+    return false, "no such Wasm plugin"
+  end
+
+  local schema = {
+    name = name,
+    fields = {
+      { name = { type = "string" } },
+      { consumer = typedefs.no_consumer },
+      { protocols = typedefs.protocols_http },
+      { config = load_filter_config_schema() },
+    },
+    entity_checks = {
+      { mutually_exclusive = { "service", "route", } },
+      { at_least_one_of = { "service", "route", } },
+    },
+  }
+
+  return true, schema
+end
+
+
+return plugins
diff --git a/spec/02-integration/20-wasm/09-filter-meta_spec.lua b/spec/02-integration/20-wasm/09-filter-meta_spec.lua
index 183802f8552c..cd77e7222f8f 100644
--- a/spec/02-integration/20-wasm/09-filter-meta_spec.lua
+++ b/spec/02-integration/20-wasm/09-filter-meta_spec.lua
@@ -104,6 +104,7 @@ describe("filter metadata [#" .. strategy .. "]", function()
 
   describe("config validation -", function()
     local create_filter_chain
+    local create_filter_plugin
 
     if strategy == "off" then
       create_filter_chain = function(route_host, filter_chain)
@@ -122,8 +123,24 @@ describe("filter metadata [#" .. strategy .. "]", function()
         })
       end
 
+      create_filter_plugin = function(route_host, filter)
+        return post_config(admin, {
+          services = {
+            { name = random_name(),
+              url = helpers.mock_upstream_url,
+              routes = {
+                { name = random_name(),
+                  hosts = { route_host },
+                  plugins = { filter },
+                },
+              },
+            },
+          },
+        })
+      end
+
     else
-      create_filter_chain = function(route_host, filter_chain)
+      local function create_service_and_route(route_host)
         local res = admin:post("/services", json {
           name = random_name(),
           url = helpers.mock_upstream_url,
@@ -141,10 +158,25 @@ describe("filter metadata [#" .. strategy .. "]", function()
 
         assert.response(res).has.status(201)
 
-        local route = assert.response(res).has.jsonbody()
+        return assert.response(res).has.jsonbody()
+      end
 
-        res = admin:post("/routes/" .. route.id .. "/filter-chains",
-                         json(filter_chain))
+      create_filter_chain = function(route_host, filter_chain)
+        local route = create_service_and_route(route_host)
+
+        local res = admin:post("/routes/" .. route.id .. "/filter-chains",
+                               json(filter_chain))
+
+        assert.response(res).has.jsonbody()
+
+        return res
+      end
+
+      create_filter_plugin = function(route_host, filter)
+        local route = create_service_and_route(route_host)
+
+        local res = admin:post("/routes/" .. route.id .. "/plugins",
+                               json(filter))
 
         assert.response(res).has.jsonbody()
 
@@ -217,6 +249,56 @@ describe("filter metadata [#" .. strategy .. "]", function()
       end).has_no_error()
     end)
 
+    it("config schemas are validated for filter plugins", function()
+      local res = create_filter_plugin(random_name(), {
+        name = "rt_with_validation",
+        config = {}, -- empty
+      })
+
+      assert.response(res).has.status(400)
+      local body = assert.response(res).has.jsonbody()
+
+      if strategy == "off" then
+        assert.is_table(body.flattened_errors)
+        assert.same(1, #body.flattened_errors)
+
+        local err = body.flattened_errors[1]
+        assert.is_table(err)
+        assert.same("plugin", err.entity_type)
+        assert.same({
+          {
+            field = "config",
+            message = "property add is required",
+            type = "field"
+          }
+        }, err.errors)
+
+      else
+        assert.same({ config = "property add is required" },
+                    body.fields)
+      end
+
+      local host = random_name() .. ".test"
+      res = create_filter_plugin(host, {
+        name = "rt_with_validation",
+        config = {
+          add = {
+            headers = {
+              "x-foo:123",
+            },
+          },
+        },
+      })
+
+      assert.response(res).has.status(201)
+
+      assert.eventually(function()
+        res = proxy:get("/status/200", { headers = { host = host } })
+        assert.response(res).has.status(200)
+        assert.response(res).has.header("x-foo")
+      end).has_no_error()
+    end)
+
     it("filters without config schemas can only accept a string", function()
       local host = random_name() .. ".test"
 
diff --git a/spec/02-integration/20-wasm/12-filters-as-plugins_spec.lua b/spec/02-integration/20-wasm/12-filters-as-plugins_spec.lua
new file mode 100644
index 000000000000..19755bb4cdac
--- /dev/null
+++ b/spec/02-integration/20-wasm/12-filters-as-plugins_spec.lua
@@ -0,0 +1,344 @@
+local helpers = require "spec.helpers"
+local cjson = require "cjson"
+local rand = require "kong.tools.rand"
+local uuid = require "kong.tools.uuid"
+
+local FILTER_HEADER = "X-Wasm-Filter-Plugin"
+local PLUGIN_HEADER = "X-Lua-Plugin"
+local FILTER_CHAIN_HEADER = "X-Wasm-Filter-Chain"
+
+local FILTER_PATH = assert(helpers.test_conf.wasm_filters_path)
+
+local function json_request(data)
+  return {
+    headers = {
+      ["Content-Type"] = "application/json",
+    },
+    body = cjson.encode(data),
+  }
+end
+
+local function rt_config(header, value)
+  value = value or rand.random_string()
+  return {
+      append = {
+        headers = {
+          header .. ":" .. value,
+        },
+      },
+    }
+end
+
+local function new_filter(value)
+  return {
+    name = "response_transformer",
+    config = cjson.encode(rt_config(FILTER_HEADER, value)),
+  }
+end
+
+local function new_plugin(value)
+  return {
+    name = "response-transformer",
+    config = rt_config(PLUGIN_HEADER, value),
+  }
+end
+
+local function new_filter_chain(value)
+  return {
+    filters = {
+      {
+        name = "response_transformer",
+        config = cjson.encode(rt_config(FILTER_CHAIN_HEADER, value)),
+      },
+    },
+  }
+end
+
+for _, strategy in helpers.each_strategy({ "postgres" }) do
+
+describe("#wasm filters as plugins (#" .. strategy .. ")", function()
+  local bp, db
+  local admin, proxy
+  local service, route
+
+  local function assert_bad_response(res)
+    assert.response(res).has.status(400)
+    return assert.response(res).has.jsonbody()
+  end
+
+  local function check_response(header_name, header_value, context)
+    helpers.wait_for_all_config_update()
+
+    local res, err = proxy:send({ path = "/status/200" })
+    assert(res and err == nil, tostring(err) .. ": " .. context)
+
+    assert.response(res).has.status(200)
+    local got = assert.response(res).has.header(header_name)
+    assert.equals(header_value, got, context)
+  end
+
+  local function check_filter_response(header_value, context)
+    return check_response(FILTER_HEADER, header_value, context)
+  end
+
+  local function admin_request(method, path, data)
+    local params = (data ~= nil and json_request(data)) or {}
+    params.method = method
+    params.path = path
+
+    local res, err = admin:send(params)
+    assert.is_nil(err)
+    return res
+  end
+
+  local function create_plugin(plugin)
+    local res = admin:post("/plugins", json_request(plugin))
+    assert.response(res).has.status(201)
+    return assert.response(res).has.jsonbody()
+  end
+
+  local function update_plugin(id, plugin)
+    local res = admin:patch("/plugins/" .. id, json_request(plugin))
+    assert.response(res).has.status(200)
+    return assert.response(res).has.jsonbody()
+  end
+
+  local function get_plugin(endpoint, id)
+    local res = admin:get("/plugins/" .. id)
+    assert.response(res).has.status(200)
+    return assert.response(res).has.jsonbody()
+  end
+
+  local function create_filter_chain(fc)
+    local res = admin:post("/filter-chains", json_request(fc))
+    assert.response(res).has.status(201)
+    return assert.response(res).has.jsonbody()
+  end
+
+
+  lazy_setup(function()
+    require("kong.runloop.wasm").enable({
+      { name = "response_transformer",
+        path = FILTER_PATH .. "/response_transformer.wasm",
+      },
+    })
+
+    bp, db = helpers.get_db_utils(strategy, {
+      "routes",
+      "services",
+      "filter_chains",
+      "plugins",
+    })
+
+    helpers.start_kong({
+      database = strategy,
+      nginx_conf = "spec/fixtures/custom_nginx.template",
+      nginx_main_worker_processes = "2",
+      wasm = true,
+      wasm_filters = "response_transformer",
+      plugins = "response-transformer",
+    })
+
+    admin = helpers.admin_client()
+    proxy = helpers.proxy_client()
+    proxy.reopen = true
+  end)
+
+  lazy_teardown(function()
+    if admin then
+      admin:close()
+    end
+
+    if proxy then
+      proxy:close()
+    end
+
+    helpers.stop_kong()
+  end)
+
+  before_each(function()
+    -- create a scratch service/route to use
+    service = assert(bp.services:insert({}))
+    route = assert(bp.routes:insert({
+      paths = { "/" },
+      service = service,
+    }))
+  end)
+
+  after_each(function()
+    bp.filter_chains:truncate()
+    bp.plugins:truncate()
+
+    db.routes:delete(route)
+    db.services:delete(service)
+  end)
+
+  describe("/plugins", function()
+    describe("POST", function()
+      it("creates a plugin instance from a filter", function()
+        local value = rand.random_string()
+        local filter = new_filter(value)
+        filter.route = { id = route.id }
+        create_plugin(filter)
+        check_filter_response(value, "POST /plugins")
+      end)
+
+      it("validates foreign references", function()
+        local filter = new_filter()
+
+        filter.route = { id = uuid.uuid() }
+        local res = admin_request("POST", "/plugins", filter)
+        local err = assert_bad_response(res)
+        assert.equals("foreign key violation", err.name)
+        assert.same({ route = filter.route }, err.fields)
+
+        filter.route = nil
+        filter.service = { id = uuid.uuid() }
+        local res = admin_request("POST", "/plugins", filter)
+        local err = assert_bad_response(res)
+        assert.equals("foreign key violation", err.name)
+        assert.same({ service = filter.service }, err.fields)
+      end)
+    end)
+
+    describe("GET", function()
+      it("returns both wasm filters and Lua plugins", function()
+        local route_filter = new_filter()
+        route_filter.route = { id = route.id }
+
+        local service_filter = new_filter()
+        service_filter.service = { id = service.id }
+
+        local route_plugin = new_plugin()
+        route_plugin.route = { id = route.id }
+
+        local service_plugin = new_plugin()
+        service_plugin.service = { id = service.id }
+
+        route_filter = create_plugin(route_filter)
+        service_filter = create_plugin(service_filter)
+
+        route_plugin = create_plugin(route_plugin)
+        service_plugin = create_plugin(service_plugin)
+
+        local res = admin:get("/plugins")
+        assert.response(res).has.status(200)
+        local json = assert.response(res).has.jsonbody()
+        assert.is_table(json)
+        assert.is_table(json.data)
+
+        local expected = 4
+        assert.equals(expected, #json.data)
+        helpers.intercept(json.data)
+        local found = 0
+
+        for _, plugin in ipairs(json.data) do
+          if   plugin.id == route_filter.id
+            or plugin.id == service_filter.id
+            or plugin.id == route_plugin.id
+            or plugin.id == service_plugin.id
+          then
+            found = found + 1
+          end
+        end
+
+        assert.equals(expected, found, "GET /plugins didn't return expected entities")
+      end)
+    end)
+  end)
+
+  describe("/plugins/:id", function()
+    local plugin
+    local path
+
+    before_each(function()
+      local value = rand.random_string()
+      local filter = new_filter(value)
+      filter.route = { id = route.id }
+      plugin = create_plugin(filter)
+      path = "/plugins/" .. plugin.id
+      check_filter_response(value, "POST /plugins")
+    end)
+
+    after_each(function()
+      db.plugins:delete(plugin)
+    end)
+
+    describe("GET", function()
+      it("retrieves a wasm filter plugin instance", function()
+        local got = get_plugin("/plugins", plugin.id)
+        assert.same(plugin, got)
+      end)
+    end)
+
+    describe("PATCH", function()
+      it("updates a wasm filter plugin instance", function()
+        local value = rand.random_string()
+        local updated = update_plugin(plugin.id, {
+          config = cjson.encode(rt_config(FILTER_HEADER, value)),
+        })
+
+        assert.not_same(plugin, updated)
+        check_filter_response(value, "PATCH /plugins/:id")
+      end)
+    end)
+
+    describe("DELETE", function()
+      it("removes a wasm filter plugin instance", function()
+        local res = admin:delete(path)
+        assert.response(res).has.status(204)
+
+        res = admin:get(path)
+        assert.response(res).status(404)
+      end)
+    end)
+  end)
+
+  describe("filter plugins and lua plugins", function()
+    it("can coexist", function()
+      local filter_value = rand.random_string()
+      local filter = new_filter(filter_value)
+      filter.route = { id = route.id }
+
+      local plugin_value = rand.random_string()
+      local plugin = new_plugin(plugin_value)
+      plugin.route = { id = route.id }
+
+      create_plugin(filter)
+      create_plugin(plugin)
+
+      helpers.wait_for_all_config_update()
+
+      local res = proxy:get("/status/200")
+      assert.response(res).has.status(200)
+
+      assert.equals(filter_value, assert.response(res).has.header(FILTER_HEADER))
+      assert.equals(plugin_value, assert.response(res).has.header(PLUGIN_HEADER))
+    end)
+  end)
+
+  describe("filter plugins and filter chains", function()
+    it("can coexist", function()
+      local filter_value = rand.random_string()
+      local filter = new_filter(filter_value)
+      filter.route = { id = route.id }
+
+      local fc_value = rand.random_string()
+      local fc = new_filter_chain(fc_value)
+      fc.route = { id = route.id }
+
+      create_plugin(filter)
+      create_filter_chain(fc)
+
+      helpers.wait_for_all_config_update()
+
+      local res = proxy:get("/status/200")
+      assert.response(res).has.status(200)
+
+      assert.equals(filter_value, assert.response(res).has.header(FILTER_HEADER))
+      assert.equals(fc_value, assert.response(res).has.header(FILTER_CHAIN_HEADER))
+    end)
+  end)
+end)
+
+end -- each strategy