Great question — this is a classic “fan-out, do a small per-item chain, fan-in” pattern. In Prefect 3.x you can do this cleanly and control parallelism in two complementary ways: - Per-flow cap: limit how many tasks run concurrently within this flow run (ThreadPoolTaskRunner) - Workspace-wide cap: ensure a global max across all runs/workers (Prefect concurrency limits) Below are both options with a complete example. Option A — cap concurrency within the flow (simple and local) - Use a ThreadPoolTaskRunner to limit concurrent tasks in this single flow run - Chain task2 → task3 per item so the right outputs feed the right inputs

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def make_items() -> list[str]:
    # produce your list
    return [f"item-{i}" for i in range(25)]

@task
def task2(item: str) -> str:
    # invoke your AWS Lambda and wait for completion
    # e.g., boto3.client("lambda").invoke(...) then read payload
    return f"lambda-output-for-{item}"

@task
def task3(task2_output: str) -> str:
    # post-process task2 output
    return f"processed-{task2_output}"

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))  # cap to 5 concurrent tasks
def orchestrate():
    items = make_items().result()  # resolve the list
    futures = [None] * len(items)
    for i, item in enumerate(items):
        t2_future = task2.submit(item)
        t3_future = task3.submit(t2_future)  # depends on t2 for the same item
        futures[i] = t3_future               # keeps results in input order
    results = [f.result() for f in futures]
    return results

if __name__ == "__main__":
    orchestrate()

Notes: - Because

task3

depends on

task2

for the same item, they’ll never overlap for that item, but multiple item-chains will run in parallel up to

max_workers

. - For network-bound work (like waiting on Lambda), threads are ideal. If you were CPU-bound, consider

ProcessPoolTaskRunner

. Option B — cap concurrency across your whole workspace (recommended for Lambda) If you need to ensure a global maximum number of Lambda invocations across all flows/workers, use Prefect’s concurrency limits. There are two good ways to do it: 1) Tag-based concurrency limits (no code changes inside tasks) - Add a shared tag to your Lambda-related tasks - Create a concurrency limit for that tag Code:

@task(tags=["lambda-invoke"])
def task2(item: str) -> str:
    # call Lambda and wait
    return f"lambda-output-for-{item}"

@task(tags=["lambda-invoke"])
def task3(task2_output: str) -> str:
    # post-process
    return f"processed-{task2_output}"

CLI (verified):

prefect concurrency-limit create lambda-invoke 5

This ensures at most 5 tasks with tag

lambda-invoke

are Running at once across the entire workspace. Both

task2

and

task3

share that cap. 2) Global concurrency limits via code (fine-grained control inside a task) - Create a global concurrency limit (GCL) by name - Acquire slots in code when you do the Lambda call CLI (verified):

prefect global-concurrency-limit create aws-lambda --limit 5

Code:

from prefect.concurrency.sync import concurrency

@task
def task2(item: str) -> str:
    with concurrency("aws-lambda"):
        # call Lambda and wait for completion
        return f"lambda-output-for-{item}"

@task
def task3(task2_output: str) -> str:
    with concurrency("aws-lambda"):
        # post-process (if you want this to count against the same cap)
        return f"processed-{task2_output}"

How to choose - Only need a per-run cap? Use ThreadPoolTaskRunner. - Need a hard global cap across all runs/workers? Use tag-based concurrency limits or the global concurrency limit context manager. - You can combine both: e.g., ThreadPoolTaskRunner(max_workers=10) locally and a global limit of 5 to protect Lambda. Follow-up questions to tailor this for you - Do you need the cap across all flows/workers, or just within this one run?

- What’s your desired max concurrency for Lambda invocations? - Roughly how many items might task1 return (tens, thousands)? If very large, I can show a streaming version using

prefect.futures.as_completed

to avoid queuing thousands of tasks at once.

You’re seeing that error because you’re trying to index the Prefect future itself. A future is a handle to a result, not the result. Get the result first, then index the dict. Three correct patterns: 1) Resolve in the flow, then index

fut = some_task.submit(...)
data = fut.result()           # blocks until the task completes
value = data["your_key"]      # now safe to index
next_task.submit(value)

