Repository: PrefectHQ/prefect

Issue:
Error while deploying flow: FileNotFoundError(2, "No such file or directory: 'prefect'")
## Description

Getting "No such file or directory: 'prefect'" error when running a simple task from the web UI. (Note there are no directories within my project named "prefect")


## Expected Behavior

Expected the task to run and display "success" on the web page.


## Reproduction



    from prefect import task, Flow

    @task
    def the_simple():
        print('hello')


    if __name__ == '__main__':

        with Flow('the-simplest') as flow:
            the_simple()

        flow.register('test-amazon')




## Environment


Operating system is Amazon Linux 2.
Have also tried running on Kali linux, and the same error shows up.

`$ env/bin/prefect diagnostics
{
  "config_overrides": {},
  "env_vars": [],
  "system_information": {
    "platform": "Linux-4.14.203-156.332.amzn2.x86_64-x86_64-with-glibc2.2.5",
    "prefect_backend": "server",
    "prefect_version": "0.13.18",
    "python_version": "3.7.9"
  }
}`


Discussion:

Patch:
diff --git a/changes/pr3917.yaml b/changes/pr3917.yaml
deleted file mode 100644
index b3f8be12fa..0000000000
--- a/changes/pr3917.yaml
+++ /dev/null
@@ -1,2 +0,0 @@
-fix:
-  - "Remove unused `description` field on `Task` serializer - [#3917](https://github.com/PrefectHQ/prefect/pull/3917)"
diff --git a/changes/pr3918.yaml b/changes/pr3918.yaml
new file mode 100644
index 0000000000..277a6e42ce
--- /dev/null
+++ b/changes/pr3918.yaml
@@ -0,0 +1,2 @@
+enhancement:
+  - "Support executing Prefect agents/flows without having the `prefect` CLI on path - [#3918](https://github.com/PrefectHQ/prefect/pull/3918)"
diff --git a/src/prefect/__main__.py b/src/prefect/__main__.py
new file mode 100644
index 0000000000..98dcca0c28
--- /dev/null
+++ b/src/prefect/__main__.py
@@ -0,0 +1,4 @@
+from .cli import cli
+
+if __name__ == "__main__":
+    cli()
diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
index ba7f0502d6..9b2ab0efcb 100644
--- a/src/prefect/agent/local/agent.py
+++ b/src/prefect/agent/local/agent.py
@@ -157,7 +157,7 @@ class LocalAgent(Agent):
         # show flow logs, these log entries will continue to stream to the users terminal
         # until these child processes exit, even if the agent has already exited.
         p = Popen(
-            ["prefect", "execute", "flow-run"],
+            [sys.executable, "-m", "prefect", "execute", "flow-run"],
             stdout=stdout,
             stderr=STDOUT,
             env=env,
diff --git a/src/prefect/engine/cloud/flow_runner.py b/src/prefect/engine/cloud/flow_runner.py
index 25b0ca1fb6..91e2b26ab9 100644
--- a/src/prefect/engine/cloud/flow_runner.py
+++ b/src/prefect/engine/cloud/flow_runner.py
@@ -1,5 +1,6 @@
 import os
 import signal
+import sys
 import threading
 from time import sleep as time_sleep
 from typing import Any, Callable, Dict, Iterable, Optional, Iterator
@@ -73,7 +74,15 @@ class CloudFlowRunner(FlowRunner):
             # use empty string for testing purposes
             flow_run_id = prefect.context.get("flow_run_id", "")  # type: str
             self.client.update_flow_run_heartbeat(flow_run_id)
-            self.heartbeat_cmd = ["prefect", "heartbeat", "flow-run", "-i", flow_run_id]
+            self.heartbeat_cmd = [
+                sys.executable,
+                "-m",
+                "prefect",
+                "heartbeat",
+                "flow-run",
+                "-i",
+                flow_run_id,
+            ]
 
             query = {
                 "query": {
diff --git a/src/prefect/serialization/task.py b/src/prefect/serialization/task.py
index 6618823abc..76204da3b5 100644
--- a/src/prefect/serialization/task.py
+++ b/src/prefect/serialization/task.py
@@ -71,6 +71,7 @@ class TaskSchema(TaskMethodsMixin, ObjectSchema):
     type = fields.Function(lambda task: to_qualified_name(type(task)), lambda x: x)
     name = fields.String(allow_none=True)
     slug = fields.String(allow_none=True)
+    description = fields.String(allow_none=True)
     tags = SortedList(fields.String())
     max_retries = fields.Integer(allow_none=True)
     retry_delay = fields.TimeDelta(allow_none=True)
@@ -122,5 +123,6 @@ class ParameterSchema(TaskMethodsMixin, ObjectSchema):
     slug = fields.String(allow_none=True)
     default = JSONCompatible(allow_none=True)
     required = fields.Boolean(allow_none=True)
+    description = fields.String(allow_none=True)
     tags = SortedList(fields.String())
     outputs = fields.Method("load_outputs", allow_none=True)


Test patch:
diff --git a/tests/agent/test_local_agent.py b/tests/agent/test_local_agent.py
index 42766e06e2..88e2418bd8 100644
--- a/tests/agent/test_local_agent.py
+++ b/tests/agent/test_local_agent.py
@@ -1,5 +1,6 @@
 import os
 import socket
+import sys
 from unittest.mock import MagicMock
 
 import pytest
@@ -446,7 +447,7 @@ def test_local_agent_heartbeat(monkeypatch, returncode, show_flow_logs, logs):
     popen = MockPopen()
     # expect a process to be called with the following command (with specified behavior)
     popen.set_command(
-        "prefect execute flow-run",
+        [sys.executable, "-m", "prefect", "execute", "flow-run"],
         stdout=b"awesome output!",
         stderr=b"blerg, eRroR!",
         returncode=returncode,
diff --git a/tests/engine/cloud/test_cloud_flow_runner.py b/tests/engine/cloud/test_cloud_flow_runner.py
index c0de291cee..6ff2b29a1e 100644
--- a/tests/engine/cloud/test_cloud_flow_runner.py
+++ b/tests/engine/cloud/test_cloud_flow_runner.py
@@ -1,5 +1,6 @@
 import datetime
 import itertools
+import sys
 import time
 import threading
 from datetime import timedelta
@@ -413,7 +414,15 @@ def test_flow_runner_heartbeat_sets_command(monkeypatch, setting_available):
         res = runner._heartbeat()
 
     assert res is True
-    assert runner.heartbeat_cmd == ["prefect", "heartbeat", "flow-run", "-i", "foo"]
+    assert runner.heartbeat_cmd == [
+        sys.executable,
+        "-m",
+        "prefect",
+        "heartbeat",
+        "flow-run",
+        "-i",
+        "foo",
+    ]
 
 
 def test_flow_runner_does_not_have_heartbeat_if_disabled(monkeypatch):