Short answer: yes. In Prefect 3.x Cloud you can stream account-level events (which include audit activity) and ship them to S3. There isn’t a built-in “send to S3” button, but it’s straightforward to forward either in real time or in batches. Options - Real-time via CLI (quick start) - Stream account-wide events (includes audit activity) and write to a file; rotate and upload to S3 with your preferred mechanism.

prefect events stream --account --format json --output-file /var/log/prefect/audit-events.jsonl

Tip: run this under a supervisor and rotate the file hourly (e.g., with logrotate or by wrapping with timeout), then upload rotated files to S3. - Real-time via Python (recommended for direct S3 writes)

import asyncio, json, os
  import boto3
  from datetime import datetime, timezone
  from prefect.events.clients import PrefectCloudAccountEventSubscriber
  from prefect.events import EventFilter

  # Requires PREFECT_API_KEY, PREFECT_API_URL env vars for Cloud auth
  S3_BUCKET = os.environ["S3_BUCKET"]
  S3_PREFIX = os.environ.get("S3_PREFIX", "prefect/audit-events/")

  async def main():
      s3 = boto3.client("s3")
      subscriber = PrefectCloudAccountEventSubscriber(filter=EventFilter())
      buf, count, start = [], 0, datetime.now(timezone.utc)

      async with subscriber as events:
          async for event in events:
              buf.append(event.model_dump())
              count += 1
              if count >= 500:
                  ts = start.strftime("%Y/%m/%d/%H/%M")
                  key = f"{S3_PREFIX}{ts}.jsonl"
                  body = "\n".join(json.dumps(e) for e in buf)
                  s3.put_object(Bucket=S3_BUCKET, Key=key, Body=body.encode())
                  buf, count, start = [], 0, datetime.now(timezone.utc)

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

Notes: - Uses the account-level subscriber to include audit-related events. - Batch/rotate however you like (e.g., by size or time). - Batch export (historical backfill) via REST API - Paginate events over a time window and write to S3.

# POST <https://api.prefect.cloud/api/accounts/{account_id}/workspaces/{workspace_id}/events/filter>
  # Use occurred.start/occurred.end, paginate with next_page.

Docs: Events REST API Important notes - Permissions: use a Cloud service account/API key with events read access. For account-wide (audit) events, ensure the key has the necessary account-level scopes. - There’s no separate “audit log export” API; account-level event streaming/filtering is the supported path. - If you’re running open-source Prefect Server (not Cloud), account-level audit logs aren’t available; you can still export workspace events and run/task logs. Helpful links - Events concepts: Events overview - CLI streaming (verified): prefect events stream - REST API (events): Read events - Audit logs UI (context): Audit logs in Cloud Quick question to tailor the setup: - Are you on Prefect Cloud or self-hosted? - Do you want continuous real-time streaming, or periodic bulk exports? - Any preferences for file partitioning in S3 (e.g., hourly/day-partitioned prefixes)?