2) Pass the future directly to the next task (let Prefect resolve it) Do the dict access inside the downstream task so Prefect resolves inputs for you:

from prefect import task

@task
def pick_key(d: dict, key: str):
    return d[key]

fut = some_task.submit(...)
value_future = pick_key.submit(fut, "your_key")  # fut is resolved by Prefect
next_task.submit(value_future)

3) Chain futures without resolving until the end (for many items)

t2_futures = [task2.submit(item) for item in items]
t3_futures = [task3.submit(f) for f in t2_futures]  # Prefect resolves per item
results = [f.result() for f in t3_futures]          # resolve at the end

Tips: -

.result(timeout=None, raise_on_failure=True)

will raise the task’s exception if it failed; set

raise_on_failure=False

if you want to inspect state first. - For a bunch of futures, you can also use:

from prefect.futures import resolve_futures_to_results
  results = resolve_futures_to_results(t3_futures)

- If you only need a single field from the dict, doing the extraction inside a small task (pattern #2) keeps your graph declarative and avoids blocking in the flow.

PrefectFuture.result()

returns whatever your task returned. If you’re getting a bool, it means your task function is returning a bool — not a dict. That’s why

[...]

indexing fails with “'bool' object is not subscriptable”. Quick checklist: - Make sure the task actually returns a dict - Don’t index the future; index the resolved result Example with an AWS Lambda invoke parsed to a dict:

from prefect import task, flow
import json
import boto3

lambda_client = boto3.client("lambda")

@task
def invoke_lambda(item: str) -> dict:
    resp = lambda_client.invoke(
        FunctionName="my-func",
        Payload=json.dumps({"item": item}).encode("utf-8"),
    )
    # boto3 returns a streaming body in Payload; parse it to a dict
    body_bytes = resp["Payload"].read()
    body = json.loads(body_bytes.decode("utf-8"))
    return body  # ensure this is a dict

@flow
def main():
    fut = invoke_lambda.submit("item-1")
    result_dict = fut.result()          # returns the dict your task returned
    value = result_dict["your_key"]     # now safe to index
    print(value)

If you want to keep things fully async in the graph, do the key access in a small task and pass the future directly:

@task
def get_key(d: dict, key: str):
    return d[key]

fut = invoke_lambda.submit("item-1")
value_future = get_key.submit(fut, "your_key")   # Prefect resolves fut for you
print(value_future.result())

Common gotchas that lead to a bool: - Returning

response.ok

instead of

response.json()

from an HTTP client - Returning an expression that evaluates to a bool (e.g.,

'key' in d

) - For Lambda, not parsing

resp["Payload"]

and instead returning something like

resp["ResponseMetadata"]["HTTPStatusCode"] == 200

If you’re still seeing a bool, can you share the task body and the flow lines where you call

.submit()

and

.result()

? I’ll pinpoint exactly where the bool is coming from.

You’re hitting a botocore read timeout while waiting for the Lambda response. A few things to fix/improve: 1) Small bug in your snippet You’ve got a double equals on FunctionName. It should be a keyword arg:

resp = client.invoke(
    FunctionName=function_name,           # not `FunctionName==...`
    InvocationType="RequestResponse",
    Payload=json.dumps(event).encode(),
)

2) Increase the boto3 client timeouts (and pool size if you parallelize) By default, botocore’s read_timeout can be too low for longer Lambdas. Set a higher read_timeout/connect_timeout and, if you’re doing parallel calls, increase max_pool_connections to avoid pool exhaustion: Pure boto3:

from botocore.config import Config

config = Config(
    read_timeout=300,            # > your Lambda’s max duration
    connect_timeout=10,
    retries={"max_attempts": 3, "mode": "standard"},
    max_pool_connections=50,     # bump if running many concurrent calls
)

session = aws.get_boto3_session()
client = session.client("lambda", region_name="us-east-2", config=config)

resp = client.invoke(
    FunctionName=function_name,
    InvocationType="RequestResponse",
    Payload=json.dumps(event).encode(),
)
body_bytes = resp["Payload"].read()

3) Prefer the Prefect AWS block for cleaner config (optional but nice) Using the Prefect AWS integration lets you centralize credentials and client config. It’s not required, but it’s convenient:

from prefect_aws.credentials import AwsCredentials
from prefect_aws.lambda_function import LambdaFunction

aws = AwsCredentials(
    region_name="us-east-2",
    aws_client_parameters={
        "config": {
            "read_timeout": 300,
            "connect_timeout": 10,
            "retries": {"max_attempts": 3, "mode": "standard"},
            "max_pool_connections": 50,
        }
    },
)

lambda_fn = LambdaFunction(
    function_name=function_name,
    aws_credentials=aws,
)

resp = lambda_fn.invoke(
    payload=event,
    invocation_type="RequestResponse",
    tail=True,   # optional – include last 4KB of logs (base64 in LogResult)
)
body_bytes = resp["Payload"].read()

4) Align timeouts and retries - Lambda function timeout: set in AWS. Make sure your botocore read_timeout > function timeout (plus some buffer). - Prefect task retries: your log “Retry 1/3…” is Prefect retrying the task. That’s fine for orchestration-level resilience. You can keep 2–3 retries in Prefect and also use botocore retries for transient network hiccups. - If you only want to retry on timeouts, you can add a

retry_condition_fn

to the task to filter which exceptions trigger retries. Example with error handling in a task:

from prefect import task, get_run_logger
from botocore.exceptions import ReadTimeoutError, ClientError

@task(retries=3, retry_delay_seconds=2)
def invoke_lambda(aws, function_name: str, event: dict) -> dict:
    logger = get_run_logger()
    session = aws.get_boto3_session()
    client = session.client(
        "lambda",
        region_name="us-east-2",
        config=Config(
            read_timeout=300,
            connect_timeout=10,
            retries={"max_attempts": 3, "mode": "standard"},
            max_pool_connections=50,
        ),
    )
    try:
        resp = client.invoke(
            FunctionName=function_name,
            InvocationType="RequestResponse",
            Payload=json.dumps(event).encode(),
        )
        if resp.get("FunctionError"):
            raise RuntimeError(f"Lambda error: {resp['FunctionError']}")
        return json.loads(resp["Payload"].read().decode())
    except ReadTimeoutError as e:
        logger.warning(f"Lambda read timeout: {e}")
        raise
    except ClientError as e:
        logger.error(f"Lambda client error: {e}")
        raise

5) If you’re invoking many Lambdas in parallel - Use a ThreadPoolTaskRunner with a reasonable

max_workers

and/or Prefect concurrency limits to avoid overwhelming the client or Lambda concurrency.

- Bump

max_pool_connections

as shown above. If you’d like, I can help adapt this to your exact flow and set a safe concurrency cap so you don’t run into connection pool or Lambda concurrency throttling.

Glad that helped! To run 100 Lambdas but keep only 50 running at a time, think in two layers: 1) Inside a single flow run (per-run cap) - Set your flow’s task runner to limit concurrency:

from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=50))
def orchestrate():
    ...

This ensures that, within this flow run, no more than 50 Lambda-invoking tasks run at once. 2) Workspace-wide protection (cross-run cap, strongly recommended for Lambda) Use Prefect’s concurrency limits so you never exceed 50 Lambdas across all flows/workers: Option A: Tag-based limit (no code changes beyond tags)

@task(tags=["lambda-invoke"])
def invoke_lambda(...):
    ...

CLI (verified):

prefect concurrency-limit create lambda-invoke 50

This caps all tasks tagged

lambda-invoke

to 50 Running at a time across your workspace. Option B: Global concurrency limit (code-enforced)

from prefect.concurrency.sync import concurrency

@task
def invoke_lambda(...):
    with concurrency("aws-lambda"):
        ...

CLI (verified):

prefect global-concurrency-limit create aws-lambda --limit 50

About “what should the workers be set to?” - If you’re running ONE flow that internally fans out 100 tasks, your ThreadPoolTaskRunner(max_workers=50) is the key knob; a single worker is fine for that flow. - If you’re running MANY flows concurrently, set a work pool concurrency to control how many flow runs can be active at once:

prefect work-pool set-concurrency-limit "my-pool" 10

