Short answer: In Prefect 3.x there isn’t a supported way to keep running 100+ individual Prefect task runs locally (for caching/concurrency) but suppress only their task-level events to Cloud. Every task state transition (including Cached) emits events, and there’s no per-task filtering, sampling, or “flow-only” reporting toggle. What you can do instead (trade-offs below): - Don’t schedule tasks that you already know are cached - Prefect’s task caching still creates a task run and emits a Cached state event. If you can cheaply check a local/persistent cache before calling

task.submit

, you can skip creating the task run entirely for cached cases and avoid those events. - Example pattern:

def cache_has(item) -> bool:
        # your quick local/disk/kv check here
        ...

    @task(cache_key_fn=lambda p: p["key"], persist_result=True)
    def process_one(params):
        ...

    def fan_out(items):
        futures = []
        for it in items:
            if cache_has(it):   # bypass Prefect task entirely
                continue
            futures.append(process_one.submit({"key": it.key, "data": it.data}))
        return futures

- Batch inside a single (or small number of) Prefect tasks - Keep Prefect’s visibility at the batch boundary but do your wide fan-out internally with a thread/async pool and your own per-item cache. Prefect will see 1 task run (dramatically fewer events), while you keep worker-level concurrency and caching inside that task. - You lose Prefect-managed retries/visibility per item, but you can implement retries in code if needed. - Sketch:

from concurrent.futures import ThreadPoolExecutor, as_completed
    import hashlib, json, os

    def disk_cache_get(key):
        p = f"/tmp/cache/{key}.json"
        try:
            with open(p) as f:
                return json.load(f)
        except FileNotFoundError:
            return None

    def disk_cache_set(key, val):
        os.makedirs("/tmp/cache", exist_ok=True)
        with open(f"/tmp/cache/{key}.json", "w") as f:
            json.dump(val, f)

    def key_for_case(case):
        return hashlib.sha1(json.dumps(case, sort_keys=True).encode()).hexdigest()

    def run_one(case):
        k = key_for_case(case)
        cached = disk_cache_get(k)
        if cached is not None:
            return cached
        # your expensive work here
        result = do_work(case)
        disk_cache_set(k, result)
        return result

    @task
    def process_batch(cases, max_workers=32):
        results = []
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futs = {pool.submit(run_one, c): c for c in cases}
            for f in as_completed(futs):
                results.append(f.result())
        return results

- All-or-nothing: programmatically drop all events from a process - If you absolutely need to keep the same task granularity but stop sending events, you can override the events client to a Null client in that worker process. This will suppress all events (flow, task, concurrency, heartbeats) from that process. You will lose UI timeline/automations and other event-driven features; use with caution. - Example:

from prefect.events.clients import NullEventsClient
    from prefect.events.worker import EventsWorker

    # Call once at process start, e.g., inside your worker image entrypoint
    EventsWorker.set_client_override(NullEventsClient)

- Note: This is process-wide and unspecific; Prefect does not support selectively dropping just task events today. - Consider raising your event limit - If the above trade-offs don’t work, we can discuss increasing your Cloud event rate limit for your workspace/account. Reply here and we’ll connect you with our team. A few additional notes: - Concurrency-limiter operations also emit events when acquiring/releasing limits; if you have many tags/limits, that adds to volume.

- Subflows still emit events; moving work to subflows won’t reduce event counts. - API request usage and event ingestion are metered separately; reducing events typically requires changing what Prefect “sees” as runs, or suppressing events entirely. References: - Events overview: Events in Prefect - Concurrency limits: Concurrency limits - Task caching overview: Task caching If you can share which constraint matters most (keep Prefect retries per item vs. keep UI visibility vs. reduce events), I can suggest a concrete pattern tailored to your flow.