We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fargate agent warning is too aggressive
## Description
This warning can be relaxed because it does not suit all use cases



## Expected Behavior
Elevate warning only under certain conditions



## Reproduction
Run the fargate agent without ARNs


--- 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/fargate/agent.py`
Content:
```
1 import os
2 
3 from prefect import config
4 from prefect.agent import Agent
5 from prefect.environments.storage import Docker
6 from prefect.serialization.storage import StorageSchema
7 from prefect.utilities.graphql import GraphQLResult
8 
9 
10 class FargateAgent(Agent):
11     """
12     Agent which deploys flow runs as tasks using Fargate. This agent can run anywhere as
13     long as the proper access configuration variables are set.  Information on using the
14     Fargate Agent can be found at https://docs.prefect.io/cloud/agent/fargate.html
15 
16     Args:
17         - name (str, optional): An optional name to give this agent. Can also be set through
18             the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to "agent"
19         - aws_access_key_id (str, optional): AWS access key id for connecting the boto3
20             client. Defaults to the value set in the environment variable
21             `AWS_ACCESS_KEY_ID`.
22         - aws_secret_access_key (str, optional): AWS secret access key for connecting
23             the boto3 client. Defaults to the value set in the environment variable
24             `AWS_SECRET_ACCESS_KEY`.
25         - aws_session_token (str, optional): AWS session key for connecting the boto3
26             client. Defaults to the value set in the environment variable
27             `AWS_SESSION_TOKEN`.
28         - task_role_arn (str, optional): AWS Amazon Resource Name (ARN) of the AWS
29             Identity and Access Management (IAM) role that grants containers in the
30             task permission to call AWS APIs on your behalf. Defaults to the value set
31              in the environment variable `TASK_ROLE_ARN`.
32         - execution_role_arn (str, optional): AWS Amazon Resource Name (ARN) of the
33             task execution role that containers in this task can assume. Defaults to
34              the value set in the environment variable `EXECUTION_ROLE_ARN`.
35         - region_name (str, optional): AWS region name for connecting the boto3 client.
36             Defaults to the value set in the environment variable `REGION_NAME`.
37         - cluster (str, optional): The Fargate cluster to deploy tasks. Defaults to the
38             value set in the environment variable `CLUSTER`.
39         - subnets (list, optional): A list of AWS VPC subnets to use for the tasks that
40             are deployed on Fargate. Defaults to the value set in the environment
41             variable `SUBNETS`.
42         - security_groups (list, optional): A list of security groups to associate with
43             the deployed tasks. Defaults to the value set in the environment variable
44             `SECURITY_GROUPS`.
45         - repository_credentials (str, optional): An Amazon Resource Name (ARN) of the
46             secret containing the private repository credentials. Defaults to the value
47             set in the environment variable `REPOSITORY_CREDENTIALS`.
48         - assign_public_ip (str, optional): Whether the task's elastic network interface
49             receives a public IP address. Defaults to the value set in the environment
50             variable `ASSIGN_PUBLIC_IP` or `ENABLED` otherwise.
51         - task_cpu (str, optional): The number of cpu units reserved for the container.
52             Defaults to the value set in the environment variable `TASK_CPU` or `256`
53             otherwise.
54         - task_memory (str, optional): The hard limit (in MiB) of memory to present to
55             the container. Defaults to the value set in the environment variable
56             `TASK_MEMORY` or `512` otherwise.
57     """
58 
59     def __init__(
60         self,
61         name: str = None,
62         aws_access_key_id: str = None,
63         aws_secret_access_key: str = None,
64         aws_session_token: str = None,
65         task_role_arn: str = None,
66         execution_role_arn: str = None,
67         region_name: str = None,
68         cluster: str = None,
69         subnets: list = None,
70         security_groups: list = None,
71         repository_credentials: str = None,
72         assign_public_ip: str = None,
73         task_cpu: str = None,
74         task_memory: str = None,
75     ) -> None:
76         super().__init__(name=name)
77 
78         from boto3 import client as boto3_client
79 
80         # Config used for boto3 client initialization
81         aws_access_key_id = aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID")
82         aws_secret_access_key = aws_secret_access_key or os.getenv(
83             "AWS_SECRET_ACCESS_KEY"
84         )
85         aws_session_token = aws_session_token or os.getenv("AWS_SESSION_TOKEN")
86         region_name = region_name or os.getenv("REGION_NAME")
87 
88         # Agent task config
89         self.task_role_arn = task_role_arn or os.getenv("TASK_ROLE_ARN")
90         self.logger.debug("Task role arn {}".format(self.task_role_arn))
91 
92         self.execution_role_arn = execution_role_arn or os.getenv("EXECUTION_ROLE_ARN")
93         if not self.execution_role_arn:
94             self.logger.warning(
95                 "Fargate requires task definition to have execution role ARN to support ECR images"
96             )
97         self.logger.debug("Execution role arn {}".format(self.execution_role_arn))
98 
99         self.cluster = cluster or os.getenv("CLUSTER", "default")
100         self.logger.debug("Cluster {}".format(self.cluster))
101 
102         if subnets:
103             self.subnets = subnets
104         elif os.getenv("SUBNETS"):
105             self.subnets = str(os.getenv("SUBNETS")).split(",")
106         else:
107             self.subnets = []
108         self.logger.debug("Subnets {}".format(self.subnets))
109 
110         if security_groups:
111             self.security_groups = security_groups
112         elif os.getenv("SECURITY_GROUPS"):
113             self.security_groups = str(os.getenv("SECURITY_GROUPS")).split(",")
114         else:
115             self.security_groups = []
116         self.logger.debug("Security groups {}".format(self.security_groups))
117 
118         self.repository_credentials = repository_credentials or os.getenv(
119             "REPOSITORY_CREDENTIALS"
120         )
121         self.logger.debug(
122             "Repository credentials {}".format(self.repository_credentials)
123         )
124 
125         self.assign_public_ip = assign_public_ip or os.getenv(
126             "ASSIGN_PUBLIC_IP", "ENABLED"
127         )
128         self.logger.debug("Assign public IP {}".format(self.assign_public_ip))
129 
130         self.task_cpu = task_cpu or os.getenv("TASK_CPU", "256")
131         self.logger.debug("Task CPU {}".format(self.task_cpu))
132 
133         self.task_memory = task_memory or os.getenv("TASK_MEMORY", "512")
134         self.logger.debug("Task Memory {}".format(self.task_memory))
135 
136         # Client initialization
137         self.boto3_client = boto3_client(
138             "ecs",
139             aws_access_key_id=aws_access_key_id,
140             aws_secret_access_key=aws_secret_access_key,
141             aws_session_token=aws_session_token,
142             region_name=region_name,
143         )
144 
145         # Look for default subnets with `MapPublicIpOnLaunch` disabled
146         if not self.subnets:
147             self.logger.debug("No subnets provided, finding defaults")
148             ec2 = boto3_client(
149                 "ec2",
150                 aws_access_key_id=aws_access_key_id,
151                 aws_secret_access_key=aws_secret_access_key,
152                 aws_session_token=aws_session_token,
153                 region_name=region_name,
154             )
155             for subnet in ec2.describe_subnets()["Subnets"]:
156                 if not subnet.get("MapPublicIpOnLaunch"):
157                     self.subnets.append(subnet.get("SubnetId"))
158                     self.logger.debug("Subnet {} found".format(subnet.get("SubnetId")))
159 
160     def deploy_flows(self, flow_runs: list) -> None:
161         """
162         Deploy flow runs to Fargate
163 
164         Args:
165             - flow_runs (list): A list of GraphQLResult flow run objects
166         """
167         for flow_run in flow_runs:
168             self.logger.debug(
169                 "Deploying flow run {}".format(flow_run.id)  # type: ignore
170             )
171 
172             # Require Docker storage
173             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):
174                 self.logger.error(
175                     "Storage for flow run {} is not of type Docker.".format(flow_run.id)
176                 )
177                 continue
178 
179             # check if task definition exists
180             self.logger.debug("Checking for task definition")
181             if not self._verify_task_definition_exists(flow_run):
182                 self.logger.debug("No task definition found")
183                 self._create_task_definition(flow_run)
184 
185             # run task
186             self._run_task(flow_run)
187 
188     def _verify_task_definition_exists(self, flow_run: GraphQLResult) -> bool:
189         """
190         Check if a task definition already exists for the flow
191 
192         Args:
193             - flow_runs (list): A list of GraphQLResult flow run objects
194 
195         Returns:
196             - bool: whether or not a preexisting task definition is found for this flow
197         """
198         from botocore.exceptions import ClientError
199 
200         try:
201             self.boto3_client.describe_task_definition(
202                 taskDefinition="prefect-task-{}".format(
203                     flow_run.flow.id[:8]  # type: ignore
204                 )
205             )
206             self.logger.debug(
207                 "Task definition {} found".format(flow_run.flow.id[:8])  # type: ignore
208             )
209         except ClientError:
210             return False
211 
212         return True
213 
214     def _create_task_definition(self, flow_run: GraphQLResult) -> None:
215         """
216         Create a task definition for the flow that each flow run will use. This function
217         is only called when a flow is run for the first time.
218 
219         Args:
220             - flow_runs (list): A list of GraphQLResult flow run objects
221         """
222         self.logger.debug(
223             "Using image {} for task definition".format(
224                 StorageSchema().load(flow_run.flow.storage).name  # type: ignore
225             )
226         )
227         container_definitions = [
228             {
229                 "name": "flow",
230                 "image": StorageSchema()
231                 .load(flow_run.flow.storage)  # type: ignore
232                 .name,
233                 "command": ["/bin/sh", "-c", "prefect execute cloud-flow"],
234                 "environment": [
235                     {
236                         "name": "PREFECT__CLOUD__API",
237                         "value": config.cloud.api or "https://api.prefect.io",
238                     },
239                     {"name": "PREFECT__CLOUD__USE_LOCAL_SECRETS", "value": "false"},
240                     {"name": "PREFECT__LOGGING__LOG_TO_CLOUD", "value": "true"},
241                     {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},
242                     {
243                         "name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",
244                         "value": "prefect.engine.cloud.CloudFlowRunner",
245                     },
246                     {
247                         "name": "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS",
248                         "value": "prefect.engine.cloud.CloudTaskRunner",
249                     },
250                 ],
251                 "essential": True,
252             }
253         ]
254 
255         # Assign repository credentials if they are specified
256         if self.repository_credentials:
257             self.logger.debug("Assigning repository credentials")
258             container_definitions[0]["repositoryCredentials"] = {
259                 "credentialsParameter": self.repository_credentials
260             }
261 
262         # Register task definition
263         self.logger.debug(
264             "Registering task definition {}".format(
265                 flow_run.flow.id[:8]  # type: ignore
266             )
267         )
268         self.boto3_client.register_task_definition(
269             family="prefect-task-{}".format(flow_run.flow.id[:8]),  # type: ignore
270             containerDefinitions=container_definitions,
271             requiresCompatibilities=["FARGATE"],
272             networkMode="awsvpc",
273             cpu=self.task_cpu,
274             memory=self.task_memory,
275             taskRoleArn=self.task_role_arn,
276             executionRoleArn=self.execution_role_arn,
277         )
278 
279     def _run_task(self, flow_run: GraphQLResult) -> None:
280         """
281         Run a task using the flow run.
282 
283         Args:
284             - flow_runs (list): A list of GraphQLResult flow run objects
285         """
286         container_overrides = [
287             {
288                 "name": "flow",
289                 "environment": [
290                     {
291                         "name": "PREFECT__CLOUD__AUTH_TOKEN",
292                         "value": config.cloud.agent.auth_token,
293                     },
294                     {
295                         "name": "PREFECT__CONTEXT__FLOW_RUN_ID",
296                         "value": flow_run.id,  # type: ignore
297                     },
298                 ],
299             }
300         ]
301 
302         network_configuration = {
303             "awsvpcConfiguration": {
304                 "subnets": self.subnets,
305                 "assignPublicIp": self.assign_public_ip,
306             }
307         }
308 
309         # Asssign task security groups if they are specified
310         if self.security_groups:
311             self.logger.debug("Assigning security groups")
312             network_configuration["awsvpcConfiguration"][
313                 "securityGroups"
314             ] = self.security_groups
315 
316         # Run task
317         self.logger.debug(
318             "Running task using task definition {}".format(
319                 flow_run.flow.id[:8]  # type: ignore
320             )
321         )
322         self.boto3_client.run_task(
323             cluster=self.cluster,
324             taskDefinition="prefect-task-{}".format(
325                 flow_run.flow.id[:8]  # type: ignore
326             ),
327             overrides={"containerOverrides": container_overrides},
328             launchType="FARGATE",
329             networkConfiguration=network_configuration,
330         )
331 
332 
333 if __name__ == "__main__":
334     FargateAgent().start()
335
```
--- 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/fargate/agent.py b/src/prefect/agent/fargate/agent.py
--- a/src/prefect/agent/fargate/agent.py
+++ b/src/prefect/agent/fargate/agent.py
@@ -86,14 +86,12 @@
         region_name = region_name or os.getenv("REGION_NAME")
 
         # Agent task config
-        self.task_role_arn = task_role_arn or os.getenv("TASK_ROLE_ARN")
+        self.task_role_arn = task_role_arn or os.getenv("TASK_ROLE_ARN", "")
         self.logger.debug("Task role arn {}".format(self.task_role_arn))
 
-        self.execution_role_arn = execution_role_arn or os.getenv("EXECUTION_ROLE_ARN")
-        if not self.execution_role_arn:
-            self.logger.warning(
-                "Fargate requires task definition to have execution role ARN to support ECR images"
-            )
+        self.execution_role_arn = execution_role_arn or os.getenv(
+            "EXECUTION_ROLE_ARN", ""
+        )
         self.logger.debug("Execution role arn {}".format(self.execution_role_arn))
 
         self.cluster = cluster or os.getenv("CLUSTER", "default")

{"golden_diff": "diff --git a/src/prefect/agent/fargate/agent.py b/src/prefect/agent/fargate/agent.py\n--- a/src/prefect/agent/fargate/agent.py\n+++ b/src/prefect/agent/fargate/agent.py\n@@ -86,14 +86,12 @@\n         region_name = region_name or os.getenv(\"REGION_NAME\")\n \n         # Agent task config\n-        self.task_role_arn = task_role_arn or os.getenv(\"TASK_ROLE_ARN\")\n+        self.task_role_arn = task_role_arn or os.getenv(\"TASK_ROLE_ARN\", \"\")\n         self.logger.debug(\"Task role arn {}\".format(self.task_role_arn))\n \n-        self.execution_role_arn = execution_role_arn or os.getenv(\"EXECUTION_ROLE_ARN\")\n-        if not self.execution_role_arn:\n-            self.logger.warning(\n-                \"Fargate requires task definition to have execution role ARN to support ECR images\"\n-            )\n+        self.execution_role_arn = execution_role_arn or os.getenv(\n+            \"EXECUTION_ROLE_ARN\", \"\"\n+        )\n         self.logger.debug(\"Execution role arn {}\".format(self.execution_role_arn))\n \n         self.cluster = cluster or os.getenv(\"CLUSTER\", \"default\")\n", "issue": "Fargate agent warning is too aggressive\n## Description\r\nThis warning can be relaxed because it does not suit all use cases\r\n\r\n\r\n\r\n## Expected Behavior\r\nElevate warning only under certain conditions\r\n\r\n\r\n\r\n## Reproduction\r\nRun the fargate agent without ARNs\r\n\n", "before_files": [{"content": "import os\n\nfrom prefect import config\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass FargateAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs as tasks using Fargate. This agent can run anywhere as\n    long as the proper access configuration variables are set.  Information on using the\n    Fargate Agent can be found at https://docs.prefect.io/cloud/agent/fargate.html\n\n    Args:\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        - aws_access_key_id (str, optional): AWS access key id for connecting the boto3\n            client. Defaults to the value set in the environment variable\n            `AWS_ACCESS_KEY_ID`.\n        - aws_secret_access_key (str, optional): AWS secret access key for connecting\n            the boto3 client. Defaults to the value set in the environment variable\n            `AWS_SECRET_ACCESS_KEY`.\n        - aws_session_token (str, optional): AWS session key for connecting the boto3\n            client. Defaults to the value set in the environment variable\n            `AWS_SESSION_TOKEN`.\n        - task_role_arn (str, optional): AWS Amazon Resource Name (ARN) of the AWS\n            Identity and Access Management (IAM) role that grants containers in the\n            task permission to call AWS APIs on your behalf. Defaults to the value set\n             in the environment variable `TASK_ROLE_ARN`.\n        - execution_role_arn (str, optional): AWS Amazon Resource Name (ARN) of the\n            task execution role that containers in this task can assume. Defaults to\n             the value set in the environment variable `EXECUTION_ROLE_ARN`.\n        - region_name (str, optional): AWS region name for connecting the boto3 client.\n            Defaults to the value set in the environment variable `REGION_NAME`.\n        - cluster (str, optional): The Fargate cluster to deploy tasks. Defaults to the\n            value set in the environment variable `CLUSTER`.\n        - subnets (list, optional): A list of AWS VPC subnets to use for the tasks that\n            are deployed on Fargate. Defaults to the value set in the environment\n            variable `SUBNETS`.\n        - security_groups (list, optional): A list of security groups to associate with\n            the deployed tasks. Defaults to the value set in the environment variable\n            `SECURITY_GROUPS`.\n        - repository_credentials (str, optional): An Amazon Resource Name (ARN) of the\n            secret containing the private repository credentials. Defaults to the value\n            set in the environment variable `REPOSITORY_CREDENTIALS`.\n        - assign_public_ip (str, optional): Whether the task's elastic network interface\n            receives a public IP address. Defaults to the value set in the environment\n            variable `ASSIGN_PUBLIC_IP` or `ENABLED` otherwise.\n        - task_cpu (str, optional): The number of cpu units reserved for the container.\n            Defaults to the value set in the environment variable `TASK_CPU` or `256`\n            otherwise.\n        - task_memory (str, optional): The hard limit (in MiB) of memory to present to\n            the container. Defaults to the value set in the environment variable\n            `TASK_MEMORY` or `512` otherwise.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str = None,\n        aws_access_key_id: str = None,\n        aws_secret_access_key: str = None,\n        aws_session_token: str = None,\n        task_role_arn: str = None,\n        execution_role_arn: str = None,\n        region_name: str = None,\n        cluster: str = None,\n        subnets: list = None,\n        security_groups: list = None,\n        repository_credentials: str = None,\n        assign_public_ip: str = None,\n        task_cpu: str = None,\n        task_memory: str = None,\n    ) -> None:\n        super().__init__(name=name)\n\n        from boto3 import client as boto3_client\n\n        # Config used for boto3 client initialization\n        aws_access_key_id = aws_access_key_id or os.getenv(\"AWS_ACCESS_KEY_ID\")\n        aws_secret_access_key = aws_secret_access_key or os.getenv(\n            \"AWS_SECRET_ACCESS_KEY\"\n        )\n        aws_session_token = aws_session_token or os.getenv(\"AWS_SESSION_TOKEN\")\n        region_name = region_name or os.getenv(\"REGION_NAME\")\n\n        # Agent task config\n        self.task_role_arn = task_role_arn or os.getenv(\"TASK_ROLE_ARN\")\n        self.logger.debug(\"Task role arn {}\".format(self.task_role_arn))\n\n        self.execution_role_arn = execution_role_arn or os.getenv(\"EXECUTION_ROLE_ARN\")\n        if not self.execution_role_arn:\n            self.logger.warning(\n                \"Fargate requires task definition to have execution role ARN to support ECR images\"\n            )\n        self.logger.debug(\"Execution role arn {}\".format(self.execution_role_arn))\n\n        self.cluster = cluster or os.getenv(\"CLUSTER\", \"default\")\n        self.logger.debug(\"Cluster {}\".format(self.cluster))\n\n        if subnets:\n            self.subnets = subnets\n        elif os.getenv(\"SUBNETS\"):\n            self.subnets = str(os.getenv(\"SUBNETS\")).split(\",\")\n        else:\n            self.subnets = []\n        self.logger.debug(\"Subnets {}\".format(self.subnets))\n\n        if security_groups:\n            self.security_groups = security_groups\n        elif os.getenv(\"SECURITY_GROUPS\"):\n            self.security_groups = str(os.getenv(\"SECURITY_GROUPS\")).split(\",\")\n        else:\n            self.security_groups = []\n        self.logger.debug(\"Security groups {}\".format(self.security_groups))\n\n        self.repository_credentials = repository_credentials or os.getenv(\n            \"REPOSITORY_CREDENTIALS\"\n        )\n        self.logger.debug(\n            \"Repository credentials {}\".format(self.repository_credentials)\n        )\n\n        self.assign_public_ip = assign_public_ip or os.getenv(\n            \"ASSIGN_PUBLIC_IP\", \"ENABLED\"\n        )\n        self.logger.debug(\"Assign public IP {}\".format(self.assign_public_ip))\n\n        self.task_cpu = task_cpu or os.getenv(\"TASK_CPU\", \"256\")\n        self.logger.debug(\"Task CPU {}\".format(self.task_cpu))\n\n        self.task_memory = task_memory or os.getenv(\"TASK_MEMORY\", \"512\")\n        self.logger.debug(\"Task Memory {}\".format(self.task_memory))\n\n        # Client initialization\n        self.boto3_client = boto3_client(\n            \"ecs\",\n            aws_access_key_id=aws_access_key_id,\n            aws_secret_access_key=aws_secret_access_key,\n            aws_session_token=aws_session_token,\n            region_name=region_name,\n        )\n\n        # Look for default subnets with `MapPublicIpOnLaunch` disabled\n        if not self.subnets:\n            self.logger.debug(\"No subnets provided, finding defaults\")\n            ec2 = boto3_client(\n                \"ec2\",\n                aws_access_key_id=aws_access_key_id,\n                aws_secret_access_key=aws_secret_access_key,\n                aws_session_token=aws_session_token,\n                region_name=region_name,\n            )\n            for subnet in ec2.describe_subnets()[\"Subnets\"]:\n                if not subnet.get(\"MapPublicIpOnLaunch\"):\n                    self.subnets.append(subnet.get(\"SubnetId\"))\n                    self.logger.debug(\"Subnet {} found\".format(subnet.get(\"SubnetId\")))\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs to Fargate\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n            self.logger.debug(\n                \"Deploying flow run {}\".format(flow_run.id)  # type: ignore\n            )\n\n            # Require Docker storage\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                self.logger.error(\n                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n                )\n                continue\n\n            # check if task definition exists\n            self.logger.debug(\"Checking for task definition\")\n            if not self._verify_task_definition_exists(flow_run):\n                self.logger.debug(\"No task definition found\")\n                self._create_task_definition(flow_run)\n\n            # run task\n            self._run_task(flow_run)\n\n    def _verify_task_definition_exists(self, flow_run: GraphQLResult) -> bool:\n        \"\"\"\n        Check if a task definition already exists for the flow\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n\n        Returns:\n            - bool: whether or not a preexisting task definition is found for this flow\n        \"\"\"\n        from botocore.exceptions import ClientError\n\n        try:\n            self.boto3_client.describe_task_definition(\n                taskDefinition=\"prefect-task-{}\".format(\n                    flow_run.flow.id[:8]  # type: ignore\n                )\n            )\n            self.logger.debug(\n                \"Task definition {} found\".format(flow_run.flow.id[:8])  # type: ignore\n            )\n        except ClientError:\n            return False\n\n        return True\n\n    def _create_task_definition(self, flow_run: GraphQLResult) -> None:\n        \"\"\"\n        Create a task definition for the flow that each flow run will use. This function\n        is only called when a flow is run for the first time.\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        self.logger.debug(\n            \"Using image {} for task definition\".format(\n                StorageSchema().load(flow_run.flow.storage).name  # type: ignore\n            )\n        )\n        container_definitions = [\n            {\n                \"name\": \"flow\",\n                \"image\": StorageSchema()\n                .load(flow_run.flow.storage)  # type: ignore\n                .name,\n                \"command\": [\"/bin/sh\", \"-c\", \"prefect execute cloud-flow\"],\n                \"environment\": [\n                    {\n                        \"name\": \"PREFECT__CLOUD__API\",\n                        \"value\": config.cloud.api or \"https://api.prefect.io\",\n                    },\n                    {\"name\": \"PREFECT__CLOUD__USE_LOCAL_SECRETS\", \"value\": \"false\"},\n                    {\"name\": \"PREFECT__LOGGING__LOG_TO_CLOUD\", \"value\": \"true\"},\n                    {\"name\": \"PREFECT__LOGGING__LEVEL\", \"value\": \"DEBUG\"},\n                    {\n                        \"name\": \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\",\n                        \"value\": \"prefect.engine.cloud.CloudFlowRunner\",\n                    },\n                    {\n                        \"name\": \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\",\n                        \"value\": \"prefect.engine.cloud.CloudTaskRunner\",\n                    },\n                ],\n                \"essential\": True,\n            }\n        ]\n\n        # Assign repository credentials if they are specified\n        if self.repository_credentials:\n            self.logger.debug(\"Assigning repository credentials\")\n            container_definitions[0][\"repositoryCredentials\"] = {\n                \"credentialsParameter\": self.repository_credentials\n            }\n\n        # Register task definition\n        self.logger.debug(\n            \"Registering task definition {}\".format(\n                flow_run.flow.id[:8]  # type: ignore\n            )\n        )\n        self.boto3_client.register_task_definition(\n            family=\"prefect-task-{}\".format(flow_run.flow.id[:8]),  # type: ignore\n            containerDefinitions=container_definitions,\n            requiresCompatibilities=[\"FARGATE\"],\n            networkMode=\"awsvpc\",\n            cpu=self.task_cpu,\n            memory=self.task_memory,\n            taskRoleArn=self.task_role_arn,\n            executionRoleArn=self.execution_role_arn,\n        )\n\n    def _run_task(self, flow_run: GraphQLResult) -> None:\n        \"\"\"\n        Run a task using the flow run.\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        container_overrides = [\n            {\n                \"name\": \"flow\",\n                \"environment\": [\n                    {\n                        \"name\": \"PREFECT__CLOUD__AUTH_TOKEN\",\n                        \"value\": config.cloud.agent.auth_token,\n                    },\n                    {\n                        \"name\": \"PREFECT__CONTEXT__FLOW_RUN_ID\",\n                        \"value\": flow_run.id,  # type: ignore\n                    },\n                ],\n            }\n        ]\n\n        network_configuration = {\n            \"awsvpcConfiguration\": {\n                \"subnets\": self.subnets,\n                \"assignPublicIp\": self.assign_public_ip,\n            }\n        }\n\n        # Asssign task security groups if they are specified\n        if self.security_groups:\n            self.logger.debug(\"Assigning security groups\")\n            network_configuration[\"awsvpcConfiguration\"][\n                \"securityGroups\"\n            ] = self.security_groups\n\n        # Run task\n        self.logger.debug(\n            \"Running task using task definition {}\".format(\n                flow_run.flow.id[:8]  # type: ignore\n            )\n        )\n        self.boto3_client.run_task(\n            cluster=self.cluster,\n            taskDefinition=\"prefect-task-{}\".format(\n                flow_run.flow.id[:8]  # type: ignore\n            ),\n            overrides={\"containerOverrides\": container_overrides},\n            launchType=\"FARGATE\",\n            networkConfiguration=network_configuration,\n        )\n\n\nif __name__ == \"__main__\":\n    FargateAgent().start()\n", "path": "src/prefect/agent/fargate/agent.py"}], "after_files": [{"content": "import os\n\nfrom prefect import config\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass FargateAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs as tasks using Fargate. This agent can run anywhere as\n    long as the proper access configuration variables are set.  Information on using the\n    Fargate Agent can be found at https://docs.prefect.io/cloud/agent/fargate.html\n\n    Args:\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        - aws_access_key_id (str, optional): AWS access key id for connecting the boto3\n            client. Defaults to the value set in the environment variable\n            `AWS_ACCESS_KEY_ID`.\n        - aws_secret_access_key (str, optional): AWS secret access key for connecting\n            the boto3 client. Defaults to the value set in the environment variable\n            `AWS_SECRET_ACCESS_KEY`.\n        - aws_session_token (str, optional): AWS session key for connecting the boto3\n            client. Defaults to the value set in the environment variable\n            `AWS_SESSION_TOKEN`.\n        - task_role_arn (str, optional): AWS Amazon Resource Name (ARN) of the AWS\n            Identity and Access Management (IAM) role that grants containers in the\n            task permission to call AWS APIs on your behalf. Defaults to the value set\n             in the environment variable `TASK_ROLE_ARN`.\n        - execution_role_arn (str, optional): AWS Amazon Resource Name (ARN) of the\n            task execution role that containers in this task can assume. Defaults to\n             the value set in the environment variable `EXECUTION_ROLE_ARN`.\n        - region_name (str, optional): AWS region name for connecting the boto3 client.\n            Defaults to the value set in the environment variable `REGION_NAME`.\n        - cluster (str, optional): The Fargate cluster to deploy tasks. Defaults to the\n            value set in the environment variable `CLUSTER`.\n        - subnets (list, optional): A list of AWS VPC subnets to use for the tasks that\n            are deployed on Fargate. Defaults to the value set in the environment\n            variable `SUBNETS`.\n        - security_groups (list, optional): A list of security groups to associate with\n            the deployed tasks. Defaults to the value set in the environment variable\n            `SECURITY_GROUPS`.\n        - repository_credentials (str, optional): An Amazon Resource Name (ARN) of the\n            secret containing the private repository credentials. Defaults to the value\n            set in the environment variable `REPOSITORY_CREDENTIALS`.\n        - assign_public_ip (str, optional): Whether the task's elastic network interface\n            receives a public IP address. Defaults to the value set in the environment\n            variable `ASSIGN_PUBLIC_IP` or `ENABLED` otherwise.\n        - task_cpu (str, optional): The number of cpu units reserved for the container.\n            Defaults to the value set in the environment variable `TASK_CPU` or `256`\n            otherwise.\n        - task_memory (str, optional): The hard limit (in MiB) of memory to present to\n            the container. Defaults to the value set in the environment variable\n            `TASK_MEMORY` or `512` otherwise.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str = None,\n        aws_access_key_id: str = None,\n        aws_secret_access_key: str = None,\n        aws_session_token: str = None,\n        task_role_arn: str = None,\n        execution_role_arn: str = None,\n        region_name: str = None,\n        cluster: str = None,\n        subnets: list = None,\n        security_groups: list = None,\n        repository_credentials: str = None,\n        assign_public_ip: str = None,\n        task_cpu: str = None,\n        task_memory: str = None,\n    ) -> None:\n        super().__init__(name=name)\n\n        from boto3 import client as boto3_client\n\n        # Config used for boto3 client initialization\n        aws_access_key_id = aws_access_key_id or os.getenv(\"AWS_ACCESS_KEY_ID\")\n        aws_secret_access_key = aws_secret_access_key or os.getenv(\n            \"AWS_SECRET_ACCESS_KEY\"\n        )\n        aws_session_token = aws_session_token or os.getenv(\"AWS_SESSION_TOKEN\")\n        region_name = region_name or os.getenv(\"REGION_NAME\")\n\n        # Agent task config\n        self.task_role_arn = task_role_arn or os.getenv(\"TASK_ROLE_ARN\", \"\")\n        self.logger.debug(\"Task role arn {}\".format(self.task_role_arn))\n\n        self.execution_role_arn = execution_role_arn or os.getenv(\n            \"EXECUTION_ROLE_ARN\", \"\"\n        )\n        self.logger.debug(\"Execution role arn {}\".format(self.execution_role_arn))\n\n        self.cluster = cluster or os.getenv(\"CLUSTER\", \"default\")\n        self.logger.debug(\"Cluster {}\".format(self.cluster))\n\n        if subnets:\n            self.subnets = subnets\n        elif os.getenv(\"SUBNETS\"):\n            self.subnets = str(os.getenv(\"SUBNETS\")).split(\",\")\n        else:\n            self.subnets = []\n        self.logger.debug(\"Subnets {}\".format(self.subnets))\n\n        if security_groups:\n            self.security_groups = security_groups\n        elif os.getenv(\"SECURITY_GROUPS\"):\n            self.security_groups = str(os.getenv(\"SECURITY_GROUPS\")).split(\",\")\n        else:\n            self.security_groups = []\n        self.logger.debug(\"Security groups {}\".format(self.security_groups))\n\n        self.repository_credentials = repository_credentials or os.getenv(\n            \"REPOSITORY_CREDENTIALS\"\n        )\n        self.logger.debug(\n            \"Repository credentials {}\".format(self.repository_credentials)\n        )\n\n        self.assign_public_ip = assign_public_ip or os.getenv(\n            \"ASSIGN_PUBLIC_IP\", \"ENABLED\"\n        )\n        self.logger.debug(\"Assign public IP {}\".format(self.assign_public_ip))\n\n        self.task_cpu = task_cpu or os.getenv(\"TASK_CPU\", \"256\")\n        self.logger.debug(\"Task CPU {}\".format(self.task_cpu))\n\n        self.task_memory = task_memory or os.getenv(\"TASK_MEMORY\", \"512\")\n        self.logger.debug(\"Task Memory {}\".format(self.task_memory))\n\n        # Client initialization\n        self.boto3_client = boto3_client(\n            \"ecs\",\n            aws_access_key_id=aws_access_key_id,\n            aws_secret_access_key=aws_secret_access_key,\n            aws_session_token=aws_session_token,\n            region_name=region_name,\n        )\n\n        # Look for default subnets with `MapPublicIpOnLaunch` disabled\n        if not self.subnets:\n            self.logger.debug(\"No subnets provided, finding defaults\")\n            ec2 = boto3_client(\n                \"ec2\",\n                aws_access_key_id=aws_access_key_id,\n                aws_secret_access_key=aws_secret_access_key,\n                aws_session_token=aws_session_token,\n                region_name=region_name,\n            )\n            for subnet in ec2.describe_subnets()[\"Subnets\"]:\n                if not subnet.get(\"MapPublicIpOnLaunch\"):\n                    self.subnets.append(subnet.get(\"SubnetId\"))\n                    self.logger.debug(\"Subnet {} found\".format(subnet.get(\"SubnetId\")))\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs to Fargate\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n            self.logger.debug(\n                \"Deploying flow run {}\".format(flow_run.id)  # type: ignore\n            )\n\n            # Require Docker storage\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                self.logger.error(\n                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n                )\n                continue\n\n            # check if task definition exists\n            self.logger.debug(\"Checking for task definition\")\n            if not self._verify_task_definition_exists(flow_run):\n                self.logger.debug(\"No task definition found\")\n                self._create_task_definition(flow_run)\n\n            # run task\n            self._run_task(flow_run)\n\n    def _verify_task_definition_exists(self, flow_run: GraphQLResult) -> bool:\n        \"\"\"\n        Check if a task definition already exists for the flow\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n\n        Returns:\n            - bool: whether or not a preexisting task definition is found for this flow\n        \"\"\"\n        from botocore.exceptions import ClientError\n\n        try:\n            self.boto3_client.describe_task_definition(\n                taskDefinition=\"prefect-task-{}\".format(\n                    flow_run.flow.id[:8]  # type: ignore\n                )\n            )\n            self.logger.debug(\n                \"Task definition {} found\".format(flow_run.flow.id[:8])  # type: ignore\n            )\n        except ClientError:\n            return False\n\n        return True\n\n    def _create_task_definition(self, flow_run: GraphQLResult) -> None:\n        \"\"\"\n        Create a task definition for the flow that each flow run will use. This function\n        is only called when a flow is run for the first time.\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        self.logger.debug(\n            \"Using image {} for task definition\".format(\n                StorageSchema().load(flow_run.flow.storage).name  # type: ignore\n            )\n        )\n        container_definitions = [\n            {\n                \"name\": \"flow\",\n                \"image\": StorageSchema()\n                .load(flow_run.flow.storage)  # type: ignore\n                .name,\n                \"command\": [\"/bin/sh\", \"-c\", \"prefect execute cloud-flow\"],\n                \"environment\": [\n                    {\n                        \"name\": \"PREFECT__CLOUD__API\",\n                        \"value\": config.cloud.api or \"https://api.prefect.io\",\n                    },\n                    {\"name\": \"PREFECT__CLOUD__USE_LOCAL_SECRETS\", \"value\": \"false\"},\n                    {\"name\": \"PREFECT__LOGGING__LOG_TO_CLOUD\", \"value\": \"true\"},\n                    {\"name\": \"PREFECT__LOGGING__LEVEL\", \"value\": \"DEBUG\"},\n                    {\n                        \"name\": \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\",\n                        \"value\": \"prefect.engine.cloud.CloudFlowRunner\",\n                    },\n                    {\n                        \"name\": \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\",\n                        \"value\": \"prefect.engine.cloud.CloudTaskRunner\",\n                    },\n                ],\n                \"essential\": True,\n            }\n        ]\n\n        # Assign repository credentials if they are specified\n        if self.repository_credentials:\n            self.logger.debug(\"Assigning repository credentials\")\n            container_definitions[0][\"repositoryCredentials\"] = {\n                \"credentialsParameter\": self.repository_credentials\n            }\n\n        # Register task definition\n        self.logger.debug(\n            \"Registering task definition {}\".format(\n                flow_run.flow.id[:8]  # type: ignore\n            )\n        )\n        self.boto3_client.register_task_definition(\n            family=\"prefect-task-{}\".format(flow_run.flow.id[:8]),  # type: ignore\n            containerDefinitions=container_definitions,\n            requiresCompatibilities=[\"FARGATE\"],\n            networkMode=\"awsvpc\",\n            cpu=self.task_cpu,\n            memory=self.task_memory,\n            taskRoleArn=self.task_role_arn,\n            executionRoleArn=self.execution_role_arn,\n        )\n\n    def _run_task(self, flow_run: GraphQLResult) -> None:\n        \"\"\"\n        Run a task using the flow run.\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        container_overrides = [\n            {\n                \"name\": \"flow\",\n                \"environment\": [\n                    {\n                        \"name\": \"PREFECT__CLOUD__AUTH_TOKEN\",\n                        \"value\": config.cloud.agent.auth_token,\n                    },\n                    {\n                        \"name\": \"PREFECT__CONTEXT__FLOW_RUN_ID\",\n                        \"value\": flow_run.id,  # type: ignore\n                    },\n                ],\n            }\n        ]\n\n        network_configuration = {\n            \"awsvpcConfiguration\": {\n                \"subnets\": self.subnets,\n                \"assignPublicIp\": self.assign_public_ip,\n            }\n        }\n\n        # Asssign task security groups if they are specified\n        if self.security_groups:\n            self.logger.debug(\"Assigning security groups\")\n            network_configuration[\"awsvpcConfiguration\"][\n                \"securityGroups\"\n            ] = self.security_groups\n\n        # Run task\n        self.logger.debug(\n            \"Running task using task definition {}\".format(\n                flow_run.flow.id[:8]  # type: ignore\n            )\n        )\n        self.boto3_client.run_task(\n            cluster=self.cluster,\n            taskDefinition=\"prefect-task-{}\".format(\n                flow_run.flow.id[:8]  # type: ignore\n            ),\n            overrides={\"containerOverrides\": container_overrides},\n            launchType=\"FARGATE\",\n            networkConfiguration=network_configuration,\n        )\n\n\nif __name__ == \"__main__\":\n    FargateAgent().start()\n", "path": "src/prefect/agent/fargate/agent.py"}]}