This limits concurrent flow runs, not tasks within a flow. Use this alongside the task-level concurrency limit (tags/global) if needed. AWS-side considerations - Set your Lambda function’s reserved concurrency to at least 50 if you need guaranteed capacity (and consider provisioned concurrency if cold-starts are a concern). - Bump your botocore client pool and timeouts so 50 concurrent requests don’t exhaust the pool:

from botocore.config import Config
config = Config(
    read_timeout=300,
    connect_timeout=10,
    retries={"max_attempts": 3, "mode": "standard"},
    max_pool_connections=50,
)
client = session.client("lambda", region_name="us-east-2", config=config)

Putting it all together - In the flow: ThreadPoolTaskRunner(max_workers=50) - In the tasks: tag with

lambda-invoke

and/or wrap with

with concurrency("aws-lambda")

- In the workspace: create either the tag-based limit or the global concurrency limit at 50 - In AWS: ensure reserved concurrency >= 50; set client max_pool_connections and timeouts accordingly If you share whether you run a single flow or many concurrent flows in a pool, I can suggest the exact combo (task runner size, pool limit, tags/global limit) to match your setup.

Short answer: by default, no — if a given task2 fails, the corresponding task3 for that item will not run. Only that one branch is skipped; the other items continue. The downstream future for task3 will end up in an upstream-failed/cancelled state, and calling

.result()

on it will raise an

UpstreamTaskError

. If you want task3 to still run for cleanup or custom handling even when task2 fails, use

allow_failure

on that edge and handle the state in task3:

from prefect import flow, task, allow_failure
from prefect.states import State, get_state_result, get_state_exception

@task
def task2(item: str) -> dict:
    # may raise
    ...

@task
def task3(t2_output):
    # When allow_failure is used, a failed upstream arrives as a State instead of raising
    if isinstance(t2_output, State):
        if t2_output.is_failed():
            err = get_state_exception(t2_output)
            # handle/skip/log as you need
            return {"status": "skipped", "error": str(err)}
        else:
            t2_output = get_state_result(t2_output)
    # normal processing
    return {"status": "ok", "value": t2_output}

@flow
def orchestrate(items: list[str]):
    t2_futs = [task2.submit(i) for i in items]
    t3_futs = [task3.submit(allow_failure(f)) for f in t2_futs]
    results = [f.result() for f in t3_futs]
    return results

Other patterns you can use: - Retries on task2 to reduce failures:

@task(retries=3, retry_delay_seconds=2)
  def task2(...): ...

- Catch exceptions inside task2 and return a structured result like

{"ok": False, "error": ...}

so task3 can branch on it without

allow_failure

. - When gathering many results and you expect some failures, prefer resolving to states first and inspecting:

from prefect.futures import resolve_futures_to_states
  states = resolve_futures_to_states(t3_futs)
  # inspect each state, extract results or exceptions as needed

Yes — you can let task2 fail and also have task3 fail for that item, while keeping the rest of the run moving. The key is to avoid raising exceptions in the flow body before you’ve submitted all work. Most “everything stopped” symptoms come from resolving a failed future too early in the flow, which raises and aborts the flow (cancelling tasks that haven’t started). What to do - Submit first, resolve later: - Submit all task2s and task3s (with allow_failure on the edge) before calling

.result()

. - When you finally resolve, either: - Resolve to states (no raises) and inspect them, or - Resolve results with

raise_on_failure=False

. - Use allow_failure so task3 runs even when task2 failed and can decide whether to fail too. Pattern

from prefect import flow, task, allow_failure
from prefect.futures import resolve_futures_to_states
from prefect.states import get_state_result, get_state_exception

@task(retries=2, retry_delay_seconds=2)
def task2(item: str) -> dict:
    # may raise -> marks this task run as Failed (good for analytics)
    ...

@task
def task3(t2_value_or_state):
    # If task2 failed and was passed via allow_failure, we receive a State
    from prefect.states import State
    if isinstance(t2_value_or_state, State):
        if t2_value_or_state.is_failed():
            # decide: we can also fail task3 to reflect double-failure in analytics
            err = get_state_exception(t2_value_or_state)
            raise RuntimeError(f"Upstream task2 failed: {err}")
        else:
            t2_value_or_state = get_state_result(t2_value_or_state)
    # normal processing
    return {"status": "ok"}

