Hi Community, We need help with Prefect agent setup. We have hosted Prefect server in EC2 instance and created prefect agent in ECS Faregate. We are using docker image as “prefecthq/prefect:2.10.8-python3.10”. We configured environment variables "‘PREFECT_API_URL’ , ‘PREFECT_CLOUD_API’ ‘PREFECT_LOGGING_LEVEL’: ‘INFO’,‘AWS_RETRY_MODE’‘adaptive’,‘AWS MAX ATTEMPTS’‘10’ while creating ECS Fargate Agent Service and running ""prefect agent start -p default-agent-pool -q default"" command through command line. In ECS Fargate while running services and task we are getting below error. If someone can point out which environment variable setup missing then it help us to complete setup. Error logs : Agent started! Looking for work from work pool ‘default-agent-pool’… Failed the last 3 attempts. Please check your environment and configuration. Examples of recent errors: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 111, in connect_tcp stream: anyio.abc.ByteStream = await anyio.connect_tcp( File “/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py”, line 213, in connect_tcp async with create_task_group() as tg: File “/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py”, line 658, in aexit raise CancelledError asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 10, in map_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 110, in connect_tcp with anyio.fail_after(timeout): File “/usr/local/lib/python3.10/site-packages/anyio/_core/_tasks.py”, line 118, in exit raise TimeoutError TimeoutError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 60, in map_httpcore_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 353, in handle_async_request resp = await self._pool.handle_async_request(req) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 253, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 237, in handle_async_request response = await connection.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 86, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 63, in handle_async_request stream = await self._connect(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 111, in _connect stream = await self._network_backend.connect_tcp(**kwargs) File “/usr/local/lib/python3.10/site-packages/httpcore/backends/auto.py”, line 29, in connect_tcp return await self._backend.connect_tcp( File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 109, in connect_tcp with map_exceptions(exc_map): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 14, in map_exceptions raise to_exc(exc) httpcore.ConnectTimeout The above exception was the direct cause of the following exception: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/prefect/utilities/services.py”, line 46, in critical_service_loop await workload() File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 196, in get_and_submit_flow_runs work_queue_names=, File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 196, in work_queue_names=, File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 144, in get_work_queues work_queue = await self.client.read_work_queue_by_name( File “/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py”, line 844, in read_work_queue_by_name response = await self._client.get( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1754, in get return await self.request( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1530, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 246, in send response = await self._send_with_retry( File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 192, in _send_with_retry response = await request() File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1617, in send response = await self._send_handling_auth( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1645, in _send_handling_auth response = await self._send_handling_redirects( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1682, in _send_handling_redirects response = await self._send_single_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1719, in _send_single_request response = await transport.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 352, in handle_async_request with map_httpcore_exceptions(): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.ConnectTimeout Failed the last 3 attempts. Please check your environment and configuration. Examples of recent errors: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 111, in connect_tcp stream: anyio.abc.ByteStream = await anyio.connect_tcp( File “/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py”, line 213, in connect_tcp async with create_task_group() as tg: File “/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py”, line 658, in aexit raise CancelledError asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 10, in map_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 110, in connect_tcp with anyio.fail_after(timeout): File “/usr/local/lib/python3.10/site-packages/anyio/_core/_tasks.py”, line 118, in exit raise TimeoutError TimeoutError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 60, in map_httpcore_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 353, in handle_async_request resp = await self._pool.handle_async_request(req) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 253, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 237, in handle_async_request response = await connection.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 86, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 63, in handle_async_request stream = await self._connect(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 111, in _connect stream = await self._network_backend.connect_tcp(**kwargs) File “/usr/local/lib/python3.10/site-packages/httpcore/backends/auto.py”, line 29, in connect_tcp return await self._backend.connect_tcp( File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 109, in connect_tcp with map_exceptions(exc_map): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 14, in map_exceptions raise to_exc(exc) httpcore.ConnectTimeout The above exception was the direct cause of the following exception: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/prefect/utilities/services.py”, line 46, in critical_service_loop await workload() File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 261, in check_for_cancelled_flow_runs async for work_queue in self.get_work_queues(): File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 144, in get_work_queues work_queue = await self.client.read_work_queue_by_name( File “/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py”, line 844, in read_work_queue_by_name response = await self._client.get( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1754, in get return await self.request( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1530, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 246, in send response = await self._send_with_retry( File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 192, in _send_with_retry response = await request() File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1617, in send response = await self._send_handling_auth( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1645, in _send_handling_auth response = await self._send_handling_redirects( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1682, in _send_handling_redirects response = await self._send_single_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1719, in _send_single_request response = await transport.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 352, in handle_async_request with map_httpcore_exceptions(): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.ConnectTimeout Agent stopped!

