POST /v1/events/subscriptions

curl -X POST "https://your-company.open-metadata.org/api/v1/events/subscriptions" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dq-failed-tests-to-webhook",
    "displayName": "DQ failed tests to webhook",
    "description": "Notify the endpoint when a test case fails.",
    "alertType": "Observability",
    "resources": ["testCase"],
    "enabled": true,
    "input": {
      "filters": [
        {
          "name": "filterByFqn",
          "effect": "include",
          "prefixCondition": "AND",
          "arguments": [{ "name": "fqnList", "input": ["svc.db.schema.orders"] }]
        }
      ],
      "actions": [
        {
          "name": "GetTestCaseStatusUpdates",
          "effect": "include",
          "prefixCondition": "AND",
          "arguments": [{ "name": "testResultList", "input": ["Failed"] }]
        }
      ]
    },
    "destinations": [
      {
        "category": "External",
        "type": "Webhook",
        "timeout": 10,
        "readTimeout": 12,
        "enabled": true,
        "config": {
          "endpoint": "https://hooks.example.com/openmetadata",
          "httpMethod": "POST",
          "headers": { "X-Source": "openmetadata" },
          "authType": { "type": "bearer", "secretKey": "<hmac-secret>" }
        }
      }
    ]
  }'

from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.generated.schema.events.api.createEventSubscription import (
    CreateEventSubscription,
)
from metadata.generated.schema.events.eventSubscription import (
    AlertType,
    Destination,
    SubscriptionCategory,
    SubscriptionType,
)

server_config = OpenMetadataConnection(
    hostPort="https://your-company.open-metadata.org/api",
    authProvider="openmetadata",
    securityConfig=OpenMetadataJWTClientConfig(jwtToken="<YOUR-JWT-TOKEN>"),
)
metadata = OpenMetadata(server_config)

request = CreateEventSubscription(
    name="dq-failed-tests-to-webhook",
    alertType=AlertType.Observability,
    resources=["testCase"],
    input={
        "filters": [
            {
                "name": "filterByFqn",
                "effect": "include",
                "prefixCondition": "AND",
                "arguments": [{"name": "fqnList", "input": ["svc.db.schema.orders"]}],
            }
        ],
        "actions": [
            {
                "name": "GetTestCaseStatusUpdates",
                "effect": "include",
                "prefixCondition": "AND",
                "arguments": [{"name": "testResultList", "input": ["Failed"]}],
            }
        ],
    },
    destinations=[
        Destination(
            category=SubscriptionCategory.External,
            type=SubscriptionType.Webhook,
            config={
                "endpoint": "https://hooks.example.com/openmetadata",
                "authType": {"type": "bearer", "secretKey": "<hmac-secret>"},
            },
        )
    ],
    enabled=True,
)

# PUT creates the alert, or updates it when the name already exists
alert = metadata.create_or_update(data=request)
print(alert.fullyQualifiedName)

POST

events

subscriptions

POST /v1/events/subscriptions

curl -X POST "https://your-company.open-metadata.org/api/v1/events/subscriptions" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dq-failed-tests-to-webhook",
    "displayName": "DQ failed tests to webhook",
    "description": "Notify the endpoint when a test case fails.",
    "alertType": "Observability",
    "resources": ["testCase"],
    "enabled": true,
    "input": {
      "filters": [
        {
          "name": "filterByFqn",
          "effect": "include",
          "prefixCondition": "AND",
          "arguments": [{ "name": "fqnList", "input": ["svc.db.schema.orders"] }]
        }
      ],
      "actions": [
        {
          "name": "GetTestCaseStatusUpdates",
          "effect": "include",
          "prefixCondition": "AND",
          "arguments": [{ "name": "testResultList", "input": ["Failed"] }]
        }
      ]
    },
    "destinations": [
      {
        "category": "External",
        "type": "Webhook",
        "timeout": 10,
        "readTimeout": 12,
        "enabled": true,
        "config": {
          "endpoint": "https://hooks.example.com/openmetadata",
          "httpMethod": "POST",
          "headers": { "X-Source": "openmetadata" },
          "authType": { "type": "bearer", "secretKey": "<hmac-secret>" }
        }
      }
    ]
  }'

from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.generated.schema.events.api.createEventSubscription import (
    CreateEventSubscription,
)
from metadata.generated.schema.events.eventSubscription import (
    AlertType,
    Destination,
    SubscriptionCategory,
    SubscriptionType,
)

server_config = OpenMetadataConnection(
    hostPort="https://your-company.open-metadata.org/api",
    authProvider="openmetadata",
    securityConfig=OpenMetadataJWTClientConfig(jwtToken="<YOUR-JWT-TOKEN>"),
)
metadata = OpenMetadata(server_config)

