diff --git a/.formatter.exs b/.formatter.exs
index e4245671..7dd7b3f1 100644
--- a/.formatter.exs
+++ b/.formatter.exs
@@ -2,6 +2,8 @@ spark_locals_without_parens = [
   accept: 1,
   access_type: 1,
   action: 1,
+  action: 2,
+  action: 3,
   allow: 1,
   allow_expr?: 1,
   allow_nil?: 1,
@@ -169,6 +171,7 @@ spark_locals_without_parens = [
   require_attributes: 1,
   required?: 1,
   returns: 1,
+  run: 1,
   run_flow: 2,
   run_flow: 3,
   select: 1,
diff --git a/documentation/topics/actions.md b/documentation/topics/actions.md
index 0a578b7e..88c8ffd3 100644
--- a/documentation/topics/actions.md
+++ b/documentation/topics/actions.md
@@ -2,7 +2,14 @@
 
 ## Action Types
 
-Ash has 4 action types `:read`, `:create`, `:update`, `:destroy`. The purpose of these action types is to provide expectations about what is required to run those actions, and what is returned from them.
+Ash has 5 action types `:read`, `:create`, `:update`, `:destroy` and `:action`. The purpose of these action types is to provide expectations about what is required to run those actions, and what is returned from them.
+
+### Basic Actions
+
+The `:action` type is a special type of action that can do essentially whatever you want. We refer to it as a "basic" action, because there are no special rules about how it works, and minimal structure surrounding it.
+A basic action takes arguments and returns a value. The struct used for building input for a basic action is `Ash.ActionInput`. For the rest of this document we will discuss the four main action types.
+
+### Create/Read/Update/Destroy
 
 The actions do not need to do _exactly_ what their action type implies however. Using manual actions, you can define a create action that actually updates something, or using the `soft?` option for `destroy` actions you can treat them as updates. The important part to consider is their interface. More action types may be added in the future.
 
diff --git a/lib/ash/action_input.ex b/lib/ash/action_input.ex
new file mode 100644
index 00000000..bfc23450
--- /dev/null
+++ b/lib/ash/action_input.ex
@@ -0,0 +1,242 @@
+defmodule Ash.ActionInput do
+  @moduledoc """
+  Input for a custom action
+  """
+
+  alias Ash.Error.Action.InvalidArgument
+
+  defstruct [:action, :api, :resource, arguments: %{}, params: %{}, context: %{}, valid?: true]
+
+  @type t :: %__MODULE__{
+          arguments: map(),
+          params: map(),
+          action: Ash.Resource.Actions.Action.t(),
+          resource: Ash.Resource.t(),
+          context: map(),
+          api: Ash.Api.t(),
+          valid?: boolean()
+        }
+
+  @doc """
+  Creates a new input for a basic action
+  """
+  @spec for_action(
+          resource_or_input :: Ash.Resource.t() | t(),
+          action :: atom,
+          params :: map,
+          opts :: Keyword.t()
+        ) :: t()
+  def for_action(resource_or_input, action, params, opts \\ []) do
+    input =
+      case resource_or_input do
+        resource when is_atom(resource) ->
+          action = Ash.Resource.Info.action(resource, action)
+          %__MODULE__{resource: resource, action: action}
+
+        input ->
+          input
+      end
+
+    {input, _opts} = Ash.Actions.Helpers.add_process_context(input.api, input, opts)
+
+    cast_params(input, params)
+  end
+
+  @doc "Set an argument value"
+  @spec set_argument(input :: t(), name :: atom, value :: term()) :: t()
+  def set_argument(input, argument, value) do
+    if input.action do
+      argument =
+        Enum.find(
+          input.action.arguments,
+          &(&1.name == argument || to_string(&1.name) == argument)
+        )
+
+      if argument do
+        with {:ok, casted} <-
+               Ash.Type.Helpers.cast_input(argument.type, value, argument.constraints, input),
+             {:constrained, {:ok, casted}, argument} when not is_nil(casted) <-
+               {:constrained,
+                Ash.Type.apply_constraints(argument.type, casted, argument.constraints),
+                argument} do
+          %{input | arguments: Map.put(input.arguments, argument.name, casted)}
+        else
+          {:constrained, {:ok, nil}, _argument} ->
+            %{input | arguments: Map.put(input.arguments, argument.name, nil)}
+
+          {:constrained, {:error, error}, argument} ->
+            input = %{
+              input
+              | arguments: Map.put(input.arguments, argument.name, value)
+            }
+
+            add_invalid_errors(value, input, argument, error)
+
+          {:error, error} ->
+            input = %{
+              input
+              | arguments: Map.put(input.arguments, argument.name, value)
+            }
+
+            add_invalid_errors(value, input, argument, error)
+        end
+      else
+        %{input | arguments: Map.put(input.arguments, argument, value)}
+      end
+    else
+      %{input | arguments: Map.put(input.arguments, argument, value)}
+    end
+  end
+
+  @doc """
+  Deep merges the provided map into the input context that can be used later
+
+  Do not use the `private` key in your custom context, as that is reserved for internal use.
+  """
+  @spec set_context(t(), map | nil) :: t()
+  def set_context(input, nil), do: input
+
+  def set_context(input, map) do
+    %{input | context: Ash.Helpers.deep_merge_maps(input.context, map)}
+  end
+
+  defp cast_params(input, params) do
+    input = %{
+      input
+      | params: Map.merge(input.params, Enum.into(params, %{}))
+    }
+
+    Enum.reduce(params, input, fn {name, value}, input ->
+      if has_argument?(input.action, name) do
+        set_argument(input, name, value)
+      else
+        input
+      end
+    end)
+  end
+
+  defp has_argument?(action, name) when is_atom(name) do
+    Enum.any?(action.arguments, &(&1.private? == false && &1.name == name))
+  end
+
+  defp has_argument?(action, name) when is_binary(name) do
+    Enum.any?(action.arguments, &(&1.private? == false && to_string(&1.name) == name))
+  end
+
+  defp add_invalid_errors(value, input, attribute, message) do
+    messages =
+      if Keyword.keyword?(message) do
+        [message]
+      else
+        List.wrap(message)
+      end
+
+    Enum.reduce(messages, input, fn message, input ->
+      if Exception.exception?(message) do
+        error =
+          message
+          |> Ash.Error.to_ash_error()
+
+        errors =
+          case error do
+            %class{errors: errors}
+            when class in [
+                   Ash.Error.Invalid,
+                   Ash.Error.Unknown,
+                   Ash.Error.Forbidden,
+                   Ash.Error.Framework
+                 ] ->
+              errors
+
+            error ->
+              [error]
+          end
+
+        Enum.reduce(errors, input, fn error, input ->
+          add_error(input, Ash.Error.set_path(error, attribute.name))
+        end)
+      else
+        opts = Ash.Type.Helpers.error_to_exception_opts(message, attribute)
+
+        Enum.reduce(opts, input, fn opts, input ->
+          error =
+            InvalidArgument.exception(
+              value: value,
+              field: Keyword.get(opts, :field),
+              message: Keyword.get(opts, :message),
+              vars: opts
+            )
+
+          error =
+            if opts[:path] do
+              Ash.Error.set_path(error, opts[:path])
+            else
+              error
+            end
+
+          add_error(input, error)
+        end)
+      end
+    end)
+  end
+
+  @doc "Adds an error to the input errors list, and marks the input as `valid?: false`"
+  @spec add_error(t(), term | String.t() | list(term | String.t())) :: t()
+  def add_error(input, errors, path \\ [])
+
+  def add_error(input, errors, path) when is_list(errors) do
+    if Keyword.keyword?(errors) do
+      errors
+      |> to_change_errors()
+      |> Ash.Error.set_path(path)
+      |> handle_error(input)
+    else
+      Enum.reduce(errors, input, &add_error(&2, &1, path))
+    end
+  end
+
+  def add_error(input, error, path) when is_binary(error) do
+    add_error(
+      input,
+      InvalidArgument.exception(message: error),
+      path
+    )
+  end
+
+  def add_error(input, error, path) do
+    error
+    |> Ash.Error.set_path(path)
+    |> handle_error(input)
+  end
+
+  defp handle_error(error, input) do
+    %{input | valid?: false, errors: [error | input.errors]}
+  end
+
+  defp to_change_errors(keyword) do
+    errors =
+      if keyword[:fields] && keyword[:fields] != [] do
+        Enum.map(keyword[:fields], fn field ->
+          InvalidArgument.exception(
+            field: field,
+            message: keyword[:message],
+            value: keyword[:value],
+            vars: keyword
+          )
+        end)
+      else
+        InvalidArgument.exception(
+          field: keyword[:field],
+          message: keyword[:message],
+          value: keyword[:value],
+          vars: keyword
+        )
+      end
+
+    if keyword[:path] do
+      Enum.map(errors, &Ash.Error.set_path(&1, keyword[:path]))
+    else
+      errors
+    end
+  end
+end
diff --git a/lib/ash/actions/action.ex b/lib/ash/actions/action.ex
new file mode 100644
index 00000000..4d188f6d
--- /dev/null
+++ b/lib/ash/actions/action.ex
@@ -0,0 +1,194 @@
+defmodule Ash.Actions.Action do
+  @moduledoc false
+
+  require Ash.Tracer
+
+  def run(api, input, opts) do
+    {input, opts} = Ash.Actions.Helpers.add_process_context(api, input, opts)
+
+    context =
+      Map.merge(input.context, %{
+        actor: opts[:actor],
+        tenant: opts[:tenant],
+        authorize?: opts[:authorize?],
+        api: opts[:api]
+      })
+
+    {module, run_opts} = input.action.run
+
+    Ash.Tracer.span :action,
+                    Ash.Api.Info.span_name(
+                      api,
+                      input.resource,
+                      input.action.name
+                    ),
+                    opts[:tracer] do
+      metadata = %{
+        api: api,
+        resource: input.resource,
+        resource_short_name: Ash.Resource.Info.short_name(input.resource),
+        actor: opts[:actor],
+        tenant: opts[:tenant],
+        action: input.action.name,
+        authorize?: opts[:authorize?]
+      }
+
+      Ash.Tracer.set_metadata(opts[:tracer], :action, metadata)
+
+      Ash.Tracer.telemetry_span [:ash, Ash.Api.Info.short_name(api), :create],
+                                metadata do
+        if input.action.transaction? do
+          resources =
+            input.resource
+            |> List.wrap()
+            |> Enum.concat(input.action.touches_resources)
+            |> Enum.uniq()
+
+          notify? =
+            if Process.get(:ash_started_transaction?) do
+              false
+            else
+              Process.put(:ash_started_transaction?, true)
+              true
+            end
+
+          try do
+            resources
+            |> Enum.reject(&Ash.DataLayer.in_transaction?/1)
+            |> Ash.DataLayer.transaction(fn ->
+              case authorize(api, opts[:actor], input) do
+                :ok ->
+                  case module.run(input, run_opts, context) do
+                    {:ok, result} ->
+                      {:ok, result, []}
+
+                    other ->
+                      other
+                  end
+
+                {:error, error} ->
+                  {:error, error}
+              end
+            end)
+            |> case do
+              {:ok, {:ok, value, notifications}} ->
+                notifications =
+                  if notify? && !opts[:return_notifications?] do
+                    Enum.concat(
+                      notifications || [],
+                      Process.delete(:ash_notifications) || []
+                    )
+                  else
+                    notifications || []
+                  end
+
+                remaining = Ash.Notifier.notify(notifications)
+
+                Ash.Actions.Helpers.warn_missed!(input.resource, input.action, %{
+                  resource_notifications: remaining
+                })
+
+                {:ok, value}
+
+              {:error, error} ->
+                {:error, error}
+            end
+          after
+            if notify? do
+              Process.delete(:ash_started_transaction?)
+            end
+          end
+        else
+          case authorize(api, opts[:actor], input) do
+            :ok ->
+              case module.run(input, run_opts, context) do
+                {:ok, result} ->
+                  {:ok, result}
+
+                {:ok, result, notifications} ->
+                  remaining = Ash.Notifier.notify(notifications)
+
+                  Ash.Actions.Helpers.warn_missed!(input.resource, input.action, %{
+                    resource_notifications: remaining
+                  })
+
+                  {:ok, result}
+
+                {:error, error} ->
+                  {:error, error}
+              end
+
+            {:error, error} ->
+              {:error, error}
+          end
+        end
+      end
+    end
+  end
+
+  defp authorize(api, actor, input) do
+    input.resource
+    |> Ash.Resource.Info.authorizers()
+    |> Enum.reduce_while(
+      :ok,
+      fn authorizer, :ok ->
+        authorizer_state =
+          authorizer.initial_state(
+            actor,
+            input.resource,
+            input.action,
+            false
+          )
+
+        context = %{
+          api: api,
+          action_input: input,
+          query: nil,
+          changeset: nil
+        }
+
+        case authorizer.strict_check(authorizer_state, context) do
+          {:error, %{class: :forbidden} = e} when is_exception(e) ->
+            {:halt, {:error, e}}
+
+          {:error, error} ->
+            {:halt, {:error, error}}
+
+          {:authorized, _} ->
+            {:cont, :ok}
+
+          {:filter, _authorizer, filter} ->
+            raise """
+            Cannot use filter checks with basic actions
+
+            Received #{inspect(filter)} when authorizing #{inspect(input.resource)}.#{input.action.name}
+            """
+
+          {:filter, filter} ->
+            raise """
+            Cannot use filter checks with basic actions
+
+            Received #{inspect(filter)} when authorizing #{inspect(input.resource)}.#{input.action.name}
+            """
+
+          {:continue, _state} ->
+            raise """
+            Cannot use runtime checks with basic actions
+
+            Must use only simple checks or other checks that can be resolved without returning results #{inspect(input.resource)}.#{input.action.name}
+            """
+
+          {:filter_and_continue, filter, _} ->
+            raise """
+            Cannot use filter checks with basic actions
+
+            Received #{inspect(filter)} when authorizing #{inspect(input.resource)}.#{input.action.name}
+            """
+
+          :forbidden ->
+            {:halt, {:error, Ash.Authorizer.exception(authorizer, :forbidden, authorizer_state)}}
+        end
+      end
+    )
+  end
+end
diff --git a/lib/ash/actions/helpers.ex b/lib/ash/actions/helpers.ex
index f00f97ee..ce7f579b 100644
--- a/lib/ash/actions/helpers.ex
+++ b/lib/ash/actions/helpers.ex
@@ -21,6 +21,9 @@ defmodule Ash.Actions.Helpers do
   defp set_context(%{__struct__: Ash.Query} = query, context),
     do: Ash.Query.set_context(query, context)
 
+  defp set_context(%{__struct__: Ash.ActionInput} = action_input, context),
+    do: Ash.ActionInput.set_context(action_input, context)
+
   def add_process_context(api, query_or_changeset, opts) do
     query_or_changeset = set_context(query_or_changeset, opts[:context] || %{})
     api = api || query_or_changeset.api
@@ -81,6 +84,11 @@ defmodule Ash.Actions.Helpers do
     private_context = Map.new(Keyword.take(opts, [:actor, :authorize?]))
 
     case query_or_changeset do
+      %{__struct__: Ash.ActionInput} ->
+        query_or_changeset
+        |> Ash.ActionInput.set_context(context)
+        |> Ash.ActionInput.set_context(%{private: private_context})
+
       %{__struct__: Ash.Query} ->
         query_or_changeset
         |> Ash.Query.set_context(context)
diff --git a/lib/ash/api/api.ex b/lib/ash/api/api.ex
index 6bd6b84b..bc895dd2 100644
--- a/lib/ash/api/api.ex
+++ b/lib/ash/api/api.ex
@@ -708,6 +708,60 @@ defmodule Ash.Api do
     ]
   ]
 
