add plugin-based send_event() API for backend-agnostic event triggering

Register event providers (ArgoEvent, SFNEvent, AirflowEvent) in the
Metaflow plugin system via EVENT_PROVIDERS_DESC.  The new send_event()
function in metaflow.integrations auto-detects the configured backend
or accepts an explicit backend= override.

- Add "event_provider" category to _plugin_categories
- Add EVENT_PROVIDERS_DESC + resolve_plugins("event_provider")
- Create SFNEvent (EventBridge) and AirflowEvent (REST API) providers
- Fix ArgoEvent.is_configured() to read env var at call time
- Fix ArgoEvent.__init__ url resolution for late-bound env vars
- Update Phase 2 trigger tests to use plugin-based send_event

Author: Nissan Pow

metaflow/extension_support/plugins.py

@@ -189,6 +189,7 @@ def resolve_plugins(category, path_only=False):
     "secrets_provider": lambda x: x.TYPE,
     "gcp_client_provider": lambda x: x.name,
     "deployer_impl_provider": lambda x: x.TYPE,
+    "event_provider": lambda x: x.TYPE,
     "azure_client_provider": lambda x: x.name,
     "sidecar": None,
     "logging_sidecar": None,

metaflow/integrations.py

@@ -17,7 +17,11 @@
 #  - name: name of the integration alias
 #  - obj: object it points to
 #
-ALIASES_DESC = [("ArgoEvent", ".plugins.argo.argo_events.ArgoEvent")]
+ALIASES_DESC = [
+    ("ArgoEvent", ".plugins.argo.argo_events.ArgoEvent"),
+    ("SFNEvent", ".plugins.aws.step_functions.sfn_event.SFNEvent"),
+    ("AirflowEvent", ".plugins.airflow.airflow_event.AirflowEvent"),
+]
 
 # Aliases can be enabled or disabled through configuration or extensions:
 #  - ENABLED_INTEGRATION_ALIAS: list of alias names to enable.
@@ -27,5 +31,70 @@
 #    ENABLED_INTEGRATION_ALIAS from the concatenation of the names in
 #    ALIASES_DESC (concatenation of the names here as well as in the extensions).
 
+
+def _resolve_event_providers():
+    """Lazily resolve all registered event provider classes from the plugin system."""
+    from metaflow.plugins import EVENT_PROVIDERS
+
+    return EVENT_PROVIDERS
+
+
+def send_event(name, payload=None, backend=None, **kwargs):
+    """Send a trigger event to wake a deployed @trigger flow.
+
+    Discovers event providers via the plugin system. Each provider class
+    exposes ``TYPE`` (e.g. ``"argo-workflows"``) and ``is_configured()``
+    which checks whether the required environment variables are set.
+
+    When *backend* is ``None``, the first provider whose ``is_configured()``
+    returns ``True`` is used.  When *backend* is given, it is matched against
+    the provider's ``TYPE`` (dashes are normalised to underscores).
+
+    New backends are added by registering an ``EVENT_PROVIDERS_DESC`` entry
+    in ``metaflow/plugins/__init__.py`` — no changes to this function needed.
+
+    Parameters
+    ----------
+    name : str
+        Event name (must match the @trigger event name on the deployed flow).
+    payload : dict, optional
+        Key-value pairs delivered with the event, used to set parameters.
+    backend : str, optional
+        Override auto-detection.  Matched against provider TYPE
+        (e.g. ``"argo-workflows"``, ``"step-functions"``, ``"airflow"``).
+        Underscores are accepted as well (``"argo_workflows"``).
+    **kwargs
+        Passed through to the provider class constructor.
+
+    Returns
+    -------
+    str or None
+        Event ID if published successfully.
+    """
+    providers = _resolve_event_providers()
+
+    if backend is not None:
+        # Normalise so "argo_workflows" matches "argo-workflows"
+        norm = backend.replace("_", "-")
+        for provider_class in providers:
+            if provider_class.TYPE == norm:
+                return provider_class(name, payload=payload, **kwargs).publish()
+        available = [p.TYPE for p in providers]
+        raise ValueError(
+            "Unknown event backend %r. Available: %s" % (backend, available)
+        )
+
+    # Auto-detect: pick the first configured provider
+    for provider_class in providers:
+        if hasattr(provider_class, "is_configured") and provider_class.is_configured():
+            return provider_class(name, payload=payload, **kwargs).publish()
+
+    available = [p.TYPE for p in providers]
+    raise RuntimeError(
+        "Cannot auto-detect event backend — no provider is configured. "
+        "Set the 'backend' parameter explicitly or configure one of: %s" % available
+    )
+
+
 # Keep this line and make sure ALIASES_DESC is above this line.
 process_integration_aliases(globals())