request = CreateEventSubscription(
    name="dq-failed-tests-to-webhook",
    alertType=AlertType.Observability,
    resources=["testCase"],
    input={
        "filters": [
            {
                "name": "filterByFqn",
                "effect": "include",
                "prefixCondition": "AND",
                "arguments": [{"name": "fqnList", "input": ["svc.db.schema.orders"]}],
            }
        ],
        "actions": [
            {
                "name": "GetTestCaseStatusUpdates",
                "effect": "include",
                "prefixCondition": "AND",
                "arguments": [{"name": "testResultList", "input": ["Failed"]}],
            }
        ],
    },
    destinations=[
        Destination(
            category=SubscriptionCategory.External,
            type=SubscriptionType.Webhook,
            config={
                "endpoint": "https://hooks.example.com/openmetadata",
                "authType": {"type": "bearer", "secretKey": "<hmac-secret>"},
            },
        )
    ],
    enabled=True,
)

# PUT creates the alert, or updates it when the name already exists
alert = metadata.create_or_update(data=request)
print(alert.fullyQualifiedName)

Create an Alert

Create an alert with POST, or use PUT to create it if it doesn’t exist and update it if it does. With PUT, OpenMetadata matches on name. The request body is described in the Alerts overview.

Examples

The example creates an Observability alert on the Test Case source. It fires when a test result with status Failed is posted for any test case under svc.db.schema.orders, and it sends the event to a webhook signed with a hash-based message authentication code (HMAC) secret.

POST /v1/events/subscriptions

curl -X POST "https://your-company.open-metadata.org/api/v1/events/subscriptions" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dq-failed-tests-to-webhook",
    "displayName": "DQ failed tests to webhook",
    "description": "Notify the endpoint when a test case fails.",
    "alertType": "Observability",
    "resources": ["testCase"],
    "enabled": true,
    "input": {
      "filters": [
        {
          "name": "filterByFqn",
          "effect": "include",
          "prefixCondition": "AND",
          "arguments": [{ "name": "fqnList", "input": ["svc.db.schema.orders"] }]
        }
      ],
      "actions": [
        {
          "name": "GetTestCaseStatusUpdates",
          "effect": "include",
          "prefixCondition": "AND",
          "arguments": [{ "name": "testResultList", "input": ["Failed"] }]
        }
      ]
    },
    "destinations": [
      {
        "category": "External",
        "type": "Webhook",
        "timeout": 10,
        "readTimeout": 12,
        "enabled": true,
        "config": {
          "endpoint": "https://hooks.example.com/openmetadata",
          "httpMethod": "POST",
          "headers": { "X-Source": "openmetadata" },
          "authType": { "type": "bearer", "secretKey": "<hmac-secret>" }
        }
      }
    ]
  }'

from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.generated.schema.events.api.createEventSubscription import (
    CreateEventSubscription,
)
from metadata.generated.schema.events.eventSubscription import (
    AlertType,
    Destination,
    SubscriptionCategory,
    SubscriptionType,
)

server_config = OpenMetadataConnection(
    hostPort="https://your-company.open-metadata.org/api",
    authProvider="openmetadata",
    securityConfig=OpenMetadataJWTClientConfig(jwtToken="<YOUR-JWT-TOKEN>"),
)
metadata = OpenMetadata(server_config)

request = CreateEventSubscription(
    name="dq-failed-tests-to-webhook",
    alertType=AlertType.Observability,
    resources=["testCase"],
    input={
        "filters": [
            {
                "name": "filterByFqn",
                "effect": "include",
                "prefixCondition": "AND",
                "arguments": [{"name": "fqnList", "input": ["svc.db.schema.orders"]}],
            }
        ],
        "actions": [
            {
                "name": "GetTestCaseStatusUpdates",
                "effect": "include",
                "prefixCondition": "AND",
                "arguments": [{"name": "testResultList", "input": ["Failed"]}],
            }
        ],
    },
    destinations=[
        Destination(
            category=SubscriptionCategory.External,
            type=SubscriptionType.Webhook,
            config={
                "endpoint": "https://hooks.example.com/openmetadata",
                "authType": {"type": "bearer", "secretKey": "<hmac-secret>"},
            },
        )
    ],
    enabled=True,
)

# PUT creates the alert, or updates it when the name already exists
alert = metadata.create_or_update(data=request)
print(alert.fullyQualifiedName)

Sources, Filters, and Triggers

The valid sources, filters, triggers, and arguments depend on the alert type and sources. To list them, see Available Sources, Filters, and Triggers. The table shows the names for each Observability source. For what each one does, see the Alert Behavior Reference.

Source (`resources`)	Trigger names (`input.actions`)	Filter names (`input.filters`)
`table`	`GetTableSchemaChanges`, `GetTableMetricsUpdates`	`filterByFqn`, `filterByDomain`, `filterByOwner`, `filterByUpdaterIsBot`
`topic`	`GetTopicSchemaChanges`	same as `table`
`container`	`GetContainerSchemaChanges`	same as `table`
`pipeline`	`GetPipelineStatusUpdates`	same as `table`
`ingestionPipeline`	`GetIngestionPipelineStatusUpdates`	same as `table`
`testCase`	`GetTestCaseStatusUpdates`, `GetTestCaseStatusUpdatesUnderSuite`	`filterByFqn`, `filterByDomain`, `filterByOwner`, `filterByTableNameTestCaseBelongsTo`, `filterByUpdaterIsBot`
`testSuite`	`GetTestSuiteStatusUpdates`	`filterByFqn`, `filterByDomain`, `filterByOwner`, `filterByUpdaterIsBot`
`dataContract`	`GetDataContractStatusUpdates`	`filterByFqn`, `filterByDomain`, `filterByOwner`, `filterByEntityName`, `filterByUpdaterIsBot`