+  @run_action_opts [
+    actor: [
+      type: :any,
+      doc: """
+      The actor for handling `^actor/1` templates, supplied to calculation context.
+      """
+    ],
+    tenant: [
+      type: :any,
+      doc: """
+      The tenant, supplied to calculation context.
+      """
+    ],
+    authorize?: [
+      type: :boolean,
+      doc: """
+      Wether or not the request should be authorized.
+      """
+    ],
+    tracer: [
+      type: :any,
+      doc: """
+      A tracer, provided to the calculation context.
+      """
+    ]
+  ]
+
+  @spec run_action!(api :: Ash.Api.t(), input :: Ash.ActionInput.t(), opts :: Keyword.t()) ::
+          term | no_return
+  def run_action!(api, input, opts \\ []) do
+    api
+    |> run_action(input, opts)
+    |> unwrap_or_raise!(opts[:stacktraces?])
+  end
+
+  @doc """
+  Runs a basic action.
+
+  Options:
+
+  #{Spark.OptionsHelpers.docs(@run_action_opts)}
+  """
+  @spec run_action(api :: Ash.Api.t(), input :: Ash.ActionInput.t(), opts :: Keyword.t()) ::
+          {:ok, term} | {:error, Ash.Error.t()}
+  def run_action(api, input, opts \\ []) do
+    case Spark.OptionsHelpers.validate(opts, @run_action_opts) do
+      {:ok, opts} ->
+        Ash.Actions.Action.run(api, input, opts)
+
+      {:error, error} ->
+        {:error, error}
+    end
+  end
+
   @doc false
   def calculate_opts, do: @calculate_opts
 
