> ## Documentation Index
> Fetch the complete documentation index at: /docs/llms.txt
> Use this file to discover all available pages before exploring further.
# ecs_worker
# `prefect_aws.workers.ecs_worker`
Prefect worker for executing flow runs as ECS tasks.
Get started by creating a work pool:
```
$ prefect work-pool create --type ecs my-ecs-pool
```
Then, you can start a worker for the pool:
```
$ prefect worker start --pool my-ecs-pool
```
It's common to deploy the worker as an ECS task as well. However, you can run the worker
locally to get started.
The worker may work without any additional configuration, but it is dependent on your
specific AWS setup and we'd recommend opening the work pool editor in the UI to see the
available options.
By default, the worker will register a task definition for each flow run and run a task
in your default ECS cluster using AWS Fargate. Fargate requires tasks to configure
subnets, which we will infer from your default VPC. If you do not have a default VPC,
you must provide a VPC ID or manually setup the network configuration for your tasks.
Note, the worker caches task definitions for each deployment to avoid excessive
registration. The worker will check that the cached task definition is compatible with
your configuration before using it.
The launch type option can be used to run your tasks in different modes. For example,
`FARGATE_SPOT` can be used to use spot instances for your Fargate tasks or `EC2` can be
used to run your tasks on a cluster backed by EC2 instances.
Generally, it is very useful to enable CloudWatch logging for your ECS tasks; this can
help you debug task failures. To enable CloudWatch logging, you must provide an
execution role ARN with permissions to create and write to log streams. See the
`configure_cloudwatch_logs` field documentation for details.
The worker can be configured to use an existing task definition by setting the task
definition arn variable or by providing a "taskDefinition" in the task run request. When
a task definition is provided, the worker will never create a new task definition which
may result in variables that are templated into the task definition payload being
ignored.
## Functions
### `parse_identifier` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L233](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L233)
```python theme={null}
parse_identifier(identifier: str) -> ECSIdentifier
```
Splits identifier into its cluster and task components, e.g.
input "cluster\_name::task\_arn" outputs ("cluster\_name", "task\_arn").
### `mask_sensitive_env_values` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L242](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L242)
```python theme={null}
mask_sensitive_env_values(task_run_request: dict, values: List[str], keep_length = 3, replace_with = '***')
```
### `mask_api_key` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L261](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L261)
```python theme={null}
mask_api_key(task_run_request)
```
## Classes
### `ECSIdentifier` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L154](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L154)
The identifier for a running ECS task.
### `CapacityProvider` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L269](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L269)
The capacity provider strategy to use when running the task.
### `ECSJobConfiguration` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L279](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L279)
Job configuration for an ECS worker.
**Methods:**
#### `at_least_one_container_is_essential` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L411](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L411)
```python theme={null}
at_least_one_container_is_essential(self) -> Self
```
Ensures that at least one container will be marked as essential
in the task definition.
#### `cloudwatch_logs_options_requires_configure_cloudwatch_logs` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L450](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L450)
```python theme={null}
cloudwatch_logs_options_requires_configure_cloudwatch_logs(self) -> Self
```
Enforces that an execution role arn is provided (or could be provided by a
runtime task definition) when configuring logging.
#### `configure_cloudwatch_logs_requires_execution_role_arn` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L427](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L427)
```python theme={null}
configure_cloudwatch_logs_requires_execution_role_arn(self) -> Self
```
Enforces that an execution role arn is provided (or could be provided by a
runtime task definition) when configuring logging.
#### `container_name_default_from_task_definition` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L395](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L395)
```python theme={null}
container_name_default_from_task_definition(self) -> Self
```
Infers the container name from the task definition if not provided.
#### `json_template` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L332](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L332)
```python theme={null}
json_template(cls) -> dict[str, Any]
```
Returns a dict with job configuration as keys and the corresponding templates as values
Defaults to using the job configuration parameter name as the template variable name.
e.g.
```python theme={null}
{
key1: '{{ key1 }}', # default variable template
key2: '{{ template2 }}', # `template2` specifically provide as template
}
```
#### `network_configuration_requires_vpc_id` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L465](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L465)
```python theme={null}
network_configuration_requires_vpc_id(self) -> Self
```
Enforces a `vpc_id` is provided when custom network configuration mode is
enabled for network settings.
#### `prepare_for_flow_run` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L358](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L358)
```python theme={null}
prepare_for_flow_run(self, flow_run: 'FlowRun', deployment: 'DeploymentResponse | None' = None, flow: 'APIFlow | None' = None, work_pool: 'WorkPool | None' = None, worker_name: str | None = None, worker_id: 'UUID | None' = None) -> None
```
#### `task_run_request_requires_arn_if_no_task_definition_given` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L379](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L379)
```python theme={null}
task_run_request_requires_arn_if_no_task_definition_given(self) -> Self
```
If no task definition is provided, a task definition ARN must be present on the
task run request.
### `ECSVariables` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L477](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L477)
Variables for templating an ECS job.
### `ECSWorkerResult` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L683](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L683)
The result of an ECS job.
### `ECSWorker` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L689](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L689)
A Prefect worker to run flow runs as ECS tasks.
**Methods:**
#### `kill_infrastructure` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L1685](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L1685)
```python theme={null}
kill_infrastructure(self, infrastructure_pid: str, configuration: ECSJobConfiguration, grace_seconds: int = 30) -> None
```
Stop an ECS task.
**Args:**
* `infrastructure_pid`: The infrastructure identifier in format
"cluster::task\_arn".
* `configuration`: The job configuration used to connect to AWS.
* `grace_seconds`: Not used for ECS (ECS handles graceful shutdown internally).
**Raises:**
* `InfrastructureNotFound`: If the task doesn't exist.
#### `run` [https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect\_aws/workers/ecs\_worker.py#L727](https://github.com/PrefectHQ/prefect/blob/main/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py#L727)
```python theme={null}
run(self, flow_run: 'FlowRun', configuration: ECSJobConfiguration, task_status: Optional[anyio.abc.TaskStatus] = None) -> ECSWorkerResult
```
Runs a given flow run on the current worker.