Hi all, I am trying to run my Prefect Agent using a Rancher container. This is set to pull from the image I pushed to my Azure CR, and I think it is pulling the image correctly since it is able to set the Prefect_API_URL which is set in the dockerfile, and it tries to run a flow when I hit execute on the orchestrator portal. However, I am getting an error that says something where it is unable to connect to Docker. I'm not sure what this means exactly.. I am trying to determine what is missing in my setup, it seems like something isn't installed or something, but this is really hard for me to understand what is off. At some point it does say file not found, so maybe it doesn't have the right version of Python or something? See attached for the error logs. Does anyone know what this is saying?

Hi all, I am trying to follow the tutorial https://docs.prefect.io/latest/guides/deployment/helm-worker/ but I do not understand where/how to find the User Account ID : May I ask for help in this channel or should I elsewhere ?

Hi all, I am attempting to use Prefect as a tool for executing a calendar-day backfill of a sequential series of steps (that is, for each calendar day, run a bunch of steps in series), with a concurrency limit on the number of simultaneous backfill executions. The series of steps fits well into the flow and task model w/ SequentialTaskRunner. Where I am getting hung up a bit is on the backfill mechanism, and also, the parallel execution of flows with the concurrency limit. I've looked at Prefect "deployments", but that seems more suited to some sort of prolonged forward fill. Is there a way to queue up many flow-runs whose parallelism is governed by a work pool? Each time I invoke a flow, it's blocking on execution completion. Apologies in advance if it's blatantly obvious in your docs and I've missed it 🙂

hi all may i know how to delete all flow-runs using cli or command line, the 'prefect flow-run delete' required flow-run uuid. i want to delete all at one

Hi, is it possible to upload a deployment in a github repo ? Is is documented as read-only here https://docs.prefect.io/latest/concepts/filesystems/#github But used just after withthe -sb github/block tag And I have this error, with this block

Hi all! How can I submit a flow as an async job so that my Fast API request can resolve?

Hello everyone, I'm looking to install Prefect locally to do some testing and demos but I'm getting the following error. Has anyone ever encountered this before? If so, how do I resolve it? Thanks!

Hello everyone! I'm trying to get Prefect2 running on one of our remote VMs. However I'm unable to connect to it and as far as I can tell it must be something with prefect. What I do is simply run

prefect server start

and then point my browser to

<server-ip>:4200

to get the UI but only get "site can't be reached". However if I simple start a python http.server I can access that machine without a problem. I read online that previously there was a

--expose

flag that you could run but it doesn't seem to be a part of prefect2. Does anyone have any experience with getting this working, maybe I'm just missing something?

Hello,

mypy

is complaining when attempting to unpack a tuple returned from a task. In the example below, when attempting to unpack

a

&

b

from the call to

get_numbers

,

mypy

raises the error

"None" object is not iterable

. However, the type definition for the

get_numbers

function shows that

None

should never be returned from the task. The main issue is that the type of

get_numbers

within the flow is

Task[(a: int, b: int), tuple[int, int]]

, instead of

tuple[int, int]

as

mypy

is expecting. Any idea how to avoid this type error without completely ignoring it? (Prefect v2.10.10, mypy v1.3.0)

from prefect import flow, task


@task
def get_numbers(a: int, b: int) -> tuple[int, int]:
    return a, b


@flow
def get_numbers_flow() -> None:
    a, b = get_numbers(1, 2)   # <- "None" object is not iterable.

Hello everyone, I'm on an M1 Mac trying to get Prefect installed but I'm running into the following error when I simply install via the cli: `incompatible architecture (have 'arm64', need 'x86_64')).`I have switched my terminal to Rosetta to no avail. Has anyone encountered this issue?

Is there a way to make Prefect 2 use the actual function name as default flow name without explicitly setting it via

name=

?