@flow
def orchestrate(items: list[str]):
    # 1) Submit everything first
    t2_futs = [task2.submit(i) for i in items]
    t3_futs = [task3.submit(allow_failure(f)) for f in t2_futs]

    # 2) Resolve to states (does not raise), so the flow does not abort early
    states = resolve_futures_to_states(t3_futs)

    # 3) Optional: fail the flow at the end if any t3 failed, after all work ran
    had_failures = any(s.is_failed() for s in states)
    if had_failures:
        raise RuntimeError("One or more item chains failed")

    return [get_state_result(s) for s in states if s.is_completed()]

Why this keeps going - Failing task2 marks that task run as Failed. Because we pass it via

allow_failure

, task3 still runs for that item and can also fail. Other items’ chains are unaffected and continue. - We don’t call

.result()

on any future until after all submissions. We resolve to states, which lets the flow finish scheduling and running the entire batch before we optionally fail the flow at the end. If you already need to read results as they finish - Use

.result(raise_on_failure=False)

so a failure doesn’t blow up the flow body:

r = some_future.result(raise_on_failure=False)
  # r is the task’s return value if completed; otherwise you can check the future's state

- Or stick to the states pattern shown above and aggregate at the end. Do you also want the flow to succeed even if some items fail? - Then skip the final raise; the failed task runs will be counted in analytics as Failed, but the flow can still end Completed. If you want the flow to be marked Failed only if any failures occurred (but still run everything), use the final raise after all tasks finished, as shown.

In Prefect 3.x, flow “inputs” are just regular function parameters on your flow. Prefect infers a parameter schema from your signature and renders it in the UI so you can pass values when you start a run. 1) Define the flow with parameters and parse them inside - Keep parameters JSON-serializable (str, int, float, bool, list, dict). - If you want structure/validation, parse into a Pydantic model in the flow. Simple example:

from prefect import flow, task
from typing import Optional

@task
def do_work(source: str, limit: int, options: dict) -> dict:
    # your task logic here
    return {"source": source, "count": limit, "opts": options}

@flow
def my_flow(
    source: str,                # required
    limit: int = 100,           # default
    options: Optional[dict] = None  # optional nested config
):
    options = options or {}
    # do any parsing/normalization you need
    # e.g., turn flags into correct types, fill defaults, etc.
    result = do_work(source, limit, options)
    return result

With Pydantic validation:

from pydantic import BaseModel

class RunConfig(BaseModel):
    country: str
    start_date: str   # or a date; pass ISO strings from UI
    dry_run: bool = False
    max_items: int = 100

@flow
def my_flow(config: dict):
    cfg = RunConfig.model_validate(config)  # validate + coerce
    # use cfg.country, cfg.start_date, cfg.dry_run, cfg.max_items

2) Add inputs via the UI - Create a deployment for your flow (so it appears in the UI’s Deployments page). - In the UI: Deployments > select your deployment > Run. - In the Parameters panel, enter JSON matching your flow signature. - For the first example:

{
  "source": "<s3://my-bucket/path>",
  "limit": 50,
  "options": {
    "dry_run": true,
    "priority": "high"
  }
}

- For the Pydantic example:

{
  "config": {
    "country": "US",
    "start_date": "2025-01-01",
    "dry_run": true,
    "max_items": 500
  }
}

- Click Run. Prefect will pass those values to your flow. 3) Set default parameters on the deployment (optional) You can bake defaults into the deployment so the UI is pre-filled: - If you deploy in Python:

# if using from_source, you can set parameters when deploying
  from prefect import flow
  # ... define my_flow ...
  my_flow.from_source("<https://github.com/owner/repo>", entrypoint="flows/my.py:my_flow").deploy(
      name="my-deployment",
      work_pool_name="my-pool",
      parameters={"limit": 200, "options": {"dry_run": false}}
  )

- You can still override them at run time in the UI. Notes - The UI accepts JSON; dates should be strings (e.g., "2025-01-01") unless you parse them yourself. - Keep sensitive values out of parameters; use blocks (e.g., Secrets) or environment variables instead. - If you need to collect input mid-run (human-in-the-loop), Prefect also supports pausing a flow to “request input” via the Run Input feature; happy to share an example if that’s what you mean by “parse it within the flow.”

