diff --git a/flytekit/core/array_node_map_task.py b/flytekit/core/array_node_map_task.py
index 94454f417b0..82bb0cbe683 100644
--- a/flytekit/core/array_node_map_task.py
+++ b/flytekit/core/array_node_map_task.py
@@ -29,6 +29,7 @@
 from flytekit.tools.module_loader import load_object_from_module
 from flytekit.types.pickle import pickle
 from flytekit.types.pickle.pickle import FlytePickleTransformer
+from flytekit.utils.asyn import loop_manager
 
 
 class ArrayNodeMapTask(PythonTask):
@@ -253,7 +254,7 @@ def _literal_map_to_python_input(
                 v = literal_map.literals[k]
                 # If the input is offloaded, we need to unwrap it
                 if v.offloaded_metadata:
-                    v = TypeEngine.unwrap_offloaded_literal(ctx, v)
+                    v = loop_manager.run_sync(TypeEngine.unwrap_offloaded_literal, ctx, v)
                 if k not in self.bound_inputs:
                     # assert that v.collection is not None
                     if not v.collection or not isinstance(v.collection.literals, list):
diff --git a/flytekit/core/base_task.py b/flytekit/core/base_task.py
index 682749c2734..b284a4c6e4b 100644
--- a/flytekit/core/base_task.py
+++ b/flytekit/core/base_task.py
@@ -86,6 +86,7 @@
 from flytekit.models.documentation import Description, Documentation
 from flytekit.models.interface import Variable
 from flytekit.models.security import SecurityContext
+from flytekit.utils.asyn import run_sync
 
 DYNAMIC_PARTITIONS = "_uap"
 MODEL_CARD = "_ucm"
@@ -608,7 +609,7 @@ def _literal_map_to_python_input(
     ) -> Dict[str, Any]:
         return TypeEngine.literal_map_to_kwargs(ctx, literal_map, self.python_interface.inputs)
 
-    def _output_to_literal_map(self, native_outputs: Dict[int, Any], ctx: FlyteContext):
+    async def _output_to_literal_map(self, native_outputs: Dict[int, Any], ctx: FlyteContext):
         expected_output_names = list(self._outputs_interface.keys())
         if len(expected_output_names) == 1:
             # Here we have to handle the fact that the task could've been declared with a typing.NamedTuple of
@@ -629,27 +630,35 @@ def _output_to_literal_map(self, native_outputs: Dict[int, Any], ctx: FlyteConte
         with timeit("Translate the output to literals"):
             literals = {}
             omt = ctx.output_metadata_tracker
+            # Here is where we iterate through the outputs, need to call new type engine.
             for i, (k, v) in enumerate(native_outputs_as_map.items()):
                 literal_type = self._outputs_interface[k].type
                 py_type = self.get_type_for_output_var(k, v)
 
                 if isinstance(v, tuple):
                     raise TypeError(f"Output({k}) in task '{self.name}' received a tuple {v}, instead of {py_type}")
-                try:
-                    lit = TypeEngine.to_literal(ctx, v, py_type, literal_type)
-                    literals[k] = lit
-                except Exception as e:
+                literals[k] = asyncio.create_task(TypeEngine.async_to_literal(ctx, v, py_type, literal_type))
+
+            await asyncio.gather(*literals.values(), return_exceptions=True)
+
+            for i, (k2, v2) in enumerate(literals.items()):
+                if v2.exception() is not None:
                     # only show the name of output key if it's user-defined (by default Flyte names these as "o<n>")
-                    key = k if k != f"o{i}" else i
+                    key = k2 if k2 != f"o{i}" else i
+                    e: BaseException = v2.exception()  # type: ignore  # we know this is not optional
+                    py_type = self.get_type_for_output_var(k2, native_outputs_as_map[k2])
                     e.args = (
                         f"Failed to convert outputs of task '{self.name}' at position {key}.\n"
                         f"Failed to convert type {type(native_outputs_as_map[expected_output_names[i]])} to type {py_type}.\n"
                         f"Error Message: {e.args[0]}.",
                     )
-                    raise
-                # Now check if there is any output metadata associated with this output variable and attach it to the
-                # literal
-                if omt is not None:
+                    raise e
+                literals[k2] = v2.result()
+
+            if omt is not None:
+                for i, (k, v) in enumerate(native_outputs_as_map.items()):
+                    # Now check if there is any output metadata associated with this output variable and attach it to the
+                    # literal
                     om = omt.get(v)
                     if om:
                         metadata = {}
@@ -669,7 +678,7 @@ def _output_to_literal_map(self, native_outputs: Dict[int, Any], ctx: FlyteConte
                             encoded = b64encode(s).decode("utf-8")
                             metadata[DYNAMIC_PARTITIONS] = encoded
                         if metadata:
-                            lit.set_metadata(metadata)
+                            literals[k].set_metadata(metadata)  # type: ignore  # we know these have been resolved
 
         return _literal_models.LiteralMap(literals=literals), native_outputs_as_map
 
@@ -697,7 +706,7 @@ def _write_decks(self, native_inputs, native_outputs_as_map, ctx, new_user_param
     async def _async_execute(self, native_inputs, native_outputs, ctx, exec_ctx, new_user_params):
         native_outputs = await native_outputs
         native_outputs = self.post_execute(new_user_params, native_outputs)
-        literals_map, native_outputs_as_map = self._output_to_literal_map(native_outputs, exec_ctx)
+        literals_map, native_outputs_as_map = await self._output_to_literal_map(native_outputs, exec_ctx)
         self._write_decks(native_inputs, native_outputs_as_map, ctx, new_user_params)
         return literals_map
 
@@ -787,7 +796,10 @@ def dispatch_execute(
                 return native_outputs
 
             try:
-                literals_map, native_outputs_as_map = self._output_to_literal_map(native_outputs, exec_ctx)
+                with timeit("dispatch execute"):
+                    literals_map, native_outputs_as_map = run_sync(
+                        self._output_to_literal_map, native_outputs, exec_ctx
+                    )
                 self._write_decks(native_inputs, native_outputs_as_map, ctx, new_user_params)
             except (FlyteUploadDataException, FlyteDownloadDataException):
                 raise
diff --git a/flytekit/core/promise.py b/flytekit/core/promise.py
index cea96ac0137..afd51cd7ccf 100644
--- a/flytekit/core/promise.py
+++ b/flytekit/core/promise.py
@@ -44,9 +44,10 @@
 from flytekit.models.literals import Binary, Literal, Primitive, Scalar
 from flytekit.models.task import Resources
 from flytekit.models.types import SimpleType
+from flytekit.utils.asyn import loop_manager, run_sync
 
 
-def translate_inputs_to_literals(
+async def _translate_inputs_to_literals(
     ctx: FlyteContext,
     incoming_values: Dict[str, Any],
     flyte_interface_types: Dict[str, _interface_models.Variable],
@@ -94,8 +95,8 @@ def my_wf(in1: int, in2: int) -> int:
         t = native_types[k]
         try:
             if type(v) is Promise:
-                v = resolve_attr_path_in_promise(v)
-            result[k] = TypeEngine.to_literal(ctx, v, t, var.type)
+                v = await resolve_attr_path_in_promise(v)
+            result[k] = await TypeEngine.async_to_literal(ctx, v, t, var.type)
         except TypeTransformerFailedError as exc:
             exc.args = (f"Failed argument '{k}': {exc.args[0]}",)
             raise
@@ -103,7 +104,10 @@ def my_wf(in1: int, in2: int) -> int:
     return result
 
 
-def resolve_attr_path_in_promise(p: Promise) -> Promise:
+translate_inputs_to_literals = loop_manager.synced(_translate_inputs_to_literals)
+
+
+async def resolve_attr_path_in_promise(p: Promise) -> Promise:
     """
     resolve_attr_path_in_promise resolves the attribute path in a promise and returns a new promise with the resolved value
     This is for local execution only. The remote execution will be resolved in flytepropeller.
@@ -145,7 +149,9 @@ def resolve_attr_path_in_promise(p: Promise) -> Promise:
             new_st = resolve_attr_path_in_pb_struct(st, attr_path=p.attr_path[used:])
             literal_type = TypeEngine.to_literal_type(type(new_st))
             # Reconstruct the resolved result to flyte literal (because the resolved result might not be struct)
-            curr_val = TypeEngine.to_literal(FlyteContextManager.current_context(), new_st, type(new_st), literal_type)
+            curr_val = await TypeEngine.async_to_literal(
+                FlyteContextManager.current_context(), new_st, type(new_st), literal_type
+            )
         elif type(curr_val.value.value) is Binary:
             binary_idl_obj = curr_val.value.value
             if binary_idl_obj.tag == _common_constants.MESSAGEPACK:
@@ -786,7 +792,7 @@ def __rshift__(self, other: Any):
     return Output(*promises)  # type: ignore
 
 
-def binding_data_from_python_std(
+async def binding_data_from_python_std(
     ctx: _flyte_context.FlyteContext,
     expected_literal_type: _type_models.LiteralType,
     t_value: Any,
@@ -821,7 +827,8 @@ def binding_data_from_python_std(
         # If the value is not a container type, then we can directly convert it to a scalar in the Union case.
         # This pushes the handling of the Union types to the type engine.
         if not isinstance(t_value, list) and not isinstance(t_value, dict):
-            scalar = TypeEngine.to_literal(ctx, t_value, t_value_type or type(t_value), expected_literal_type).scalar
+            lit = await TypeEngine.async_to_literal(ctx, t_value, t_value_type or type(t_value), expected_literal_type)
+            scalar = lit.scalar
             return _literals_models.BindingData(scalar=scalar)
 
         # If it is a container type, then we need to iterate over the variants in the Union type, try each one. This is
@@ -831,7 +838,7 @@ def binding_data_from_python_std(
             try:
                 lt_type = expected_literal_type.union_type.variants[i]
                 python_type = get_args(t_value_type)[i] if t_value_type else None
-                return binding_data_from_python_std(ctx, lt_type, t_value, python_type, nodes)
+                return await binding_data_from_python_std(ctx, lt_type, t_value, python_type, nodes)
             except Exception:
                 logger.debug(
                     f"failed to bind data {t_value} with literal type {expected_literal_type.union_type.variants[i]}."
@@ -844,7 +851,9 @@ def binding_data_from_python_std(
         sub_type: Optional[type] = ListTransformer.get_sub_type_or_none(t_value_type)
         collection = _literals_models.BindingDataCollection(
             bindings=[
-                binding_data_from_python_std(ctx, expected_literal_type.collection_type, t, sub_type or type(t), nodes)
+                await binding_data_from_python_std(
+                    ctx, expected_literal_type.collection_type, t, sub_type or type(t), nodes
+                )
                 for t in t_value
             ]
         )
@@ -860,13 +869,13 @@ def binding_data_from_python_std(
                 f"this should be a Dictionary type and it is not: {type(t_value)} vs {expected_literal_type}"
             )
         if expected_literal_type.simple == _type_models.SimpleType.STRUCT:
-            lit = TypeEngine.to_literal(ctx, t_value, type(t_value), expected_literal_type)
+            lit = await TypeEngine.async_to_literal(ctx, t_value, type(t_value), expected_literal_type)
             return _literals_models.BindingData(scalar=lit.scalar)
         else:
             _, v_type = DictTransformer.extract_types_or_metadata(t_value_type)
             m = _literals_models.BindingDataMap(
                 bindings={
-                    k: binding_data_from_python_std(
+                    k: await binding_data_from_python_std(
                         ctx, expected_literal_type.map_value_type, v, v_type or type(v), nodes
                     )
                     for k, v in t_value.items()
@@ -883,8 +892,8 @@ def binding_data_from_python_std(
         )
 
     # This is the scalar case - e.g. my_task(in1=5)
-    scalar = TypeEngine.to_literal(ctx, t_value, t_value_type or type(t_value), expected_literal_type).scalar
-    return _literals_models.BindingData(scalar=scalar)
+    lit = await TypeEngine.async_to_literal(ctx, t_value, t_value_type or type(t_value), expected_literal_type)
+    return _literals_models.BindingData(scalar=lit.scalar)
 
 
 def binding_from_python_std(
@@ -895,7 +904,8 @@ def binding_from_python_std(
     t_value_type: type,
 ) -> Tuple[_literals_models.Binding, List[Node]]:
     nodes: List[Node] = []
-    binding_data = binding_data_from_python_std(
+    binding_data = run_sync(
+        binding_data_from_python_std,
         ctx,
         expected_literal_type,
         t_value,
diff --git a/flytekit/core/type_engine.py b/flytekit/core/type_engine.py
index 18eba962a22..94e824a30b8 100644
--- a/flytekit/core/type_engine.py
+++ b/flytekit/core/type_engine.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import collections
 import copy
 import dataclasses
@@ -47,6 +48,7 @@
 from flytekit.models.core import types as _core_types
 from flytekit.models.literals import Binary, Literal, LiteralCollection, LiteralMap, Primitive, Scalar, Union, Void
 from flytekit.models.types import LiteralType, SimpleType, TypeStructure, UnionType
+from flytekit.utils.asyn import loop_manager
 
 T = typing.TypeVar("T")
 DEFINITIONS = "definitions"
@@ -145,6 +147,10 @@ def python_type(self) -> Type[T]:
         """
         return self._t
 
+    @property
+    def is_async(self) -> bool:
+        return False
+
     @property
     def type_assertions_enabled(self) -> bool:
         """
@@ -249,6 +255,56 @@ def __str__(self):
         return str(self.__repr__())
 
 
+class AsyncTypeTransformer(TypeTransformer[T]):
+    def __init__(self, name: str, t: Type[T], enable_type_assertions: bool = True):
+        super().__init__(name, t, enable_type_assertions)
+
+    @property
+    def is_async(self) -> bool:
+        return True
+
+    @abstractmethod
+    async def async_to_literal(
+        self, ctx: FlyteContext, python_val: T, python_type: Type[T], expected: LiteralType
+    ) -> Literal:
+        """
+        Converts a given python_val to a Flyte Literal, assuming the given python_val matches the declared python_type.
+        Implementers should refrain from using type(python_val) instead rely on the passed in python_type. If these
+        do not match (or are not allowed) the Transformer implementer should raise an AssertionError, clearly stating
+        what was the mismatch
+        :param ctx: A FlyteContext, useful in accessing the filesystem and other attributes
+        :param python_val: The actual value to be transformed
+        :param python_type: The assumed type of the value (this matches the declared type on the function)
+        :param expected: Expected Literal Type
+        """
+
+        raise NotImplementedError(f"Conversion to Literal for python type {python_type} not implemented")
+
+    @abstractmethod
+    async def async_to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[T]) -> Optional[T]:
+        """
+        Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised
+        :param ctx: FlyteContext
+        :param lv: The received literal Value
+        :param expected_python_type: Expected native python type that should be returned
+        """
+        raise NotImplementedError(
+            f"Conversion to python value expected type {expected_python_type} from literal not implemented"
+        )
+
+    def to_literal(
+        self, ctx: FlyteContext, python_val: typing.Any, python_type: Type[T], expected: LiteralType
+    ) -> Literal:
+        synced = loop_manager.synced(self.async_to_literal)
+        result = synced(ctx, python_val, python_type, expected)
+        return result
+
+    def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[T]) -> Optional[T]:
+        synced = loop_manager.synced(self.async_to_python_value)
+        result = synced(ctx, lv, expected_python_type)
+        return result
+
+
 class SimpleTransformer(TypeTransformer[T]):
     """
     A Simple implementation of a type transformer that uses simple lambdas to transform and reduces boilerplate
@@ -949,14 +1005,14 @@ def register_restricted_type(
         cls.register(RestrictedTypeTransformer(name, type))  # type: ignore
 
     @classmethod
-    def register_additional_type(cls, transformer: TypeTransformer, additional_type: Type, override=False):
+    def register_additional_type(cls, transformer: TypeTransformer[T], additional_type: Type[T], override=False):
         if additional_type not in cls._REGISTRY or override:
             cls._REGISTRY[additional_type] = transformer
 
     @classmethod
     def get_transformer(cls, python_type: Type) -> TypeTransformer[T]:
         """
-        The TypeEngine hierarchy for flyteKit. This method looksup and selects the type transformer. The algorithm is
+        The TypeEngine hierarchy for flyteKit. This method looks up and selects the type transformer. The algorithm is
         as follows
 
           d = dictionary of registered transformers, where is a python `type`
@@ -1123,7 +1179,7 @@ def lazy_import_transformers(cls):
                     logger.debug("Transformer for snowflake is already registered.")
 
     @classmethod
-    def to_literal_type(cls, python_type: Type) -> LiteralType:
+    def to_literal_type(cls, python_type: Type[T]) -> LiteralType:
         """
         Converts a python type into a flyte specific ``LiteralType``
         """
@@ -1146,15 +1202,9 @@ def to_literal_type(cls, python_type: Type) -> LiteralType:
         return res
 
     @classmethod
-    def to_literal(cls, ctx: FlyteContext, python_val: typing.Any, python_type: Type, expected: LiteralType) -> Literal:
-        """
-        Converts a python value of a given type and expected ``LiteralType`` into a resolved ``Literal`` value.
-        """
-        from flytekit.core.promise import Promise, VoidPromise
+    def to_literal_checks(cls, python_val: typing.Any, python_type: Type[T], expected: LiteralType):
+        from flytekit.core.promise import VoidPromise
 
-        if isinstance(python_val, Promise):
-            # In the example above, this handles the "in2=a" type of argument
-            return python_val.val
         if isinstance(python_val, VoidPromise):
             raise AssertionError(
                 f"Outputs of a non-output producing task {python_val.task_name} cannot be passed to another task."
@@ -1168,32 +1218,86 @@ def to_literal(cls, ctx: FlyteContext, python_val: typing.Any, python_type: Type
             )
         if (python_val is None and python_type != type(None)) and expected and expected.union_type is None:
             raise TypeTransformerFailedError(f"Python value cannot be None, expected {python_type}/{expected}")
-        transformer = cls.get_transformer(python_type)
-        if transformer.type_assertions_enabled:
-            transformer.assert_type(python_type, python_val)
 
+    @classmethod
+    def calculate_hash(cls, python_val: typing.Any, python_type: Type[T]) -> Optional[str]:
         # In case the value is an annotated type we inspect the annotations and look for hash-related annotations.
-        hash = None
+        hsh = None
         if is_annotated(python_type):
             # We are now dealing with one of two cases:
             # 1. The annotated type is a `HashMethod`, which indicates that we should produce the hash using
             #    the method indicated in the annotation.
-            # 2. The annotated type is being used for a different purpose other than calculating hash values, in which case
-            #    we should just continue.
+            # 2. The annotated type is being used for a different purpose other than calculating hash values,
+            #    in which case we should just continue.
             for annotation in get_args(python_type)[1:]:
                 if not isinstance(annotation, HashMethod):
                     continue
-                hash = annotation.calculate(python_val)
+                hsh = annotation.calculate(python_val)
                 break
+        return hsh
+
+    @classmethod
+    def to_literal(
+        cls, ctx: FlyteContext, python_val: typing.Any, python_type: Type[T], expected: LiteralType
+    ) -> Literal:
+        """
+        The current dance is because we are allowing users to call from an async function, this synchronous
+        to_literal function, and allowing this to_literal function, to then invoke yet another async functionl,
+        namely an async transformer.
+        """
+        from flytekit.core.promise import Promise
+
+        cls.to_literal_checks(python_val, python_type, expected)
+        if isinstance(python_val, Promise):
+            # In the example above, this handles the "in2=a" type of argument
+            return python_val.val
+
+        transformer = cls.get_transformer(python_type)
+
+        if transformer.type_assertions_enabled:
+            transformer.assert_type(python_type, python_val)
+
+        if isinstance(transformer, AsyncTypeTransformer):
+            synced = loop_manager.synced(transformer.async_to_literal)
+            lv = synced(ctx, python_val, python_type, expected)
+        else:
+            lv = transformer.to_literal(ctx, python_val, python_type, expected)
+
+        modify_literal_uris(lv)
+        lv.hash = cls.calculate_hash(python_val, python_type)
+        return lv
+
+    @classmethod
+    async def async_to_literal(
+        cls, ctx: FlyteContext, python_val: typing.Any, python_type: Type[T], expected: LiteralType
+    ) -> Literal:
+        """
+        Converts a python value of a given type and expected ``LiteralType`` into a resolved ``Literal`` value.
+        """
+        from flytekit.core.promise import Promise
+
+        cls.to_literal_checks(python_val, python_type, expected)
+
+        if isinstance(python_val, Promise):
+            # In the example above, this handles the "in2=a" type of argument
+            return python_val.val
+
+        transformer = cls.get_transformer(python_type)
+        if transformer.type_assertions_enabled:
+            transformer.assert_type(python_type, python_val)
+
+        if isinstance(transformer, AsyncTypeTransformer):
+            lv = await transformer.async_to_literal(ctx, python_val, python_type, expected)
+        else:
+            lv = transformer.to_literal(ctx, python_val, python_type, expected)
 
-        lv = transformer.to_literal(ctx, python_val, python_type, expected)
         modify_literal_uris(lv)
-        if hash is not None:
-            lv.hash = hash
+        lv.hash = cls.calculate_hash(python_val, python_type)
+
         return lv
 
     @classmethod
-    def unwrap_offloaded_literal(cls, ctx: FlyteContext, lv: Literal) -> Literal:
+    async def unwrap_offloaded_literal(cls, ctx: FlyteContext, lv: Literal) -> Literal:
         if not lv.offloaded_metadata:
             return lv
 
@@ -1210,9 +1314,27 @@ def to_python_value(cls, ctx: FlyteContext, lv: Literal, expected_python_type: T
         """
         # Initiate the process of loading the offloaded literal if offloaded_metadata is set
         if lv.offloaded_metadata:
-            lv = cls.unwrap_offloaded_literal(ctx, lv)
+            synced = loop_manager.synced(cls.unwrap_offloaded_literal)
+            lv = synced(ctx, lv)
         transformer = cls.get_transformer(expected_python_type)
-        return transformer.to_python_value(ctx, lv, expected_python_type)
+
+        if isinstance(transformer, AsyncTypeTransformer):
+            synced = loop_manager.synced(transformer.async_to_python_value)
+            return synced(ctx, lv, expected_python_type)
+        else:
+            res = transformer.to_python_value(ctx, lv, expected_python_type)
+            return res
+
+    @classmethod
+    async def async_to_python_value(cls, ctx: FlyteContext, lv: Literal, expected_python_type: Type) -> typing.Any:
+        if lv.offloaded_metadata:
+            lv = await cls.unwrap_offloaded_literal(ctx, lv)
+        transformer = cls.get_transformer(expected_python_type)
+        if isinstance(transformer, AsyncTypeTransformer):
+            pv = await transformer.async_to_python_value(ctx, lv, expected_python_type)
+        else:
+            pv = transformer.to_python_value(ctx, lv, expected_python_type)
+        return pv
 
     @classmethod
     def to_html(cls, ctx: FlyteContext, python_val: typing.Any, expected_python_type: Type[typing.Any]) -> str:
@@ -1245,6 +1367,18 @@ def literal_map_to_kwargs(
         lm: LiteralMap,
         python_types: typing.Optional[typing.Dict[str, type]] = None,
         literal_types: typing.Optional[typing.Dict[str, _interface_models.Variable]] = None,
+    ) -> typing.Dict[str, typing.Any]:
+        synced = loop_manager.synced(cls._literal_map_to_kwargs)
+        return synced(ctx, lm, python_types, literal_types)
+
+    @classmethod
+    @timeit("AsyncTranslate literal to python value")
+    async def _literal_map_to_kwargs(
+        cls,
+        ctx: FlyteContext,
+        lm: LiteralMap,
+        python_types: typing.Optional[typing.Dict[str, type]] = None,
+        literal_types: typing.Optional[typing.Dict[str, _interface_models.Variable]] = None,
     ) -> typing.Dict[str, typing.Any]:
         """
         Given a ``LiteralMap`` (usually an input into a task - intermediate), convert to kwargs for the task
@@ -1265,12 +1399,17 @@ def literal_map_to_kwargs(
                 f" than allowed by the input spec {len(python_interface_inputs)}"
             )
         kwargs = {}
-        for i, k in enumerate(lm.literals):
-            try:
-                kwargs[k] = TypeEngine.to_python_value(ctx, lm.literals[k], python_interface_inputs[k])
-            except TypeTransformerFailedError as exc:
-                exc.args = (f"Error converting input '{k}' at position {i}:\n  {exc.args[0]}",)
-                raise
+        try:
+            for i, k in enumerate(lm.literals):
+                kwargs[k] = asyncio.create_task(
+                    TypeEngine.async_to_python_value(ctx, lm.literals[k], python_interface_inputs[k])
+                )
+            await asyncio.gather(*kwargs.values())
+        except TypeTransformerFailedError as exc:
+            exc.args = (f"Error converting input '{k}' at position {i}:\n  {exc.args[0]}",)
+            raise
+
+        kwargs = {k: v.result() for k, v in kwargs.items() if v is not None}
         return kwargs
 
     @classmethod
@@ -1279,6 +1418,16 @@ def dict_to_literal_map(
         ctx: FlyteContext,
         d: typing.Dict[str, typing.Any],
         type_hints: Optional[typing.Dict[str, type]] = None,
+    ) -> LiteralMap:
+        synced = loop_manager.synced(cls._dict_to_literal_map)
+        return synced(ctx, d, type_hints)
+
+    @classmethod
+    async def _dict_to_literal_map(
+        cls,
+        ctx: FlyteContext,
+        d: typing.Dict[str, typing.Any],
+        type_hints: Optional[typing.Dict[str, type]] = None,
     ) -> LiteralMap:
         """
         Given a dictionary mapping string keys to python values and a dictionary containing guessed types for such string keys,
@@ -1291,25 +1440,35 @@ def dict_to_literal_map(
             # to account for the type erasure that happens in the case of built-in collection containers, such as
             # `list` and `dict`.
             python_type = type_hints.get(k, type(v))
-            try:
-                literal_map[k] = TypeEngine.to_literal(
+            literal_map[k] = asyncio.create_task(
+                TypeEngine.async_to_literal(
                     ctx=ctx,
                     python_val=v,
                     python_type=python_type,
                     expected=TypeEngine.to_literal_type(python_type),
                 )
-            except TypeError:
-                raise user_exceptions.FlyteTypeException(type(v), python_type, received_value=v)
+            )
+        await asyncio.gather(*literal_map.values(), return_exceptions=True)
+        for idx, (k, v) in enumerate(literal_map.items()):
+            if literal_map[k].exception() is not None:
+                python_type = type_hints.get(k, type(d[k]))
+                e: BaseException = literal_map[k].exception()  # type: ignore
+                if isinstance(e, TypeError):
+                    raise user_exceptions.FlyteTypeException(type(v), python_type, received_value=v)
+                else:
+                    raise e
+            literal_map[k] = v.result()
+
         return LiteralMap(literal_map)
 
     @classmethod
-    def dict_to_literal_map_pb(
+    async def dict_to_literal_map_pb(
         cls,
         ctx: FlyteContext,
         d: typing.Dict[str, typing.Any],
         type_hints: Optional[typing.Dict[str, type]] = None,
     ) -> Optional[literals_pb2.LiteralMap]:
-        literal_map = cls.dict_to_literal_map(ctx, d, type_hints)
+        literal_map = await cls._dict_to_literal_map(ctx, d, type_hints)
         return literal_map.to_flyte_idl()
 
     @classmethod
@@ -1351,7 +1510,7 @@ def guess_python_type(cls, flyte_type: LiteralType) -> type:
         raise ValueError(f"No transformers could reverse Flyte literal type {flyte_type}")
 
 
-class ListTransformer(TypeTransformer[T]):
+class ListTransformer(AsyncTypeTransformer[T]):
     """
     Transformer that handles a univariate typing.List[T]
     """
@@ -1411,7 +1570,9 @@ def is_batchable(t: Type):
                 return True
         return False
 
-    def to_literal(self, ctx: FlyteContext, python_val: T, python_type: Type[T], expected: LiteralType) -> Literal:
+    async def async_to_literal(
+        self, ctx: FlyteContext, python_val: T, python_type: Type[T], expected: LiteralType
+    ) -> Literal:
         if type(python_val) != list:
             raise TypeTransformerFailedError("Expected a list")
 
@@ -1434,10 +1595,14 @@ def to_literal(self, ctx: FlyteContext, python_val: T, python_type: Type[T], exp
                 lit_list = []
         else:
             t = self.get_sub_type(python_type)
-            lit_list = [TypeEngine.to_literal(ctx, x, t, expected.collection_type) for x in python_val]  # type: ignore
+            lit_list = [TypeEngine.async_to_literal(ctx, x, t, expected.collection_type) for x in python_val]
+            lit_list = await asyncio.gather(*lit_list)
+
         return Literal(collection=LiteralCollection(literals=lit_list))
 
-    def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[T]) -> typing.List[typing.Any]:  # type: ignore
+    async def async_to_python_value(  # type: ignore
+        self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[T]
+    ) -> typing.Optional[typing.List[T]]:
         if lv and lv.scalar and lv.scalar.binary is not None:
             return self.from_binary_idl(lv.scalar.binary, expected_python_type)  # type: ignore
 
@@ -1455,13 +1620,15 @@ def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type:
 
             batch_list = [TypeEngine.to_python_value(ctx, batch, FlytePickle) for batch in lits]
             if len(batch_list) > 0 and type(batch_list[0]) is list:
-                # Make it have backward compatibility. The upstream task may use old version of Flytekit that
-                # won't merge the elements in the list. Therefore, we should check if the batch_list[0] is the list first.
+                # Make it have backward compatibility. The upstream task may use old version of Flytekit that won't
+                # merge the elements in the list. Therefore, we should check if the batch_list[0] is the list first.
                 return [item for batch in batch_list for item in batch]
             return batch_list
         else:
             st = self.get_sub_type(expected_python_type)
-            return [TypeEngine.to_python_value(ctx, x, st) for x in lits]
+            result = [TypeEngine.async_to_python_value(ctx, x, st) for x in lits]
+            result = await asyncio.gather(*result)
+            return result  # type: ignore  # should be a list, thinks its a tuple
 
     def guess_python_type(self, literal_type: LiteralType) -> list:  # type: ignore
         if literal_type.collection_type:
@@ -1578,7 +1745,7 @@ def _is_union_type(t):
     return t is typing.Union or get_origin(t) is typing.Union or UnionType and isinstance(t, UnionType)
 
 
-class UnionTransformer(TypeTransformer[T]):
+class UnionTransformer(AsyncTypeTransformer[T]):
     """
     Transformer that handles a typing.Union[T1, T2, ...]
     """
@@ -1628,7 +1795,9 @@ def get_literal_type(self, t: Type[T]) -> Optional[LiteralType]:
         except Exception as e:
             raise ValueError(f"Type of Generic Union type is not supported, {e}")
 
-    def to_literal(self, ctx: FlyteContext, python_val: T, python_type: Type[T], expected: LiteralType) -> Literal:
+    async def async_to_literal(
+        self, ctx: FlyteContext, python_val: T, python_type: Type[T], expected: LiteralType
+    ) -> typing.Union[Literal, asyncio.Future]:
         python_type = get_underlying_type(python_type)
 
         found_res = False
@@ -1640,13 +1809,20 @@ def to_literal(self, ctx: FlyteContext, python_val: T, python_type: Type[T], exp
             try:
                 t = get_args(python_type)[i]
                 trans: TypeTransformer[T] = TypeEngine.get_transformer(t)
-                res = trans.to_literal(ctx, python_val, t, expected.union_type.variants[i])
-                res_type = _add_tag_to_type(trans.get_literal_type(t), trans.name)
+                if isinstance(trans, AsyncTypeTransformer):
+                    attempt = trans.async_to_literal(ctx, python_val, t, expected.union_type.variants[i])
+                    res = await attempt
+                else:
+                    res = trans.to_literal(ctx, python_val, t, expected.union_type.variants[i])
                 if found_res:
+                    print(f"Current type {get_args(python_type)[i]} old res {res_type}")
                     is_ambiguous = True
+                res_type = _add_tag_to_type(trans.get_literal_type(t), trans.name)
                 found_res = True
             except Exception as e:
-                logger.debug(f"Failed to convert from {python_val} to {t} with error: {e}", exc_info=True)
+                logger.warning(
+                    f"UnionTransformer failed attempt to convert from {python_val} to {t} error: {e}",
+                )
                 continue
 
         if is_ambiguous:
@@ -1657,7 +1833,9 @@ def to_literal(self, ctx: FlyteContext, python_val: T, python_type: Type[T], exp
 
         raise TypeTransformerFailedError(f"Cannot convert from {python_val} to {python_type}")
 
-    def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[T]) -> Optional[typing.Any]:
+    async def async_to_python_value(
+        self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[T]
+    ) -> Optional[typing.Any]:
         expected_python_type = get_underlying_type(expected_python_type)
 
         if lv.scalar is not None and lv.scalar.binary is not None:
@@ -1675,6 +1853,7 @@ def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type:
         cur_transformer = ""
         res = None
         res_tag = None
+        # This is serial, not actually async, but should be okay since it's more reasonable for Unions.
         for v in get_args(expected_python_type):
             try:
                 trans: TypeTransformer[T] = TypeEngine.get_transformer(v)
@@ -1689,13 +1868,22 @@ def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type:
                     assert lv.scalar is not None  # type checker
                     assert lv.scalar.union is not None  # type checker
 
-                    res = trans.to_python_value(ctx, lv.scalar.union.value, v)
+                    if isinstance(trans, AsyncTypeTransformer):
+                        res = await trans.async_to_python_value(ctx, lv.scalar.union.value, v)
+                    else:
+                        res = trans.to_python_value(ctx, lv.scalar.union.value, v)
+                        if isinstance(res, asyncio.Future):
+                            res = await res
+
                     if found_res:
                         is_ambiguous = True
                         cur_transformer = trans.name
                         break
                 else:
-                    res = trans.to_python_value(ctx, lv, v)
+                    if isinstance(trans, AsyncTypeTransformer):
+                        res = await trans.async_to_python_value(ctx, lv, v)
+                    else:
+                        res = trans.to_python_value(ctx, lv, v)
                     if found_res:
                         is_ambiguous = True
                         cur_transformer = trans.name
@@ -1723,7 +1911,7 @@ def guess_python_type(self, literal_type: LiteralType) -> type:
         raise ValueError(f"Union transformer cannot reverse {literal_type}")
 
 
-class DictTransformer(TypeTransformer[dict]):
+class DictTransformer(AsyncTypeTransformer[dict]):
     """
     Transformer that transforms an univariate dictionary Dict[str, T] to a Literal Map or
     transforms an untyped dictionary to a Binary Scalar Literal with a Struct Literal Type.
@@ -1810,7 +1998,7 @@ def get_literal_type(self, t: Type[dict]) -> LiteralType:
                     raise ValueError(f"Type of Generic List type is not supported, {e}")
         return _type_models.LiteralType(simple=_type_models.SimpleType.STRUCT)
 
-    def to_literal(
+    async def async_to_literal(
         self, ctx: FlyteContext, python_val: typing.Any, python_type: Type[dict], expected: LiteralType
     ) -> Literal:
         if type(python_val) != dict:
@@ -1836,10 +2024,17 @@ def to_literal(
             else:
                 _, v_type = self.extract_types_or_metadata(python_type)
 
-            lit_map[k] = TypeEngine.to_literal(ctx, v, cast(type, v_type), expected.map_value_type)
+            lit_map[k] = asyncio.create_task(
+                TypeEngine.async_to_literal(ctx, v, cast(type, v_type), expected.map_value_type)
+            )
+
+        await asyncio.gather(*lit_map.values())
+        for k, v in lit_map.items():
+            lit_map[k] = v.result()
+
         return Literal(map=LiteralMap(literals=lit_map))
 
-    def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[dict]) -> dict:
+    async def async_to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[dict]) -> dict:
         if lv and lv.scalar and lv.scalar.binary is not None:
             return self.from_binary_idl(lv.scalar.binary, expected_python_type)  # type: ignore
 
@@ -1856,7 +2051,13 @@ def to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type:
                 raise TypeError("TypeMismatch. Destination dictionary does not accept 'str' key")
             py_map = {}
             for k, v in lv.map.literals.items():
-                py_map[k] = TypeEngine.to_python_value(ctx, v, cast(Type, tp[1]))
+                fut = asyncio.create_task(TypeEngine.async_to_python_value(ctx, v, cast(Type, tp[1])))
+                py_map[k] = fut
+
+            await asyncio.gather(*py_map.values())
+            for k, v in py_map.items():
+                py_map[k] = v.result()
+
             return py_map
 
         # for empty generic we have to explicitly test for lv.scalar.generic is not None as empty dict
diff --git a/flytekit/extend/backend/agent_service.py b/flytekit/extend/backend/agent_service.py
index 9b444d101e9..5919ab06117 100644
--- a/flytekit/extend/backend/agent_service.py
+++ b/flytekit/extend/backend/agent_service.py
@@ -133,8 +133,8 @@ async def GetTask(self, request: GetTaskRequest, context: grpc.ServicerContext)
             agent = AgentRegistry.get_agent(request.task_type)
         logger.info(f"{agent.name} start checking the status of the job")
         res = await mirror_async_methods(agent.get, resource_meta=agent.metadata_type.decode(request.resource_meta))
-
-        return GetTaskResponse(resource=res.to_flyte_idl())
+        resource = await res.to_flyte_idl()
+        return GetTaskResponse(resource=resource)
 
     @record_agent_metrics
     async def DeleteTask(self, request: DeleteTaskRequest, context: grpc.ServicerContext) -> DeleteTaskResponse:
@@ -167,7 +167,8 @@ async def ExecuteTaskSync(
                     agent.do, task_template=template, inputs=literal_map, output_prefix=output_prefix
                 )
 
-                header = ExecuteTaskSyncResponseHeader(resource=res.to_flyte_idl())
+                resource = await res.to_flyte_idl()
+                header = ExecuteTaskSyncResponseHeader(resource=resource)
                 yield ExecuteTaskSyncResponse(header=header)
             request_success_count.labels(task_type=task_type, operation=do_operation).inc()
         except Exception as e:
diff --git a/flytekit/extend/backend/base_agent.py b/flytekit/extend/backend/base_agent.py
index f8264edc922..2f973e94f07 100644
--- a/flytekit/extend/backend/base_agent.py
+++ b/flytekit/extend/backend/base_agent.py
@@ -31,7 +31,6 @@
 from flytekit.exceptions.user import FlyteUserException
 from flytekit.extend.backend.utils import is_terminal_phase, mirror_async_methods, render_task_template
 from flytekit.loggers import set_flytekit_log_properties
-from flytekit.models import common
 from flytekit.models.literals import LiteralMap
 from flytekit.models.task import TaskExecutionMetadata, TaskTemplate
 
@@ -80,7 +79,7 @@ def decode(cls, data: bytes) -> "ResourceMeta":
 
 
 @dataclass
-class Resource(common.FlyteIdlEntity):
+class Resource:
     """
     This is the output resource of the job.
 
@@ -97,14 +96,19 @@ class Resource(common.FlyteIdlEntity):
     outputs: Optional[Union[LiteralMap, typing.Dict[str, Any]]] = None
     custom_info: Optional[typing.Dict[str, Any]] = None
 
-    def to_flyte_idl(self) -> _Resource:
+    async def to_flyte_idl(self) -> _Resource:
+        """
+        This function is async to call the async type engine functions. This is okay to do because this is not a
+        normal model class that inherits from FlyteIdlEntity
+        """
         if self.outputs is None:
             outputs = None
         elif isinstance(self.outputs, LiteralMap):
             outputs = self.outputs.to_flyte_idl()
         else:
             ctx = FlyteContext.current_context()
-            outputs = TypeEngine.dict_to_literal_map_pb(ctx, self.outputs)
+
+            outputs = await TypeEngine.dict_to_literal_map_pb(ctx, self.outputs)
 
         return _Resource(
             phase=self.phase,
@@ -301,7 +305,7 @@ async def _do(
     ) -> Resource:
         try:
             ctx = FlyteContext.current_context()
-            literal_map = TypeEngine.dict_to_literal_map(ctx, inputs or {}, self.get_input_types())
+            literal_map = await TypeEngine._dict_to_literal_map(ctx, inputs or {}, self.get_input_types())
             return await mirror_async_methods(
                 agent.do, task_template=template, inputs=literal_map, output_prefix=output_prefix
             )
@@ -347,7 +351,7 @@ def execute(self: PythonTask, **kwargs) -> LiteralMap:
             return LiteralMap.from_flyte_idl(output_proto)
 
         if resource.outputs and not isinstance(resource.outputs, LiteralMap):
-            return TypeEngine.dict_to_literal_map(ctx, resource.outputs)
+            return TypeEngine.dict_to_literal_map(ctx, resource.outputs)  # type: ignore
 
         return resource.outputs
 
@@ -361,7 +365,7 @@ async def _create(
 
             with FlyteContextManager.with_context(cb) as ctx:
                 # Write the inputs to a remote file, so that the remote task can read the inputs from this file.
-                literal_map = TypeEngine.dict_to_literal_map(ctx, inputs or {}, self.get_input_types())
+                literal_map = await TypeEngine._dict_to_literal_map(ctx, inputs or {}, self.get_input_types())
                 path = ctx.file_access.get_random_local_path()
                 utils.write_proto_to_file(literal_map.to_flyte_idl(), path)
                 ctx.file_access.put_data(path, f"{output_prefix}/inputs.pb")
diff --git a/flytekit/remote/remote.py b/flytekit/remote/remote.py
index 46054805421..31b69cee1a7 100644
--- a/flytekit/remote/remote.py
+++ b/flytekit/remote/remote.py
@@ -87,7 +87,6 @@
 from flytekit.remote.lazy_entity import LazyEntity
 from flytekit.remote.remote_callable import RemoteEntity
 from flytekit.remote.remote_fs import get_flyte_fs
-from flytekit.tools.asyn import run_sync
 from flytekit.tools.fast_registration import FastPackageOptions, fast_package
 from flytekit.tools.interactive import ipython_check
 from flytekit.tools.script_mode import _find_project_root, compress_scripts, get_all_modules, hash_file
@@ -98,6 +97,7 @@
     get_serializable,
     get_serializable_launch_plan,
 )
+from flytekit.utils.asyn import run_sync
 
 if typing.TYPE_CHECKING:
     try:
diff --git a/flytekit/sensor/sensor_engine.py b/flytekit/sensor/sensor_engine.py
index ac718abe359..36e4a4aaa03 100644
--- a/flytekit/sensor/sensor_engine.py
+++ b/flytekit/sensor/sensor_engine.py
@@ -25,7 +25,7 @@ async def create(self, task_template: TaskTemplate, inputs: Optional[LiteralMap]
             python_interface_inputs = {
                 name: TypeEngine.guess_python_type(lt.type) for name, lt in task_template.interface.inputs.items()
             }
-            native_inputs = TypeEngine.literal_map_to_kwargs(ctx, inputs, python_interface_inputs)
+            native_inputs = await TypeEngine._literal_map_to_kwargs(ctx, inputs, python_interface_inputs)
             sensor_metadata.inputs = native_inputs
 
         return sensor_metadata
diff --git a/flytekit/tools/repo.py b/flytekit/tools/repo.py
index 0617c871aea..2d10355f2b8 100644
--- a/flytekit/tools/repo.py
+++ b/flytekit/tools/repo.py
@@ -330,7 +330,7 @@ def _raw_register(cp_entity: FlyteControlPlaneEntity):
             secho(og_id, "failed")
 
     async def _register(entities: typing.List[task.TaskSpec]):
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         tasks = []
         for entity in entities:
             tasks.append(loop.run_in_executor(None, functools.partial(_raw_register, entity)))
diff --git a/flytekit/tools/script_mode.py b/flytekit/tools/script_mode.py
index 7188b5b90dd..fa8634361df 100644
--- a/flytekit/tools/script_mode.py
+++ b/flytekit/tools/script_mode.py
@@ -118,6 +118,7 @@ def ls_files(
     else:
         all_files = list_all_files(source_path, deref_symlinks, ignore_group)
 
+    all_files.sort()
     hasher = hashlib.md5()
     for abspath in all_files:
         relpath = os.path.relpath(abspath, source_path)
diff --git a/flytekit/types/file/file.py b/flytekit/types/file/file.py
index bacde7089f2..602f5bc12eb 100644
--- a/flytekit/types/file/file.py
+++ b/flytekit/types/file/file.py
@@ -17,7 +17,12 @@
 
 from flytekit.core.constants import MESSAGEPACK
 from flytekit.core.context_manager import FlyteContext, FlyteContextManager
-from flytekit.core.type_engine import TypeEngine, TypeTransformer, TypeTransformerFailedError, get_underlying_type
+from flytekit.core.type_engine import (
+    AsyncTypeTransformer,
+    TypeEngine,
+    TypeTransformerFailedError,
+    get_underlying_type,
+)
 from flytekit.exceptions.user import FlyteAssertion
 from flytekit.loggers import logger
 from flytekit.models.core import types as _core_types
@@ -350,7 +355,7 @@ def __str__(self):
         return self.path
 
 
-class FlyteFilePathTransformer(TypeTransformer[FlyteFile]):
+class FlyteFilePathTransformer(AsyncTypeTransformer[FlyteFile]):
     def __init__(self):
         super().__init__(name="FlyteFilePath", t=FlyteFile)
 
@@ -428,7 +433,7 @@ def validate_file_type(
             if real_type not in expected_type:
                 raise ValueError(f"Incorrect file type, expected {expected_type}, got {real_type}")
 
-    def to_literal(
+    async def async_to_literal(
         self,
         ctx: FlyteContext,
         python_val: typing.Union[FlyteFile, os.PathLike, str],
@@ -549,7 +554,7 @@ def from_binary_idl(
         else:
             raise TypeTransformerFailedError(f"Unsupported binary format: `{binary_idl_object.tag}`")
 
-    def to_python_value(
+    async def async_to_python_value(
         self, ctx: FlyteContext, lv: Literal, expected_python_type: typing.Union[typing.Type[FlyteFile], os.PathLike]
     ) -> FlyteFile:
         # Handle dataclass attribute access
diff --git a/flytekit/utils/__init__.py b/flytekit/utils/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/flytekit/tools/asyn.py b/flytekit/utils/asyn.py
similarity index 94%
rename from flytekit/tools/asyn.py
rename to flytekit/utils/asyn.py
index a1264056728..c447db052f1 100644
--- a/flytekit/tools/asyn.py
+++ b/flytekit/utils/asyn.py
@@ -11,15 +11,20 @@ async def async_add(a: int, b: int) -> int:
 
 import asyncio
 import atexit
+import functools
 import os
 import threading
 from contextlib import contextmanager
 from typing import Any, Awaitable, Callable, TypeVar
 
+from typing_extensions import ParamSpec
+
 from flytekit.loggers import logger
 
 T = TypeVar("T")
 
+P = ParamSpec("P")
+
 
 @contextmanager
 def _selector_policy():
@@ -87,13 +92,13 @@ def run_sync(self, coro_func: Callable[..., Awaitable[T]], *args, **kwargs) -> T
             self._runner_map[name] = _TaskRunner()
         return self._runner_map[name].run(coro)
 
-    def synced(self, coro_func: Callable[..., Awaitable[T]]) -> Callable[..., T]:
+    def synced(self, coro_func: Callable[P, Awaitable[T]]) -> Callable[P, T]:
         """Make loop run coroutine until it returns. Runs in other thread"""
 
+        @functools.wraps(coro_func)
         def wrapped(*args: Any, **kwargs: Any) -> T:
             return self.run_sync(coro_func, *args, **kwargs)
 
-        wrapped.__doc__ = coro_func.__doc__
         return wrapped
 
 
diff --git a/plugins/flytekit-flyteinteractive/tests/test_flyteinteractive_vscode.py b/plugins/flytekit-flyteinteractive/tests/test_flyteinteractive_vscode.py
index 54aba7a63a4..de1a386a8b0 100644
--- a/plugins/flytekit-flyteinteractive/tests/test_flyteinteractive_vscode.py
+++ b/plugins/flytekit-flyteinteractive/tests/test_flyteinteractive_vscode.py
@@ -202,7 +202,7 @@ def wf(a: int, b: int):
     mock_process.assert_called_once()
     mock_exit_handler.assert_called_once()
     mock_prepare_interactive_python.assert_called_once()
-    mock_signal.assert_called_once()
+    assert mock_signal.call_count >= 1
     mock_prepare_resume_task_python.assert_called_once()
     mock_prepare_launch_json.assert_called_once()
 
diff --git a/tests/flytekit/unit/cli/pyflyte/test_script_mode.py b/tests/flytekit/unit/cli/pyflyte/test_script_mode.py
index 74d8aeab73d..b063091075e 100644
--- a/tests/flytekit/unit/cli/pyflyte/test_script_mode.py
+++ b/tests/flytekit/unit/cli/pyflyte/test_script_mode.py
@@ -6,6 +6,7 @@
 from flytekit.tools.script_mode import ls_files
 from flytekit.constants import CopyFileDetection
 
+
 # a pytest fixture that creates a tmp directory and creates
 # a small file structure in it
 @pytest.fixture
@@ -39,7 +40,7 @@ def test_list_dir(dummy_dir_structure):
     files, d = ls_files(str(dummy_dir_structure), CopyFileDetection.ALL)
     assert len(files) == 5
     if os.name != "nt":
-        assert d == "c092f1b85f7c6b2a71881a946c00a855"
+        assert d == "b6907fd823a45e26c780a4ba62111243"
 
 
 def test_list_filtered_on_modules(dummy_dir_structure):
diff --git a/tests/flytekit/unit/core/test_array_node.py b/tests/flytekit/unit/core/test_array_node.py
index c861d392684..2f4c3145bab 100644
--- a/tests/flytekit/unit/core/test_array_node.py
+++ b/tests/flytekit/unit/core/test_array_node.py
@@ -3,7 +3,8 @@
 
 import pytest
 
-from flytekit import LaunchPlan, current_context, task, workflow
+from flytekit import LaunchPlan, task, workflow
+from flytekit.core.context_manager import FlyteContextManager
 from flytekit.configuration import Image, ImageConfig, SerializationSettings
 from flytekit.core.array_node import array_node
 from flytekit.core.array_node_map_task import map_task
@@ -35,7 +36,8 @@ def parent_wf(a: int, b: typing.Union[int, str], c: int = 2) -> int:
     return multiply(val=a, val1=b, val2=c)
 
 
-lp = LaunchPlan.get_default_launch_plan(current_context(), parent_wf)
+ctx = FlyteContextManager.current_context()
+lp = LaunchPlan.get_default_launch_plan(ctx, parent_wf)
 
 
 @workflow
@@ -103,7 +105,8 @@ def ex_task(val: int) -> int:
     def ex_wf(val: int) -> int:
         return ex_task(val=val)
 
-    ex_lp = LaunchPlan.get_default_launch_plan(current_context(), ex_wf)
+    ctx = FlyteContextManager.current_context()
+    ex_lp = LaunchPlan.get_default_launch_plan(ctx, ex_wf)
 
     @workflow
     def grandparent_ex_wf() -> typing.List[typing.Optional[int]]:
diff --git a/tests/flytekit/unit/core/test_data.py b/tests/flytekit/unit/core/test_data.py
index aa308d7929f..2de6e8c1969 100644
--- a/tests/flytekit/unit/core/test_data.py
+++ b/tests/flytekit/unit/core/test_data.py
@@ -3,18 +3,20 @@
 import shutil
 import tempfile
 from uuid import UUID
-
+import typing
 import fsspec
 import mock
 import pytest
 from s3fs import S3FileSystem
 
 from flytekit.configuration import Config, DataConfig, S3Config
-from flytekit.core.context_manager import FlyteContextManager
+from flytekit.core.context_manager import FlyteContextManager, FlyteContext
 from flytekit.core.data_persistence import FileAccessProvider, get_fsspec_storage_options, s3_setup_args
 from flytekit.core.type_engine import TypeEngine
 from flytekit.types.directory.types import FlyteDirectory
 from flytekit.types.file import FlyteFile
+from flytekit.utils.asyn import loop_manager
+from flytekit.models.literals import Literal
 
 local = fsspec.filesystem("file")
 root = os.path.abspath(os.sep)
@@ -446,3 +448,47 @@ def test_s3_metadata():
         res = [(x, y) for x, y in res]
         files = [os.path.join(x, y) for x, y in res]
         assert len(files) == 2
+
+
+async def dummy_output_to_literal_map(ctx: FlyteContext, ff: typing.List[FlyteFile]) -> Literal:
+    lt = TypeEngine.to_literal_type(typing.List[FlyteFile])
+    lit = await TypeEngine.async_to_literal(ctx, ff, typing.List[FlyteFile], lt)
+    return lit
+
+
+@pytest.mark.sandbox_test
+def test_async_local_copy_to_s3():
+    import time
+    import datetime
+
+    f1 = "/Users/ytong/go/src/github.com/unionai/debugyt/user/ytong/src/yt_dbg/fr/rand.file"
+    f2 = "/Users/ytong/go/src/github.com/unionai/debugyt/user/ytong/src/yt_dbg/fr/rand2.file"
+    f3 = "/Users/ytong/go/src/github.com/unionai/debugyt/user/ytong/src/yt_dbg/fr/rand3.file"
+
+    ff1 = FlyteFile(path=f1)
+    ff2 = FlyteFile(path=f2)
+    ff3 = FlyteFile(path=f3)
+    ff = [ff1, ff2, ff3]
+
+    ctx = FlyteContextManager.current_context()
+    dc = Config.for_sandbox().data_config
+    random_folder = UUID(int=random.getrandbits(64)).hex
+    raw_output = f"s3://my-s3-bucket/testing/upload_test/{random_folder}"
+    print(f"Uploading to {raw_output}")
+    provider = FileAccessProvider(local_sandbox_dir="/tmp/unittest", raw_output_prefix=raw_output, data_config=dc)
+
+    start_time = datetime.datetime.now(datetime.timezone.utc)
+    start_wall_time = time.perf_counter()
+    start_process_time = time.process_time()
+
+    with FlyteContextManager.with_context(ctx.with_file_access(provider)) as ctx:
+        lit = loop_manager.run_sync(dummy_output_to_literal_map, ctx, ff)
+        print(lit)
+
+    end_time = datetime.datetime.now(datetime.timezone.utc)
+    end_wall_time = time.perf_counter()
+    end_process_time = time.process_time()
+
+    print(f"Time taken: {end_time - start_time}")
+    print(f"Wall time taken: {end_wall_time - start_wall_time}")
+    print(f"Process time taken: {end_process_time - start_process_time}")
diff --git a/tests/flytekit/unit/core/test_flyte_file.py b/tests/flytekit/unit/core/test_flyte_file.py
index cce4f35afc9..7e09e918aea 100644
--- a/tests/flytekit/unit/core/test_flyte_file.py
+++ b/tests/flytekit/unit/core/test_flyte_file.py
@@ -105,20 +105,25 @@ def my_wf(path: FlyteFile[typing.TypeVar("txt")]) -> FlyteFile:
     with open(res, "r") as fh:
         assert fh.read() == "Hello World"
 
+
 def test_flytefile_in_dataclass(local_dummy_txt_file):
     TxtFile = FlyteFile[typing.TypeVar("txt")]
+
     @dataclass
     class DC:
         f: TxtFile
+
     @task
     def t1(path: TxtFile) -> DC:
         return DC(f=path)
+
     @workflow
     def my_wf(path: TxtFile) -> DC:
         dc = t1(path=path)
         return dc
 
     txt_file = TxtFile(local_dummy_txt_file)
+    my_wf.compile()
     dc1 = my_wf(path=txt_file)
     with open(dc1.f, "r") as fh:
         assert fh.read() == "Hello World"
@@ -126,6 +131,7 @@ def my_wf(path: TxtFile) -> DC:
     dc2 = DC(f=txt_file)
     assert dc1 == dc2
 
+
 @pytest.mark.skipif(not can_import("magic"), reason="Libmagic is not installed")
 def test_mismatching_file_types(local_dummy_txt_file):
     @task
diff --git a/tests/flytekit/unit/core/test_promise.py b/tests/flytekit/unit/core/test_promise.py
index ca8d6e95c2e..455d53a5eb5 100644
--- a/tests/flytekit/unit/core/test_promise.py
+++ b/tests/flytekit/unit/core/test_promise.py
@@ -224,8 +224,9 @@ def my_workflow() -> (str, str, str):
     assert o3 == "b"
 
 
+@pytest.mark.asyncio
 @pytest.mark.skipif("pandas" not in sys.modules, reason="Pandas is not installed.")
-def test_resolve_attr_path_in_promise():
+async def test_resolve_attr_path_in_promise():
     @dataclass_json
     @dataclass
     class Foo:
@@ -242,15 +243,16 @@ class Foo:
     src_promise = Promise("val1", src_lit)
 
     # happy path
-    tgt_promise = resolve_attr_path_in_promise(src_promise["a"][0]["b"])
+    tgt_promise = await resolve_attr_path_in_promise(src_promise["a"][0]["b"])
     assert "foo" == TypeEngine.to_python_value(FlyteContextManager.current_context(), tgt_promise.val, str)
 
     # exception
     with pytest.raises(FlytePromiseAttributeResolveException):
-        tgt_promise = resolve_attr_path_in_promise(src_promise["c"])
+        await resolve_attr_path_in_promise(src_promise["c"])
 
 
-def test_prom_with_union_literals():
+@pytest.mark.asyncio
+async def test_prom_with_union_literals():
     ctx = FlyteContextManager.current_context()
     pt = typing.Union[str, int]
     lt = TypeEngine.to_literal_type(pt)
@@ -259,9 +261,9 @@ def test_prom_with_union_literals():
         LiteralType(simple=SimpleType.INTEGER, structure=TypeStructure(tag="int")),
     ]
 
-    bd = binding_data_from_python_std(ctx, lt, 3, pt, [])
+    bd = await binding_data_from_python_std(ctx, lt, 3, pt, [])
     assert bd.scalar.union.stored_type.structure.tag == "int"
-    bd = binding_data_from_python_std(ctx, lt, "hello", pt, [])
+    bd = await binding_data_from_python_std(ctx, lt, "hello", pt, [])
     assert bd.scalar.union.stored_type.structure.tag == "str"
 
 def test_pickling_promise_object():
diff --git a/tests/flytekit/unit/core/test_type_hints.py b/tests/flytekit/unit/core/test_type_hints.py
index 3b0eeb13a5b..1d0165751fd 100644
--- a/tests/flytekit/unit/core/test_type_hints.py
+++ b/tests/flytekit/unit/core/test_type_hints.py
@@ -737,8 +737,9 @@ def print_expr(expr):
 
 
 def test_comparison_lits():
-    px = Promise("x", TypeEngine.to_literal(None, 5, int, None))
-    py = Promise("y", TypeEngine.to_literal(None, 8, int, None))
+    ctx = context_manager.FlyteContextManager.current_context()
+    px = Promise("x", TypeEngine.to_literal(ctx, 5, int, None))
+    py = Promise("y", TypeEngine.to_literal(ctx, 8, int, None))
 
     def eval_expr(expr, expected: bool):
         print(f"{expr} evals to {expr.eval()}")
diff --git a/tests/flytekit/unit/extend/test_agent.py b/tests/flytekit/unit/extend/test_agent.py
index 946bf3a7789..f16369dae6d 100644
--- a/tests/flytekit/unit/extend/test_agent.py
+++ b/tests/flytekit/unit/extend/test_agent.py
@@ -21,7 +21,6 @@
 from flyteidl.core.identifier_pb2 import ResourceType
 
 from flytekit import PythonFunctionTask, task
-from flytekit.clis.sdk_in_container.serve import print_agents_metadata
 from flytekit.configuration import (
     FastSerializationSettings,
     Image,
@@ -59,6 +58,7 @@
 from flytekit.models.security import Identity
 from flytekit.models.task import TaskExecutionMetadata, TaskTemplate
 from flytekit.tools.translator import get_serializable
+from flytekit.utils.asyn import loop_manager
 
 dummy_id = "dummy_id"
 
@@ -440,7 +440,7 @@ def test_resource_type():
     o = Resource(
         phase=TaskExecution.SUCCEEDED,
     )
-    v = o.to_flyte_idl()
+    v = loop_manager.run_sync(o.to_flyte_idl)
     assert v
     assert v.phase == TaskExecution.SUCCEEDED
     assert len(v.log_links) == 0
@@ -458,7 +458,7 @@ def test_resource_type():
         outputs={"o0": 1},
         custom_info={"custom": "info", "num": 1},
     )
-    v = o.to_flyte_idl()
+    v = loop_manager.run_sync(o.to_flyte_idl)
     assert v
     assert v.phase == TaskExecution.SUCCEEDED
     assert v.log_links[0].name == "console"
diff --git a/tests/flytekit/unit/interactive/test_flyteinteractive_vscode.py b/tests/flytekit/unit/interactive/test_flyteinteractive_vscode.py
index 20f1c7c10c7..4102ee84b54 100644
--- a/tests/flytekit/unit/interactive/test_flyteinteractive_vscode.py
+++ b/tests/flytekit/unit/interactive/test_flyteinteractive_vscode.py
@@ -2,6 +2,7 @@
 
 import mock
 import pytest
+import sys
 from flytekit.interactive import (
     DEFAULT_CODE_SERVER_DIR_NAMES,
     DEFAULT_CODE_SERVER_EXTENSIONS,
@@ -90,11 +91,12 @@ def wf():
     mock_process.assert_called_once()
     mock_exit_handler.assert_called_once()
     mock_prepare_interactive_python.assert_called_once()
-    mock_signal.assert_called_once()
+    assert mock_signal.call_count >=1
     mock_prepare_resume_task_python.assert_called_once()
     mock_prepare_launch_json.assert_called_once()
 
 
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="asyncio signal behavior diff")
 def test_vscode_remote_execution_but_disable(vscode_patches, mock_remote_execution):
     (
         mock_process,
@@ -120,7 +122,7 @@ def wf():
     mock_process.assert_not_called()
     mock_exit_handler.assert_not_called()
     mock_prepare_interactive_python.assert_not_called()
-    mock_signal.assert_not_called()
+    assert mock_signal.call_count == 0
     mock_prepare_resume_task_python.assert_not_called()
     mock_prepare_launch_json.assert_not_called()
 
@@ -150,7 +152,6 @@ def wf():
     mock_process.assert_not_called()
     mock_exit_handler.assert_not_called()
     mock_prepare_interactive_python.assert_not_called()
-    mock_signal.assert_not_called()
     mock_prepare_resume_task_python.assert_not_called()
     mock_prepare_launch_json.assert_not_called()
 
@@ -170,6 +171,7 @@ def wf(a: int, b: int) -> int:
     assert res == 15
 
 
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="asyncio signal behavior diff")
 def test_vscode_run_task_first_fail(vscode_patches, mock_remote_execution):
     (
         mock_process,
@@ -196,7 +198,7 @@ def wf(a: int, b: int):
     mock_process.assert_called_once()
     mock_exit_handler.assert_called_once()
     mock_prepare_interactive_python.assert_called_once()
-    mock_signal.assert_called_once()
+    assert mock_signal.call_count >= 1
     mock_prepare_resume_task_python.assert_called_once()
     mock_prepare_launch_json.assert_called_once()
 
@@ -222,6 +224,7 @@ def test_vscode_config():
     assert config.extension_remote_paths == DEFAULT_CODE_SERVER_EXTENSIONS
 
 
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="asyncio signal behavior diff")
 def test_vscode_with_args(vscode_patches, mock_remote_execution):
     (
         mock_process,
@@ -248,7 +251,7 @@ def wf():
     mock_process.assert_called_once()
     mock_exit_handler.assert_called_once()
     mock_prepare_interactive_python.assert_called_once()
-    mock_signal.assert_called_once()
+    assert mock_signal.call_count >= 1
     mock_prepare_resume_task_python.assert_called_once()
     mock_prepare_launch_json.assert_called_once()
 
diff --git a/tests/flytekit/unit/tools/test_asyn.py b/tests/flytekit/unit/utils/test_asyn.py
similarity index 98%
rename from tests/flytekit/unit/tools/test_asyn.py
rename to tests/flytekit/unit/utils/test_asyn.py
index 0a3ffeb2d3c..db74ac6f539 100644
--- a/tests/flytekit/unit/tools/test_asyn.py
+++ b/tests/flytekit/unit/utils/test_asyn.py
@@ -4,7 +4,7 @@
 from typing import List, Dict, Optional
 from asyncio import get_running_loop
 from functools import partial
-from flytekit.tools.asyn import run_sync, loop_manager
+from flytekit.utils.asyn import run_sync, loop_manager
 
 from contextvars import ContextVar