@@ -919,6 +973,17 @@ defmodule Ash.Api do
   @callback calculate!(resource :: Ash.Resource.t(), calculation :: atom, opts :: Keyword.t()) ::
               term | no_return
 
+  @doc "Runs a basic action, raising on errors"
+  @callback run_action!(input :: Ash.ActionInput.t(), opts :: Keyword.t()) ::
+              term | no_return
+
+  @doc "Runs a basic action"
+  @callback run_action(input :: Ash.ActionInput.t(), opts :: Keyword.t()) ::
+              {:ok, term} | {:error, term}
+
+  @callback calculate!(resource :: Ash.Resource.t(), calculation :: atom, opts :: Keyword.t()) ::
+              term | no_return
+
   @doc """
   Get a record by a primary key. See `c:get/3` for more.
   """
diff --git a/lib/ash/api/interface.ex b/lib/ash/api/interface.ex
index 458c477a..ae82bfc6 100644
--- a/lib/ash/api/interface.ex
+++ b/lib/ash/api/interface.ex
@@ -31,6 +31,14 @@ defmodule Ash.Api.Interface do
         Api.can(__MODULE__, action_or_query_or_changeset, actor, opts)
       end
 
+      def run_action!(input, opts \\ []) do
+        Api.run_action!(__MODULE__, input, opts)
+      end
+
+      def run_action(input, opts \\ []) do
+        Api.run_action(__MODULE__, input, opts)
+      end
+
       def calculate!(resource, calculation, opts \\ []) do
         case calculate(resource, calculation, opts) do
           {:ok, result} ->
