Short answer: In Prefect 3.x there isn’t a public

PrefectClient

method for querying events. To fetch events you’ll call the Events REST API directly. You can optionally use the server “events client” classes, but they’re in the

prefect.server

namespace and aren’t considered part of the stable, public Python client. Useful docs: - Read events (REST): Read events - Stream events (Server-Sent Events): Stream workspace events - Python “events clients” (server internals): prefect.server.events.clients Example: fetching a page of events via REST (works for Prefect Cloud or OSS Server)

import os
import httpx

base_url = os.environ.get("PREFECT_API_URL", "<http://127.0.0.1:4200/api>")
headers = {}
api_key = os.environ.get("PREFECT_API_KEY")
if api_key:
    headers["Authorization"] = f"Bearer {api_key}"

with httpx.Client(timeout=30) as client:
    # Adjust params per the REST docs (e.g., limit, page token, filters, time range)
    resp = client.get(f"{base_url}/events", params={"limit": 50}, headers=headers)
    resp.raise_for_status()
    data = resp.json()
    print(data)

Example: streaming events in real time via SSE

import os
import httpx

base_url = os.environ.get("PREFECT_API_URL", "<http://127.0.0.1:4200/api>")
headers = {}
api_key = os.environ.get("PREFECT_API_KEY")
if api_key:
    headers["Authorization"] = f"Bearer {api_key}"

with httpx.Client(timeout=None) as client:
    with client.stream("GET", f"{base_url}/events/stream", headers=headers) as r:
        r.raise_for_status()
        for line in r.iter_lines():
            if line:
                print(line)

If you prefer a Python abstraction, you can experiment with

prefect.server.events.clients.PrefectServerEventsAPIClient

, but note it’s not a public/stable API. If you share whether you’re on Prefect Cloud or OSS and what filters/time range you need, I can tailor the exact request parameters for you.