Short answer: no special requirement. If you want default parameter values for a deployment, add a parameters map under that deployment in prefect.yaml. Prefect will render these defaults in the UI and use them for runs unless you override them at run-time. Minimal example:

deployments:
  - name: "my-deployment"
    entrypoint: "flows/my_flow.py:my_flow"
    parameters:
      source: "<s3://my-bucket/path>"
      limit: 100
      options:
        dry_run: false

Notes: - These are defaults; you can override them in the UI when you click Run. - You don’t need to declare types in YAML; Prefect infers from your flow signature. Keep values JSON-serializable. - If you use schedules, you can also set schedule-specific parameter overrides:

schedule:
    cron: "0 9 * * *"
    timezone: "US/Eastern"
    parameters:
      dry_run: true

- Everything else (work_pool, entrypoint, schedule) is standard. No extra keys are needed just for parameters. If you prefer to set parameters in Python instead of YAML, you can set

parameters={...}

in

flow.from_source(...).deploy(...)

, and they’ll show up in the UI the same way.

You’re likely calling it in an async context.

get_state_result

is sync-compatible: - In sync code, call it directly. - In async code, you must await it. Examples: - Downstream sync task

from prefect import task
from prefect.states import State, get_state_result

@task
def task3(t2_value_or_state):
    if isinstance(t2_value_or_state, State):
        value = get_state_result(t2_value_or_state)  # returns your dict
    else:
        value = t2_value_or_state
    return value["some_key"]

- Downstream async task

from prefect import task
from prefect.states import State, get_state_result

@task
async def task3(t2_value_or_state):
    if isinstance(t2_value_or_state, State):
        value = await get_state_result(t2_value_or_state)  # await in async
    else:
        value = t2_value_or_state
    return value["some_key"]

- In the flow body - Sync flow:

d = get_state_result(state)  # dict

- Async flow:

d = await get_state_result(state)  # dict

If you don’t need to use

State

directly, you can also stick to futures: - Sync flow/task:

val = fut.result(raise_on_failure=False)

- Async flow/task:

val = await fut.result(raise_on_failure=False)

Finally, ensure the upstream task actually returns a dict. If it returns bytes (e.g., raw Lambda payload), decode/JSON-load inside that task before returning, or you’ll get bytes instead of a dict.

You’re hitting that KeyError because with allow_failure the downstream input can be either: - the upstream task’s return value (on success), or - a State (on failure) Your code only handles the State branch, so when agent2 succeeds you leave prev_result as {} and then index ['success']. Fix: handle both cases and validate the structure before indexing. Drop-in patch:

from prefect.states import State, get_state_result, get_state_exception

@task(name="Agent3", retries=1, retry_delay_seconds=2, timeout_seconds=900, log_prints=True)
def agent3_scrape_and_process_articles(prev_result_value_or_state, event: dict, credentials: dict) -> dict:
    print(f"Starting Agent3 - Scrape and Process Articles with Name ID {task_run.get_name()}")

    # Normalize upstream input to a dict 'prev_result'
    if isinstance(prev_result_value_or_state, State):
        if prev_result_value_or_state.is_failed():
            err = get_state_exception(prev_result_value_or_state)
            raise RuntimeError(f"Upstream Agent 2 failed with error: {err}")
        prev_result = get_state_result(prev_result_value_or_state)  # sync call
    else:
        prev_result = prev_result_value_or_state  # agent2 succeeded: this is the raw return

    # Defensive checks
    if not isinstance(prev_result, dict):
        raise TypeError(f"Agent2 returned {type(prev_result)!r}; expected dict with 'success' and 'output_state_id'")

    success = prev_result.get("success", False)
    if not success:
        print(f"Not running Agent3 {task_run.get_name()} because previous state success was false")
        raise RuntimeError(
            f"Previous state failed so cannot continue, prev id: {prev_result.get('output_state_id')}"
        )

    # Proceed
    event["input_state_id"] = prev_result["output_state_id"]
    ...
    return {...}

And your submit pattern is good:

a2_future = agent2_find_article_urls.submit(copyEvent, credentials)
a3_future = agent3_scrape_and_process_articles.submit(allow_failure(a2_future), copyEvent, credentials)

Make sure agent2 returns the shape you expect:

# In agent2
return {
    "success": True,               # or False on failure you handle internally
    "output_state_id": some_id,
    ...
}

If agent2 actually raises on failure (recommended), you’ll receive a Failed State in agent3 (handled above). If agent2 sometimes returns a non-dict (e.g., bytes from Lambda), parse to dict inside agent2 before returning, so agent3 can safely read keys.

Great question! Assuming you’re on Prefect 3.x, here are the two simplest ways to keep

prefect server

running in the background on Ubuntu (EC2): Option A — Quick and dirty (good for testing) - Use tmux or screen so the process survives your SSH session:

tmux new -s prefect
  prefect server start --host 0.0.0.0 --port 4200

Detach with Ctrl+b d and reattach with

tmux attach -t prefect

. - Or use nohup (less robust):

nohup prefect server start --host 0.0.0.0 --port 4200 > /var/log/prefect-server.log 2>&1 &
  disown

Option B — Recommended for reliability (systemd service) - Create a systemd unit so the server starts on boot and restarts on failure. 1) Create an environment file for settings (recommended, especially if you use Postgres):

sudo mkdir -p /etc/prefect
sudo tee /etc/prefect/server.env >/dev/null <<'EOF'
# Point clients/workers at this server
PREFECT_API_URL=http://YOUR_EC2_PUBLIC_DNS_OR_IP:4200/api

# Optional: use Postgres in production for persistence
# Make sure the DB exists and the user has permissions
# Requires asyncpg driver in your environment (pip install "prefect[postgres]")
# PREFECT_API_DATABASE_CONNECTION_URL=postgresql+asyncpg://user:pass@db-host:5432/prefect

# Logging level
PREFECT_SERVER_LOGGING_LEVEL=INFO
EOF

2) Create the systemd service: - Replace /path/to/venv/bin/prefect with the full path to your Prefect CLI (or adjust PATH). - Replace User/Group/WorkingDirectory with your setup.

sudo tee /etc/systemd/system/prefect-server.service >/dev/null <<'EOF'
[Unit]
Description=Prefect Server
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=prefect
Group=prefect
WorkingDirectory=/home/prefect
EnvironmentFile=/etc/prefect/server.env
# If using a virtualenv, point ExecStart to the venv's prefect binary
ExecStart=/path/to/venv/bin/prefect server start --host 0.0.0.0 --port 4200 --log-level ${PREFECT_SERVER_LOGGING_LEVEL}
Restart=always
RestartSec=5
# Ensure the service has the venv on PATH if you prefer that approach
# Environment=PATH=/path/to/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin

[Install]
WantedBy=multi-user.target
EOF

3) Reload, enable, and start:

sudo systemctl daemon-reload
sudo systemctl enable --now prefect-server
sudo systemctl status prefect-server
# Logs:
journalctl -u prefect-server -f

Notes and best practices - Port and host: The UI/API default to port 4200. Use

--host 0.0.0.0

if you want to reach it from other machines. Ensure your EC2 security group allows inbound 4200 or put it behind a reverse proxy. - Persistence: For production, use Postgres via

PREFECT_API_DATABASE_CONNECTION_URL

so your state survives restarts. You’ll need the async driver installed in the environment that runs the service (e.g.,

pip install "prefect[postgres]"

). - Environment loading: When starting via systemd, environment variables from your shell/profile will not be loaded. Put settings in

Environment=

entries or an

EnvironmentFile

. Some users have reported that DB settings set in a Prefect profile aren’t used by

prefect server start

unless they’re set in the OS environment. See: PREFECT_API_DATABASE_CONNECTION_URL ignored from profile - Pointing clients/workers: On any machine that needs to talk to this server, set:

prefect config set PREFECT_API_URL=<http://YOUR_EC2_PUBLIC_DNS_OR_IP:4200/api>

If you’d like, share how you’ve installed Prefect (system Python vs. venv/conda) and whether you want Postgres. I can tailor the exact unit file and env for your setup.

