diff --git a/godot-core/src/init/mod.rs b/godot-core/src/init/mod.rs
index 94cfaeb27..49aeeaeb0 100644
--- a/godot-core/src/init/mod.rs
+++ b/godot-core/src/init/mod.rs
@@ -172,6 +172,9 @@ fn gdext_on_level_deinit(level: InitLevel) {
         // If lowest level is unloaded, call global deinitialization.
         // No business logic by itself, but ensures consistency if re-initialization (hot-reload on Linux) occurs.
 
+        #[cfg(since_api = "4.2")]
+        crate::task::cleanup();
+
         // Garbage-collect various statics.
         // SAFETY: this is the last time meta APIs are used.
         unsafe {
diff --git a/godot-core/src/lib.rs b/godot-core/src/lib.rs
index 567d1258e..133121610 100644
--- a/godot-core/src/lib.rs
+++ b/godot-core/src/lib.rs
@@ -27,6 +27,10 @@ pub mod init;
 pub mod meta;
 pub mod obj;
 pub mod registry;
+#[cfg(since_api = "4.2")]
+pub mod task;
+#[cfg(before_api = "4.2")]
+pub mod task {}
 pub mod tools;
 
 mod storage;
diff --git a/godot-core/src/registry/signal/typed_signal.rs b/godot-core/src/registry/signal/typed_signal.rs
index 7f7901b13..fe44435e4 100644
--- a/godot-core/src/registry/signal/typed_signal.rs
+++ b/godot-core/src/registry/signal/typed_signal.rs
@@ -207,4 +207,8 @@ impl<'c, C: WithBaseField, Ps: meta::ParamTuple> TypedSignal<'c, C, Ps> {
             c.done();
         });
     }
+
+    pub(crate) fn to_untyped(&self) -> crate::builtin::Signal {
+        crate::builtin::Signal::from_object_signal(&self.receiver_object(), &*self.name)
+    }
 }
