diff --git a/src/promptflow/promptflow/_cli/_utils.py b/src/promptflow/promptflow/_cli/_utils.py
index d6cf454d61c..fd675a014ef 100644
--- a/src/promptflow/promptflow/_cli/_utils.py
+++ b/src/promptflow/promptflow/_cli/_utils.py
@@ -23,6 +23,7 @@
 
 from promptflow._sdk._utils import print_red_error, print_yellow_warning
 from promptflow._utils.utils import is_in_ci_pipeline
+from promptflow._utils.exception_utils import ExceptionPresenter
 from promptflow.exceptions import ErrorTarget, PromptflowException, UserErrorException
 
 AzureMLWorkspaceTriad = namedtuple("AzureMLWorkspace", ["subscription_id", "resource_group_name", "workspace_name"])
@@ -334,6 +335,12 @@ def pretty_print_dataframe_as_table(df: pd.DataFrame) -> None:
     print(tabulate(df, headers="keys", tablefmt="grid", maxcolwidths=column_widths, maxheadercolwidths=column_widths))
 
 
+def is_format_exception():
+    if os.environ.get("PROMPTFLOW_STRUCTURE_EXCEPTION_OUTPUT", "false").lower() == "true":
+        return True
+    return False
+
+
 def exception_handler(command: str):
     """Catch known cli exceptions."""
 
@@ -342,9 +349,18 @@ def decorator(func):
         def wrapper(*args, **kwargs):
             try:
                 return func(*args, **kwargs)
-            except PromptflowException as e:
-                print_red_error(f"{command} failed with {e.__class__.__name__}: {str(e)}")
-                exit(1)
+            except Exception as e:
+                if is_format_exception():
+                    # When the flag format_exception is set in command,
+                    # it will write a json with exception info and command to stderr.
+                    error_msg = ExceptionPresenter.create(e).to_dict(include_debug_info=True)
+                    error_msg["command"] = " ".join(sys.argv)
+                    sys.stderr.write(json.dumps(error_msg))
+                if isinstance(e, PromptflowException):
+                    print_red_error(f"{command} failed with {e.__class__.__name__}: {str(e)}")
+                    exit(1)
+                else:
+                    raise e
 
         return wrapper
 
diff --git a/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py b/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py
index e32631c66f3..a241a745060 100644
--- a/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py
+++ b/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py
@@ -1102,3 +1102,46 @@ def test_pf_run_no_stream_log(self):
         assert "Executing node print_val. node run id:" not in f.getvalue()
         # executor logs won't stream
         assert "Node print_val completes." not in f.getvalue()
+
+    def test_format_cli_exception(self, capsys):
+        from promptflow._sdk.operations._connection_operations import ConnectionOperations
+
+        with patch.dict(os.environ, {"PROMPTFLOW_STRUCTURE_EXCEPTION_OUTPUT": "true"}):
+            with pytest.raises(SystemExit):
+                run_pf_command(
+                    "connection",
+                    "show",
+                    "--name",
+                    "invalid_connection_name",
+                )
+            outerr = capsys.readouterr()
+            assert outerr.err
+            error_msg = json.loads(outerr.err)
+            assert error_msg["code"] == "ConnectionNotFoundError"
+
+            def mocked_connection_get(*args, **kwargs):
+                raise Exception("mock exception")
+
+            with patch.object(ConnectionOperations, "get") as mock_connection_get:
+                mock_connection_get.side_effect = mocked_connection_get
+                with pytest.raises(Exception):
+                    run_pf_command(
+                        "connection",
+                        "show",
+                        "--name",
+                        "invalid_connection_name",
+                    )
+                outerr = capsys.readouterr()
+                assert outerr.err
+                error_msg = json.loads(outerr.err)
+                assert error_msg["code"] == "SystemError"
+
+        with pytest.raises(SystemExit):
+            run_pf_command(
+                "connection",
+                "show",
+                "--name",
+                "invalid_connection_name",
+            )
+        outerr = capsys.readouterr()
+        assert not outerr.err