Short answer: yes—set memory on the infrastructure your flow run uses via the deployment’s

work_pool.job_variables

. In Prefect 3.x, memory is configured per “job” (container/pod/task) launched by your worker, not per individual

@task

. Which worker/work pool are you using (Kubernetes, Docker, ECS, or Process)? The exact keys differ. Examples: - Docker work pool (prefect-docker) - Use

mem_limit

(and optionally

memswap_limit

) in your deployment YAML:

deployments:
    - name: my-deployment
      entrypoint: flows/my_flow.py:my_flow
      work_pool:
        name: my-docker-pool
        job_variables:
          mem_limit: "2g"         # e.g. "512m", "1g"
          memswap_limit: "4g"     # optional; -1 disables swap

Docs: Docker deployment guide, prefect-docker integration - Kubernetes work pool (prefect-kubernetes) - Add memory requests/limits in the work pool’s Base Job Template (Advanced settings), then pass values via

job_variables

in the deployment. Example base job template fragment:

{
  "variables": {
    "memory_request": {
      "type": "string",
      "default": "256Mi"
    },
    "memory_limit": {
      "type": "string",
      "default": "512Mi"
    }
  },
  "job_configuration": {
    "spec": {
      "template": {
        "spec": {
          "containers": [
            {
              "name": "prefect-job",
              "resources": {
                "requests": {
                  "memory": "{{ memory_request }}"
                },
                "limits": {
                  "memory": "{{ memory_limit }}"
                }
              }
            }
          ]
        }
      }
    }
  }
}

Then in your deployment:

deployments:
    - name: my-deployment
      entrypoint: flows/my_flow.py:my_flow
      work_pool:
        name: my-k8s-pool
        job_variables:
          memory_request: "1Gi"
          memory_limit: "2Gi"

Note: use Kubernetes units like `Mi`/`Gi`. Docs: Kubernetes deployment guide, Customize job variables, Manage work pools - AWS ECS work pool (prefect-aws) - Use integer

cpu

and

memory

(MB) in `job_variables`:

deployments:
    - name: my-deployment
      entrypoint: flows/my_flow.py:my_flow
      work_pool:
        name: my-ecs-pool
        job_variables:
          cpu: 1024         # CPU units
          memory: 2048      # MB
          launch_type: "FARGATE"

Docs: ECS worker guide - Process work pool (local subprocess) - There isn’t a Prefect-level memory setting. Use OS/container limits (e.g., run the process worker inside Docker/Kubernetes and control memory there). A couple of notes: - These settings apply to the entire flow run container/pod/task. Prefect doesn’t enforce per-task memory inside a single process. If you need per-task isolation, run tasks in separate containers/pods (e.g., a task that launches a job) or use an orchestrator like Kubernetes directly. - You can also set these in Python with

flow.deploy(..., job_variables={...})

. If you tell me which worker/work pool you’re using, I can tailor the exact YAML or template you need.

Yes — in Prefect 3.x you can fail a task immediately (no more retries) by using a retry condition. The

retry_condition_fn

lets you decide, per exception, whether to retry or not. - If you never want retries: set

retries=0

(that’s the default). - If you only want to skip retries for certain expected errors: use

retry_condition_fn

to return

False

for those exceptions. Example: fail fast on an expected error, but retry on others

from prefect import flow, task

class ExpectedError(Exception):
    pass

def retry_condition(task, task_run, state) -> bool:
    # Return True to retry, False to stop retrying
    try:
        state.result()  # re-raise the task exception
    except ExpectedError:
        return False  # don't retry this one
    except Exception:
        return True   # retry other exceptions

@task(retries=3, retry_condition_fn=retry_condition)
def my_task(x: int):
    if x < 0:
        raise ExpectedError("Invalid input; fail immediately without retry")
    # simulate some other transient error
    raise RuntimeError("Transient issue; please retry")

@flow
def my_flow():
    my_task(-1)  # will fail once and not retry

Key points: -

retry_condition_fn

runs whenever the task fails; returning

False

ends retries and leaves the task in a Failed state. - If you never want retries at all, just don’t set

retries

(default is 0). Docs: - Retries (Prefect 3.x) - States overview