diff --git a/.vscode/settings.json b/.vscode/settings.json
index 1ee0dd7b..dc5cf55a 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -1,11 +1,11 @@
 {
-    "python.testing.pytestArgs": [
-        "operator"
-    ],
-    "python.testing.unittestEnabled": false,
-    "python.testing.pytestEnabled": true,
-    "python.formatting.provider": "black",
-    "gefyra.up": {
-        "minikube": false
-    },
-}
\ No newline at end of file
+  "python.testing.pytestArgs": ["tests/"],
+  "python.testing.cwd": "${workspaceFolder}/client/",
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true,
+  "python.formatting.provider": "black",
+  "gefyra.up": {
+    "minikube": false
+  },
+  "python.analysis.include": ["client"]
+}
diff --git a/client/gefyra/api/run.py b/client/gefyra/api/run.py
index d77ec989..649bdb67 100644
--- a/client/gefyra/api/run.py
+++ b/client/gefyra/api/run.py
@@ -2,8 +2,9 @@
 import os
 import sys
 from threading import Thread, Event
-from typing import Dict, List, Optional, TYPE_CHECKING
-from gefyra.cluster.utils import retrieve_pod_and_container
+from typing import Dict, List, Optional, TYPE_CHECKING, Tuple
+from gefyra.cli.utils import FileFromArgument
+from gefyra.cluster.utils import get_file_from_pod_container, retrieve_pod_and_container
 
 if TYPE_CHECKING:
     from docker.models.containers import Container
@@ -30,6 +31,62 @@ def check_input():
         stop_thread.set()
 
 
