That error typically happens when the async client isn’t being closed cleanly, so httpx/asyncio tries to schedule work as the interpreter is shutting down. In Prefect 2.20, make sure you: - Use the async client within an async context manager - Run your coroutine with asyncio.run from a proper main guard - Do not reuse a client after its context exits Here’s a minimal, correct pattern:

import asyncio
from prefect import get_client

async def main():
    async with get_client() as client:
        # If you have the deployment ID directly:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id="00000000-0000-0000-0000-000000000000",
            # optional:
            # parameters={"x": 1},
            # tags=["triggered-by-api"],
        )
        print(f"Created flow run: {flow_run.id}")

if __name__ == "__main__":
    asyncio.run(main())

If you only know the deployment name, resolve it first:

import asyncio
from prefect import get_client

async def main():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name("flow_name/deployment_name")
        flow_run = await client.create_flow_run_from_deployment(deployment_id=deployment.id)
        print(flow_run.id)

if __name__ == "__main__":
    asyncio.run(main())

Common pitfalls that can cause “cannot schedule new futures after interpreter shutdown”: - Creating the client at module import time or outside an

async with

, then the process exits before cleanup - Calling this code from an `atexit`/shutdown handler or after the event loop is already closing - Reusing a client after it has been closed (after the

async with

block) - Running this in a worker thread without its own event loop If you need synchronous code, use the sync client:

from prefect import get_client

def main():
    with get_client(sync_client=True) as client:
        flow_run = client.create_flow_run_from_deployment(
            deployment_id="00000000-0000-0000-0000-000000000000"
        )
        print(flow_run.id)

if __name__ == "__main__":
    main()

If you’re already doing the above and still see the error, can you share: - A minimal snippet of how you’re invoking

create_flow_run_from_deployment

- Whether this runs in a short-lived script, a web server (e.g., FastAPI), a notebook, or another thread/process - Your Python version and where in the program lifecycle this is called (e.g., normal path vs. shutdown handler)

Sorry, I encountered an error while processing your request. Please try again.

Sorry, I encountered an error while processing your request. Please try again.

In Prefect 2.20, flow runs that are created but remain in Pending almost always point to orchestration not finding a worker to pick them up or being blocked by a pause/limit. Run through these in order: Top causes - No worker is running for the target work pool (or worker is pointed at a different pool/queue) - Work pool or work queue is paused - Deployment references a non-existent/mismatched work pool or work queue - Concurrency limits are reached (work pool limit or tag-based concurrency limits) - The run is scheduled for the future (less common if you didn’t set a future time) - Worker can’t submit infrastructure (Docker/K8s/Process) due to an error Quick checks 1) Verify a worker is running and attached to the right work pool

prefect worker ls
# if none are running:
prefect worker start -p "your-pool-name"

If your worker filters specific queues, ensure it includes the deployment’s queue. 2) Check the work pool status (not paused, correct type, limits)

prefect work-pool inspect "your-pool-name"
# If paused:
prefect work-pool resume "your-pool-name"

3) Check the work queues in that pool (exist, not paused)

prefect work-queue ls -p "your-pool-name"
prefect work-queue inspect -p "your-pool-name" -q "your-queue-name"
# If paused:
prefect work-queue resume -p "your-pool-name" -q "your-queue-name"

4) Verify the deployment is targeting the correct pool/queue and not paused

prefect deployment inspect "flow_name/deployment_name"
# Look for:
# - work_pool_name matches an existing pool
# - work_queue_name exists in that pool (if set)
# - paused is false

5) Check concurrency limits - Work pool concurrency limit:

prefect work-pool inspect "your-pool-name"
# If needed:
prefect work-pool update "your-pool-name" --concurrency-limit <higher-number>

- Tag-based concurrency limits:

prefect concurrency-limit ls
# If a tag is saturated:
prefect concurrency-limit inspect <tag>
# Options:
prefect concurrency-limit reset <tag>
# or increase the limit:
prefect concurrency-limit create <tag> <new-limit>

6) Check worker logs for infra submission errors - Start your worker with debug logs and watch for messages like “Failed to submit flow run …”

export PREFECT_LOGGING_LEVEL=DEBUG
prefect worker start -p "your-pool-name"

- For Docker/K8s workers, confirm the runtime is reachable (docker daemon up, kubectl works, etc.) If you want, I can help pinpoint it quickly—please share: - The output of: -

prefect worker ls

-

prefect work-pool inspect "your-pool-name"

-

prefect work-queue ls -p "your-pool-name"

-

prefect deployment inspect "flow_name/deployment_name"

-

prefect concurrency-limit ls

- How your worker is started (command and flags), and any relevant worker logs around when the run was created.