diff --git a/godot-core/src/task/async_runtime.rs b/godot-core/src/task/async_runtime.rs
new file mode 100644
index 000000000..5f683bcfd
--- /dev/null
+++ b/godot-core/src/task/async_runtime.rs
@@ -0,0 +1,485 @@
+/*
+ * Copyright (c) godot-rust; Bromeon and contributors.
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+use std::cell::RefCell;
+use std::future::Future;
+use std::marker::PhantomData;
+use std::panic::AssertUnwindSafe;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll, Wake, Waker};
+use std::thread::{self, LocalKey, ThreadId};
+
+use crate::builtin::{Callable, Variant};
+use crate::private::handle_panic;
+
+// ----------------------------------------------------------------------------------------------------------------------------------------------
+// Public interface
+
+/// Create a new async background task.
+///
+/// This function allows creating a new async task in which Godot signals can be awaited, like it is possible in GDScript. The
+/// [`TaskHandle`] that is returned provides synchronous introspection into the current state of the task.
+///
+/// Refer to [`Signal::to_future`](crate::builtin::Signal::to_future) and [`Signal::to_fallible_future`](crate::builtin::Signal::to_fallible_future)
+/// for details on how to await a signal.
+///
+/// # Panics
+/// - If called from any other thread than the main-thread.
+///
+/// # Example
+/// ```no_run
+/// # use godot::builtin::Signal;
+/// # use godot::classes::Node;
+/// # use godot::obj::NewAlloc;
+/// let node = Node::new_alloc();
+/// let signal = Signal::from_object_signal(&node, "signal");
+///
+/// godot::task::spawn(async move {
+///     println!("starting task...");
+///
+///     signal.to_future::<()>().await;
+///
+///     println!("node has changed: {}", node.get_name());
+/// });
+/// ```
+pub fn spawn(future: impl Future<Output = ()> + 'static) -> TaskHandle {
+    // Spawning new tasks is only allowed on the main thread for now.
+    // We can not accept Sync + Send futures since all object references (i.e. Gd<T>) are not thread-safe. So a future has to remain on the
+    // same thread it was created on. Godots signals on the other hand can be emitted on any thread, so it can't be guaranteed on which thread
+    // a future will be polled.
+    // By limiting async tasks to the main thread we can redirect all signal callbacks back to the main thread via `call_deferred`.
+    //
+    // Once thread-safe futures are possible the restriction can be lifted.
+    assert!(
+        crate::init::is_main_thread(),
+        "godot_task() can only be used on the main thread"
+    );
+
+    let (task_handle, godot_waker) = ASYNC_RUNTIME.with_runtime_mut(move |rt| {
+        let task_handle = rt.add_task(Box::pin(future));
+        let godot_waker = Arc::new(GodotWaker::new(
+            task_handle.index,
+            task_handle.id,
+            thread::current().id(),
+        ));
+
+        (task_handle, godot_waker)
+    });
+
+    poll_future(godot_waker);
+    task_handle
+}
+
+/// Handle for an active background task.
+///
+/// This handle provides introspection into the current state of the task, as well as providing a way to cancel it.
+///
+/// The associated task will **not** be canceled if this handle is dropped.
+pub struct TaskHandle {
+    index: usize,
+    id: u64,
+    _no_send_sync: PhantomData<*const ()>,
+}
+
+impl TaskHandle {
+    fn new(index: usize, id: u64) -> Self {
+        Self {
+            index,
+            id,
+            _no_send_sync: PhantomData,
+        }
+    }
+
+    /// Cancels the task if it is still pending and does nothing if it is already completed.
+    pub fn cancel(self) {
+        ASYNC_RUNTIME.with_runtime_mut(|rt| {
+            let Some(task) = rt.tasks.get(self.index) else {
+                // Getting the task from the runtime might return None if the runtime has already been deinitialized. In this case, we just
+                // ignore the cancel request, as the entire runtime has already been canceled.
+                return;
+            };
+
+            let alive = match task.value {
+                FutureSlotState::Empty => {
+                    panic!("Future slot is empty when canceling it! This is a bug!")
+                }
+                FutureSlotState::Gone => false,
+                FutureSlotState::Pending(_) => task.id == self.id,
+                FutureSlotState::Polling => panic!("Can not cancel future from inside it!"),
+            };
+
+            if !alive {
+                return;
+            }
+
+            rt.clear_task(self.index);
+        })
+    }
+
+    /// Synchronously checks if the task is still pending or has already completed.
+    pub fn is_pending(&self) -> bool {
+        ASYNC_RUNTIME.with_runtime(|rt| {
+            let slot = rt
+                .tasks
+                .get(self.index)
+                .unwrap_or_else(|| unreachable!("missing future slot at index {}", self.index));
+
+            if slot.id != self.id {
+                return false;
+            }
+
+            matches!(
+                slot.value,
+                FutureSlotState::Pending(_) | FutureSlotState::Polling
+            )
+        })
+    }
+}
+
+// ----------------------------------------------------------------------------------------------------------------------------------------------
+// Async Runtime
+
+const ASYNC_RUNTIME_DEINIT_PANIC_MESSAGE: &str = "The async runtime is being accessed after it has been deinitialized. This should not be possible and is most likely a bug.";
+
+thread_local! {
+    /// The thread local is only initialized the first time it's used. This means the async runtime won't be allocated until a task is
+    /// spawned.
+    static ASYNC_RUNTIME: RefCell<Option<AsyncRuntime>> = RefCell::new(Some(AsyncRuntime::new()));
+}
+
+/// Will be called during engine shutdown.
+///
+/// We have to drop all the remaining Futures during engine shutdown. This avoids them being dropped at process termination where they would
+/// try to access engine resources, which leads to SEGFAULTs.
+pub(crate) fn cleanup() {
+    ASYNC_RUNTIME.set(None);
+}
+
+#[cfg(feature = "trace")]
+pub fn has_godot_task_panicked(task_handle: TaskHandle) -> bool {
+    ASYNC_RUNTIME.with_runtime(|rt| rt.panicked_tasks.contains(&task_handle.id))
+}
+
+/// The current state of a future inside the async runtime.
+enum FutureSlotState<T> {
+    /// Slot is currently empty.
+    Empty,
+    /// Slot was previously occupied but the future has been canceled or the slot reused.
+    Gone,
+    /// Slot contains a pending future.
+    Pending(T),
+    /// Slot contains a future which is currently being polled.
+    Polling,
+}
+
+/// Wrapper around a future that is being stored in the async runtime.
+///
+/// This wrapper contains additional metadata for the async runtime.
+struct FutureSlot<T> {
+    value: FutureSlotState<T>,
+    id: u64,
+}
+
+impl<T> FutureSlot<T> {
+    /// Create a new slot with a pending future.
+    fn pending(id: u64, value: T) -> Self {
+        Self {
+            value: FutureSlotState::Pending(value),
+            id,
+        }
+    }
+
+    /// Checks if the future slot is either still empty or has become unoccupied due to a future completing.
+    fn is_empty(&self) -> bool {
+        matches!(self.value, FutureSlotState::Empty | FutureSlotState::Gone)
+    }
+
+    /// Drop the future from this slot.
+    ///
+    /// This transitions the slot into the [`FutureSlotState::Gone`] state.
+    fn clear(&mut self) {
+        self.value = FutureSlotState::Gone;
+    }
+
+    /// Attempts to extract the future with the given ID from the slot.
+    ///
+    /// Puts the slot into [`FutureSlotState::Polling`] state after taking the future out. It is expected that the future is either parked
+    /// again or the slot is cleared.
+    /// In cases were the slot state is not [`FutureSlotState::Pending`], a copy of the state is returned but the slot remains untouched.
+    fn take_for_polling(&mut self, id: u64) -> FutureSlotState<T> {
+        match self.value {
+            FutureSlotState::Empty => FutureSlotState::Empty,
+            FutureSlotState::Polling => FutureSlotState::Polling,
+            FutureSlotState::Gone => FutureSlotState::Gone,
+            FutureSlotState::Pending(_) if self.id != id => FutureSlotState::Gone,
+            FutureSlotState::Pending(_) => {
+                std::mem::replace(&mut self.value, FutureSlotState::Polling)
+            }
+        }
+    }
+
+    /// Parks the future in this slot again.
+    ///
+    /// # Panics
+    /// - If the slot is not in state [`FutureSlotState::Polling`].
+    fn park(&mut self, value: T) {
+        match self.value {
+            FutureSlotState::Empty | FutureSlotState::Gone => {
+                panic!("cannot park future in slot which is unoccupied")
+            }
+            FutureSlotState::Pending(_) => {
+                panic!(
+                    "cannot park future in slot, which is already occupied by a different future"
+                )
+            }
+            FutureSlotState::Polling => {
+                self.value = FutureSlotState::Pending(value);
+            }
+        }
+    }
+}
+
+/// The storage for the pending tasks of the async runtime.
+#[derive(Default)]
+struct AsyncRuntime {
+    tasks: Vec<FutureSlot<Pin<Box<dyn Future<Output = ()>>>>>,
+    next_task_id: u64,
+    #[cfg(feature = "trace")]
+    panicked_tasks: std::collections::HashSet<u64>,
+}
+
+impl AsyncRuntime {
+    fn new() -> Self {
+        Self {
+            // We only create a new async runtime inside a thread_local, which has lazy initialization on first use.
+            tasks: Vec::with_capacity(16),
+            next_task_id: 0,
+            #[cfg(feature = "trace")]
+            panicked_tasks: std::collections::HashSet::default(),
+        }
+    }
+
+    /// Get the next task ID.
+    fn next_id(&mut self) -> u64 {
+        let id = self.next_task_id;
+        self.next_task_id += 1;
+        id
+    }
+
+    /// Store a new async task in the runtime.
+    ///
+    /// First, a linear search is performed to locate an already existing but currently unoccupied slot in the task buffer. If there is no
+    /// free slot, a new slot is added which may grow the underlying [`Vec`].
+    ///
+    /// The future storage always starts out with a capacity of 10 tasks.
+    fn add_task<F: Future<Output = ()> + 'static>(&mut self, future: F) -> TaskHandle {
+        let id = self.next_id();
+        let index_slot = self
+            .tasks
+            // If we find an available slot, we will assign the new future to it.
+            .iter_mut()
+            .enumerate()
+            .find(|(_, slot)| slot.is_empty());
+
+        let boxed = Box::pin(future);
+
+        let index = match index_slot {
+            Some((index, slot)) => {
+                *slot = FutureSlot::pending(id, boxed);
+                index
+            }
+            None => {
+                self.tasks.push(FutureSlot::pending(id, boxed));
+                self.tasks.len() - 1
+            }
+        };
+
+        TaskHandle::new(index, id)
+    }
+
+    /// Extract a pending task from the storage.
+    ///
+    /// Attempts to extract a future with the given ID from the specified index and leaves the slot in state [`FutureSlotState::Polling`].
+    /// In cases were the slot state is not [`FutureSlotState::Pending`], a copy of the state is returned but the slot remains untouched.
+    fn take_task_for_polling(
+        &mut self,
+        index: usize,
+        id: u64,
+    ) -> FutureSlotState<Pin<Box<dyn Future<Output = ()> + 'static>>> {
+        let slot = self.tasks.get_mut(index);
+        slot.map(|inner| inner.take_for_polling(id))
+            .unwrap_or(FutureSlotState::Empty)
+    }
+
+    /// Remove a future from the storage and free up its slot.
+    ///
+    /// The slot is left in the [`FutureSlotState::Gone`] state.
+    fn clear_task(&mut self, index: usize) {
+        self.tasks[index].clear();
+    }
+
+    /// Move a future back into its slot.
+    ///
+    /// # Panic
+    /// - If the underlying slot is not in the [`FutureSlotState::Polling`] state.
+    fn park_task(&mut self, index: usize, future: Pin<Box<dyn Future<Output = ()>>>) {
+        self.tasks[index].park(future);
+    }
+
+    /// Track that a future caused a panic.
+    ///
+    /// This is only available for itest.
+    #[cfg(feature = "trace")]
+    fn track_panic(&mut self, task_id: u64) {
+        self.panicked_tasks.insert(task_id);
+    }
+}
+
+trait WithRuntime {
+    fn with_runtime<R>(&'static self, f: impl FnOnce(&AsyncRuntime) -> R) -> R;
+    fn with_runtime_mut<R>(&'static self, f: impl FnOnce(&mut AsyncRuntime) -> R) -> R;
+}
+
+impl WithRuntime for LocalKey<RefCell<Option<AsyncRuntime>>> {
+    fn with_runtime<R>(&'static self, f: impl FnOnce(&AsyncRuntime) -> R) -> R {
+        self.with_borrow(|rt| {
+            let rt_ref = rt.as_ref().expect(ASYNC_RUNTIME_DEINIT_PANIC_MESSAGE);
+
+            f(rt_ref)
+        })
+    }
+
+    fn with_runtime_mut<R>(&'static self, f: impl FnOnce(&mut AsyncRuntime) -> R) -> R {
+        self.with_borrow_mut(|rt| {
+            let rt_ref = rt.as_mut().expect(ASYNC_RUNTIME_DEINIT_PANIC_MESSAGE);
+
+            f(rt_ref)
+        })
+    }
+}
+
+/// Use a godot waker to poll it's associated future.
+///
+/// # Panics
+/// - If called from a thread other than the main-thread.
+fn poll_future(godot_waker: Arc<GodotWaker>) {
+    let current_thread = thread::current().id();
+
+    assert_eq!(
+        godot_waker.thread_id,
+        current_thread,
+        "trying to poll future on a different thread!\n  Current thread: {:?}\n  Future thread: {:?}",
+        current_thread,
+        godot_waker.thread_id,
+    );
+
+    let waker = Waker::from(godot_waker.clone());
+    let mut ctx = Context::from_waker(&waker);
+
+    // Move future out of the runtime while we are polling it to avoid holding a mutable reference for the entire runtime.
+    let future = ASYNC_RUNTIME.with_runtime_mut(|rt| {
+        match rt.take_task_for_polling(godot_waker.runtime_index, godot_waker.task_id) {
+            FutureSlotState::Empty => {
+                panic!("Future slot is empty when waking it! This is a bug!");
+            }
+
+            FutureSlotState::Gone => None,
+
+            FutureSlotState::Polling => {
+                unreachable!("the same GodotWaker has been called recursively");
+            }
+
+            FutureSlotState::Pending(future) => Some(future),
+        }
+    });
+
+    let Some(future) = future else {
+        // Future has been canceled while the waker was already triggered.
+        return;
+    };
+
+    let error_context = || "Godot async task failed";
+
+    // If Future::poll() panics, the future is immediately dropped and cannot be accessed again,
+    // thus any state that may not have been unwind-safe cannot be observed later.
+    let mut future = AssertUnwindSafe(future);
+
+    let panic_result = handle_panic(error_context, move || {
+        (future.as_mut().poll(&mut ctx), future)
+    });
+
+    let Ok((poll_result, future)) = panic_result else {
+        // Polling the future caused a panic. The task state has to be cleaned up and we want track the panic if the trace feature is enabled.
+        ASYNC_RUNTIME.with_runtime_mut(|rt| {
+            #[cfg(feature = "trace")]
+            rt.track_panic(godot_waker.task_id);
+            rt.clear_task(godot_waker.runtime_index);
+        });
+
+        return;
+    };
+
+    // Update the state of the Future in the runtime.
+    ASYNC_RUNTIME.with_runtime_mut(|rt| match poll_result {
+        // Future is still pending, so we park it again.
+        Poll::Pending => rt.park_task(godot_waker.runtime_index, future.0),
+
+        // Future has resolved, so we remove it from the runtime.
+        Poll::Ready(()) => rt.clear_task(godot_waker.runtime_index),
+    });
+}
+
+/// Implementation of a [`Waker`] to poll futures with the engine.
+struct GodotWaker {
+    runtime_index: usize,
+    task_id: u64,
+    thread_id: ThreadId,
+}
+
+impl GodotWaker {
+    fn new(index: usize, task_id: u64, thread_id: ThreadId) -> Self {
+        Self {
+            runtime_index: index,
+            thread_id,
+            task_id,
+        }
+    }
+}
+
+// Uses a deferred callable to poll the associated future, i.e. at the end of the current frame.
+impl Wake for GodotWaker {
+    fn wake(self: Arc<Self>) {
+        let mut waker = Some(self);
+
+        /// Enforce the passed closure is generic over its lifetime. The compiler gets confused about the livetime of the argument otherwise.
+        /// This appears to be a common issue: https://github.com/rust-lang/rust/issues/89976
+        fn callback_type_hint<F>(f: F) -> F
+        where
+            F: for<'a> FnMut(&'a [&Variant]) -> Result<Variant, ()>,
+        {
+            f
+        }
+
+        #[cfg(not(feature = "experimental-threads"))]
+        let create_callable = Callable::from_local_fn;
+
+        #[cfg(feature = "experimental-threads")]
+        let create_callable = Callable::from_sync_fn;
+
+        let callable = create_callable(
+            "GodotWaker::wake",
+            callback_type_hint(move |_args| {
+                poll_future(waker.take().expect("Callable will never be called again"));
+                Ok(Variant::nil())
+            }),
+        );
+
+        // Schedule waker to poll the Future at the end of the frame.
+        callable.call_deferred(&[]);
+    }
+}
diff --git a/godot-core/src/task/futures.rs b/godot-core/src/task/futures.rs
new file mode 100644
index 000000000..7a13417fc
--- /dev/null
+++ b/godot-core/src/task/futures.rs
@@ -0,0 +1,330 @@
+/*
+ * Copyright (c) godot-rust; Bromeon and contributors.
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+use core::panic;
+use std::fmt::Display;
+use std::future::{Future, IntoFuture};
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll, Waker};
+
+use crate::builtin::{Callable, RustCallable, Signal, Variant};
+use crate::classes::object::ConnectFlags;
+use crate::meta::ParamTuple;
+use crate::obj::{EngineBitfield, WithBaseField};
+use crate::registry::signal::TypedSignal;
+
+/// The panicking counter part to the [`FallibleSignalFuture`].
+///
+/// This future works in the same way as `FallibleSignalFuture`, but panics when the signal object is freed, instead of resolving to a
+/// [`Result::Err`].
+pub struct SignalFuture<R: ParamTuple + Sync + Send>(FallibleSignalFuture<R>);
+
+impl<R: ParamTuple + Sync + Send> SignalFuture<R> {
+    fn new(signal: Signal) -> Self {
+        Self(FallibleSignalFuture::new(signal))
+    }
+}
+
+impl<R: ParamTuple + Sync + Send> Future for SignalFuture<R> {
+    type Output = R;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let poll_result = self.get_mut().0.poll(cx);
+
+        match poll_result {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Ok(value)) => Poll::Ready(value),
+            Poll::Ready(Err(FallibleSignalFutureError)) => panic!(
+                "the signal object of a SignalFuture was freed, while the future was still waiting for the signal to be emitted"
+            ),
+        }
+    }
+}
+
+// Not derived, otherwise an extra bound `Output: Default` is required.
+struct SignalFutureData<T> {
+    state: SignalFutureState<T>,
+    waker: Option<Waker>,
+}
+
+impl<T> Default for SignalFutureData<T> {
+    fn default() -> Self {
+        Self {
+            state: Default::default(),
+            waker: None,
+        }
+    }
+}
+
+// Only public for itest.
+#[cfg_attr(feature = "trace", derive(Default))]
+pub struct SignalFutureResolver<R> {
+    data: Arc<Mutex<SignalFutureData<R>>>,
+}
+
+impl<R> Clone for SignalFutureResolver<R> {
+    fn clone(&self) -> Self {
+        Self {
+            data: self.data.clone(),
+        }
+    }
+}
+
+impl<R> SignalFutureResolver<R> {
+    fn new(data: Arc<Mutex<SignalFutureData<R>>>) -> Self {
+        Self { data }
+    }
+}
+
+impl<R> std::hash::Hash for SignalFutureResolver<R> {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        state.write_usize(Arc::as_ptr(&self.data) as usize);
+    }
+}
+
+impl<R> PartialEq for SignalFutureResolver<R> {
+    fn eq(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.data, &other.data)
+    }
+}
+
+impl<R: ParamTuple + Sync + Send> RustCallable for SignalFutureResolver<R> {
+    fn invoke(&mut self, args: &[&Variant]) -> Result<Variant, ()> {
+        let waker = {
+            let mut data = self.data.lock().unwrap();
+            data.state = SignalFutureState::Ready(R::from_variant_array(args));
+
+            // We no longer need the waker after we resolved. If the future is polled again, we'll also get a new waker.
+            data.waker.take()
+        };
+
+        if let Some(waker) = waker {
+            waker.wake();
+        }
+
+        Ok(Variant::nil())
+    }
+}
+
+impl<R> Display for SignalFutureResolver<R> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "SignalFutureResolver::<{}>", std::any::type_name::<R>())
+    }
+}
+
+// This resolver will change the futures state when it's being dropped (i.e. the engine removes all connected signal callables). By marking
+// the future as dead we can resolve it to an error value the next time it gets polled.
+impl<R> Drop for SignalFutureResolver<R> {
+    fn drop(&mut self) {
+        let mut data = self.data.lock().unwrap();
+
+        if !matches!(data.state, SignalFutureState::Pending) {
+            // The future is no longer pending, so no clean up is required.
+            return;
+        }
+
+        // We mark the future as dead, so the next time it gets polled we can react to it's inability to resolve.
+        data.state = SignalFutureState::Dead;
+
+        // If we got a waker we trigger it to get the future polled. If there is no waker, then the future has not been polled yet and we
+        // simply wait for the runtime to perform the first poll.
+        if let Some(ref waker) = data.waker {
+            waker.wake_by_ref();
+        }
+    }
+}
+
+#[derive(Default)]
+enum SignalFutureState<T> {
+    #[default]
+    Pending,
+    Ready(T),
+    Dead,
+    Dropped,
+}
+
+impl<T> SignalFutureState<T> {
+    fn take(&mut self) -> Self {
+        let new_value = match self {
+            Self::Pending => Self::Pending,
+            Self::Ready(_) | Self::Dead => Self::Dead,
+            Self::Dropped => Self::Dropped,
+        };
+
+        std::mem::replace(self, new_value)
+    }
+}
+
+/// A future that tries to resolve as soon as the provided Godot signal was emitted.
+///
+/// The future might resolve to an error if the signal object is freed before the signal is emitted.
+pub struct FallibleSignalFuture<R: ParamTuple + Sync + Send> {
+    data: Arc<Mutex<SignalFutureData<R>>>,
+    callable: SignalFutureResolver<R>,
+    signal: Signal,
+}
+
+impl<R: ParamTuple + Sync + Send> FallibleSignalFuture<R> {
+    fn new(signal: Signal) -> Self {
+        debug_assert!(
+            !signal.is_null(),
+            "Failed to create a future for an invalid Signal!\nEither the signal object was already freed or the signal was not registered in the object before using it.",
+        );
+
+        let data = Arc::new(Mutex::new(SignalFutureData::default()));
+
+        // The callable currently requires that the return value is Sync + Send.
+        let callable = SignalFutureResolver::new(data.clone());
+
+        signal.connect(
+            &Callable::from_custom(callable.clone()),
+            ConnectFlags::ONE_SHOT.ord() as i64,
+        );
+
+        Self {
+            data,
+            callable,
+            signal,
+        }
+    }
+    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Result<R, FallibleSignalFutureError>> {
+        let mut data = self.data.lock().unwrap();
+
+        data.waker.replace(cx.waker().clone());
+
+        let value = data.state.take();
+
+        match value {
+            SignalFutureState::Pending => Poll::Pending,
+            SignalFutureState::Dropped => unreachable!(),
+            SignalFutureState::Dead => Poll::Ready(Err(FallibleSignalFutureError)),
+            SignalFutureState::Ready(value) => Poll::Ready(Ok(value)),
+        }
+    }
+}
+
+/// Error that might be returned  by the [`FallibleSignalFuture`].
+///
+/// This error is being resolved to when the signal object is freed before the awaited singal is emitted.
+#[derive(Debug)]
+pub struct FallibleSignalFutureError;
+
+impl Display for FallibleSignalFutureError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "The signal object was freed before the awaited signal was emitted"
+        )
+    }
+}
+
+impl std::error::Error for FallibleSignalFutureError {}
+
+impl<R: ParamTuple + Sync + Send> Future for FallibleSignalFuture<R> {
+    type Output = Result<R, FallibleSignalFutureError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.get_mut().poll(cx)
+    }
+}
+
+impl<R: ParamTuple + Sync + Send> Drop for FallibleSignalFuture<R> {
+    fn drop(&mut self) {
+        // The callable might alredy be destroyed, this occurs during engine shutdown.
+        if self.signal.object().is_none() {
+            return;
+        }
+
+        let mut data_lock = self.data.lock().unwrap();
+
+        data_lock.state = SignalFutureState::Dropped;
+
+        drop(data_lock);
+
+        // We create a new Godot Callable from our RustCallable so we get independent reference counting.
+        let gd_callable = Callable::from_custom(self.callable.clone());
+
+        // is_connected will return true if the signal was never emited before the future is dropped.
+        if self.signal.is_connected(&gd_callable) {
+            self.signal.disconnect(&gd_callable);
+        }
+    }
+}
+
+impl Signal {
+    /// Creates a fallible future for this signal.
+    ///
+    /// The future will resolve the next time the signal is emitted.
+    /// See [`TrySignalFuture`] for details.
+    ///
+    /// Since the `Signal` type does not contain information on the signal argument types, the future output type has to be inferred from
+    /// the call to this function.
+    pub fn to_fallible_future<R: ParamTuple + Sync + Send>(&self) -> FallibleSignalFuture<R> {
+        FallibleSignalFuture::new(self.clone())
+    }
+
+    /// Creates a future for this signal.
+    ///
+    /// The future will resolve the next time the signal is emitted, but might panic if the signal object is freed.
+    /// See [`SignalFuture`] for details.
+    ///
+    /// Since the `Signal` type does not contain information on the signal argument types, the future output type has to be inferred from
+    /// the call to this function.
+    pub fn to_future<R: ParamTuple + Sync + Send>(&self) -> SignalFuture<R> {
+        SignalFuture::new(self.clone())
+    }
+}
+
+impl<C: WithBaseField, R: ParamTuple + Sync + Send> TypedSignal<'_, C, R> {
+    /// Creates a fallible future for this signal.
+    ///
+    /// The future will resolve the next time the signal is emitted.
+    /// See [`FallibleSignalFuture`] for details.
+    pub fn to_fallible_future(&self) -> FallibleSignalFuture<R> {
+        FallibleSignalFuture::new(self.to_untyped())
+    }
+
+    /// Creates a future for this signal.
+    ///
+    /// The future will resolve the next time the signal is emitted, but might panic if the signal object is freed.
+    /// See [`SignalFuture`] for details.
+    pub fn to_future(&self) -> SignalFuture<R> {
+        SignalFuture::new(self.to_untyped())
+    }
+}
+
+impl<C: WithBaseField, R: ParamTuple + Sync + Send> IntoFuture for &TypedSignal<'_, C, R> {
+    type Output = R;
+
+    type IntoFuture = SignalFuture<R>;
+
+    fn into_future(self) -> Self::IntoFuture {
+        self.to_future()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::sys;
+    use std::sync::Arc;
+
+    use super::SignalFutureResolver;
+
+    /// Test that the hash of a cloned future resolver is equal to its original version. With this equality in place, we can create new
+    /// Callables that are equal to their original version but have separate reference counting.
+    #[test]
+    fn future_resolver_cloned_hash() {
+        let resolver_a = SignalFutureResolver::<u8>::new(Arc::default());
+        let resolver_b = resolver_a.clone();
+
+        let hash_a = sys::hash_value(&resolver_a);
+        let hash_b = sys::hash_value(&resolver_b);
+
+        assert_eq!(hash_a, hash_b);
+    }
+}
diff --git a/godot-core/src/task/mod.rs b/godot-core/src/task/mod.rs
new file mode 100644
index 000000000..0834fbdf4
--- /dev/null
+++ b/godot-core/src/task/mod.rs
@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) godot-rust; Bromeon and contributors.
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+//! Integrates async rust code with the engine.
+//!
+//! This module contains:
+//! - Implementations of [`Future`](std::future::Future) for [`Signal`](crate::builtin::Signal) and [`TypedSignal`](crate::registry::signal::TypedSignal).
+//! - A way to [`spawn`] new async tasks by using the engine as the async runtime.
+
+mod async_runtime;
+mod futures;
+
+pub(crate) use async_runtime::cleanup;
+
+pub use async_runtime::{spawn, TaskHandle};
+pub use futures::{FallibleSignalFuture, FallibleSignalFutureError, SignalFuture};
+
+// Only exported for itest.
+#[cfg(feature = "trace")]
+pub use async_runtime::has_godot_task_panicked;
+#[cfg(feature = "trace")]
+pub use futures::SignalFutureResolver;
diff --git a/godot-macros/src/itest.rs b/godot-macros/src/itest.rs
index d0d23b696..6c0bb06bb 100644
--- a/godot-macros/src/itest.rs
+++ b/godot-macros/src/itest.rs
@@ -8,7 +8,7 @@
 use proc_macro2::TokenStream;
 use quote::{quote, ToTokens};
 
-use crate::util::{bail, path_ends_with, KvParser};
+use crate::util::{bail, extract_typename, ident, path_ends_with, KvParser};
 use crate::ParseResult;
 
 pub fn attribute_itest(input_item: venial::Item) -> ParseResult<TokenStream> {
@@ -17,20 +17,21 @@ pub fn attribute_itest(input_item: venial::Item) -> ParseResult<TokenStream> {
         _ => return bail!(&input_item, "#[itest] can only be applied to functions"),
     };
 
+    let mut attr = KvParser::parse_required(&func.attributes, "itest", &func.name)?;
+    let skipped = attr.handle_alone("skip")?;
+    let focused = attr.handle_alone("focus")?;
+    let is_async = attr.handle_alone("async")?;
+    attr.finish()?;
+
     // Note: allow attributes for things like #[rustfmt] or #[clippy]
     if func.generic_params.is_some()
         || func.params.len() > 1
-        || func.return_ty.is_some()
+        || (func.return_ty.is_some() && !is_async)
         || func.where_clause.is_some()
     {
         return bad_signature(&func);
     }
 
-    let mut attr = KvParser::parse_required(&func.attributes, "itest", &func.name)?;
-    let skipped = attr.handle_alone("skip")?;
-    let focused = attr.handle_alone("focus")?;
-    attr.finish()?;
-
     if skipped && focused {
         return bail!(
             func.name,
@@ -47,9 +48,13 @@ pub fn attribute_itest(input_item: venial::Item) -> ParseResult<TokenStream> {
             // Correct parameter type (crude macro check) -> reuse parameter name
             if path_ends_with(&param.ty.tokens, "TestContext") {
                 param.to_token_stream()
+            } else if is_async {
+                return bad_async_signature(&func);
             } else {
                 return bad_signature(&func);
             }
+        } else if is_async {
+            return bad_async_signature(&func);
         } else {
             return bad_signature(&func);
         }
@@ -57,14 +62,35 @@ pub fn attribute_itest(input_item: venial::Item) -> ParseResult<TokenStream> {
         quote! { __unused_context: &crate::framework::TestContext }
     };
 
+    if is_async
+        && func
+            .return_ty
+            .as_ref()
+            .and_then(extract_typename)
+            .map_or(true, |segment| segment.ident != "TaskHandle")
+    {
+        return bad_async_signature(&func);
+    }
+
     let body = &func.body;
 
+    let (return_tokens, test_case_ty, plugin_name);
+    if is_async {
+        return_tokens = quote! { -> TaskHandle };
+        test_case_ty = quote! { crate::framework::AsyncRustTestCase };
+        plugin_name = ident("__GODOT_ASYNC_ITEST");
+    } else {
+        return_tokens = TokenStream::new();
+        test_case_ty = quote! { crate::framework::RustTestCase };
+        plugin_name = ident("__GODOT_ITEST");
+    };
+
     Ok(quote! {
-        pub fn #test_name(#param) {
+        pub fn #test_name(#param) #return_tokens {
             #body
         }
 
-        ::godot::sys::plugin_add!(crate::framework::__GODOT_ITEST; crate::framework::RustTestCase {
+        ::godot::sys::plugin_add!(crate::framework::#plugin_name; #test_case_ty {
             name: #test_name_str,
             skipped: #skipped,
             focused: #focused,
@@ -84,3 +110,13 @@ fn bad_signature(func: &venial::Function) -> Result<TokenStream, venial::Error>
         f = func.name,
     )
 }
+
+fn bad_async_signature(func: &venial::Function) -> Result<TokenStream, venial::Error> {
+    bail!(
+        func,
+        "#[itest(async)] function must have one of these signatures:\
+        \n  fn {f}() -> TaskHandle {{ ... }}\
+        \n  fn {f}(ctx: &TestContext) -> TaskHandle {{ ... }}",
+        f = func.name,
+    )
+}
diff --git a/godot/src/lib.rs b/godot/src/lib.rs
index c8409382d..52a84c73d 100644
--- a/godot/src/lib.rs
+++ b/godot/src/lib.rs
@@ -35,6 +35,7 @@
 //! * [`tools`], higher-level utilities that extend the generated code, e.g. `load<T>()`.
 //! * [`meta`], fundamental information about types, properties and conversions.
 //! * [`init`], entry point and global library configuration.
+//! * [`task`], integration with async code.
 //!
 //! The [`prelude`] contains often-imported symbols; feel free to `use godot::prelude::*` in your code.
 //! <br><br>
@@ -160,7 +161,7 @@ compile_error!("The feature `double-precision` currently requires `api-custom` d
 // Modules
 
 #[doc(inline)]
-pub use godot_core::{builtin, classes, global, meta, obj, tools};
+pub use godot_core::{builtin, classes, global, meta, obj, task, tools};
 
 #[doc(hidden)]
 pub use godot_core::possibly_docs as docs;
diff --git a/itest/godot/TestRunner.gd b/itest/godot/TestRunner.gd
index 0f204863a..0f27f059d 100644
--- a/itest/godot/TestRunner.gd
+++ b/itest/godot/TestRunner.gd
@@ -66,20 +66,24 @@ func _ready():
 
 	var property_tests = load("res://gen/GenPropertyTests.gd").new()
 
-	var success: bool = rust_runner.run_all_tests(
+	# Run benchmarks after all synchronous and asynchronous tests have completed.
+	var run_benchmarks = func (success: bool):
+		if success:
+			rust_runner.run_all_benchmarks(self)
+
+		var exit_code: int = 0 if success else 1
+		get_tree().quit(exit_code)
+
+	rust_runner.run_all_tests(
 		gdscript_tests,
 		gdscript_suites.size(),
 		allow_focus,
 		self,
 		filters,
-		property_tests
+		property_tests,
+		run_benchmarks
 	)
 
-	if success:
-		rust_runner.run_all_benchmarks(self)
-
-	var exit_code: int = 0 if success else 1
-	get_tree().quit(exit_code)
 
 
 class GDScriptTestCase:
diff --git a/itest/rust/src/engine_tests/async_test.rs b/itest/rust/src/engine_tests/async_test.rs
new file mode 100644
index 000000000..3e62d5ede
--- /dev/null
+++ b/itest/rust/src/engine_tests/async_test.rs
@@ -0,0 +1,112 @@
+/*
+ * Copyright (c) godot-rust; Bromeon and contributors.
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+use std::ops::Deref;
+
+use godot::builtin::{Callable, Signal, Variant};
+use godot::classes::{Object, RefCounted};
+use godot::meta::ToGodot;
+use godot::obj::{Base, Gd, NewAlloc, NewGd};
+use godot::prelude::{godot_api, GodotClass};
+use godot::task::{self, SignalFuture, SignalFutureResolver, TaskHandle};
+
+use crate::framework::{itest, TestContext};
+
+#[derive(GodotClass)]
+#[class(init)]
+struct AsyncRefCounted {
+    base: Base<RefCounted>,
+}
+
+#[godot_api]
+impl AsyncRefCounted {
+    #[signal]
+    fn custom_signal(value: u32);
+}
+
+#[itest(async)]
+fn start_async_task() -> TaskHandle {
+    let mut object = RefCounted::new_gd();
+    let object_ref = object.clone();
+    let signal = Signal::from_object_signal(&object, "custom_signal");
+
+    object.add_user_signal("custom_signal");
+
+    let task_handle = task::spawn(async move {
+        let signal_future: SignalFuture<(u8,)> = signal.to_future();
+        let (result,) = signal_future.await;
+
+        assert_eq!(result, 10);
+        drop(object_ref);
+    });
+
+    object.emit_signal("custom_signal", &[10.to_variant()]);
+
+    task_handle
+}
+
+#[itest]
+fn cancel_async_task(ctx: &TestContext) {
+    let tree = ctx.scene_tree.get_tree().unwrap();
+    let signal = Signal::from_object_signal(&tree, "process_frame");
+
+    let handle = task::spawn(async move {
+        let _: () = signal.to_future().await;
+
+        unreachable!();
+    });
+
+    handle.cancel();
+}
+
+#[itest(async)]
+fn async_task_fallible_signal_future() -> TaskHandle {
+    let mut obj = Object::new_alloc();
+
+    let signal = Signal::from_object_signal(&obj, "script_changed");
+
+    let handle = task::spawn(async move {
+        let result = signal.to_fallible_future::<()>().await;
+
+        assert!(result.is_err());
+    });
+
+    obj.call_deferred("free", &[]);
+
+    handle
+}
+
+// Test that two callables created from the same future resolver (but cloned) are equal, while they are not equal to an unrelated
+// callable.
+#[itest]
+fn resolver_callabable_equality() {
+    let resolver = SignalFutureResolver::<(u8,)>::default();
+
+    let callable = Callable::from_custom(resolver.clone());
+    let cloned_callable = Callable::from_custom(resolver.clone());
+    let unrelated_callable = Callable::from_local_fn("unrelated", |_| Ok(Variant::nil()));
+
+    assert_eq!(callable, cloned_callable);
+    assert_ne!(callable, unrelated_callable);
+    assert_ne!(cloned_callable, unrelated_callable);
+}
+
+#[itest(async)]
+fn async_typed_signal() -> TaskHandle {
+    let object = AsyncRefCounted::new_gd();
+    let object_ref = object.clone();
+
+    let task_handle = task::spawn(async move {
+        let (result,) = object.signals().custom_signal().deref().await;
+
+        assert_eq!(result, 66);
+    });
+
+    object_ref.signals().custom_signal().emit(66);
+
+    task_handle
+}
diff --git a/itest/rust/src/engine_tests/mod.rs b/itest/rust/src/engine_tests/mod.rs
index 3419b5599..f20a21160 100644
--- a/itest/rust/src/engine_tests/mod.rs
+++ b/itest/rust/src/engine_tests/mod.rs
@@ -5,6 +5,8 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/.
  */
 