+def copy_files_or_directories_from_cluster_to_local_container(
+    config,
+    namespace: str,
+    local_container_name: str,
+    file_from: Tuple[FileFromArgument],
+):
+    from kubernetes.client import ApiException
+
+    for file_from_argument in file_from:
+        try:
+            file_from_pod, file_from_container = retrieve_pod_and_container(
+                file_from_argument.workload, namespace=namespace, config=config
+            )
+        except ApiException as e:
+            logger.error(
+                f"Cannot copy file/directory from pod: {e.reason} ({e.status})."
+            )
+            continue
+
+        logger.debug(
+            f"Copying file/directory from {file_from_pod}/{file_from_container}"
+        )
+
+        try:
+            tar_buffer = get_file_from_pod_container(
+                config,
+                file_from_pod,
+                namespace,
+                file_from_container,
+                file_from_argument.source,
+            )
+        except (RuntimeError, ApiException) as e:
+            logger.info(
+                f"Cannot copy file/directory from pod:"
+                f" --file-from {str(file_from_argument.workload)}:"
+                f"{file_from_argument.source}:{file_from_argument.destination}"
+            )
+            logger.debug(e)
+            continue
+
+        try:
+            client = config.DOCKER
+            client.containers.get(local_container_name).put_archive(
+                file_from_argument.destination,
+                tar_buffer.read(),
+            )
+        except Exception as e:
+            logger.info(
+                f"Cannot copy file/directory into container:"
+                f" --file-from {str(file_from_argument.workload)}:"
+                f"{file_from_argument.source}:{file_from_argument.destination}"
+            )
+            logger.debug(e)
+            continue
+
+
 @stopwatch
 def run(
     image: str,
@@ -43,6 +100,7 @@ def run(
     namespace: str = "",
     env: Optional[List] = None,
     env_from: str = "",
+    file_from: Optional[Tuple[FileFromArgument]] = None,
     pull: str = "missing",
     platform: str = "linux/amd64",
 ) -> bool:
@@ -100,6 +158,7 @@ def run(
     except ApiException as e:
         logger.error(f"Cannot copy environment from Pod: {e.reason} ({e.status}).")
         return False
+
     if env:
         env_overrides = generate_env_dict_from_strings(env)
         env_dict.update(env_overrides)
@@ -131,6 +190,14 @@ def run(
         else:
             raise RuntimeError(e.explanation)
 
+    if file_from:
+        copy_files_or_directories_from_cluster_to_local_container(
+            config,
+            namespace=namespace,
+            local_container_name=name,
+            file_from=file_from,
+        )
+
     logger.info(
         f"Container image '{', '.join(container.image.tags)}' started with name"
         f" '{container.name}' in Kubernetes namespace '{namespace}' (from {ns_source})"
diff --git a/client/gefyra/cli/run.py b/client/gefyra/cli/run.py
index 9585cc18..7871090f 100644
--- a/client/gefyra/cli/run.py
+++ b/client/gefyra/cli/run.py
@@ -4,6 +4,7 @@
     OptionEatAll,
     check_connection_name,
     parse_env,
+    parse_file_from,
     parse_ip_port_map,
     parse_workload,
 )
@@ -40,6 +41,13 @@
     type=str,
     callback=parse_workload,
 )
+@click.option(
+    "--file-from",
+    help="Copy the file from the container in the notation 'Pod/Container'",
+    type=str,
+    callback=parse_file_from,
+    multiple=True,
+)
 @click.option(
     "-v",
     "--volume",
@@ -102,6 +110,7 @@ def run(
     auto_remove,
     expose,
     env_from,
+    file_from,
     volume,
     env,
     namespace,
@@ -116,12 +125,14 @@ def run(
 
     if command:
         command = ast.literal_eval(command)[0]
+
     api.run(
         image=image,
         name=name,
         command=command,
         namespace=namespace,
         env_from=env_from,
+        file_from=file_from,
         env=env,
         ports=expose,
         auto_remove=auto_remove,
diff --git a/client/gefyra/cli/utils.py b/client/gefyra/cli/utils.py
index 8ecb5d2d..360770e9 100644
--- a/client/gefyra/cli/utils.py
+++ b/client/gefyra/cli/utils.py
@@ -2,6 +2,7 @@
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 import click
 from click import ClickException
+from typing import NamedTuple
 
 
 def standard_error_handler(func):
@@ -237,6 +238,47 @@ def parse_workload(ctx, param, workload: str) -> str:
     return workload
 
 
+class FileFromArgument(NamedTuple):
+    workload: str
+    source: str
+    destination: str = "/"
+
+
+def parse_file_from(
+    ctx, param, file_from_arguments
+) -> Optional[Tuple[FileFromArgument]]:
+    if not file_from_arguments:
+        return None
+
+    file_from = []
+    for file_from_argument in file_from_arguments:
+        split_argument = file_from_argument.split(":")
+        if len(split_argument) not in (2, 3) or "/" not in split_argument[0]:
+            raise ValueError(
+                (
+                    "Invalid argument format. Please provide the file "
+                    "in the format 'type/name[/container-name]:src[:dest]'."
+                )
+            )
+
+        if len(split_argument) == 2:
+            workload, source = split_argument
+            destination = source
+        else:
+            workload, source, destination = split_argument
+
+        file_from_argument_parsed = FileFromArgument(
+            workload=workload,
+            source=source,
+            destination=destination,
+        )
+
+        if file_from_argument_parsed not in file_from:
+            file_from.append(file_from_argument_parsed)
+
+    return tuple(file_from)
+
+
 def check_connection_name(ctx, param, selected: Optional[str] = None) -> str:
     from gefyra import api
 
diff --git a/client/gefyra/cluster/utils.py b/client/gefyra/cluster/utils.py
index 1f0d7787..e6beda1b 100644
--- a/client/gefyra/cluster/utils.py
+++ b/client/gefyra/cluster/utils.py
@@ -2,6 +2,7 @@
 from time import sleep
 from typing import Tuple, TYPE_CHECKING
 from gefyra.api.utils import get_workload_type
+import io
 
 if TYPE_CHECKING:
     from kubernetes.client.models import V1Pod
@@ -52,6 +53,70 @@ def get_env_from_pod_container(
     )
 
 
+def get_file_from_pod_container(
+    config: ClientConfiguration,
+    pod_name: str,
+    namespace: str,
+    container_name: str,
+    source: str,
+) -> io.BytesIO:
+    from kubernetes.client import ApiException
+    from kubernetes.stream import stream
+
+    retries = 10
+    counter = 0
+    interval = 1
+    while counter < retries:
+        try:
+            import os
+
+            exec_command = [
+                "/bin/sh",
+                "-c",
+                f"cd {os.path.dirname(source)} && tar cf - {os.path.basename(source)}",
+            ]
+
+            resp = stream(
+                config.K8S_CORE_API.connect_get_namespaced_pod_exec,
+                pod_name,
+                namespace,
+                container=container_name,
+                command=exec_command,
+                stderr=True,
+                stdin=False,
+                stdout=True,
+                tty=False,
+                _preload_content=False,
+            )
+
+            tar_buffer = io.BytesIO()
+            while resp.is_open():
+                resp.update(timeout=1)
+                if resp.peek_stdout():
+                    tar_buffer.write(resp.read_stdout().encode("utf-8"))
+                if resp.peek_stderr():
+                    logger.debug("Error:", resp.read_stderr())
+
+            resp.close()
+
+            tar_buffer.seek(0)
+            return tar_buffer
+        except ApiException as e:
+            if "500 Internal Server Error" in e.reason:
+                sleep(interval)
+                counter += 1
+                logger.debug(
+                    f"Failed to copy file from pod {pod_name} in namespace {namespace} on"
+                    f" try {counter}."
+                )
+            else:
+                raise e
+    raise RuntimeError(
+        f"Failed to copy file from pod {pod_name} in namespace {namespace} after"
+        f" {retries} tries."
+    )
+
+
 def get_container(pod, container_name: str):
     for container in pod.spec.containers:
         if container.name == container_name:
diff --git a/client/tests/unit/test_file_from.py b/client/tests/unit/test_file_from.py
new file mode 100644
index 00000000..239462ae
--- /dev/null
+++ b/client/tests/unit/test_file_from.py
@@ -0,0 +1,77 @@
+import unittest
+from gefyra.cli.utils import FileFromArgument, parse_file_from
+
+
+class ParseFileFromTest(unittest.TestCase):
+    def test_invalid_input_string(self):
+        with self.assertRaises(ValueError):
+            parse_file_from(None, None, ["invalid"])
+
+    def test_invalid_input_format(self):
+        with self.assertRaises(ValueError):
+            parse_file_from(
+                None,
+                None,
+                ["deployment/hello-world:/home/test.txt:/home/test.txt:/home/test.txt"],
+            )
+
+    def test_workload_source_destination(self):
+        file_from_arguments = parse_file_from(
+            None,
+            None,
+            ["deployment/hello-world/container:/home/test.txt:/home/test.txt"],
+        )
+        expected = (
+            FileFromArgument(
+                workload="deployment/hello-world/container",
+                source="/home/test.txt",
+                destination="/home/test.txt",
+            ),
+        )
+        assert file_from_arguments == expected
+
+    def test_workload_source(self):
+        file_from_arguments = parse_file_from(
+            None, None, ["deployment/hello-world/container:/home/test.txt"]
+        )
+        expected = (
+            FileFromArgument(
+                workload="deployment/hello-world/container",
+                source="/home/test.txt",
+                destination="/home/test.txt",
+            ),
+        )
+        assert file_from_arguments == expected
+
+    def test_duplicate_input(self):
+        file_from_arguments = parse_file_from(
+            None,
+            None,
+            [
+                "deployment/hello-world/container:/home/test.txt",
+                "deployment/hello-world/container:/home/test.txt",
+            ],
+        )
+        expected = (
+            FileFromArgument(
+                workload="deployment/hello-world/container",
+                source="/home/test.txt",
+                destination="/home/test.txt",
+            ),
+        )
+        assert file_from_arguments == expected
+
+    def test_folder(self):
+        file_from_arguments = parse_file_from(
+            None,
+            None,
+            ["deployment/hello-world/container:/home:/home"],
+        )
+        expected = (
+            FileFromArgument(
+                workload="deployment/hello-world/container",
+                source="/home",
+                destination="/home",
+            ),
+        )
+        assert file_from_arguments == expected