Agent uploads input.pb even if task input is None (#2054)

Signed-off-by: Kevin Su <pingsutw@apache.org>

flytekit/core/data_persistence.py

@@ -128,7 +128,7 @@ def __init__(
         self._local = fsspec.filesystem(None)
 
         self._data_config = data_config if data_config else DataConfig.auto()
-        self._default_protocol = get_protocol(raw_output_prefix)
+        self._default_protocol = get_protocol(str(raw_output_prefix))
         self._default_remote = cast(fsspec.AbstractFileSystem, self.get_filesystem(self._default_protocol))
         if os.name == "nt" and raw_output_prefix.startswith("file://"):
             raise FlyteAssertion("Cannot use the file:// prefix on Windows.")

flytekit/extend/backend/base_agent.py

@@ -204,11 +204,10 @@ async def _create(
         literals = inputs or {}
         for k, v in inputs.items():
             literals[k] = TypeEngine.to_literal(ctx, v, type(v), self._entity.interface.inputs[k].type)
-        literal_map = LiteralMap(literals) if literals else None
-        if literal_map and isinstance(self, PythonFunctionTask):
+        if isinstance(self, PythonFunctionTask):
             # Write the inputs to a remote file, so that the remote task can read the inputs from this file.
             path = ctx.file_access.get_random_local_path()
-            utils.write_proto_to_file(literal_map.to_flyte_idl(), path)
+            utils.write_proto_to_file(LiteralMap(literals).to_flyte_idl(), path)
             ctx.file_access.put_data(path, f"{output_prefix}/inputs.pb")
             task_template = render_task_template(task_template, output_prefix)