@flow
def do_things():  # flow will be named do-things not do_things as expected
    pass

When I apply the task decorator to a method within a class, I'm getting the error that the method is expecting the "self" parameter, which it doesn't get. Is something like this supported, or should I keep all my flows and tasks in the main class?

Hi, I am a prefect user and just implemented it at a client. I don’t like releasing flows through the Git integration and doing so through docker is much better, however the documentation is really bad. Has anyone been able to push flows through docker or better yet have them run through Kubernetes Jobs?

Is there any way to actually rename a flow run in Prefect 2 with

@flow(flow_run_name="foobar")

? Even the examples from the docs copied verbatim do not work (see replies).

Hello there

I am running in a EC2 instance, 4 containers of docker that will act as prefect agents. In each container we will run using Process

from prefect.infrastructure.process import Process

the workflows

How can I make sure that my process will always run into a clean working directory, so what clean mechanisms can I work with ?

I am pulling a github storage block to run the code and then I want to run and clean everything afterwards

Any help is welcomed! I just want to have a cleaning mechanism to make sure I have clean workflows run .. by default it seems it doesn't clean anything and it gets us problems ( especially Github storage include_git_objects set to True, lot of problems )

Thanks a lot for your time helping me with this!

may i know why the prefect cli command when output to file the output get truncated. ive tried to increased the history size in windows command prompt settings (windows 11) but it still happening. tried to use powershell, git bash terminal same issue still persist eg command prefect flow-run ls > output.txt

Hi everyone, I have a question regarding the priority queue in the workpool So let suppose i have 2 queue (q1, q2) in a workpool (w) where q1 in a priority queue with highest priority. Now let say i have to multiple input (input1, input2 and so on) where "input 23" has the highest priority and need to processed as soon as it came to "q1" My question here is: 1. The scenario i mentioned above is okay ? 2. if input 16 come to q1 then all the inputs which are in q2 will be put on to hold till input 16 of q1 is processed ? Can anyone please help me on this.

Hiii I am new to open source and I don't know ow how to deal with it.. It will be so helpful to make me walk through it

Hi everyone, is it possible for one flow to be processed in 2 (or more) k8s jobs? For example to run subflows in separate k8s jobs or separate pods?

Hi everyone, I'm trying to persist the results of my tasks. Referred to the docs and I found that we can rename the persisted file using the result storage key. However, when I tried to use it in my task decorator, the error is

TypeError: task() got an unexpected keyword argument 'result_storage_key'

. I'm using Prefect version 2.8.2.

Hey team, potentially a simple answer here but I’m trying to get Prefect set up on EC2. I’m wanting flow code to be on Github and downloaded at flow runtime (essentially CI/CD). I’m wanting to run both the server and the agents in Docker containers. The repo has all flows in a subdirectory (

repo-sub-path

below) but the code is not in the container. I’m running this command from within the container, and am getting this message:

prefect deployment build flow_test.py:flow_test --name test --storage-block github/flow-repository --path repo-sub-path

. What am I doing wrong? Do I need to have the code sitting in the container locally to set up the deployment? Or can I have it pull from Github when first generating the deployment yaml? Is this actually possible or do I need to push my code to a docker image and manually pull that each time I update my flows? Appreciate any insight here.

Hey everyone, I am working on some flows using prefect 2.7.12, specifically sending an email when a flow fails. I'm having trouble accessing the name of the flow run instance (adjective, noun). I would like to be able to include this information in the email. I've tried a handful of things including: flow_run_name =context['flow_run_name'] flow_run_name = prefect.context.get("flow_run_name") flow_run_name = prefect.context.flow_run.name flow_name = get_flow_name() and probably a few that I am forgetting. Does anyone have any suggestions or advise? Anything you can provide is much appreciated!

Hi! Is there a way to start a flow or a task without using a decorator? E.g. calling a function of the prefect Python package. I am asking because some of the Python functions I would like to run in a flow or task are already decorated (e.g. because of hydra-core) and using Prefect's decorators on the same functions throws a RuntimeError.

Hey everyone, I'm having some trouble, and I'm not sure what the cause is. I have my cron schedule set in my deployment file, and it is showing up on my UI Dashboard. However, Prefect is not adhering to it. Any ideas what might cause this? I know the time zone is correct, that's about all I can think to check.