diff --git a/lib/ash/code_interface.ex b/lib/ash/code_interface.ex
index b33a63e1..6517710f 100644
--- a/lib/ash/code_interface.ex
+++ b/lib/ash/code_interface.ex
@@ -325,6 +325,81 @@ defmodule Ash.CodeInterface do
         """
 
         case action.type do
+          :action ->
+            @doc doc
+            @dialyzer {:nowarn_function, {interface.name, Enum.count(args) + 2}}
+            def unquote(interface.name)(
+                  unquote_splicing(arg_vars_function),
+                  params_or_opts \\ %{},
+                  opts \\ []
+                ) do
+              if opts == [] && Keyword.keyword?(params_or_opts) do
+                apply(__MODULE__, elem(__ENV__.function, 0), [
+                  unquote_splicing(arg_vars),
+                  %{},
+                  params_or_opts
+                ])
+              else
+                input =
+                  unquote(args)
+                  |> Enum.zip([unquote_splicing(arg_vars)])
+                  |> Enum.reduce(params_or_opts, fn {key, value}, params_or_opts ->
+                    Map.put(params_or_opts, key, value)
+                  end)
+
+                action_input =
+                  opts[:input]
+                  |> Kernel.||(unquote(resource))
+                  |> Ash.ActionInput.for_action(
+                    unquote(action.name),
+                    input,
+                    Keyword.take(opts, [:actor, :tenant, :authorize?, :tracer])
+                  )
+
+                unquote(api).run_action(
+                  action_input,
+                  Keyword.drop(opts, [:actor, :changeset, :tenant, :authorize?, :tracer])
+                )
+              end
+            end
+
+            @doc doc
+            @dialyzer {:nowarn_function, {:"#{interface.name}!", Enum.count(args) + 2}}
+            # sobelow_skip ["DOS.BinToAtom"]
+            def unquote(:"#{interface.name}!")(
+                  unquote_splicing(arg_vars_function),
+                  params_or_opts \\ %{},
+                  opts \\ []
+                ) do
+              if opts == [] && Keyword.keyword?(params_or_opts) do
+                apply(__MODULE__, elem(__ENV__.function, 0), [
+                  unquote_splicing(arg_vars),
+                  %{},
+                  params_or_opts
+                ])
+              else
+                input =
+                  unquote(args)
+                  |> Enum.zip([unquote_splicing(arg_vars)])
+                  |> Enum.reduce(params_or_opts, fn {key, value}, params_or_opts ->
+                    Map.put(params_or_opts, key, value)
+                  end)
+
+                action_input =
+                  (opts[:input] || unquote(resource))
+                  |> Ash.ActionInput.for_action(
+                    unquote(action.name),
+                    input,
+                    Keyword.take(opts, [:actor, :tenant, :authorize?, :tracer])
+                  )
+
+                unquote(api).run_action!(
+                  action_input,
+                  Keyword.drop(opts, [:actor, :changeset, :authorize?, :tracer])
+                )
+              end
+            end
+
           :read ->
             @doc doc
             @dialyzer {:nowarn_function, {interface.name, Enum.count(args) + 2}}
diff --git a/lib/ash/error/action/invalid_argument.ex b/lib/ash/error/action/invalid_argument.ex
new file mode 100644
index 00000000..b8be2837
--- /dev/null
+++ b/lib/ash/error/action/invalid_argument.ex
@@ -0,0 +1,29 @@
+defmodule Ash.Error.Action.InvalidArgument do
+  @moduledoc "Used when an invalid value is provided for an action argument"
+  use Ash.Error.Exception
+
+  def_ash_error([:field, :message, :value], class: :invalid)
+
+  defimpl Ash.ErrorKind do
+    def id(_), do: Ash.UUID.generate()
+
+    def code(_), do: "invalid_argument"
+
+    def message(error) do
+      """
+      Invalid value provided#{for_field(error)}#{do_message(error)}
+
+      #{inspect(error.value)}
+      """
+    end
+
+    defp for_field(%{field: field}) when not is_nil(field), do: " for #{field}"
+    defp for_field(_), do: ""
+
+    defp do_message(%{message: message}) when not is_nil(message) do
+      ": #{message}."
+    end
+
+    defp do_message(_), do: "."
+  end
+end
diff --git a/lib/ash/policy/authorizer.ex b/lib/ash/policy/authorizer.ex
index 758ac69c..ce40626f 100644
--- a/lib/ash/policy/authorizer.ex
+++ b/lib/ash/policy/authorizer.ex
@@ -4,6 +4,7 @@ defmodule Ash.Policy.Authorizer do
     :resource,
     :query,
     :changeset,
+    :action_input,
     :data,
     :action,
     :api,
@@ -347,7 +348,7 @@ defmodule Ash.Policy.Authorizer do
 
   @impl true
   def strict_check_context(_authorizer) do
-    [:query, :changeset, :api, :resource]
+    [:query, :changeset, :api, :resource, :action_input]
   end
 
   @impl true
@@ -366,6 +367,7 @@ defmodule Ash.Policy.Authorizer do
       authorizer
       | query: context.query,
         changeset: context.changeset,
+        action_input: context[:action_input],
         api: context.api
     }
     |> get_policies()
diff --git a/lib/ash/policy/check/accessing_from.ex b/lib/ash/policy/check/accessing_from.ex
index fa9be822..122e8843 100644
--- a/lib/ash/policy/check/accessing_from.ex
+++ b/lib/ash/policy/check/accessing_from.ex
@@ -29,4 +29,6 @@ defmodule Ash.Policy.Check.AccessingFrom do
       false
     end
   end
+
+  def match?(_, _, _), do: false
 end
diff --git a/lib/ash/policy/check/context_equals.ex b/lib/ash/policy/check/context_equals.ex
index 5447f312..c88f4d62 100644
--- a/lib/ash/policy/check/context_equals.ex
+++ b/lib/ash/policy/check/context_equals.ex
@@ -20,7 +20,8 @@ defmodule Ash.Policy.Check.ContextEquals do
 
   @impl true
   def match?(_, context, opts) do
-    changeset_or_query = Map.get(context, :changeset) || Map.get(context, :query)
+    changeset_or_query =
+      Map.get(context, :changeset) || Map.get(context, :query) || Map.get(context, :action_input)
 
     if is_nil(changeset_or_query) do
       false
diff --git a/lib/ash/resource/actions/action/action.ex b/lib/ash/resource/actions/action/action.ex
new file mode 100644
index 00000000..c95f3256
--- /dev/null
+++ b/lib/ash/resource/actions/action/action.ex
@@ -0,0 +1,60 @@
+defmodule Ash.Resource.Actions.Action do
+  @moduledoc "Represents a custom action on a resource."
+
+  defstruct [
+    :name,
+    :description,
+    :returns,
+    :run,
+    constraints: [],
+    touches_resources: [],
+    arguments: [],
+    transaction?: false,
+    primary?: false,
+    type: :action
+  ]
+
+  @type t :: %__MODULE__{
+          type: :action,
+          name: atom,
+          description: String.t() | nil,
+          arguments: [Ash.Resource.Actions.Argument.t()],
+          touches_resources: [Ash.Resource.t()],
+          constraints: Keyword.t(),
+          run: {module, Keyword.t()},
+          returns: Ash.Type.t(),
+          primary?: boolean,
+          transaction?: boolean
+        }
+
+  import Ash.Resource.Actions.SharedOptions
+
+  @global_opts shared_options()
+  @opt_schema [
+                returns: [
+                  type: Ash.OptionsHelpers.ash_type(),
+                  doc: "The return type of the action. See `Ash.Type` for more."
+                ],
+                constraints: [
+                  type: :keyword_list,
+                  doc: """
+                  Constraints for the return type.
+                  For more information see the specific type's documentation,
+                  for general type information see `Ash.Type` and
+                  for practical example [see the constraints topic](/documentation/topics/constraints.md).
+                  """
+                ],
+                run: [
+                  type:
+                    {:spark_function_behaviour, Ash.Resource.Actions.Implementation,
+                     {Ash.Resource.Action.ImplementationFunction, 2}}
+                ]
+              ]
+              |> Spark.OptionsHelpers.merge_schemas(
+                @global_opts,
+                "Action Options"
+              )
+
+  @doc false
+  def opt_schema, do: @opt_schema
+end
diff --git a/lib/ash/resource/actions/action/implementation.ex b/lib/ash/resource/actions/action/implementation.ex
new file mode 100644
index 00000000..2fff4ca9
--- /dev/null
+++ b/lib/ash/resource/actions/action/implementation.ex
@@ -0,0 +1,21 @@
+defmodule Ash.Resource.Actions.Implementation do
+  @moduledoc """
+  An implementation of a basic action.
+  """
+  @type context :: %{
+          optional(:actor) => term,
+          optional(:tenant) => term,
+          optional(:authorize?) => boolean,
+          optional(:api) => module,
+          optional(any) => any
+        }
+
+  @callback run(Ash.ActionInput.t(), opts :: Keyword.t(), context) ::
+              {:ok, term()} | {:ok, [Ash.Notifier.Notification.t()]} | {:error, term()}
+
+  defmacro __using__(_) do
+    quote do
+      @behaviour Ash.Resource.Actions.Implementation
+    end
+  end
+end
diff --git a/lib/ash/resource/actions/action/implementation_function.ex b/lib/ash/resource/actions/action/implementation_function.ex
new file mode 100644
index 00000000..9b91a34a
--- /dev/null
+++ b/lib/ash/resource/actions/action/implementation_function.ex
@@ -0,0 +1,12 @@
+defmodule Ash.Resource.Action.ImplementationFunction do
+  @moduledoc false
+  use Ash.Resource.Actions.Implementation
+
+  def run(input, [fun: {m, f, a}], context) do
+    apply(m, f, [input, context | a])
+  end
+
+  def run(input, [fun: fun], context) do
+    fun.(input, context)
+  end
+end
diff --git a/lib/ash/resource/actions/actions.ex b/lib/ash/resource/actions/actions.ex
index c5a23bfc..fdcc16d4 100644
--- a/lib/ash/resource/actions/actions.ex
+++ b/lib/ash/resource/actions/actions.ex
@@ -1,7 +1,7 @@
 defmodule Ash.Resource.Actions do
   @moduledoc "Types for Ash actions"
-  alias Ash.Resource.Actions.{Create, Destroy, Read, Update}
+  alias Ash.Resource.Actions.{Action, Create, Destroy, Read, Update}
 
-  @type action :: Create.t() | Read.t() | Update.t() | Destroy.t()
-  @type action_type :: :read | :create | :update | :destroy
+  @type action :: Action.t() | Create.t() | Read.t() | Update.t() | Destroy.t()
+  @type action_type :: :action | :read | :create | :update | :destroy
 end
diff --git a/lib/ash/resource/dsl.ex b/lib/ash/resource/dsl.ex
index fae20314..ab2e1307 100644
--- a/lib/ash/resource/dsl.ex
+++ b/lib/ash/resource/dsl.ex
@@ -433,6 +433,36 @@ defmodule Ash.Resource.Dsl do
     args: [:validation]
   }
 
+  @action %Spark.Dsl.Entity{
+    name: :action,
+    describe: """
+    Declares a basic action. A combination of arguments, a return type and a run function.
+
+    For calling this action, see the `Ash.Api` documentation.
+    """,
+    examples: [
+      """
+      action :top_user_emails do
+        argument :limit, :integer, default: 10, allow_nil?: false
+        returns {:array, :string}
+        run fn input, context ->
+          with {:ok, top_users} <- top_users(input.limit) do
+            {:ok, Enum.map(top_users, &(&1.email))}
+          end
+        end
+      end
+      """
+    ],
+    target: Ash.Resource.Actions.Action,
+    schema: Ash.Resource.Actions.Action.opt_schema(),
+    entities: [
+      arguments: [
+        @action_argument
+      ]
+    ],
+    args: [:name, :returns]
+  }
+
   @create %Spark.Dsl.Entity{
     name: :create,
     describe: """