+#[cfg(since_api = "4.2")]
+mod async_test;
 mod codegen_enums_test;
 mod codegen_test;
 mod engine_enum_test;
diff --git a/itest/rust/src/framework/mod.rs b/itest/rust/src/framework/mod.rs
index f7f519ae7..d581c9ecf 100644
--- a/itest/rust/src/framework/mod.rs
+++ b/itest/rust/src/framework/mod.rs
@@ -24,10 +24,12 @@ pub use godot::test::{bench, itest};
 
 // Registers all the `#[itest]` tests and `#[bench]` benchmarks.
 sys::plugin_registry!(pub(crate) __GODOT_ITEST: RustTestCase);
+#[cfg(since_api = "4.2")]
+sys::plugin_registry!(pub(crate) __GODOT_ASYNC_ITEST: AsyncRustTestCase);
 sys::plugin_registry!(pub(crate) __GODOT_BENCH: RustBenchmark);
 
 /// Finds all `#[itest]` tests.
-fn collect_rust_tests(filters: &[String]) -> (Vec<RustTestCase>, usize, bool) {
+fn collect_rust_tests(filters: &[String]) -> (Vec<RustTestCase>, HashSet<&str>, bool) {
     let mut all_files = HashSet::new();
     let mut tests: Vec<RustTestCase> = vec![];
     let mut is_focus_run = false;
@@ -50,7 +52,38 @@ fn collect_rust_tests(filters: &[String]) -> (Vec<RustTestCase>, usize, bool) {
     // Sort alphabetically for deterministic run order
     tests.sort_by_key(|test| test.file);
 
-    (tests, all_files.len(), is_focus_run)
+    (tests, all_files, is_focus_run)
+}
+
+/// Finds all `#[itest(async)]` tests.
+#[cfg(since_api = "4.2")]
+fn collect_async_rust_tests(
+    filters: &[String],
+    sync_focus_run: bool,
+) -> (Vec<AsyncRustTestCase>, HashSet<&str>, bool) {
+    let mut all_files = HashSet::new();
+    let mut tests = vec![];
+    let mut is_focus_run = sync_focus_run;
+
+    sys::plugin_foreach!(__GODOT_ASYNC_ITEST; |test: &AsyncRustTestCase| {
+        // First time a focused test is encountered, switch to "focused" mode and throw everything away.
+        if !is_focus_run && test.focused {
+            tests.clear();
+            all_files.clear();
+            is_focus_run = true;
+        }
+
+        // Only collect tests if normal mode, or focus mode and test is focused.
+        if (!is_focus_run || test.focused) && passes_filter(filters, test.name) {
+            all_files.insert(test.file);
+            tests.push(*test);
+        }
+    });
+
+    // Sort alphabetically for deterministic run order
+    tests.sort_by_key(|test| test.file);
+
+    (tests, all_files, is_focus_run)
 }
 
 /// Finds all `#[bench]` benchmarks.
@@ -71,7 +104,7 @@ fn collect_rust_benchmarks() -> (Vec<RustBenchmark>, usize) {
 
 // ----------------------------------------------------------------------------------------------------------------------------------------------
 // Shared types
-
+#[derive(Clone)]
 pub struct TestContext {
     pub scene_tree: Gd<Node>,
     pub property_tests: Gd<Node>,
@@ -108,6 +141,19 @@ pub struct RustTestCase {
     pub function: fn(&TestContext),
 }
 
+#[cfg(since_api = "4.2")]
+#[derive(Copy, Clone)]
+pub struct AsyncRustTestCase {
+    pub name: &'static str,
+    pub file: &'static str,
+    pub skipped: bool,
+    /// If one or more tests are focused, only they will be executed. Helpful for debugging and working on specific features.
+    pub focused: bool,
+    #[allow(dead_code)]
+    pub line: u32,
+    pub function: fn(&TestContext) -> godot::task::TaskHandle,
+}
+
 #[derive(Copy, Clone)]
 pub struct RustBenchmark {
     pub name: &'static str,
diff --git a/itest/rust/src/framework/runner.rs b/itest/rust/src/framework/runner.rs
index 299809614..097969b64 100644
--- a/itest/rust/src/framework/runner.rs
+++ b/itest/rust/src/framework/runner.rs
@@ -7,7 +7,7 @@
 
 use std::time::{Duration, Instant};
 
-use godot::builtin::{Array, GString, Variant, VariantArray};
+use godot::builtin::{Array, Callable, GString, Variant, VariantArray};
 use godot::classes::{Engine, Node, Os};
 use godot::global::godot_error;
 use godot::meta::ToGodot;
@@ -18,19 +18,30 @@ use crate::framework::{
     bencher, passes_filter, BenchResult, RustBenchmark, RustTestCase, TestContext,
 };
 
+#[cfg(since_api = "4.2")]
+use super::AsyncRustTestCase;
+
+#[derive(Debug, Clone, Default)]
+struct TestStats {
+    total: usize,
+    passed: usize,
+    skipped: usize,
+    failed_list: Vec<String>,
+}
+
 #[derive(GodotClass, Debug)]
 #[class(init)]
 pub struct IntegrationTests {
-    total: i64,
-    passed: i64,
-    skipped: i64,
-    failed_list: Vec<String>,
+    stats: TestStats,
     focus_run: bool,
+    #[cfg(before_api = "4.2")]
+    base: godot::obj::Base<godot::classes::RefCounted>,
 }
 
 #[godot_api]
 impl IntegrationTests {
     #[allow(clippy::uninlined_format_args)]
+    #[allow(clippy::too_many_arguments)]
     #[func]
     fn run_all_tests(
         &mut self,
@@ -40,7 +51,8 @@ impl IntegrationTests {
         scene_tree: Gd<Node>,
         filters: VariantArray,
         property_tests: Gd<Node>,
-    ) -> bool {
+        on_finished: Callable,
+    ) {
         println!("{}Run{} Godot integration tests...", FMT_CYAN_BOLD, FMT_END);
         let filters: Vec<String> = filters.iter_shared().map(|v| v.to::<String>()).collect();
         let gdscript_tests = gdscript_tests
@@ -50,20 +62,19 @@ impl IntegrationTests {
                 passes_filter(filters.as_slice(), &test_name)
             })
             .collect::<Array<_>>();
-        let (rust_tests, rust_file_count, focus_run) =
-            super::collect_rust_tests(filters.as_slice());
+
+        let rust_test_cases = collect_rust_tests(&filters);
 
         // Print based on focus/not focus.
-        self.focus_run = focus_run;
-        if focus_run {
+        self.focus_run = rust_test_cases.focus_run;
+        if rust_test_cases.focus_run {
             println!("  {FMT_CYAN}Focused run{FMT_END} -- execute only selected Rust tests.")
         }
         println!(
             "  Rust: found {} tests in {} files.",
-            rust_tests.len(),
-            rust_file_count
+            rust_test_cases.rust_test_count, rust_test_cases.rust_file_count
         );
-        if !focus_run {
+        if !rust_test_cases.focus_run {
             println!(
                 "  GDScript: found {} tests in {} files.",
                 gdscript_tests.len(),
@@ -72,18 +83,74 @@ impl IntegrationTests {
         }
 
         let clock = Instant::now();
-        self.run_rust_tests(rust_tests, scene_tree, property_tests.clone());
+        self.run_rust_tests(
+            rust_test_cases.rust_tests,
+            scene_tree.clone(),
+            property_tests.clone(),
+        );
         let rust_time = clock.elapsed();
-        property_tests.free();
 
-        let gdscript_time = if !focus_run {
+        let gdscript_time = if !rust_test_cases.focus_run {
             let extra_duration = self.run_gdscript_tests(gdscript_tests);
-            Some((clock.elapsed() - rust_time) + extra_duration)
+            Some((clock.elapsed() - rust_time, extra_duration))
         } else {
             None
         };
 
-        self.conclude_tests(rust_time, gdscript_time, allow_focus)
+        #[cfg(before_api = "4.2")]
+        {
+            use godot::obj::WithBaseField;
+
+            property_tests.free();
+
+            let result = Self::conclude_tests(
+                &self.stats,
+                rust_time,
+                gdscript_time.map(|(elapsed, extra)| elapsed + extra),
+                allow_focus,
+            );
+
+            // on_finished will call back into self, so we have to make self re-entrant. We also can't call on_finished in deferred mode,
+            // since it's not available under the 4.1 API.
+            let base = self.base_mut();
+            on_finished.callv(&godot::builtin::varray![result]);
+            // We should do something with base to satisfy the compiler.
+            drop(base);
+        }
+
+        #[cfg(since_api = "4.2")]
+        {
+            let stats = self.stats.clone();
+
+            let on_finalize_test = move |stats, property_tests: Gd<Node>| {
+                let gdscript_elapsed = gdscript_time
+                    .as_ref()
+                    .map(|gdtime| gdtime.0)
+                    .unwrap_or_default();
+
+                let rust_async_time = clock.elapsed() - rust_time - gdscript_elapsed;
+
+                property_tests.free();
+
+                let result = Self::conclude_tests(
+                    &stats,
+                    rust_time + rust_async_time,
+                    gdscript_time.map(|(elapsed, extra)| elapsed + extra),
+                    allow_focus,
+                );
+
+                // Calling deferred to break a potentially synchronous call stack and avoid re-entrancy.
+                on_finished.call(&[result.to_variant()]);
+            };
+
+            Self::run_async_rust_tests(
+                stats,
+                rust_test_cases.async_rust_tests,
+                scene_tree,
+                property_tests,
+                on_finalize_test,
+            );
+        }
     }
 
     #[func]
@@ -139,14 +206,69 @@ impl IntegrationTests {
 
         let mut last_file = None;
         for test in tests {
-            print_test_pre(test.name, test.file.to_string(), &mut last_file, false);
+            print_test_pre(test.name, test.file, last_file.as_deref(), false);
+            last_file = Some(test.file.to_string());
+
             let outcome = run_rust_test(&test, &ctx);
 
-            self.update_stats(&outcome, test.file, test.name);
+            Self::update_stats(&mut self.stats, &outcome, test.file, test.name);
             print_test_post(test.name, outcome);
         }
     }
 
+    #[cfg(since_api = "4.2")]
+    fn run_async_rust_tests(
+        stats: TestStats,
+        tests: Vec<AsyncRustTestCase>,
+        scene_tree: Gd<Node>,
+        property_tests: Gd<Node>,
+        on_finalize_test: impl FnOnce(TestStats, Gd<Node>) + 'static,
+    ) {
+        let mut tests_iter = tests.into_iter();
+
+        let Some(first_test) = tests_iter.next() else {
+            return on_finalize_test(stats, property_tests);
+        };
+
+        let ctx = TestContext {
+            scene_tree,
+            property_tests,
+        };
+
+        Self::run_async_rust_tests_step(tests_iter, first_test, ctx, stats, None, on_finalize_test);
+    }
+
+    #[cfg(since_api = "4.2")]
+    fn run_async_rust_tests_step(
+        mut tests_iter: impl Iterator<Item = AsyncRustTestCase> + 'static,
+        test: AsyncRustTestCase,
+        ctx: TestContext,
+        mut stats: TestStats,
+        mut last_file: Option<String>,
+        on_finalize_test: impl FnOnce(TestStats, Gd<Node>) + 'static,
+    ) {
+        print_test_pre(test.name, test.file, last_file.as_deref(), true);
+        last_file.replace(test.file.to_string());
+
+        run_async_rust_test(&test, &ctx.clone(), move |outcome| {
+            Self::update_stats(&mut stats, &outcome, test.file, test.name);
+            print_test_post(test.name, outcome);
+
+            if let Some(next) = tests_iter.next() {
+                return Self::run_async_rust_tests_step(
+                    tests_iter,
+                    next,
+                    ctx,
+                    stats,
+                    last_file,
+                    on_finalize_test,
+                );
+            }
+
+            on_finalize_test(stats, ctx.property_tests);
+        });
+    }
+
     fn run_gdscript_tests(&mut self, tests: VariantArray) -> Duration {
         let mut last_file = None;
         let mut extra_duration = Duration::new(0, 0);
@@ -155,7 +277,9 @@ impl IntegrationTests {
             let test_file = get_property(&test, "suite_name");
             let test_case = get_property(&test, "method_name");
 
-            print_test_pre(&test_case, test_file.clone(), &mut last_file, true);
+            print_test_pre(&test_case, &test_file, last_file.as_deref(), true);
+
+            last_file = Some(test_file.clone());
 
             // If GDScript invokes Rust code that fails, the panic would break through; catch it.
             // TODO(bromeon): use try_call() once available.
@@ -191,28 +315,28 @@ impl IntegrationTests {
                 }
             };
 
-            self.update_stats(&outcome, &test_file, &test_case);
+            Self::update_stats(&mut self.stats, &outcome, &test_file, &test_case);
             print_test_post(&test_case, outcome);
         }
         extra_duration
     }
 
     fn conclude_tests(
-        &self,
+        stats: &TestStats,
         rust_time: Duration,
         gdscript_time: Option<Duration>,
         allow_focus: bool,
     ) -> bool {
-        let Self {
+        let TestStats {
             total,
             passed,
             skipped,
             ..
-        } = *self;
+        } = stats;
 
         // Consider 0 tests run as a failure too, because it's probably a problem with the run itself.
         let failed = total - passed - skipped;
-        let all_passed = failed == 0 && total != 0;
+        let all_passed = failed == 0 && *total != 0;
 
         let outcome = TestOutcome::from_bool(all_passed);
 
@@ -220,7 +344,7 @@ impl IntegrationTests {
         let gdscript_time = gdscript_time.map(|t| t.as_secs_f32());
         let focused_run = gdscript_time.is_none();
 
-        let extra = if skipped > 0 {
+        let extra = if *skipped > 0 {
             format!(", {skipped} skipped")
         } else if focused_run {
             " (focused run)".to_string()
@@ -241,12 +365,12 @@ impl IntegrationTests {
         if !all_passed {
             println!("\n  Failed tests:");
             let max = 10;
-            for test in self.failed_list.iter().take(max) {
+            for test in stats.failed_list.iter().take(max) {
                 println!("  * {test}");
             }
 
-            if self.failed_list.len() > max {
-                println!("  * ... and {} more.", self.failed_list.len() - max);
+            if stats.failed_list.len() > max {
+                println!("  * ... and {} more.", stats.failed_list.len() - max);
             }
 
             println!();
@@ -271,7 +395,9 @@ impl IntegrationTests {
 
         let mut last_file = None;
         for bench in benchmarks {
-            print_bench_pre(bench.name, bench.file.to_string(), &mut last_file);
+            print_bench_pre(bench.name, bench.file, last_file.as_deref());
+            last_file = Some(bench.file.to_string());
+
             let result = bencher::run_benchmark(bench.function, bench.repetitions);
             print_bench_post(result);
         }
@@ -282,16 +408,21 @@ impl IntegrationTests {
         println!("\nBenchmarks completed in {secs:.2}s.");
     }
 
-    fn update_stats(&mut self, outcome: &TestOutcome, test_file: &str, test_name: &str) {
-        self.total += 1;
+    fn update_stats(
+        stats: &mut TestStats,
+        outcome: &TestOutcome,
+        test_file: &str,
+        test_name: &str,
+    ) {
+        stats.total += 1;
         match outcome {
-            TestOutcome::Passed => self.passed += 1,
-            TestOutcome::Failed => self.failed_list.push(format!(
+            TestOutcome::Passed => stats.passed += 1,
+            TestOutcome::Failed => stats.failed_list.push(format!(
                 "{} > {}",
                 extract_file_subtitle(test_file),
                 test_name
             )),
-            TestOutcome::Skipped => self.skipped += 1,
+            TestOutcome::Skipped => stats.skipped += 1,
         }
     }
 }
@@ -320,7 +451,72 @@ fn run_rust_test(test: &RustTestCase, ctx: &TestContext) -> TestOutcome {
     TestOutcome::from_bool(success.is_ok())
 }
 
-fn print_test_pre(test_case: &str, test_file: String, last_file: &mut Option<String>, flush: bool) {
+#[cfg(since_api = "4.2")]
+fn run_async_rust_test(
+    test: &AsyncRustTestCase,
+    ctx: &TestContext,
+    on_test_finished: impl FnOnce(TestOutcome) + 'static,
+) {
+    if test.skipped {
+        return on_test_finished(TestOutcome::Skipped);
+    }
+
+    // Explicit type to prevent tests from returning a value
+    let err_context = || format!("itest `{}` failed", test.name);
+    let success: Result<godot::task::TaskHandle, _> =
+        godot::private::handle_panic(err_context, || (test.function)(ctx));
+
+    let Ok(task_handle) = success else {
+        return on_test_finished(TestOutcome::Failed);
+    };
+
+    check_async_test_task(task_handle, on_test_finished, ctx);
+}
+
+#[cfg(since_api = "4.2")]
+fn check_async_test_task(
+    task_handle: godot::task::TaskHandle,
+    on_test_finished: impl FnOnce(TestOutcome) + 'static,
+    ctx: &TestContext,
+) {
+    use godot::classes::object::ConnectFlags;
+    use godot::obj::EngineBitfield;
+    use godot::task::has_godot_task_panicked;
+
+    if !task_handle.is_pending() {
+        on_test_finished(TestOutcome::from_bool(!has_godot_task_panicked(
+            task_handle,
+        )));
+
+        return;
+    }
+
+    let next_ctx = ctx.clone();
+    let mut callback = Some(on_test_finished);
+    let mut probably_task_handle = Some(task_handle);
+
+    let deferred = Callable::from_local_fn("run_async_rust_test", move |_| {
+        check_async_test_task(
+            probably_task_handle
+                .take()
+                .expect("Callable will only be called once!"),
+            callback
+                .take()
+                .expect("Callable should not be called multiple times!"),
+            &next_ctx,
+        );
+        Ok(Variant::nil())
+    });
+
+    ctx.scene_tree
+        .get_tree()
+        .expect("The itest scene tree node is part of a Godot SceneTree")
+        .connect_ex("process_frame", &deferred)
+        .flags(ConnectFlags::ONE_SHOT.ord() as u32)
+        .done();
+}
+
+fn print_test_pre(test_case: &str, test_file: &str, last_file: Option<&str>, flush: bool) {
     print_file_header(test_file, last_file);
 
     print!("   -- {test_case} ... ");
@@ -331,16 +527,13 @@ fn print_test_pre(test_case: &str, test_file: String, last_file: &mut Option<Str
     }
 }
 
-fn print_file_header(file: String, last_file: &mut Option<String>) {
+fn print_file_header(file: &str, last_file: Option<&str>) {
     // Check if we need to open a new category for a file.
-    let print_file = last_file.as_ref() != Some(&file);
+    let print_file = last_file != Some(file);
 
     if print_file {
-        println!("\n   {}:", extract_file_subtitle(&file));
+        println!("\n   {}:", extract_file_subtitle(file));
     }
-
-    // State update for file-category-print
-    *last_file = Some(file);
 }
 
 fn extract_file_subtitle(file: &str) -> &str {
@@ -364,7 +557,7 @@ fn print_test_post(test_case: &str, outcome: TestOutcome) {
     }
 }
 
-fn print_bench_pre(benchmark: &str, bench_file: String, last_file: &mut Option<String>) {
+fn print_bench_pre(benchmark: &str, bench_file: &str, last_file: Option<&str>) {
     print_file_header(bench_file, last_file);
 
     let benchmark = if benchmark.len() > 26 {
@@ -402,6 +595,53 @@ fn get_errors(test: &Variant) -> Array<GString> {
         .unwrap_or_default()
 }
 
+struct RustTestCases {
+    rust_tests: Vec<RustTestCase>,
+    #[cfg(since_api = "4.2")]
+    async_rust_tests: Vec<AsyncRustTestCase>,
+    rust_test_count: usize,
+    rust_file_count: usize,
+    focus_run: bool,
+}
+
+#[cfg(before_api = "4.2")]
+fn collect_rust_tests(filters: &[String]) -> RustTestCases {
+    let (rust_tests, rust_files, focus_run) = super::collect_rust_tests(filters);
+
+    let rust_test_count = rust_tests.len();
+
+    RustTestCases {
+        rust_tests,
+        rust_test_count,
+        rust_file_count: rust_files.len(),
+        focus_run,
+    }
+}
+
+#[cfg(since_api = "4.2")]
+fn collect_rust_tests(filters: &[String]) -> RustTestCases {
+    let (mut rust_tests, mut rust_files, focus_run) = super::collect_rust_tests(filters);
+
+    let (async_rust_tests, async_rust_files, async_focus_run) =
+        super::collect_async_rust_tests(filters, focus_run);
+
+    if !focus_run && async_focus_run {
+        rust_tests.clear();
+        rust_files.clear();
+    }
+
+    let rust_test_count = rust_tests.len() + async_rust_tests.len();
+    let rust_file_count = rust_files.union(&async_rust_files).count();
+
+    RustTestCases {
+        rust_tests,
+        async_rust_tests,
+        rust_test_count,
+        rust_file_count,
+        focus_run: focus_run || async_focus_run,
+    }
+}
+
 #[must_use]
 enum TestOutcome {
     Passed,