metaflow/plugins/__init__.py

@@ -172,6 +172,12 @@
     ("airflow", ".airflow.airflow_deployer.AirflowDeployer"),
 ]
 
+EVENT_PROVIDERS_DESC = [
+    ("argo-workflows", ".argo.argo_events.ArgoEvent"),
+    ("step-functions", ".aws.step_functions.sfn_event.SFNEvent"),
+    ("airflow", ".airflow.airflow_event.AirflowEvent"),
+]
+
 TL_PLUGINS_DESC = [
     ("yaml_parser", ".parsers.yaml_parser"),
     ("requirements_txt_parser", ".pypi.parsers.requirements_txt_parser"),
@@ -220,6 +226,8 @@ def get_runner_cli_path():
 if sys.version_info >= (3, 7):
     DEPLOYER_IMPL_PROVIDERS = resolve_plugins("deployer_impl_provider")
 
+EVENT_PROVIDERS = resolve_plugins("event_provider")
+
 TL_PLUGINS = resolve_plugins("tl_plugin")
 
 from .cards.card_modules import MF_EXTERNAL_CARDS

metaflow/plugins/airflow/airflow_event.py

@@ -0,0 +1,109 @@
+import json
+import sys
+import time
+import urllib
+import uuid
+from datetime import datetime
+
+from metaflow.exception import MetaflowException
+
+
+class AirflowEventException(MetaflowException):
+    headline = "Airflow Event Exception"
+
+
+class AirflowEvent(object):
+    """
+    AirflowEvent triggers an Airflow DAG run via the Airflow REST API,
+    used to start flows deployed with @trigger.
+
+    Parameters
+    ----------
+    name : str
+        Event name — maps to the DAG ID of the deployed flow.
+    url : str, optional
+        Airflow webserver base URL. Defaults to METAFLOW_AIRFLOW_WEBSERVER_URL env var.
+    payload : dict, optional
+        Key-value pairs delivered as DAG run conf parameters.
+    """
+
+    TYPE = "airflow"
+
+    @classmethod
+    def is_configured(cls):
+        import os
+
+        return bool(os.environ.get("METAFLOW_AIRFLOW_WEBSERVER_URL"))
+
+    def __init__(self, name, url=None, payload=None):
+        import os
+
+        self._name = name
+        self._url = url or os.environ.get(
+            "METAFLOW_AIRFLOW_WEBSERVER_URL", "http://localhost:8080"
+        )
+        self._payload = payload or {}
+
+    def add_to_payload(self, key, value):
+        """Add a key-value pair to the event payload."""
+        self._payload[key] = str(value)
+        return self
+
+    def publish(self, payload=None, ignore_errors=True):
+        """
+        Trigger an Airflow DAG run via the REST API.
+
+        Parameters
+        ----------
+        payload : dict, optional
+            Additional key-value pairs to merge into the conf.
+        ignore_errors : bool, default True
+            If True, errors are silently ignored.
+
+        Returns
+        -------
+        str or None
+            The DAG run ID if triggered successfully, None otherwise.
+        """
+        if payload is None:
+            payload = {}
+
+        try:
+            dag_run_id = "metaflow__%s__%s" % (
+                self._name,
+                str(uuid.uuid4())[:8],
+            )
+            conf = {
+                "name": self._name,
+                "id": str(uuid.uuid4()),
+                "timestamp": int(time.time()),
+                "utc_date": datetime.utcnow().strftime("%Y%m%d"),
+                "generated-by-metaflow": True,
+                **self._payload,
+                **payload,
+            }
+
+            api_url = "%s/api/v1/dags/%s/dagRuns" % (
+                self._url.rstrip("/"),
+                self._name,
+            )
+            data = json.dumps({"dag_run_id": dag_run_id, "conf": conf}).encode("utf-8")
+
+            request = urllib.request.Request(
+                api_url,
+                method="POST",
+                headers={"Content-Type": "application/json"},
+                data=data,
+            )
+
+            urllib.request.urlopen(request, timeout=60)
+            print("Airflow Event (%s) published." % self._name, file=sys.stderr)
+            return dag_run_id
+
+        except Exception as e:
+            msg = "Unable to publish Airflow Event (%s): %s" % (self._name, e)
+            if ignore_errors:
+                print(msg, file=sys.stderr)
+                return None
+            else:
+                raise AirflowEventException(msg)

metaflow/plugins/argo/argo_events.py

@@ -35,9 +35,18 @@ class ArgoEvent(object):
         A set of key-value pairs delivered in this event. Used to set parameters of triggered flows.
     """
 
-    def __init__(
-        self, name, url=ARGO_EVENTS_WEBHOOK_URL, payload=None, access_token=None
-    ):
+    TYPE = "argo-workflows"
+
+    @classmethod
+    def is_configured(cls):
+        # Check the env var directly (not the module-level constant) so that
+        # env vars set after import are picked up.  Also fall back to the
+        # already-loaded config value for non-env-var configuration sources.
+        return bool(
+            os.environ.get("ARGO_EVENTS_WEBHOOK_URL") or ARGO_EVENTS_WEBHOOK_URL
+        )
+
+    def __init__(self, name, url=None, payload=None, access_token=None):
         # TODO: Introduce support for NATS
         if callable(name):
             name = name()
@@ -47,7 +56,10 @@ def __init__(
                     % type(name).__name__
                 )
         self._name = name
-        self._url = url
+        # Resolve URL: explicit arg > env var > config constant
+        self._url = (
+            url or os.environ.get("ARGO_EVENTS_WEBHOOK_URL") or ARGO_EVENTS_WEBHOOK_URL
+        )
         self._payload = payload or {}
         self._access_token = access_token
 

metaflow/plugins/aws/step_functions/sfn_event.py

@@ -0,0 +1,112 @@
+import json
+import os
+import sys
+import time
+import uuid
+from datetime import datetime
+
+from metaflow.exception import MetaflowException
+
+
+class SFNEventException(MetaflowException):
+    headline = "SFN Event Exception"
+
+
+class SFNEvent(object):
+    """
+    SFNEvent sends a trigger event via AWS EventBridge to start Step Functions
+    workflows deployed with @trigger.
+
+    Parameters
+    ----------
+    name : str
+        Event name (must match the @trigger event name on the deployed flow).
+    payload : dict, optional
+        Key-value pairs delivered with the event, used to set parameters of
+        triggered flows.
+    """
+
+    TYPE = "step-functions"
+
+    @classmethod
+    def is_configured(cls):
+        import os
+
+        return bool(
+            os.environ.get("METAFLOW_SFN_IAM_ROLE")
+            or os.environ.get("METAFLOW_SFN_STATE_MACHINE_PREFIX")
+        )
+
+    def __init__(self, name, payload=None):
+        self._name = name
+        self._payload = payload or {}
+
+    def add_to_payload(self, key, value):
+        """Add a key-value pair to the event payload."""
+        self._payload[key] = str(value)
+        return self
+
+    def publish(self, payload=None, ignore_errors=True):
+        """
+        Publish an event to EventBridge that will trigger any Step Functions
+        workflow listening for this event name.
+
+        Parameters
+        ----------
+        payload : dict, optional
+            Additional key-value pairs to merge into the payload.
+        ignore_errors : bool, default True
+            If True, errors are silently ignored.
+
+        Returns
+        -------
+        str or None
+            The event ID if published successfully, None otherwise.
+        """
+        if payload is None:
+            payload = {}
+
+        try:
+            import boto3
+
+            event_bus = os.environ.get("METAFLOW_SFN_EVENT_BUS_ARN", "default")
+
+            event_id = str(uuid.uuid4())
+            event_payload = {
+                "name": self._name,
+                "id": event_id,
+                "timestamp": int(time.time()),
+                "utc_date": datetime.utcnow().strftime("%Y%m%d"),
+                "generated-by-metaflow": True,
+                **self._payload,
+                **payload,
+            }
+
+            client = boto3.client("events")
+            response = client.put_events(
+                Entries=[
+                    {
+                        "Source": "metaflow",
+                        "DetailType": self._name,
+                        "Detail": json.dumps(event_payload),
+                        "EventBusName": event_bus,
+                    }
+                ]
+            )
+
+            if response.get("FailedEntryCount", 0) > 0:
+                err = response["Entries"][0].get("ErrorMessage", "Unknown error")
+                raise SFNEventException(
+                    "Failed to publish event %s: %s" % (self._name, err)
+                )
+
+            print("SFN Event (%s) published." % self._name, file=sys.stderr)
+            return event_id
+
+        except Exception as e:
+            msg = "Unable to publish SFN Event (%s): %s" % (self._name, e)
+            if ignore_errors:
+                print(msg, file=sys.stderr)
+                return None
+            else:
+                raise SFNEventException(msg)