@@ -671,6 +701,7 @@ defmodule Ash.Resource.Dsl do
       """
     ],
     entities: [
+      @action,
       @create,
       @read,
       @update,
diff --git a/lib/ash/resource/transformers/default_accept.ex b/lib/ash/resource/transformers/default_accept.ex
index b94e4a6b..db4d7850 100644
--- a/lib/ash/resource/transformers/default_accept.ex
+++ b/lib/ash/resource/transformers/default_accept.ex
@@ -21,6 +21,7 @@ defmodule Ash.Resource.Transformers.DefaultAccept do
 
     dsl_state
     |> Transformer.get_entities([:actions])
+    |> Enum.reject(&(&1.type == :action))
     |> Enum.reduce({:ok, dsl_state}, fn
       %{type: :read}, {:ok, _dsl_state} = acc ->
         acc
diff --git a/lib/ash/resource/transformers/validate_accept.ex b/lib/ash/resource/transformers/validate_accept.ex
index 19c10714..f4dc1bab 100644
--- a/lib/ash/resource/transformers/validate_accept.ex
+++ b/lib/ash/resource/transformers/validate_accept.ex
@@ -54,8 +54,7 @@ defmodule Ash.Resource.Transformers.ValidateAccept do
           )
         )
 
-      # read types do not have accept / reject fields
-      %{type: :read} ->
+      _ ->
         :ok
     end)
 
diff --git a/lib/ash/resource/transformers/validate_action_types_supported.ex b/lib/ash/resource/transformers/validate_action_types_supported.ex
index 567248cf..d2ab0c7e 100644
--- a/lib/ash/resource/transformers/validate_action_types_supported.ex
+++ b/lib/ash/resource/transformers/validate_action_types_supported.ex
@@ -12,7 +12,7 @@ defmodule Ash.Resource.Transformers.ValidateActionTypesSupported do
   def transform(dsl_state) do
     dsl_state
     |> Transformer.get_entities([:actions])
-    |> Enum.reject(&(&1.type == :read))
+    |> Enum.reject(&(&1.type in [:read, :action]))
     |> Enum.each(fn action ->
       data_layer = Transformer.get_persisted(dsl_state, :data_layer)
       resource = Transformer.get_persisted(dsl_state, :module)
diff --git a/lib/ash/resource/transformers/validate_manage_relationship_opts.ex b/lib/ash/resource/transformers/validate_manage_relationship_opts.ex
index a3406652..4f48dcdc 100644
--- a/lib/ash/resource/transformers/validate_manage_relationship_opts.ex
+++ b/lib/ash/resource/transformers/validate_manage_relationship_opts.ex
@@ -15,7 +15,7 @@ defmodule Ash.Resource.Transformers.ValidateManagedRelationshipOpts do
 
     dsl_state
     |> Transformer.get_entities([:actions])
-    |> Enum.reject(&(&1.type == :read))
+    |> Enum.reject(&(&1.type in [:read, :action]))
     |> Enum.each(fn action ->
       action.changes
       |> Enum.filter(
diff --git a/test/actions/basic_test.exs b/test/actions/basic_test.exs
new file mode 100644
index 00000000..18a944b0
--- /dev/null
+++ b/test/actions/basic_test.exs
@@ -0,0 +1,90 @@
+defmodule Ash.Test.Actions.BasicTest do
+  @moduledoc false
+  use ExUnit.Case, async: true
+
+  defmodule PassingFredOrGeorge do
+    use Ash.Policy.SimpleCheck
+
+    def describe(_), do: "is one of the twins"
+
+    def match?(_, %{action_input: action_input}, _) do
+      String.downcase(action_input.arguments.name) in ["fred", "george"]
+    end
+  end
+
+  defmodule Post do
+    @moduledoc false
+    use Ash.Resource, data_layer: Ash.DataLayer.Ets, authorizers: [Ash.Policy.Authorizer]
+
+    ets do
+      private?(true)
+    end
+
+    actions do
+      defaults [:create, :read, :update, :destroy]
+
+      action :hello, :string do
+        argument :name, :string, allow_nil?: false
+
+        run(fn input, _context ->
+          {:ok, "Hello #{input.arguments.name}"}
+        end)
+      end
+    end
+
+    attributes do
+      uuid_primary_key :id
+      attribute(:title, :string, allow_nil?: false)
+
+      timestamps()
+    end
+
+    policies do
+      policy action(:hello) do
+        authorize_if PassingFredOrGeorge
+      end
+    end
+  end
+
+  defmodule Registry do
+    @moduledoc false
+    use Ash.Registry
+
+    entries do
+      entry(Post)
+    end
+  end
+
+  defmodule Api do
+    @moduledoc false
+    use Ash.Api
+
+    resources do
+      registry Registry
+    end
+  end
+
+  describe "basic actions can be called" do
+    test "basic actions can be run" do
+      assert "Hello fred" =
+               Post
+               |> Ash.ActionInput.for_action(:hello, %{name: "fred"})
+               |> Api.run_action!()
+    end
+  end
+
+  describe "authorization" do
+    test "basic actions can be authorized" do
+      assert "Hello fred" =
+               Post
+               |> Ash.ActionInput.for_action(:hello, %{name: "fred"})
+               |> Api.run_action!(authorize?: true)
+
+      assert_raise Ash.Error.Forbidden, ~r/Forbidden/, fn ->
+        Post
+        |> Ash.ActionInput.for_action(:hello, %{name: "mike"})
+        |> Api.run_action!(authorize?: true)
+      end
+    end
+  end
+end
diff --git a/test/code_interface_test.exs b/test/code_interface_test.exs
index da1e9a28..91922de5 100644
--- a/test/code_interface_test.exs
+++ b/test/code_interface_test.exs
@@ -16,6 +16,7 @@ defmodule Ash.Test.CodeInterfaceTest do
       define :read_users, action: :read
       define :get_by_id, action: :read, get_by: [:id]
       define :create, args: [{:optional, :first_name}]
+      define :hello, args: [:name]
 
       define_calculation(:full_name, args: [:first_name, :last_name])
 
@@ -39,6 +40,14 @@ defmodule Ash.Test.CodeInterfaceTest do
 
         filter expr(id == ^arg(:id))
       end
+
+      action :hello, :string do
+        argument :name, :string, allow_nil?: false
+
+        run(fn input, _ ->
+          {:ok, "Hello #{input.arguments.name}"}
+        end)
+      end
     end
 
     calculations do
@@ -76,6 +85,13 @@ defmodule Ash.Test.CodeInterfaceTest do
     end
   end
 
+  describe "basic actions" do
+    test "basic actions can be invoked" do
+      assert "Hello fred" == User.hello!("fred")
+      assert {:ok, "Hello george"} == User.hello("george")
+    end
+  end
+
   describe "calculations" do
     test "calculation value can be fetched dynamically" do
       assert {:ok, "Zach Daniel"} =