We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Change "PYTHONPATH" construct for local agent
## Current behavior
Local agent has the "--import-path" feature, but its now availible only for unix enviroment, because PYTHONPATH is concatenated with ":"
current_env["PYTHONPATH"] = ":".join(python_path)

## Proposed behavior
Use  "os.pathsep" instead ":", to get the correct value aswell for Windows users



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/agent/local/agent.py`
Content:
```
1 import os
2 import socket
3 import sys
4 from subprocess import STDOUT, Popen, DEVNULL
5 from typing import Iterable, List
6 
7 from prefect import config
8 from prefect.agent import Agent
9 from prefect.environments.storage import Docker
10 from prefect.run_configs import LocalRun
11 from prefect.serialization.storage import StorageSchema
12 from prefect.serialization.run_config import RunConfigSchema
13 from prefect.utilities.agent import get_flow_run_command
14 from prefect.utilities.graphql import GraphQLResult
15 
16 
17 class LocalAgent(Agent):
18     """
19     Agent which deploys flow runs locally as subprocesses. There are a range of kwarg
20     options to control information which may be provided to these subprocesses.
21 
22     Optional import paths may be specified to append dependency modules to the PATH:
23     ```
24     prefect agent start local --import-path "/usr/local/my_module" --import-path "~/other_module"
25 
26     # Now the local scripts/packages my_module and other_module will be importable in
27     # the flow's subprocess
28     ```
29 
30     Environment variables may be set on the agent to be provided to each flow run's subprocess:
31     ```
32     prefect agent start local --env MY_SECRET_KEY=secret --env OTHER_VAR=$OTHER_VAR
33     ```
34 
35     Args:
36         - agent_config_id (str, optional): An optional agent configuration ID that can be used to set
37             configuration based on an agent from a backend API. If set all configuration values will be
38             pulled from backend agent configuration.
39         - name (str, optional): An optional name to give this agent. Can also be set through
40             the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to "agent"
41         - labels (List[str], optional): a list of labels, which are arbitrary string
42             identifiers used by Prefect Agents when polling for work
43         - env_vars (dict, optional): a dictionary of environment variables and values that will
44             be set on each flow run that this agent submits for execution
45         - max_polls (int, optional): maximum number of times the agent will poll Prefect Cloud
46             for flow runs; defaults to infinite
47         - agent_address (str, optional):  Address to serve internal api at. Currently this is
48             just health checks for use by an orchestration layer. Leave blank for no api server
49             (default).
50         - no_cloud_logs (bool, optional): Disable logging to a Prefect backend for this agent
51             and all deployed flow runs
52         - import_paths (List[str], optional): system paths which will be provided to each
53             Flow's runtime environment; useful for Flows which import from locally hosted
54             scripts or packages
55         - show_flow_logs (bool, optional): a boolean specifying whether the agent should
56             re-route Flow run logs to stdout; defaults to `False`
57         - hostname_label (boolean, optional): a boolean specifying whether this agent should
58             auto-label itself with the hostname of the machine it is running on.  Useful for
59             flows which are stored on the local filesystem.
60         - storage_labels (boolean, optional): a boolean specifying whether this agent should
61             auto-label itself with all of the storage options labels.
62     """
63 
64     def __init__(
65         self,
66         agent_config_id: str = None,
67         name: str = None,
68         labels: Iterable[str] = None,
69         env_vars: dict = None,
70         import_paths: List[str] = None,
71         show_flow_logs: bool = False,
72         hostname_label: bool = True,
73         max_polls: int = None,
74         agent_address: str = None,
75         no_cloud_logs: bool = False,
76         storage_labels: bool = True,
77     ) -> None:
78         self.processes = set()
79         self.import_paths = import_paths or []
80         self.show_flow_logs = show_flow_logs
81         super().__init__(
82             agent_config_id=agent_config_id,
83             name=name,
84             labels=labels,
85             env_vars=env_vars,
86             max_polls=max_polls,
87             agent_address=agent_address,
88             no_cloud_logs=no_cloud_logs,
89         )
90         hostname = socket.gethostname()
91 
92         # Resolve common Docker hostname by using IP
93         if hostname == "docker-desktop":
94             hostname = socket.gethostbyname(hostname)
95 
96         if hostname_label and (hostname not in self.labels):
97             assert isinstance(self.labels, list)
98             self.labels.append(hostname)
99 
100         if storage_labels:
101             all_storage_labels = [
102                 "azure-flow-storage",
103                 "gcs-flow-storage",
104                 "s3-flow-storage",
105                 "github-flow-storage",
106                 "webhook-flow-storage",
107                 "gitlab-flow-storage",
108             ]
109             for label in all_storage_labels:
110                 if label not in self.labels:
111                     self.labels.append(label)
112 
113         self.logger.debug(f"Import paths: {self.import_paths}")
114         self.logger.debug(f"Show flow logs: {self.show_flow_logs}")
115 
116     def heartbeat(self) -> None:
117         for process in list(self.processes):
118             if process.poll() is not None:
119                 self.processes.remove(process)
120                 if process.returncode:
121                     self.logger.info(
122                         "Process PID {} returned non-zero exit code".format(process.pid)
123                     )
124         super().heartbeat()
125 
126     def deploy_flow(self, flow_run: GraphQLResult) -> str:
127         """
128         Deploy flow runs on your local machine as Docker containers
129 
130         Args:
131             - flow_run (GraphQLResult): A GraphQLResult flow run object
132 
133         Returns:
134             - str: Information about the deployment
135 
136         Raises:
137             - ValueError: if deployment attempted on unsupported Storage type
138         """
139         self.logger.info("Deploying flow run {}".format(flow_run.id))  # type: ignore
140 
141         storage = StorageSchema().load(flow_run.flow.storage)
142         if isinstance(storage, Docker):
143             self.logger.error(
144                 "Flow run %s has an unsupported storage type: `%s`",
145                 flow_run.id,
146                 type(storage).__name__,
147             )
148             raise TypeError("Unsupported Storage type: %s" % type(storage).__name__)
149 
150         # If the flow is using a run_config, load it
151         if getattr(flow_run.flow, "run_config", None) is not None:
152             run_config = RunConfigSchema().load(flow_run.flow.run_config)
153             if not isinstance(run_config, LocalRun):
154                 self.logger.error(
155                     "Flow run %s has a `run_config` of type `%s`, only `LocalRun` is supported",
156                     flow_run.id,
157                     type(run_config).__name__,
158                 )
159                 raise TypeError(
160                     "Unsupported RunConfig type: %s" % type(run_config).__name__
161                 )
162         else:
163             run_config = None
164 
165         env = self.populate_env_vars(flow_run, run_config=run_config)
166 
167         working_dir = None if run_config is None else run_config.working_dir
168         if working_dir and not os.path.exists(working_dir):
169             msg = f"Flow run {flow_run.id} has a nonexistent `working_dir` configured: {working_dir}"
170             self.logger.error(msg)
171             raise ValueError(msg)
172 
173         stdout = sys.stdout if self.show_flow_logs else DEVNULL
174 
175         # note: we will allow these processes to be orphaned if the agent were to exit
176         # before the flow runs have completed. The lifecycle of the agent should not
177         # dictate the lifecycle of the flow run. However, if the user has elected to
178         # show flow logs, these log entries will continue to stream to the users terminal
179         # until these child processes exit, even if the agent has already exited.
180         p = Popen(
181             get_flow_run_command(flow_run).split(" "),
182             stdout=stdout,
183             stderr=STDOUT,
184             env=env,
185             cwd=working_dir,
186         )
187 
188         self.processes.add(p)
189         self.logger.debug(
190             "Submitted flow run {} to process PID {}".format(flow_run.id, p.pid)
191         )
192 
193         return "PID: {}".format(p.pid)
194 
195     def populate_env_vars(
196         self, flow_run: GraphQLResult, run_config: LocalRun = None
197     ) -> dict:
198         """
199         Populate metadata and variables in the environment variables for a flow run
200 
201         Args:
202             - flow_run (GraphQLResult): A flow run object
203             - run_config (LocalRun, optional): The `run_config` for the flow, if any.
204 
205         Returns:
206             - dict: a dictionary representing the populated environment variables
207         """
208         # Environment variables come from (later options override):
209 
210         # 1. Local environment
211         env = os.environ.copy()
212 
213         # 2. Logging config
214         env.update({"PREFECT__LOGGING__LEVEL": config.logging.level})
215 
216         # 3. PYTHONPATH
217         # XXX: I'm 99% sure this isn't needed, a new Python process should
218         # always add the CWD as the first entry in sys.path
219         python_path = [
220             run_config.working_dir
221             if run_config is not None and run_config.working_dir
222             else os.getcwd()
223         ]
224         if os.environ.get("PYTHONPATH"):
225             python_path.append(os.environ["PYTHONPATH"])
226         if self.import_paths:
227             python_path.extend(self.import_paths)
228         env["PYTHONPATH"] = ":".join(python_path)
229 
230         # 4. Values set on the agent via `--env`
231         env.update(self.env_vars)
232 
233         # 5. Values set on a LocalRun RunConfig (if present
234         if run_config is not None and run_config.env is not None:
235             env.update(run_config.env)
236 
237         # 6. Non-overrideable required env vars
238         env.update(
239             {
240                 "PREFECT__CLOUD__API": config.cloud.api,
241                 "PREFECT__CLOUD__AUTH_TOKEN": self.client._api_token,
242                 "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
243                 "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
244                 "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
245                 "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
246                 "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
247                 "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
248                 "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
249             }
250         )
251 
252         # Filter out None values
253         return {k: v for k, v in env.items() if v is not None}
254 
255     @staticmethod
256     def generate_supervisor_conf(
257         token: str = None,
258         labels: Iterable[str] = None,
259         import_paths: List[str] = None,
260         show_flow_logs: bool = False,
261     ) -> str:
262         """
263         Generate and output an installable supervisorctl configuration file for the agent.
264 
265         Args:
266             - token (str, optional): A `RUNNER` token to give the agent
267             - labels (List[str], optional): a list of labels, which are arbitrary string
268                 identifiers used by Prefect Agents when polling for work
269             - import_paths (List[str], optional): system paths which will be provided to each
270                 Flow's runtime environment; useful for Flows which import from locally hosted
271                 scripts or packages
272             - show_flow_logs (bool, optional): a boolean specifying whether the agent should
273                 re-route Flow run logs to stdout; defaults to `False`
274 
275         Returns:
276             - str: A string representation of the generated configuration file
277         """
278 
279         # Use defaults if not provided
280         token = token or ""
281         labels = labels or []
282         import_paths = import_paths or []
283 
284         with open(
285             os.path.join(os.path.dirname(__file__), "supervisord.conf"), "r"
286         ) as conf_file:
287             conf = conf_file.read()
288 
289         add_opts = ""
290         add_opts += "-t {token} ".format(token=token) if token else ""
291         add_opts += "-f " if show_flow_logs else ""
292         add_opts += (
293             " ".join("-l {label} ".format(label=label) for label in labels)
294             if labels
295             else ""
296         )
297         add_opts += (
298             " ".join("-p {path}".format(path=path) for path in import_paths)
299             if import_paths
300             else ""
301         )
302         conf = conf.replace("{{OPTS}}", add_opts)
303         return conf
304 
305 
306 if __name__ == "__main__":
307     LocalAgent().start()
308
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
--- a/src/prefect/agent/local/agent.py
+++ b/src/prefect/agent/local/agent.py
@@ -225,7 +225,7 @@
             python_path.append(os.environ["PYTHONPATH"])
         if self.import_paths:
             python_path.extend(self.import_paths)
