diff --git a/lib/live_data.ex b/lib/live_data.ex
index dcd9a04..9888e10 100644
--- a/lib/live_data.ex
+++ b/lib/live_data.ex
@@ -86,7 +86,7 @@ defmodule LiveData do
   the list change order, any individual properties change, or other changes.
   """
 
-  alias LiveData.{Socket, Tracked}
+  alias LiveData.{Socket, Tracked, Async, AsyncResult}
 
   @type rendered :: any()
 
@@ -109,9 +109,83 @@ defmodule LiveData do
   end
 
   def assign(%Socket{assigns: assigns} = socket, key, value) do
+    validate_assign_key!(key)
     assigns = Map.put(assigns, key, value)
     %{socket | assigns: assigns}
   end
 
+  def assign(%Socket{} = socket, keyword_or_map)
+      when is_map(keyword_or_map) or is_list(keyword_or_map) do
+    Enum.reduce(keyword_or_map, socket, fn {key, value}, acc ->
+      assign(acc, key, value)
+    end)
+  end
+
+  def assign_new(%Socket{} = socket, key, fun) when is_function(fun, 1) do
+    validate_assign_key!(key)
+    assigns = assign_new(socket.assigns, key, fun)
+    %Socket{socket | assigns: assigns}
+  end
+
+  def assign_new(%{} = assigns, key, fun) when is_function(fun, 1) do
+    validate_assign_key!(key)
+
+    case assigns do
+      %{^key => _} -> assigns
+      _ -> Map.put_new(assigns, key, fun.(assigns))
+    end
+  end
+
+  @doc """
+  Assigns keys asynchronously.
+
+  The task is linked to the caller and errors are wrapped.
+  Each key passed to `assign_async/3` will be assigned to
+  an `%AsyncResult{}` struct holding the status of the operation
+  and the result when completed.
+
+  ## Examples
+
+      def mount(%{"slug" => slug}, socket) do
+        {:ok,
+         socket
+         |> assign(:foo, "bar")
+         |> assign_async(:org, fn -> {:ok, %{org: fetch_org!(slug)}} end)
+         |> assign_async([:profile, :rank], fn -> {:ok, %{profile: ..., rank: ...}} end)}
+      end
+
+  See the moduledoc for more information.
+  """
+  def assign_async(%Socket{} = socket, key_or_keys, func)
+      when (is_atom(key_or_keys) or is_list(key_or_keys)) and
+             is_function(func, 0) do
+    Async.assign_async(socket, key_or_keys, func)
+  end
+
+  defp validate_assign_key!(key) when is_atom(key), do: :ok
+
+  defp validate_assign_key!(key) do
+    raise ArgumentError, "assigns in LiveData must be atoms, got: #{inspect(key)}"
+  end
+
+  def async_result(%AsyncResult{} = async_assign, clauses) do
+    if !Keyword.keyword?(clauses) or
+         !Enum.all?(clauses, fn {key, _} -> key in [:ok, :loading, :failed] end) do
+      raise ArgumentError,
+            "invalid clauses, expected :ok, :loading, or :failed: #{inspect(clauses)}"
+    end
+
+    cond do
+      async_assign.ok? ->
+        clauses[:ok].(async_assign.result)
+
+      async_assign.loading ->
+        clauses[:loading].()
+
+      !is_nil(async_assign.failed) ->
+        clauses[:failed].(async_assign.result)
+    end
+  end
+
   def debug_prints?, do: Application.get_env(:live_data, :deft_compiler_debug_prints, false)
 end
diff --git a/lib/live_data/async.ex b/lib/live_data/async.ex
new file mode 100644
index 0000000..c838a6a
--- /dev/null
+++ b/lib/live_data/async.ex
@@ -0,0 +1,203 @@
+defmodule LiveData.Async do
+  @moduledoc false
+
+  alias LiveData.{AsyncResult, Socket, Channel}
+
+  def start_async(%Socket{} = socket, key, func)
+      when is_atom(key) and is_function(func, 0) do
+    run_async_task(socket, key, func, :start)
+  end
+
+  def assign_async(%Socket{} = socket, key_or_keys, func)
+      when (is_atom(key_or_keys) or is_list(key_or_keys)) and
+             is_function(func, 0) do
+    keys = List.wrap(key_or_keys)
+
+    # verifies result inside task
+    wrapped_func = fn ->
+      case func.() do
+        {:ok, %{} = assigns} ->
+          if Enum.find(keys, &(not is_map_key(assigns, &1))) do
+            raise ArgumentError, """
+            expected assign_async to return map of assigns for all keys
+            in #{inspect(keys)}, but got: #{inspect(assigns)}
+            """
+          else
+            {:ok, assigns}
+          end
+
+        {:error, reason} ->
+          {:error, reason}
+
+        other ->
+          raise ArgumentError, """
+          expected assign_async to return {:ok, map} of
+          assigns for #{inspect(keys)} or {:error, reason}, got: #{inspect(other)}
+          """
+      end
+    end
+
+    new_assigns =
+      Enum.map(keys, fn key ->
+        case socket.assigns do
+          %{^key => %AsyncResult{ok?: true} = existing} ->
+            {key, AsyncResult.loading(existing, keys)}
+
+          %{} ->
+            {key, AsyncResult.loading(keys)}
+        end
+      end)
+
+    socket
+    |> LiveData.assign(new_assigns)
+    |> run_async_task(keys, wrapped_func, :assign)
+  end
+
+  defp run_async_task(%Socket{} = socket, key, func, kind) do
+    lv_pid = self()
+    cid = cid(socket)
+    {:ok, pid} = Task.start_link(fn -> do_async(lv_pid, cid, key, func, kind) end)
+
+    ref =
+      :erlang.monitor(:process, pid, alias: :reply_demonitor, tag: {__MODULE__, key, cid, kind})
+
+    send(pid, {:context, ref})
+
+    update_private_async(socket, &Map.put(&1, key, {ref, pid, kind}))
+  end
+
+  defp do_async(lv_pid, cid, key, func, async_kind) do
+    receive do
+      {:context, ref} ->
+        try do
+          result = func.()
+          Channel.report_async_result(ref, async_kind, ref, cid, key, {:ok, result})
+        catch
+          catch_kind, reason ->
+            Process.unlink(lv_pid)
+            caught_result = to_exit(catch_kind, reason, __STACKTRACE__)
+            Channel.report_async_result(ref, async_kind, ref, cid, key, caught_result)
+            :erlang.raise(catch_kind, reason, __STACKTRACE__)
+        end
+    end
+  end
+
+  def cancel_async(%Socket{} = socket, %AsyncResult{} = result, reason) do
+    case result do
+      %AsyncResult{loading: keys} when is_list(keys) ->
+        new_assigns = for key <- keys, do: {key, AsyncResult.failed(result, {:exit, reason})}
+
+        socket
+        |> LiveData.assign(new_assigns)
+        |> cancel_async(keys, reason)
+
+      %AsyncResult{} ->
+        socket
+    end
+  end
+
+  def cancel_async(%Socket{} = socket, key, reason) do
+    case get_private_async(socket, key) do
+      {_ref, pid, _kind} when is_pid(pid) ->
+        Process.unlink(pid)
+        Process.exit(pid, reason)
+        update_private_async(socket, &Map.delete(&1, key))
+
+      nil ->
+        socket
+    end
+  end
+
+  def handle_async(socket, maybe_component, kind, key, ref, result) do
+    case prune_current_async(socket, key, ref) do
+      {:ok, pruned_socket} ->
+        handle_kind(pruned_socket, maybe_component, kind, key, result)
+
+      :error ->
+        socket
+    end
+  end
+
+  def handle_trap_exit(socket, maybe_component, kind, key, ref, reason) do
+    handle_async(socket, maybe_component, kind, key, ref, {:exit, reason})
+  end
+
+  defp handle_kind(socket, maybe_component, :start, key, result) do
+    callback_mod = maybe_component || socket.view
+
+    case callback_mod.handle_async(key, result, socket) do
+      {:noreply, %Socket{} = new_socket} ->
+        new_socket
+
+      other ->
+        raise ArgumentError, """
+        expected #{inspect(callback_mod)}.handle_async/3 to return {:noreply, socket}, got:
+
+            #{inspect(other)}
+        """
+    end
+  end
+
+  defp handle_kind(socket, _maybe_component, :assign, keys, result) do
+    case result do
+      {:ok, {:ok, %{} = assigns}} ->
+        new_assigns =
+          for {key, val} <- assigns do
+            {key, AsyncResult.ok(get_current_async!(socket, key), val)}
+          end
+
+        LiveData.assign(socket, new_assigns)
+
+      {:ok, {:error, reason}} ->
+        new_assigns =
+          for key <- keys do
+            {key, AsyncResult.failed(get_current_async!(socket, key), {:error, reason})}
+          end
+
+        LiveData.assign(socket, new_assigns)
+
+      {:exit, _reason} = normalized_exit ->
+        new_assigns =
+          for key <- keys do
+            {key, AsyncResult.failed(get_current_async!(socket, key), normalized_exit)}
+          end
+
+        LiveData.assign(socket, new_assigns)
+    end
+  end
+
+  # handle race of async being canceled and then reassigned
+  defp prune_current_async(socket, key, ref) do
+    case get_private_async(socket, key) do
+      {^ref, _pid, _kind} -> {:ok, update_private_async(socket, &Map.delete(&1, key))}
+      {_ref, _pid, _kind} -> :error
+      nil -> :error
+    end
+  end
+
+  defp update_private_async(%{private: private} = socket, func) do
+    existing = Map.get(private, :live_async, %{})
+    %{socket | private: Map.put(private, :live_async, func.(existing))}
+  end
+
+  defp get_private_async(%Socket{} = socket, key) do
+    socket.private[:live_async][key]
+  end
+
+  defp get_current_async!(socket, key) do
+    # handle case where assign is temporary and needs to be rebuilt
+    case socket.assigns do
+      %{^key => %AsyncResult{} = current_async} -> current_async
+      %{^key => _other} -> AsyncResult.loading(key)
+      %{} -> raise ArgumentError, "missing async assign #{inspect(key)}"
+    end
+  end
+
+  defp to_exit(:throw, reason, stack), do: {:exit, {{:nocatch, reason}, stack}}
+  defp to_exit(:error, reason, stack), do: {:exit, {reason, stack}}
+  defp to_exit(:exit, reason, _stack), do: {:exit, reason}
+
+  defp cid(%Socket{} = socket) do
+    if myself = socket.assigns[:myself], do: myself.cid
+  end
+end
diff --git a/lib/live_data/async_result.ex b/lib/live_data/async_result.ex
new file mode 100644
index 0000000..bd44c53
--- /dev/null
+++ b/lib/live_data/async_result.ex
@@ -0,0 +1,78 @@
+defmodule LiveData.AsyncResult do
+  @moduledoc ~S'''
+  Provides a data structure for tracking the state of an async assign.
+
+  See the `Async Operations` section of the `LiveData` docs for more information.
+
+  ## Fields
+
+    * `:ok?` - When true, indicates the `:result` has been set successfully at least once.
+    * `:loading` - The current loading state
+    * `:failed` - The current failed state
+    * `:result` - The successful result of the async task
+  '''
+
+  defstruct ok?: false,
+            loading: nil,
+            failed: nil,
+            result: nil
+
+  alias LiveData.AsyncResult
+
+  @doc """
+  Updates the loading state.
+
+  When loading, the failed state will be reset to `nil`.
+
+  ## Examples
+
+      AsyncResult.loading()
+      AsyncResult.loading(my_async)
+      AsyncResult.loading(my_async, %{my: :loading_state})
+  """
+  def loading do
+    %AsyncResult{loading: true}
+  end
+
+  def loading(%AsyncResult{} = result) do
+    %AsyncResult{result | loading: true, failed: nil}
+  end
+
+  def loading(loading_state) do
+    %AsyncResult{loading: loading_state, failed: nil}
+  end
+
+  def loading(%AsyncResult{} = result, loading_state) do
+    %AsyncResult{result | loading: loading_state, failed: nil}
+  end
+
+  @doc """
+  Updates the failed state.
+
+  When failed, the loading state will be reset to `nil`.
+
+  ## Examples
+
+      AsyncResult.failed(my_async, {:exit, :boom})
+      AsyncResult.failed(my_async, {:error, reason})
+  """
+  def failed(%AsyncResult{} = result, reason) do
+    %AsyncResult{result | failed: reason, loading: nil}
+  end
+
+  @doc """
+  Updates the successful result.
+
+  The `:ok?` field will also be set to `true` to indicate this result has
+  completed successfully at least once, regardless of future state changes.
+
+  When ok'd, the loading and failed state will be reset to `nil`.
+
+  ## Examples
+
+      AsyncResult.ok(my_async, my_result)
+  """
+  def ok(%AsyncResult{} = result, value) do
+    %AsyncResult{result | failed: nil, loading: nil, ok?: true, result: value}
+  end
+end
diff --git a/lib/live_data/channel.ex b/lib/live_data/channel.ex
index 8d45874..0fbb13e 100644
--- a/lib/live_data/channel.ex
+++ b/lib/live_data/channel.ex
@@ -8,7 +8,7 @@ defmodule LiveData.Channel do
   require Logger
 
   alias Phoenix.Socket.Message
-  alias LiveData.Socket
+  alias LiveData.{Socket, Async}
   alias LiveData.Tracked.RenderDiff
   alias LiveData.Tracked.Encoding
 
@@ -49,15 +49,15 @@ defmodule LiveData.Channel do
     mount(params, from, phx_socket)
   end
 
-  #@impl true
-  #def handle_call(msg, _from, socket) do
+  # @impl true
+  # def handle_call(msg, _from, socket) do
   #  IO.inspect(msg)
   #  true = false
-  #end
+  # end
 
-  #def handle_info({:DOWN, ref, _, _, _reason}, ref) do
+  # def handle_info({:DOWN, ref, _, _, _reason}, ref) do
   #  {:stop, {:shutdown, :closed}, ref}
-  #end
+  # end
 
   def handle_info(
         {:DOWN, _ref, _typ, transport_pid, _reason},
@@ -73,6 +73,14 @@ defmodule LiveData.Channel do
     {:noreply, state}
   end
 
+  def handle_info({@prefix, :async_result, {kind, info}}, state) do
+    {ref, _cid, keys, result} = info
+    socket = Async.handle_async(state.socket, nil, kind, keys, ref, result)
+    state = %{state | socket: socket}
+    state = render_view(state)
+    {:noreply, state}
+  end
+
   def handle_info(message, state) do
     {:ok, socket} = state.view.handle_info(message, state.socket)
     state = %{state | socket: socket}
@@ -83,6 +91,7 @@ defmodule LiveData.Channel do
   defp call_handler({module, function}, params) do
     apply(module, function, [params])
   end
+
   defp call_handler(fun, params) when is_function(fun, 1) do
     fun.(params)
   end
@@ -148,7 +157,7 @@ defmodule LiveData.Channel do
 
     state = %{state | tracked_state: tracked_state, encoding_state: encoding_state}
 
-    if LiveData.debug_prints?(), do: IO.inspect encoded_ops
+    if LiveData.debug_prints?(), do: IO.inspect(encoded_ops)
     state = push(state, "o", %{"o" => encoded_ops})
     state
   end
@@ -162,6 +171,11 @@ defmodule LiveData.Channel do
     end
   end
 
+  def report_async_result(monitor_ref, kind, ref, cid, keys, result)
+      when is_reference(monitor_ref) and kind in [:assign, :start] and is_reference(ref) do
+    send(monitor_ref, {@prefix, :async_result, {kind, {ref, cid, keys, result}}})
+  end
+
   defp push(state, event, payload) do
     message = %Message{topic: state.topic, event: event, payload: payload}
     send(state.socket.transport_pid, state.serializer.encode!(message))
@@ -173,7 +187,7 @@ defmodule LiveData.Channel do
       true -> nil
       false -> :code.ensure_loaded(module)
     end
+
     function_exported?(module, function, arity)
   end
-
 end
diff --git a/lib/live_data/socket.ex b/lib/live_data/socket.ex
index 92ef063..07c4294 100644
--- a/lib/live_data/socket.ex
+++ b/lib/live_data/socket.ex
@@ -3,6 +3,7 @@ defmodule LiveData.Socket do
 
   defstruct endpoint: nil,
             transport_pid: nil,
+            private: %{},
             assigns: %{}
 
   @type assigns :: map()
diff --git a/test/phoenix_live_data/full_async_test.exs b/test/phoenix_live_data/full_async_test.exs
new file mode 100644
index 0000000..4f104a3
--- /dev/null
+++ b/test/phoenix_live_data/full_async_test.exs
@@ -0,0 +1,12 @@
+defmodule LiveData.AsyncFullTest do
+  use ExUnit.Case, async: true
+
+  import LiveData.Test
+
+  test "testing" do
+    {:ok, view, data} = live_data(LiveData.Test.AsyncTestingData)
+    assert data == %{balance: "Loading..."}
+    Process.sleep(1)
+    assert render(view) == %{balance: 0}
+  end
+end
diff --git a/test/support/data/async_testing_data.ex b/test/support/data/async_testing_data.ex
new file mode 100644
index 0000000..2953eda
--- /dev/null
+++ b/test/support/data/async_testing_data.ex
@@ -0,0 +1,19 @@
+defmodule LiveData.Test.AsyncTestingData do
+  use LiveData
+
+  def mount(_params, socket) do
+    socket = assign_async(socket, :balance, fn -> {:ok, %{balance: 0}} end)
+    {:ok, socket}
+  end
+
+  deft render(assigns) do
+    %{
+      balance:
+        async_result(assigns[:balance],
+          ok: fn result -> result end,
+          loading: fn -> "Loading..." end,
+          failed: fn result -> "Failed: #{inspect(result)}" end
+        )
+    }
+  end
+end