I’m not aware of an official Prefect ↔️ New Relic integration for Prefect 3.x, but there are a few good patterns that people use: Best options 1. Send Prefect logs to New Relic - Prefect uses standard Python logging, so you can attach a custom handler or ship stdout/stderr from your worker/container into New Relic. - Prefect logging customization docs: logging customization 2. Use flow/task state hooks to emit New Relic events - You can attach

on_completion

,

on_failure

, etc. to flows/tasks and call New Relic’s ingest API there. - Prefect hooks docs: state change hooks 3. Emit Prefect events / use automations or webhooks - If you want alerts around failures / retries / SLA-style events, Prefect events + webhooks can be a clean fit. - Custom events: custom events 4. Instrument your flow code with New Relic’s Python agent - If your main goal is APM/tracing of the code inside tasks/flows, instrument the worker process or container with New Relic’s Python agent, then run Prefect normally. A practical pattern If you want something simple and reliable, I’d usually recommend: - Use New Relic agent for app-level tracing/metrics - Use Prefect hooks for run-state notifications - Forward logs from worker/container runtime to New Relic Example flow hook pattern:

python
import os
import requests
from prefect import flow

NEW_RELIC_LICENSE_KEY = os.getenv("NEW_RELIC_LICENSE_KEY")
NEW_RELIC_ACCOUNT_ID = os.getenv("NEW_RELIC_ACCOUNT_ID")

def notify_new_relic(flow, flow_run, state):
    url = f"<https://insights-collector.newrelic.com/v1/accounts/{NEW_RELIC_ACCOUNT_ID}/events>"
    headers = {
        "Api-Key": NEW_RELIC_LICENSE_KEY,
        "Content-Type": "application/json",
    }

    payload = {
        "eventType": "PrefectFlowStateChange",
        "flow_run_id": str(flow_run.id),
        "flow_name": flow.name,
        "flow_run_name": flow_run.name,
        "state": state.name,
        "message": state.message,
    }

    <http://requests.post|requests.post>(url, json=payload, headers=headers, timeout=5)

@flow(on_failure=[notify_new_relic], on_completion=[notify_new_relic])
def my_flow():
    print("running")

What I did not find - No clear official Prefect-maintained New Relic integration package - No obvious GitHub discussion with established prior art from the Prefect repo Useful references - Prefect logging customization - Prefect state change hooks - Prefect custom events - New Relic Python agent - New Relic Event API - New Relic Log API If you want, I can put together a concrete example for your setup, e.g.: - Prefect worker in Docker/Kubernetes + New Relic - serverless worker + New Relic - log forwarding config - failure alerting only