Argument names and values:

filterByFqn takes fqnList, filterByDomain takes domainList, filterByOwner takes ownerNameList, and filterByTableNameTestCaseBelongsTo takes tableNameList. filterByUpdaterIsBot takes no arguments.
GetTestCaseStatusUpdates and GetTestSuiteStatusUpdates take testResultList with values Success, Failed, Aborted, or Queued.
GetTestCaseStatusUpdatesUnderSuite takes testStatusList and testSuiteList.
GetPipelineStatusUpdates takes pipelineStateList with values Successful, Failed, Pending, or Skipped.
GetIngestionPipelineStatusUpdates takes ingestionPipelineStateList with values queued, running, success, partialSuccess, failed, or stopped.

To watch several sources, list them all in resources. A filter must be supported by every source, and each trigger applies only to the sources that support it. This input gives an alert on pipelines and ingestion pipelines a trigger for each:

{
  "resources": ["pipeline", "ingestionPipeline"],
  "input": {
    "actions": [
      {
        "name": "GetPipelineStatusUpdates",
        "effect": "include",
        "arguments": [{ "name": "pipelineStateList", "input": ["Failed"] }]
      },
      {
        "name": "GetIngestionPipelineStatusUpdates",
        "effect": "include",
        "arguments": [{ "name": "ingestionPipelineStateList", "input": ["failed"] }]
      }
    ]
  }
}

Call the capabilities endpoint with your sources to confirm the exact filter, trigger, and argument names for your version before you automate alert creation.

Alert for Incident Events

To receive a webhook when an incident changes status, create a Notification alert on the task source, because incidents are tasks. The task source accepts only the filterByMentionedName filter, so omit input and filter the events in your receiver. Log a few real events from your environment to confirm the payload before you build on it.

{
  "name": "incident-events-to-webhook",
  "alertType": "Notification",
  "resources": ["task"],
  "destinations": [
    {
      "category": "External",
      "type": "Webhook",
      "config": { "endpoint": "https://hooks.example.com/incidents" }
    }
  ]
}

Notify Downstream Owners

To notify the owners of downstream assets, set notifyDownstream on an internal destination. This Observability alert fires when a table’s schema changes and sends a Slack message to each owning team of that table and of every table directly downstream of it. It uses the Slack endpoint set on each team’s profile, so the destination has no config.

{
  "name": "schema-change-to-downstream-owners",
  "alertType": "Observability",
  "resources": ["table"],
  "input": {
    "actions": [
      {
        "name": "GetTableSchemaChanges",
        "effect": "include",
        "prefixCondition": "AND"
      }
    ]
  },
  "destinations": [
    {
      "category": "Owners",
      "type": "Slack",
      "notifyDownstream": true,
      "downstreamDepth": 1
    }
  ]
}

Owners are skipped when their team or user has no endpoint for the destination type. For how recipients are resolved, see Notify Downstream.

Response

Returns the created alert. The response includes these fields:

id, fullyQualifiedName, and version
filteringRules: the filters and triggers as expanded conditions
statusDetails: an object on each destination

Test a Destination Before You Create the Alert

POST /v1/events/subscriptions/testDestination sends a test message to one or more destinations and returns the result for each. At least one destination is required.

{
  "destinations": [
    {
      "category": "External",
      "type": "Webhook",
      "config": { "endpoint": "https://hooks.example.com/openmetadata" }
    }
  ]
}

Errors

The endpoint can return these errors:

Status	Cause
`400`	No `resources` value, sources that can’t be combined, a filter that not every source supports, a trigger that no source supports, an empty required argument, or an invalid destination `config` (for example a missing `endpoint`).
`403`	The caller lacks Create permission on the `EventSubscription` entity.

API Reference

Data Assets

Collaboration

AI SDK

SDK

Discovery

Lineage

Data Contracts

Teams And Users

Data Governance

Data Quality

Alerts

Incidents

Main Concepts

Create an Alert

Create an Alert

Examples

Sources, Filters, and Triggers

Alert for Incident Events

Notify Downstream Owners

Response

Test a Destination Before You Create the Alert

Errors

​Create an Alert

​Examples

​Sources, Filters, and Triggers

​Alert for Incident Events

​Notify Downstream Owners

​Response

​Test a Destination Before You Create the Alert

​Errors

Create an Alert

Examples

Sources, Filters, and Triggers

Alert for Incident Events

Notify Downstream Owners

Response

Test a Destination Before You Create the Alert

Errors