-        env["PYTHONPATH"] = ":".join(python_path)
+        env["PYTHONPATH"] = os.pathsep.join(python_path)
 
         # 4. Values set on the agent via `--env`
         env.update(self.env_vars)

{"golden_diff": "diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py\n--- a/src/prefect/agent/local/agent.py\n+++ b/src/prefect/agent/local/agent.py\n@@ -225,7 +225,7 @@\n             python_path.append(os.environ[\"PYTHONPATH\"])\n         if self.import_paths:\n             python_path.extend(self.import_paths)\n-        env[\"PYTHONPATH\"] = \":\".join(python_path)\n+        env[\"PYTHONPATH\"] = os.pathsep.join(python_path)\n \n         # 4. Values set on the agent via `--env`\n         env.update(self.env_vars)\n", "issue": "Change \"PYTHONPATH\" construct for local agent\n## Current behavior\r\nLocal agent has the \"--import-path\" feature, but its now availible only for unix enviroment, because PYTHONPATH is concatenated with \":\"\r\ncurrent_env[\"PYTHONPATH\"] = \":\".join(python_path)\r\n\r\n## Proposed behavior\r\nUse  \"os.pathsep\" instead \":\", to get the correct value aswell for Windows users\r\n\r\n\n", "before_files": [{"content": "import os\nimport socket\nimport sys\nfrom subprocess import STDOUT, Popen, DEVNULL\nfrom typing import Iterable, List\n\nfrom prefect import config\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.run_configs import LocalRun\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.serialization.run_config import RunConfigSchema\nfrom prefect.utilities.agent import get_flow_run_command\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as subprocesses. There are a range of kwarg\n    options to control information which may be provided to these subprocesses.\n\n    Optional import paths may be specified to append dependency modules to the PATH:\n    ```\n    prefect agent start local --import-path \"/usr/local/my_module\" --import-path \"~/other_module\"\n\n    # Now the local scripts/packages my_module and other_module will be importable in\n    # the flow's subprocess\n    ```\n\n    Environment variables may be set on the agent to be provided to each flow run's subprocess:\n    ```\n    prefect agent start local --env MY_SECRET_KEY=secret --env OTHER_VAR=$OTHER_VAR\n    ```\n\n    Args:\n        - agent_config_id (str, optional): An optional agent configuration ID that can be used to set\n            configuration based on an agent from a backend API. If set all configuration values will be\n            pulled from backend agent configuration.\n        - name (str, optional): An optional name to give this agent. Can also be set through\n            the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to \"agent\"\n        - labels (List[str], optional): a list of labels, which are arbitrary string\n            identifiers used by Prefect Agents when polling for work\n        - env_vars (dict, optional): a dictionary of environment variables and values that will\n            be set on each flow run that this agent submits for execution\n        - max_polls (int, optional): maximum number of times the agent will poll Prefect Cloud\n            for flow runs; defaults to infinite\n        - agent_address (str, optional):  Address to serve internal api at. Currently this is\n            just health checks for use by an orchestration layer. Leave blank for no api server\n            (default).\n        - no_cloud_logs (bool, optional): Disable logging to a Prefect backend for this agent\n            and all deployed flow runs\n        - import_paths (List[str], optional): system paths which will be provided to each\n            Flow's runtime environment; useful for Flows which import from locally hosted\n            scripts or packages\n        - show_flow_logs (bool, optional): a boolean specifying whether the agent should\n            re-route Flow run logs to stdout; defaults to `False`\n        - hostname_label (boolean, optional): a boolean specifying whether this agent should\n            auto-label itself with the hostname of the machine it is running on.  Useful for\n            flows which are stored on the local filesystem.\n        - storage_labels (boolean, optional): a boolean specifying whether this agent should\n            auto-label itself with all of the storage options labels.\n    \"\"\"\n\n    def __init__(\n        self,\n        agent_config_id: str = None,\n        name: str = None,\n        labels: Iterable[str] = None,\n        env_vars: dict = None,\n        import_paths: List[str] = None,\n        show_flow_logs: bool = False,\n        hostname_label: bool = True,\n        max_polls: int = None,\n        agent_address: str = None,\n        no_cloud_logs: bool = False,\n        storage_labels: bool = True,\n    ) -> None:\n        self.processes = set()\n        self.import_paths = import_paths or []\n        self.show_flow_logs = show_flow_logs\n        super().__init__(\n            agent_config_id=agent_config_id,\n            name=name,\n            labels=labels,\n            env_vars=env_vars,\n            max_polls=max_polls,\n            agent_address=agent_address,\n            no_cloud_logs=no_cloud_logs,\n        )\n        hostname = socket.gethostname()\n\n        # Resolve common Docker hostname by using IP\n        if hostname == \"docker-desktop\":\n            hostname = socket.gethostbyname(hostname)\n\n        if hostname_label and (hostname not in self.labels):\n            assert isinstance(self.labels, list)\n            self.labels.append(hostname)\n\n        if storage_labels:\n            all_storage_labels = [\n                \"azure-flow-storage\",\n                \"gcs-flow-storage\",\n                \"s3-flow-storage\",\n                \"github-flow-storage\",\n                \"webhook-flow-storage\",\n                \"gitlab-flow-storage\",\n            ]\n            for label in all_storage_labels:\n                if label not in self.labels:\n                    self.labels.append(label)\n\n        self.logger.debug(f\"Import paths: {self.import_paths}\")\n        self.logger.debug(f\"Show flow logs: {self.show_flow_logs}\")\n\n    def heartbeat(self) -> None:\n        for process in list(self.processes):\n            if process.poll() is not None:\n                self.processes.remove(process)\n                if process.returncode:\n                    self.logger.info(\n                        \"Process PID {} returned non-zero exit code\".format(process.pid)\n                    )\n        super().heartbeat()\n\n    def deploy_flow(self, flow_run: GraphQLResult) -> str:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_run (GraphQLResult): A GraphQLResult flow run object\n\n        Returns:\n            - str: Information about the deployment\n\n        Raises:\n            - ValueError: if deployment attempted on unsupported Storage type\n        \"\"\"\n        self.logger.info(\"Deploying flow run {}\".format(flow_run.id))  # type: ignore\n\n        storage = StorageSchema().load(flow_run.flow.storage)\n        if isinstance(storage, Docker):\n            self.logger.error(\n                \"Flow run %s has an unsupported storage type: `%s`\",\n                flow_run.id,\n                type(storage).__name__,\n            )\n            raise TypeError(\"Unsupported Storage type: %s\" % type(storage).__name__)\n\n        # If the flow is using a run_config, load it\n        if getattr(flow_run.flow, \"run_config\", None) is not None:\n            run_config = RunConfigSchema().load(flow_run.flow.run_config)\n            if not isinstance(run_config, LocalRun):\n                self.logger.error(\n                    \"Flow run %s has a `run_config` of type `%s`, only `LocalRun` is supported\",\n                    flow_run.id,\n                    type(run_config).__name__,\n                )\n                raise TypeError(\n                    \"Unsupported RunConfig type: %s\" % type(run_config).__name__\n                )\n        else:\n            run_config = None\n\n        env = self.populate_env_vars(flow_run, run_config=run_config)\n\n        working_dir = None if run_config is None else run_config.working_dir\n        if working_dir and not os.path.exists(working_dir):\n            msg = f\"Flow run {flow_run.id} has a nonexistent `working_dir` configured: {working_dir}\"\n            self.logger.error(msg)\n            raise ValueError(msg)\n\n        stdout = sys.stdout if self.show_flow_logs else DEVNULL\n\n        # note: we will allow these processes to be orphaned if the agent were to exit\n        # before the flow runs have completed. The lifecycle of the agent should not\n        # dictate the lifecycle of the flow run. However, if the user has elected to\n        # show flow logs, these log entries will continue to stream to the users terminal\n        # until these child processes exit, even if the agent has already exited.\n        p = Popen(\n            get_flow_run_command(flow_run).split(\" \"),\n            stdout=stdout,\n            stderr=STDOUT,\n            env=env,\n            cwd=working_dir,\n        )\n\n        self.processes.add(p)\n        self.logger.debug(\n            \"Submitted flow run {} to process PID {}\".format(flow_run.id, p.pid)\n        )\n\n        return \"PID: {}\".format(p.pid)\n\n    def populate_env_vars(\n        self, flow_run: GraphQLResult, run_config: LocalRun = None\n    ) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n            - run_config (LocalRun, optional): The `run_config` for the flow, if any.\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        # Environment variables come from (later options override):\n\n        # 1. Local environment\n        env = os.environ.copy()\n\n        # 2. Logging config\n        env.update({\"PREFECT__LOGGING__LEVEL\": config.logging.level})\n\n        # 3. PYTHONPATH\n        # XXX: I'm 99% sure this isn't needed, a new Python process should\n        # always add the CWD as the first entry in sys.path\n        python_path = [\n            run_config.working_dir\n            if run_config is not None and run_config.working_dir\n            else os.getcwd()\n        ]\n        if os.environ.get(\"PYTHONPATH\"):\n            python_path.append(os.environ[\"PYTHONPATH\"])\n        if self.import_paths:\n            python_path.extend(self.import_paths)\n        env[\"PYTHONPATH\"] = \":\".join(python_path)\n\n        # 4. Values set on the agent via `--env`\n        env.update(self.env_vars)\n\n        # 5. Values set on a LocalRun RunConfig (if present\n        if run_config is not None and run_config.env is not None:\n            env.update(run_config.env)\n\n        # 6. Non-overrideable required env vars\n        env.update(\n            {\n                \"PREFECT__CLOUD__API\": config.cloud.api,\n                \"PREFECT__CLOUD__AUTH_TOKEN\": self.client._api_token,\n                \"PREFECT__CLOUD__AGENT__LABELS\": str(self.labels),\n                \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n                \"PREFECT__CONTEXT__FLOW_ID\": flow_run.flow.id,  # type: ignore\n                \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n                \"PREFECT__LOGGING__LOG_TO_CLOUD\": str(self.log_to_cloud).lower(),\n                \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n                \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n            }\n        )\n\n        # Filter out None values\n        return {k: v for k, v in env.items() if v is not None}\n\n    @staticmethod\n    def generate_supervisor_conf(\n        token: str = None,\n        labels: Iterable[str] = None,\n        import_paths: List[str] = None,\n        show_flow_logs: bool = False,\n    ) -> str:\n        \"\"\"\n        Generate and output an installable supervisorctl configuration file for the agent.\n\n        Args:\n            - token (str, optional): A `RUNNER` token to give the agent\n            - labels (List[str], optional): a list of labels, which are arbitrary string\n                identifiers used by Prefect Agents when polling for work\n            - import_paths (List[str], optional): system paths which will be provided to each\n                Flow's runtime environment; useful for Flows which import from locally hosted\n                scripts or packages\n            - show_flow_logs (bool, optional): a boolean specifying whether the agent should\n                re-route Flow run logs to stdout; defaults to `False`\n\n        Returns:\n            - str: A string representation of the generated configuration file\n        \"\"\"\n\n        # Use defaults if not provided\n        token = token or \"\"\n        labels = labels or []\n        import_paths = import_paths or []\n\n        with open(\n            os.path.join(os.path.dirname(__file__), \"supervisord.conf\"), \"r\"\n        ) as conf_file:\n            conf = conf_file.read()\n\n        add_opts = \"\"\n        add_opts += \"-t {token} \".format(token=token) if token else \"\"\n        add_opts += \"-f \" if show_flow_logs else \"\"\n        add_opts += (\n            \" \".join(\"-l {label} \".format(label=label) for label in labels)\n            if labels\n            else \"\"\n        )\n        add_opts += (\n            \" \".join(\"-p {path}\".format(path=path) for path in import_paths)\n            if import_paths\n            else \"\"\n        )\n        conf = conf.replace(\"{{OPTS}}\", add_opts)\n        return conf\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}], "after_files": [{"content": "import os\nimport socket\nimport sys\nfrom subprocess import STDOUT, Popen, DEVNULL\nfrom typing import Iterable, List\n\nfrom prefect import config\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.run_configs import LocalRun\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.serialization.run_config import RunConfigSchema\nfrom prefect.utilities.agent import get_flow_run_command\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as subprocesses. There are a range of kwarg\n    options to control information which may be provided to these subprocesses.\n\n    Optional import paths may be specified to append dependency modules to the PATH:\n    ```\n    prefect agent start local --import-path \"/usr/local/my_module\" --import-path \"~/other_module\"\n\n    # Now the local scripts/packages my_module and other_module will be importable in\n    # the flow's subprocess\n    ```\n\n    Environment variables may be set on the agent to be provided to each flow run's subprocess:\n    ```\n    prefect agent start local --env MY_SECRET_KEY=secret --env OTHER_VAR=$OTHER_VAR\n    ```\n\n    Args:\n        - agent_config_id (str, optional): An optional agent configuration ID that can be used to set\n            configuration based on an agent from a backend API. If set all configuration values will be\n            pulled from backend agent configuration.\n        - name (str, optional): An optional name to give this agent. Can also be set through\n            the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to \"agent\"\n        - labels (List[str], optional): a list of labels, which are arbitrary string\n            identifiers used by Prefect Agents when polling for work\n        - env_vars (dict, optional): a dictionary of environment variables and values that will\n            be set on each flow run that this agent submits for execution\n        - max_polls (int, optional): maximum number of times the agent will poll Prefect Cloud\n            for flow runs; defaults to infinite\n        - agent_address (str, optional):  Address to serve internal api at. Currently this is\n            just health checks for use by an orchestration layer. Leave blank for no api server\n            (default).\n        - no_cloud_logs (bool, optional): Disable logging to a Prefect backend for this agent\n            and all deployed flow runs\n        - import_paths (List[str], optional): system paths which will be provided to each\n            Flow's runtime environment; useful for Flows which import from locally hosted\n            scripts or packages\n        - show_flow_logs (bool, optional): a boolean specifying whether the agent should\n            re-route Flow run logs to stdout; defaults to `False`\n        - hostname_label (boolean, optional): a boolean specifying whether this agent should\n            auto-label itself with the hostname of the machine it is running on.  Useful for\n            flows which are stored on the local filesystem.\n        - storage_labels (boolean, optional): a boolean specifying whether this agent should\n            auto-label itself with all of the storage options labels.\n    \"\"\"\n\n    def __init__(\n        self,\n        agent_config_id: str = None,\n        name: str = None,\n        labels: Iterable[str] = None,\n        env_vars: dict = None,\n        import_paths: List[str] = None,\n        show_flow_logs: bool = False,\n        hostname_label: bool = True,\n        max_polls: int = None,\n        agent_address: str = None,\n        no_cloud_logs: bool = False,\n        storage_labels: bool = True,\n    ) -> None:\n        self.processes = set()\n        self.import_paths = import_paths or []\n        self.show_flow_logs = show_flow_logs\n        super().__init__(\n            agent_config_id=agent_config_id,\n            name=name,\n            labels=labels,\n            env_vars=env_vars,\n            max_polls=max_polls,\n            agent_address=agent_address,\n            no_cloud_logs=no_cloud_logs,\n        )\n        hostname = socket.gethostname()\n\n        # Resolve common Docker hostname by using IP\n        if hostname == \"docker-desktop\":\n            hostname = socket.gethostbyname(hostname)\n\n        if hostname_label and (hostname not in self.labels):\n            assert isinstance(self.labels, list)\n            self.labels.append(hostname)\n\n        if storage_labels:\n            all_storage_labels = [\n                \"azure-flow-storage\",\n                \"gcs-flow-storage\",\n                \"s3-flow-storage\",\n                \"github-flow-storage\",\n                \"webhook-flow-storage\",\n                \"gitlab-flow-storage\",\n            ]\n            for label in all_storage_labels:\n                if label not in self.labels:\n                    self.labels.append(label)\n\n        self.logger.debug(f\"Import paths: {self.import_paths}\")\n        self.logger.debug(f\"Show flow logs: {self.show_flow_logs}\")\n\n    def heartbeat(self) -> None:\n        for process in list(self.processes):\n            if process.poll() is not None:\n                self.processes.remove(process)\n                if process.returncode:\n                    self.logger.info(\n                        \"Process PID {} returned non-zero exit code\".format(process.pid)\n                    )\n        super().heartbeat()\n\n    def deploy_flow(self, flow_run: GraphQLResult) -> str:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_run (GraphQLResult): A GraphQLResult flow run object\n\n        Returns:\n            - str: Information about the deployment\n\n        Raises:\n            - ValueError: if deployment attempted on unsupported Storage type\n        \"\"\"\n        self.logger.info(\"Deploying flow run {}\".format(flow_run.id))  # type: ignore\n\n        storage = StorageSchema().load(flow_run.flow.storage)\n        if isinstance(storage, Docker):\n            self.logger.error(\n                \"Flow run %s has an unsupported storage type: `%s`\",\n                flow_run.id,\n                type(storage).__name__,\n            )\n            raise TypeError(\"Unsupported Storage type: %s\" % type(storage).__name__)\n\n        # If the flow is using a run_config, load it\n        if getattr(flow_run.flow, \"run_config\", None) is not None:\n            run_config = RunConfigSchema().load(flow_run.flow.run_config)\n            if not isinstance(run_config, LocalRun):\n                self.logger.error(\n                    \"Flow run %s has a `run_config` of type `%s`, only `LocalRun` is supported\",\n                    flow_run.id,\n                    type(run_config).__name__,\n                )\n                raise TypeError(\n                    \"Unsupported RunConfig type: %s\" % type(run_config).__name__\n                )\n        else:\n            run_config = None\n\n        env = self.populate_env_vars(flow_run, run_config=run_config)\n\n        working_dir = None if run_config is None else run_config.working_dir\n        if working_dir and not os.path.exists(working_dir):\n            msg = f\"Flow run {flow_run.id} has a nonexistent `working_dir` configured: {working_dir}\"\n            self.logger.error(msg)\n            raise ValueError(msg)\n\n        stdout = sys.stdout if self.show_flow_logs else DEVNULL\n\n        # note: we will allow these processes to be orphaned if the agent were to exit\n        # before the flow runs have completed. The lifecycle of the agent should not\n        # dictate the lifecycle of the flow run. However, if the user has elected to\n        # show flow logs, these log entries will continue to stream to the users terminal\n        # until these child processes exit, even if the agent has already exited.\n        p = Popen(\n            get_flow_run_command(flow_run).split(\" \"),\n            stdout=stdout,\n            stderr=STDOUT,\n            env=env,\n            cwd=working_dir,\n        )\n\n        self.processes.add(p)\n        self.logger.debug(\n            \"Submitted flow run {} to process PID {}\".format(flow_run.id, p.pid)\n        )\n\n        return \"PID: {}\".format(p.pid)\n\n    def populate_env_vars(\n        self, flow_run: GraphQLResult, run_config: LocalRun = None\n    ) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n            - run_config (LocalRun, optional): The `run_config` for the flow, if any.\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        # Environment variables come from (later options override):\n\n        # 1. Local environment\n        env = os.environ.copy()\n\n        # 2. Logging config\n        env.update({\"PREFECT__LOGGING__LEVEL\": config.logging.level})\n\n        # 3. PYTHONPATH\n        # XXX: I'm 99% sure this isn't needed, a new Python process should\n        # always add the CWD as the first entry in sys.path\n        python_path = [\n            run_config.working_dir\n            if run_config is not None and run_config.working_dir\n            else os.getcwd()\n        ]\n        if os.environ.get(\"PYTHONPATH\"):\n            python_path.append(os.environ[\"PYTHONPATH\"])\n        if self.import_paths:\n            python_path.extend(self.import_paths)\n        env[\"PYTHONPATH\"] = os.pathsep.join(python_path)\n\n        # 4. Values set on the agent via `--env`\n        env.update(self.env_vars)\n\n        # 5. Values set on a LocalRun RunConfig (if present\n        if run_config is not None and run_config.env is not None:\n            env.update(run_config.env)\n\n        # 6. Non-overrideable required env vars\n        env.update(\n            {\n                \"PREFECT__CLOUD__API\": config.cloud.api,\n                \"PREFECT__CLOUD__AUTH_TOKEN\": self.client._api_token,\n                \"PREFECT__CLOUD__AGENT__LABELS\": str(self.labels),\n                \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n                \"PREFECT__CONTEXT__FLOW_ID\": flow_run.flow.id,  # type: ignore\n                \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n                \"PREFECT__LOGGING__LOG_TO_CLOUD\": str(self.log_to_cloud).lower(),\n                \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n                \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n            }\n        )\n\n        # Filter out None values\n        return {k: v for k, v in env.items() if v is not None}\n\n    @staticmethod\n    def generate_supervisor_conf(\n        token: str = None,\n        labels: Iterable[str] = None,\n        import_paths: List[str] = None,\n        show_flow_logs: bool = False,\n    ) -> str:\n        \"\"\"\n        Generate and output an installable supervisorctl configuration file for the agent.\n\n        Args:\n            - token (str, optional): A `RUNNER` token to give the agent\n            - labels (List[str], optional): a list of labels, which are arbitrary string\n                identifiers used by Prefect Agents when polling for work\n            - import_paths (List[str], optional): system paths which will be provided to each\n                Flow's runtime environment; useful for Flows which import from locally hosted\n                scripts or packages\n            - show_flow_logs (bool, optional): a boolean specifying whether the agent should\n                re-route Flow run logs to stdout; defaults to `False`\n\n        Returns:\n            - str: A string representation of the generated configuration file\n        \"\"\"\n\n        # Use defaults if not provided\n        token = token or \"\"\n        labels = labels or []\n        import_paths = import_paths or []\n\n        with open(\n            os.path.join(os.path.dirname(__file__), \"supervisord.conf\"), \"r\"\n        ) as conf_file:\n            conf = conf_file.read()\n\n        add_opts = \"\"\n        add_opts += \"-t {token} \".format(token=token) if token else \"\"\n        add_opts += \"-f \" if show_flow_logs else \"\"\n        add_opts += (\n            \" \".join(\"-l {label} \".format(label=label) for label in labels)\n            if labels\n            else \"\"\n        )\n        add_opts += (\n            \" \".join(\"-p {path}\".format(path=path) for path in import_paths)\n            if import_paths\n            else \"\"\n        )\n        conf = conf.replace(\"{{OPTS}}\", add_opts)\n        return conf\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}]}