title: "Adding a new Provider" sidebarTitle: "Adding a New Provider"
This guide explains how to create a new provider for Keep. Providers are integrations that allow Keep to interact with external services for alerting, querying data, managing incidents, or building topology maps.
Table of contents
- Provider structure
- Step-by-step implementation
- Provider attributes
- Abstract methods
- Provider types and capabilities
- 身份验证 configuration
- Testing your provider
- Best practices
- Common patterns
- Complete provider example
- Checklist
Provider structure
Each provider in Keep follows a specific structure:
keep/providers/
├── yourservice_provider/
│ ├── __init__.py
│ └── yourservice_provider.pyImportant Notes:
- Keep's ProvidersFactory automatically discovers providers based on the directory naming convention (
*_provider). - You don't need to register them explicitly - just follow the naming pattern.
- 该 Provider type is automatically extracted from the class name (for example,
ServiceNowProvider→servicenow).
Step-by-step implementation
1. Create provider directory
Create a new directory under keep/providers/ with the pattern {service}_provider:
bash
mkdir keep/providers/yourservice_provider2. Create the provider module
Create yourservice_provider.py with the following structure:
python
"""
YourService Provider is a class that 允许 integration with YourService.
"""
import dataclasses
import json
import os
from typing import 可选, List, Dict, Any
import pydantic
import requests
from keep.api.models.alert import AlertDto, AlertSeverity, AlertStatus
from keep.contextmanager.contextmanager import ContextManager
from keep.providers.base.base_provider import BaseProvider
from keep.providers.models.provider_config import ProviderConfig, ProviderScope
from keep.providers.models.provider_method import ProviderMethod
@pydantic.dataclasses.dataclass
class YourserviceProviderAuthConfig:
"""YourService authentication configuration."""
api_endpoint: str = dataclasses.field(
metadata={
"required": True,
"description": "YourService API endpoint URL",
"validation": "https_url", # 可选: validates HTTPS URLs
}
)
api_key: str = dataclasses.field(
metadata={
"required": True,
"description": "API key for YourService",
"sensitive": True, # Marks field as sensitive in UI
}
)
region: str = dataclasses.field(
default="us-east-1",
metadata={
"required": False,
"description": "YourService region",
"type": "select",
"options": ["us-east-1", "eu-west-1", "ap-south-1"],
}
)
class YourserviceProvider(BaseProvider):
"""Send alerts and fetch data from YourService."""
# 必填: Display name shown in UI
PROVIDER_DISPLAY_NAME = "YourService"
# 必填: Categories for provider classification
PROVIDER_CATEGORY = ["Monitoring"]
# 可选: Tags for searchability
PROVIDER_TAGS = ["alert", "data"]
# 可选: Define required scopes/permissions
PROVIDER_SCOPES = [
ProviderScope(
name="read:alerts",
description="Read alerts from YourService",
mandatory=True,
documentation_url="https://docs.yourservice.com/permissions",
alias="Read Alerts",
),
ProviderScope(
name="write:alerts",
description="Create and update alerts",
mandatory=False,
mandatory_for_webhook=True, # 必填 only for webhook setup
),
]
# 可选: OAuth2 URL (MUST be set as class attribute, not in __init__)
OAUTH2_URL = None # Or os.environ.get("YOURSERVICE_OAUTH2_URL")
def __init__(
self, context_manager: ContextManager, provider_id: str, config: ProviderConfig
):
super().__init__(context_manager, provider_id, config)
# Initialize any client libraries or state here
# 注意： Logger is automatically available as self.logger
# Context manager 提供 access to:
# - self.context_manager.tenant_id: Current tenant ID
# - self.context_manager.workflow_id: Current workflow ID
# - self.context_manager.workflow_execution_id: Current execution ID
# - self.context_manager.get_full_context(): Full workflow context
def validate_config(self):
"""
Validates required configuration for YourService provider.
This is an abstract method that MUST be implemented.
"""
self.authentication_config = YourserviceProviderAuthConfig(
**self.config.authentication
)
def dispose(self):
"""
Cleanup any resources when provider is disposed.
This is an abstract method that MUST be implemented, even if it just passes.
"""
pass3. Create the init.py File
Create keep/providers/yourservice_provider/__init__.py:
python
from keep.providers.yourservice_provider.yourservice_provider import (
YourserviceProvider,
YourserviceProviderAuthConfig
)
__all__ = ["YourserviceProvider", "YourserviceProviderAuthConfig"]4. Add provider documentation
Create docs/providers/documentation/yourservice-provider.mdx following the documentation template.
::: note Provider configuration fields are automatically documented through auto-generated snippets. Keep generates the snippet files in docs/snippets/providers/ from the provider's AuthConfig metadata and includes them in the documentation automatically. :::
Provider architecture
Abstract methods
Every provider must implement these two abstract methods from BaseProvider:
validate_config(self)- Validates and processes the provider configurationdispose(self)- Clean up resources when the provider is disposed of
Provider capabilities
Providers expose capabilities through standard methods:
_notify(**kwargs)- Send notifications or alerts_query(**kwargs)- Query data from the provider_get_alerts()- Fetch alerts for monitoringsetup_webhook(...)- Configure webhook endpointsvalidate_scopes()- Check provider permissionsexpose()- Return parameters calculated during execution for use in workflows
::: note The public methods notify() and query() wrap the private implementations (_notify() and _query()) with additional capabilities like enrichment and error handling. Always implement the private methods. :::
Provider discovery
Keep automatically discovers providers based on naming conventions:
- Location:
keep/providers/directory - Directory naming: Must end with
_provider(for example,slack_provider) - Main file: Must match directory name with
.pyextension (for example,slack_provider.py) - No explicit registration needed - just follow the naming convention
Implementation examples
Validate_config()
python
def validate_config(self):
"""Validate and process provider configuration."""
self.authentication_config = YourserviceProviderAuthConfig(
**self.config.authentication
)Dispose()
python
def dispose(self):
"""Cleanup any resources."""
# Close connections, cleanup clients, etc.
# Can just pass if no cleanup needed
passProvider type extraction
该 Provider type is automatically extracted from your class name:
YourserviceProvider→yourserviceServiceNowProvider→service.nowDatadogProvider→datadog
This happens via the _extract_type() method in BaseProvider.
Provider attributes
Providers should define the following class attributes:
PROVIDER_DISPLAY_NAME: String used for UI display (for example, "Slack")PROVIDER_CATEGORY: List of categories from the allowed values (see Provider Categories section)PROVIDER_COMING_SOON: Boolean flag to mark providers as not ready (default: False)WEBHOOK_INSTALLATION_REQUIRED: Boolean to make webhook setup mandatory in UI (default: False)PROVIDER_TAGS: List of tags describing provider capabilities (for example, ["alert", "messaging"])PROVIDER_SCOPES: List of ProviderScope objects defining required permissionsPROVIDER_METHODS: List of ProviderMethod objects for additional capabilities (see Provider Methods)FINGERPRINT_FIELDS: List of field names used to calculate alert fingerprintsOAUTH2_URL: OAuth 2.0 authorization URL if provider supports OAuth 2.0 authentication
Provider categories
Providers must specify one or more categories from the following list:
python
PROVIDER_CATEGORY: list[Literal[
"AI", "Monitoring", "Incident Management", "Cloud Infrastructure",
"Ticketing", "Identity", "Developer Tools", "Database",
"Identity and Access Management", "Security", "Collaboration",
"Organizational Tools", "CRM", "Queues", "Orchestration", "Others"
]]Provider tags
Valid options for PROVIDER_TAGS:
"alert"- Provider handles alerts"ticketing"- Provider manages tickets"messaging"- Provider sends messages"data"- Provider queries data"queue"- Provider manages queues"topology"- Provider 提供 topology data"incident"- Provider manages incidents
Provider scope
python
@dataclass
class ProviderScope:
"""
Provider scope model.
Args:
name (str): The name of the scope.
description (可选[str]): The description of the scope.
mandatory (bool): Whether the scope is mandatory.
mandatory_for_webhook (bool): Whether the scope is mandatory for webhook auto installation.
documentation_url (可选[str]): The documentation url of the scope.
alias (可选[str]): Another alias of the scope.
"""
name: str
description: 可选[str] = None
mandatory: bool = False
mandatory_for_webhook: bool = False
documentation_url: 可选[str] = None
alias: 可选[str] = NoneProvider config
python
@dataclass
class ProviderConfig:
"""
Provider configuration model.
Args:
description (可选[str]): The description of the provider.
authentication (dict): The configuration for the provider.
"""
authentication: 可选[dict]
name: 可选[str] = None
description: 可选[str] = None
def __post_init__(self):
if not self.authentication:
return
for key, value in self.authentication.items():
if (
isinstance(value, str)
and value.startswith("{{")
and value.endswith("}}")
):
self.authentication[key] = chevron.render(value, {"env": os.environ})Base provider
python
"""
Base class for all providers.
"""
class BaseProvider(metaclass=abc.ABCMeta):
OAUTH2_URL = None
PROVIDER_SCOPES: list[ProviderScope] = []
PROVIDER_METHODS: list[ProviderMethod] = []
FINGERPRINT_FIELDS: list[str] = []
PROVIDER_TAGS: list[
Literal["alert", "ticketing", "messaging", "data", "queue", "topology", "incident"]
] = []
PROVIDER_DISPLAY_NAME: str = None
PROVIDER_CATEGORY: list[str] = []
PROVIDER_COMING_SOON: bool = False
WEBHOOK_INSTALLATION_REQUIRED: bool = False
def __init__(
self,
context_manager: ContextManager,
provider_id: str,
config: ProviderConfig,
webhook_template: 可选[str] = None,
webhook_description: 可选[str] = None,
webhook_markdown: 可选[str] = None,
provider_description: 可选[str] = None,
):
"""
Initialize a provider.
Args:
provider_id (str): 该 Provider id.
**kwargs: Provider configuration loaded from the provider yaml file.
"""
self.provider_id = provider_id
self.config = config
self.webhook_template = webhook_template
self.webhook_description = webhook_description
self.provider_description = provider_description
self.context_manager = context_manager
self.logger = context_manager.get_logger()
self.validate_config()
self.logger.debug(
"Base provider initalized", extra={"provider": self.__class__.__name__}
)
self.provider_type = self._extract_type()
self.results = []
# tb: we can have this overriden by customer configuration, when initializing the provider
self.fingerprint_fields = self.FINGERPRINT_FIELDS
def _extract_type(self):
"""
Extract the provider type from the provider class name.
Returns:
str: 该 Provider type.
"""
name = self.__class__.__name__
name_without_provider = name.replace("Provider", "")
name_with_spaces = (
re.sub("([A-Z])", r" \1", name_without_provider).lower().strip()
)
return name_with_spaces.replace(" ", ".")
@abc.abstractmethod
def dispose(self):
"""
Dispose of the provider.
"""
raise NotImplementedError("dispose() method not implemented")
@abc.abstractmethod
def validate_config(self):
"""
Validate provider configuration.
"""
raise NotImplementedError("validate_config() method not implemented")
def validate_scopes(self) -> dict[str, bool | str]:
"""
Validate provider scopes.
Returns:
dict: where key is the scope name and value is whether the scope is valid (True boolean) or string with error message.
"""
return {}
def notify(self, **kwargs):
"""
Output alert message.
Args:
**kwargs (dict): 该 Provider context (with statement)
"""
# trigger the provider
results = self._notify(**kwargs)
self.results.append(results)
# if the alert should be enriched, enrich it
enrich_alert = kwargs.get("enrich_alert", [])
if not enrich_alert or not results:
return results if results else None
self._enrich(enrich_alert, results)
return results
def _enrich(self, enrichments, results, audit_启用d=True):
"""
Enrich alert or incident with provider specific data.
This method replaces the deprecated _enrich_alert method and supports both
alert and incident enrichment.
Args:
enrichments: List of enrichment configurations
results: Results from the provider action
audit_启用d: Whether to audit the enrichment operation (default: True)
"""
self.logger.debug("Extracting the fingerprint from the alert")
if "fingerprint" in results:
fingerprint = results["fingerprint"]
elif self.context_manager.foreach_context.get("value", {}):
# TODO: if it's zipped, we need to extract the fingerprint from the zip (i.e. multiple foreach)
fingerprint = self.context_manager.foreach_context.get("value", {}).get(
"fingerprint"
)
# else, if we are in an event context, use the event fingerprint
elif self.context_manager.event_context:
# TODO: map all cases event_context is dict and update them to the DTO
# and remove this if statement
if isinstance(self.context_manager.event_context, dict):
fingerprint = self.context_manager.event_context.get("fingerprint")
# Alert DTO
else:
fingerprint = self.context_manager.event_context.fingerprint
else:
fingerprint = None
if not fingerprint:
self.logger.error(
"No fingerprint found for alert enrichment",
extra={"provider": self.provider_id},
)
raise Exception("No fingerprint found for alert enrichment")
self.logger.debug("Fingerprint extracted", extra={"fingerprint": fingerprint})
_enrichments = {}
# enrich only the requested fields
for enrichment in enrichments:
try:
if enrichment["value"].startswith("results."):
val = enrichment["value"].replace("results.", "")
parts = val.split(".")
r = copy.copy(results)
for part in parts:
r = r[part]
_enrichments[enrichment["key"]] = r
else:
_enrichments[enrichment["key"]] = enrichment["value"]
except Exception:
self.logger.error(
f"Failed to enrich alert - enrichment: {enrichment}",
extra={"fingerprint": fingerprint, "provider": self.provider_id},
)
continue
self.logger.info("Enriching alert", extra={"fingerprint": fingerprint})
try:
enrich_alert(self.context_manager.tenant_id, fingerprint, _enrichments)
except Exception as e:
self.logger.error(
"Failed to enrich alert in db",
extra={"fingerprint": fingerprint, "provider": self.provider_id},
)
raise e
self.logger.info("Alert enriched", extra={"fingerprint": fingerprint})
def _notify(self, **kwargs):
"""
Output alert message.
Args:
**kwargs (dict): 该 Provider context (with statement)
"""
raise NotImplementedError("notify() method not implemented")
def _query(self, **kwargs: dict):
"""
Query the provider using the given query
Args:
kwargs (dict): 该 Provider context (with statement)
Raises:
NotImplementedError: _description_
"""
raise NotImplementedError("query() method not implemented")
def query(self, **kwargs: dict):
# just run the query
results = self._query(**kwargs)
# now add the type of the results to the global context
if results and isinstance(results, list):
self.context_manager.dependencies.add(results[0].__class__)
elif results:
self.context_manager.dependencies.add(results.__class__)
enrich_alert = kwargs.get("enrich_alert", [])
if enrich_alert:
self._enrich(enrich_alert, results)
# and return the results
return results
@staticmethod
def _format_alert(
event: dict | list[dict], provider_instance: "BaseProvider" = None
) -> AlertDto | list[AlertDto]:
"""
Format incoming event(s) into AlertDto object(s).
Args:
event: Single event dict or list of event dicts
provider_instance: 可选 provider instance for context
Returns:
AlertDto or list of AlertDto objects
"""
raise NotImplementedError("format_alert() method not implemented")
@classmethod
def format_alert(cls, event: dict) -> AlertDto | list[AlertDto]:
logger = logging.getLogger(__name__)
logger.debug("Formatting alert")
formatted_alert = cls._format_alert(event)
logger.debug("Alert formatted")
return formatted_alert
@staticmethod
def get_alert_fingerprint(alert: AlertDto, fingerprint_fields: list = []) -> str:
"""
Get the fingerprint of an alert.
Args:
event (AlertDto): The alert to get the fingerprint of.
fingerprint_fields (list, optional): The fields we calculate the fingerprint upon. Defaults to [].
Returns:
str: hexdigest of the fingerprint or the event.name if no fingerprint_fields were given.
"""
if not fingerprint_fields:
return alert.name
fingerprint = hashlib.sha256()
event_dict = alert.dict()
for fingerprint_field in fingerprint_fields:
fingerprint_field_value = event_dict.get(fingerprint_field, None)
if isinstance(fingerprint_field_value, (list, dict)):
fingerprint_field_value = json.dumps(fingerprint_field_value)
if fingerprint_field_value:
fingerprint.update(str(fingerprint_field_value).encode())
return fingerprint.hexdigest()
def get_alerts_configuration(self, alert_id: 可选[str] = None):
"""
Get configuration of alerts from the provider.
Args:
alert_id (可选[str], optional): If given, gets a specific alert by id. Defaults to None.
"""
# todo: we'd want to have a common alert model for all providers (also for consistent output from GPT)
raise NotImplementedError("get_alerts() method not implemented")
def deploy_alert(self, alert: dict, alert_id: 可选[str] = None):
"""
Deploy an alert to the provider.
Args:
alert (dict): The alert to deploy.
alert_id (可选[str], optional): If given, deploys a specific alert by id. Defaults to None.
"""
raise NotImplementedError("deploy_alert() method not implemented")
def _get_alerts(self) -> list[AlertDto]:
"""
Get alerts from the provider.
"""
raise NotImplementedError("get_alerts() method not implemented")
def get_alerts(self) -> list[AlertDto]:
"""
Get alerts from the provider.
"""
with tracer.start_as_current_span(f"{self.__class__.__name__}-get_alerts"):
alerts = self._get_alerts()
# enrich alerts with provider id
for alert in alerts:
alert.providerId = self.provider_id
return alerts
def get_alerts_by_fingerprint(self, tenant_id: str) -> dict[str, list[AlertDto]]:
"""
Get alerts from the provider grouped by fingerprint, sorted by lastReceived.
Returns:
dict[str, list[AlertDto]]: A dict of alerts grouped by fingerprint, sorted by lastReceived.
"""
alerts = self.get_alerts()
if not alerts:
return {}
# get alerts, group by fingerprint and sort them by lastReceived
with tracer.start_as_current_span(f"{self.__class__.__name__}-get_last_alerts"):
get_attr = operator.attrgetter("fingerprint")
grouped_alerts = {
fingerprint: list(alerts)
for fingerprint, alerts in itertools.groupby(
sorted(
alerts,
key=get_attr,
),
get_attr,
)
}
# enrich alerts
with tracer.start_as_current_span(f"{self.__class__.__name__}-enrich_alerts"):
pulled_alerts_enrichments = get_enrichments(
tenant_id=tenant_id,
fingerprints=grouped_alerts.keys(),
)
for alert_enrichment in pulled_alerts_enrichments:
if alert_enrichment:
alerts_to_enrich = grouped_alerts.get(
alert_enrichment.alert_fingerprint
)
for alert_to_enrich in alerts_to_enrich:
parse_and_enrich_deleted_and_assignees(
alert_to_enrich, alert_enrichment.enrichments
)
for enrichment in alert_enrichment.enrichments:
# set the enrichment
setattr(
alert_to_enrich,
enrichment,
alert_enrichment.enrichments[enrichment],
)
return grouped_alerts
def setup_webhook(
self, tenant_id: str, keep_api_url: str, api_key: str, setup_alerts: bool = True
) -> dict | None:
"""
设置 a webhook for the provider.
Args:
tenant_id (str): The tenant ID
keep_api_url (str): The Keep API URL for webhook callbacks
api_key (str): The API key for authentication
setup_alerts (bool, optional): Whether to setup alerts. Defaults to True.
Returns:
dict | None: Dictionary of secrets to be saved if any, None otherwise
Raises:
NotImplementedError: If not implemented by the provider
"""
raise NotImplementedError("setup_webhook() method not implemented")
@staticmethod
def get_alert_schema() -> dict:
"""
Get the alert schema description for the provider.
e.g. 如何 define an alert for the provider that can be pushed via the API.
Returns:
str: The alert format description.
"""
raise NotImplementedError(
"get_alert_format_description() method not implemented"
)
@staticmethod
def oauth2_logic(**payload) -> dict:
"""
Logic for oauth2 authentication.
For example, in Slack oauth2, we need to get the code from the payload and exchange it for a token.
return: dict: The secrets to be saved as the provider configuration. (e.g. the Slack access token)
"""
raise NotImplementedError("oauth2_logic() method not implemented")
@staticmethod
def parse_event_raw_body(raw_body: bytes | dict) -> dict:
"""
Parse the raw body of an event and create an ingestible dict from it.
For instance, in parseable, the "event" is just a string
> b'Alert: Server side error triggered on teststream1\nMessage: server reporting status as 500\nFailing Condition: status column equal to abcd, 2 times'
and we want to return an object
> {'alert': 'Server side error triggered on teststream1', 'message': 'server reporting status as 500', 'failing_condition': 'status column equal to abcd, 2 times'}
If this method is not implemented for a provider, it should convert the raw body to a dict.
Args:
raw_body (bytes | dict): The raw body of the incoming event (can be bytes or dict)
Returns:
dict: Ingestible event dictionary
"""
if isinstance(raw_body, dict):
return raw_body
return raw_body
def get_logs(self, limit: int = 5) -> list:
"""
Get logs from the provider.
Args:
limit (int): The number of logs to get.
"""
raise NotImplementedError("get_logs() method not implemented")
def expose(self):
"""Expose parameters that were calculated during query time.
Each provider can expose parameters that were calculated during query time.
E.g. parameters that were supplied by the user and were rendered by the provider.
A concrete example is the "_from" and "to" of the Datadog Provider which are calculated during execution.
"""
# TODO - implement dynamically using decorators and
return {}
def start_consume(self):
"""Get the consumer for the provider.
should be implemented by the provider if it has a consumer.
for an example, see Kafka Provider
Returns:
Consumer: The consumer for the provider.
"""
return
def status(self) -> bool:
"""Return the status of the provider.
Returns:
bool: The status of the provider.
"""
return {
"status": "should be implemented by the provider if it has a consumer",
"error": "",
}
@property
def is_consumer(self) -> bool:
"""Return consumer if the inherited class has a start_consume method.
Returns:
bool: _description_
"""
return self.start_consume.__qualname__ != "BaseProvider.start_consume"
def _push_alert(self, alert: dict):
"""
Push an alert to the provider.
Args:
alert (dict): The alert to push.
"""
# if this is not a dict, try to convert it to a dict
if not isinstance(alert, dict):
try:
alert_data = json.loads(alert)
except Exception:
alert_data = alert_data
else:
alert_data = alert
# if this is still not a dict, we can't push it
if not isinstance(alert_data, dict):
self.logger.warning(
"We currently support only alert represented as a dict, dismissing alert",
extra={"alert": alert},
)
return
# now try to build the alert model
# we will have a lot of default values here to support all providers and all cases, the
# way to fine tune those would be to use the provider specific model or enforce that the event from the queue will be casted into the fields
alert_model = AlertDto(
id=alert_data.get("id", str(uuid.uuid4())),
name=alert_data.get("name", "alert-from-event-queue"),
status=alert_data.get("status", AlertStatus.FIRING),
lastReceived=alert_data.get("lastReceived", datetime.datetime.now()),
environment=alert_data.get("environment", "alert-from-event-queue"),
isDuplicate=alert_data.get("isDuplicate", False),
duplicateReason=alert_data.get("duplicateReason", None),
service=alert_data.get("service", "alert-from-event-queue"),
source=alert_data.get("source", [self.provider_type]),
message=alert_data.get("message", "alert-from-event-queue"),
description=alert_data.get("description", "alert-from-event-queue"),
severity=alert_data.get("severity", AlertSeverity.INFO),
pushed=alert_data.get("pushed", False),
event_id=alert_data.get("event_id", str(uuid.uuid4())),
url=alert_data.get("url", None),
fingerprint=alert_data.get("fingerprint", None),
)
# push the alert to the provider
url = f'{os.environ["KEEP_API_URL"]}/alerts/event'
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"X-API-KEY": self.context_manager.api_key,
}
response = requests.post(url, json=alert_model.dict(), headers=headers)
try:
response.raise_for_status()
self.logger.info("Alert pushed successfully")
except Exception:
self.logger.error(
f"Failed to push alert to {self.provider_id}: {response.content}"
)Provider types and capabilities
Base provider types
Keep supports several base provider types, each with specific capabilities:
BaseProvider (
keep/providers/base/base_provider.py)- Basic provider capabilities
- Methods:
_notify(),_query(),_get_alerts() - Use for: General integrations
BaseTopologyProvider (
keep/providers/base/base_provider.py)- Extends BaseProvider
- Methods:
pull_topology() - Use for: Services that provide infrastructure topology data
- 示例: Datadog Provider (
keep/providers/datadog_provider/datadog_provider.py)
BaseIncidentProvider (
keep/providers/base/base_provider.py)- Extends BaseProvider
- Methods:
_get_incidents(),_format_incident()(static),format_incident()(classmethod),setup_incident_webhook() - Use for: Incident management systems
- 示例: PagerDuty Provider (
keep/providers/pagerduty_provider/pagerduty_provider.py)
Common capabilities
1. Notification (_notify)
Send alerts or messages to external services:
python
def _notify(self, title: str, description: str = "", **kwargs) -> dict:
# Implementation2. Query (_query)
Fetch data from external services:
python
def _query(self, query: str, **kwargs) -> list:
# Implementation3. Alert Fetching (_get_alerts)
Pull alerts for monitoring:
python
def _get_alerts(self) -> List[AlertDto]:
# Implementation4. Webhook support
Handle incoming webhooks:
python
@staticmethod
def parse_event_raw_body(raw_body: bytes | str) -> dict:
# Parse webhook payload
@staticmethod
def _format_alert(event: dict, provider_instance: "BaseProvider" = None) -> AlertDto | list[AlertDto]:
# Format webhook events into alerts5. OAuth 2.0 support
Handle OAuth 2.0 authentication:
python
# IMPORTANT: Define OAUTH2_URL as a class attribute at the class level, NOT in __init__
class YourserviceProvider(BaseProvider):
OAUTH2_URL = os.environ.get("YOURSERVICE_OAUTH2_URL") # Must be at class level
@staticmethod
def oauth2_logic(**payload) -> dict:
# OAuth 2.0 implementation6. Consumer providers
For providers that consume messages from queues or streams:
python
def start_consume(self):
"""
Start consuming messages from the provider.
This method is called when Keep starts the provider as a consumer.
Implement long-running consumption logic here.
"""
# 示例: Kafka consumer
while True:
message = self.consumer.poll()
if message:
self._push_alert(message)
@property
def is_consumer(self) -> bool:
"""Provider is automatically detected as consumer if start_consume is implemented."""
return True # Automatically set if start_consume is overridden
def status(self) -> dict:
"""Return the status of the consumer."""
return {
"status": "running" if self.consumer_active else "stopped",
"error": self.last_error if hasattr(self, 'last_error') else ""
}Specialized base classes
Keep 提供 specialized base classes for specific provider types:
Base topology provider
For providers that manage infrastructure topology and service dependencies:
python
from keep.providers.base.base_topology_provider import BaseTopologyProvider
class MyTopologyProvider(BaseTopologyProvider):
def pull_topology(self) -> tuple[list[TopologyServiceInDto], dict]:
"""
Pull topology data from the provider.
Returns:
tuple: A tuple of (services list, edges dict)
"""
# Implement topology fetching logic
passBaseIncidentProvider
For providers that manage incidents and incident response:
python
from keep.providers.base.base_incident_provider import BaseIncidentProvider
class MyIncidentProvider(BaseIncidentProvider):
def _get_incidents(self) -> list[IncidentDto]:
"""
Fetch incidents from the provider (abstract method).
Returns:
list[IncidentDto]: List of incidents
"""
# Implement incident fetching logic
pass
@staticmethod
def _format_incident(
event: dict, provider_instance: "BaseProvider" = None
) -> IncidentDto | list[IncidentDto]:
"""
Format raw incident data into IncidentDto objects.
Args:
event: Raw incident data from webhook or API
provider_instance: 可选 provider instance for context
Returns:
IncidentDto or list of IncidentDto objects
"""
# Implement incident formatting logic
pass
def setup_incident_webhook(
self,
tenant_id: str,
keep_api_url: str,
api_key: str,
setup_alerts: bool = True,
) -> dict | None:
"""
设置 webhook for incident updates.
Args:
tenant_id: Tenant identifier
keep_api_url: Keep API URL for callbacks
api_key: API key for authentication
setup_alerts: Whether to also setup alert webhooks
Returns:
dict | None: Secrets to save if any
"""
# Implement webhook setup logic
pass注意： The get_incidents() method is automatically provided by the base class and wraps _get_incidents(). The format_incident() class method handles provider loading and calls _format_incident().
身份验证 configuration
Providers should define an authentication configuration class as a dataclass with proper field types and validation:
python
import dataclasses
import pydantic
from keep.validation.fields import HttpsUrl, NoSchemeUrl, UrlPort
@pydantic.dataclasses.dataclass
class MyProviderAuthConfig:
"""配置 for MyProvider authentication."""
api_key: str = dataclasses.field(
metadata={
"required": True,
"description": "API Key for authentication",
"sensitive": True, # Masks the field value in UI
}
)
api_url: HttpsUrl = dataclasses.field(
default="https://api.example.com",
metadata={
"required": False,
"description": "API endpoint URL (HTTPS only)",
"documentation_url": "https://docs.example.com/api",
"validation": "https_url", # Maps to HttpsUrl validator
}
)
host: NoSchemeUrl = dataclasses.field(
metadata={
"required": True,
"description": "Service hostname",
"hint": "example.com or 192.168.1.1",
"validation": "no_scheme_url", # Maps to NoSchemeUrl validator
}
)
port: UrlPort = dataclasses.field(
default=443,
metadata={
"required": False,
"description": "Service port",
"validation": "port", # Validates port range 1-65535
}
)
workspace_id: str = dataclasses.field(
metadata={
"required": True,
"description": "Workspace identifier",
"hint": "Can be found in Settings > Workspace",
}
)
region: str = dataclasses.field(
default="us-east-1",
metadata={
"required": False,
"description": "Service region",
"type": "select", # Renders as dropdown in UI
"options": ["us-east-1", "eu-west-1", "ap-south-1"],
}
)Field validation
Keep 提供 built-in field validation through custom Pydantic field types:
| Validation Type | Field Type | Description | 示例 |
|---|---|---|---|
"https_url" | HttpsUrl | Validates HTTPS URLs only | https://api.example.com |
"any_http_url" | pydantic.AnyHttpUrl | Validates any HTTP/HTTPS URL | http://example.com |
"no_scheme_url" | NoSchemeUrl | Validates URLs without scheme | example.com:8080 |
"port" | UrlPort | Validates port numbers (1-65535) | 443 |
"multihost_url" | MultiHostUrl | Validates multi-host URLs | mongodb://host1:27017,host2:27017 |
"no_scheme_multihost_url" | NoSchemeMultiHostUrl | Multi-host URLs without scheme | host1:9092,host2:9092 |
To use validation:
- Import the appropriate field type from
keep.validation.fields - Use it as the field type annotation
- Add the corresponding validation string in metadata
示例 implementations:
python
# HTTPS-only webhook URL
webhook_url: HttpsUrl = dataclasses.field(
metadata={
"required": True,
"description": "Webhook endpoint (HTTPS required)",
"sensitive": True,
"validation": "https_url",
}
)
# Database connection with multiple hosts
connection_string: MultiHostUrl = dataclasses.field(
metadata={
"required": True,
"description": "Database connection string",
"hint": "mongodb://host1:27017,host2:27017/dbname",
"validation": "multihost_url",
}
)
# SSH connection
ssh_host: NoSchemeUrl = dataclasses.field(
metadata={
"required": True,
"description": "SSH hostname or IP",
"validation": "no_scheme_url",
}
)
ssh_port: UrlPort = dataclasses.field(
default=22,
metadata={
"required": False,
"description": "SSH port",
"validation": "port",
}
)Metadata fields reference
required: Whether the field is mandatorydescription: Field description shown in UIsensitive: Whether to mask the field value (for secrets)hidden: Whether to hide the field in UIdocumentation_url: Link to relevant documentationhint: Help text for usersvalidation: Validation type string (see preceding table)type: UI input type (for example, "select" for dropdown)options: List of valid options for select fieldsconfig_main_group: Group name for organizing fields in UIconfig_sub_group: Sub-group name for nested organization
::: note The validation system ensures that configuration values are valid before Keep instantiates the provider. Invalid values are rejected with clear error messages, improving the user experience and preventing runtime errors. :::
Testing your provider
1. Unit test
Create tests/test_yourservice_provider.py:
python
import pytest
from keep.providers.yourservice_provider.yourservice_provider import YourserviceProvider
from keep.providers.models.provider_config import ProviderConfig
from keep.contextmanager.contextmanager import ContextManager
def test_yourservice_provider_init():
"""Test provider initialization."""
config = ProviderConfig(
authentication={
"api_endpoint": "https://api.yourservice.com",
"api_key": "test-key",
}
)
context_manager = ContextManager(tenant_id="test", workflow_id="test")
provider = YourserviceProvider(
context_manager=context_manager,
provider_id="test",
config=config
)
assert provider.authentication_config.api_endpoint == "https://api.yourservice.com"
assert provider.authentication_config.api_key == "test-key"
@pytest.fixture
def mock_requests(monkeypatch):
"""Mock requests module."""
import requests
class MockResponse:
def __init__(self, json_data, status_code=200):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
def raise_for_status(self):
pass
def mock_post(*args, **kwargs):
return MockResponse({"success": True})
def mock_get(*args, **kwargs):
return MockResponse({"alerts": []})
monkeypatch.setattr(requests, "post", mock_post)
monkeypatch.setattr(requests, "get", mock_get)
def test_yourservice_notify(mock_requests):
"""Test notification sending."""
config = ProviderConfig(
authentication={
"api_endpoint": "https://api.yourservice.com",
"api_key": "test-key",
}
)
context_manager = ContextManager(tenant_id="test", workflow_id="test")
provider = YourserviceProvider(
context_manager=context_manager,
provider_id="test",
config=config
)
result = provider.notify(message="Test message")
assert result["success"] is True2. Integration test
Test with the provider factory:
python
def test_provider_factory_loading():
"""Test that provider loads correctly through factory."""
from keep.providers.providers_factory import ProvidersFactory
# Get provider class
provider_class = ProvidersFactory.get_provider_class("yourservice")
assert provider_class.__name__ == "YourserviceProvider"
# Get all providers
all_providers = ProvidersFactory.get_all_providers()
yourservice = next((p for p in all_providers if p.type == "yourservice"), None)
assert yourservice is not None
assert yourservice.display_name == "YourService"3. Manual testing
你可以 test your provider by running it directly:
bash
cd keep
python -m keep.providers.yourservice_provider.yourservice_providerThe if __name__ == "__main__": block 允许 you to test provider initialization and basic capabilities.
Add a test block to your provider for direct execution:
python
if __name__ == "__main__":
# Test the provider directly
import logging
logging.basicConfig(level=logging.DEBUG, handlers=[logging.StreamHandler()])
context_manager = ContextManager(
tenant_id="singletenant",
workflow_id="test",
)
# Initialize the provider with test config
config = ProviderConfig(
authentication={
"api_endpoint": "https://api.yourservice.com",
"api_key": "test-key",
}
)
provider = YourserviceProvider(
context_manager=context_manager,
provider_id="test",
config=config
)
# Test provider methods
print("Provider initialized successfully!")
# Test specific functionality
try:
result = provider._query("test query")
print(f"Query result: {result}")
except Exception as e:
print(f"Query failed: {e}")Best practices
1. Error handling
Always handle API errors gracefully:
python
from keep.exceptions.provider_exception import ProviderException
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
raise ProviderException(f"Failed to fetch data: {str(e)}")2. Logging
Use the provider's logger:
python
self.logger.info("Fetching alerts from YourService")
self.logger.error(f"Failed to connect: {str(e)}")3. 配置 validation
Validate configuration in validate_config():
python
def validate_config(self):
self.authentication_config = YourserviceProviderAuthConfig(
**self.config.authentication
)
# Additional validation
if not self.authentication_config.api_endpoint.startswith("https://"):
raise ValueError("API endpoint must use HTTPS")4. Alert formatting
When returning alerts, use Keep's standard format:
python
from keep.api.models.alert import AlertDto, AlertSeverity, AlertStatus
alert = AlertDto(
id="unique-alert-id",
name="Alert Title",
description="Detailed description",
severity=AlertSeverity.HIGH,
status=AlertStatus.FIRING,
lastReceived=datetime.now().isoformat(),
source=["yourservice"],
fingerprint="unique-fingerprint",
labels={"key": "value"},
annotations={"runbook": "https://docs.example.com"},
)5. Secrets management
Never hardcode secrets. Use environment variables or configuration:
python
client_id = os.environ.get("YOURSERVICE_CLIENT_ID")
if not client_id:
raise ProviderException("YOURSERVICE_CLIENT_ID environment variable not set")Common patterns
1. Provider health checks
Implement health monitoring using the ProviderHealthMixin:
python
from keep.providers.base.base_provider import BaseProvider, ProviderHealthMixin
class YourserviceProvider(BaseProvider, ProviderHealthMixin):
HAS_HEALTH_CHECK = True
# The mixin 提供 automatic health checking for:
# - Topology coverage validation
# - Spammy alerts detection
# - Alerting rule usage monitoring::: note The health check mixin is particularly useful for monitoring providers that collect topology data or handle high volumes of alerts. :::
2. Pagination
Handle paginated API responses:
python
def _get_all_items(self):
items = []
page = 1
while True:
response = self._query_page(page)
items.extend(response["items"])
if not response.get("has_next"):
break
page += 1
return items3. Rate limiting
Respect API rate limits:
python
import time
from typing import Any
def _rate_limited_request(self, url: str, **kwargs) -> Any:
max_retries = 3
for attempt in range(max_retries):
try:
response = requests.get(url, **kwargs)
if response.status_code == 429: # Rate limited
retry_after = int(response.headers.get("Retry-After", 60))
self.logger.warning(f"Rate limited, waiting {retry_after}s")
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # Exponential backoff4. Caching
Cache frequently accessed data:
python
from datetime import datetime, timedelta
class YourserviceProvider(BaseProvider):
def __init__(self, context_manager, provider_id, config):
super().__init__(context_manager, provider_id, config)
self._cache = {}
self._cache_ttl = timedelta(minutes=5)
def _get_cached_data(self, key: str) -> Any:
if key in self._cache:
data, timestamp = self._cache[key]
if datetime.now() - timestamp < self._cache_ttl:
return data
return None
def _set_cached_data(self, key: str, data: Any):
self._cache[key] = (data, datetime.now())5. Webhook signature verification
Verify webhook authenticity:
python
import hmac
import hashlib
@staticmethod
def verify_webhook_signature(raw_body: bytes, signature: str, secret: str) -> bool:
expected = hmac.new(
secret.encode(),
raw_body,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)6. Exposing runtime parameters
Use the expose() method to make runtime-calculated values available to workflows:
python
class YourserviceProvider(BaseProvider):
def __init__(self, context_manager, provider_id, config):
super().__init__(context_manager, provider_id, config)
self._from_timestamp = None
self._to_timestamp = None
def _query(self, metric: str, from_time: str = "1h", **kwargs):
# Calculate actual timestamps
self._to_timestamp = datetime.now()
self._from_timestamp = self._to_timestamp - parse_duration(from_time)
# Query with calculated timestamps
return self._fetch_metrics(metric, self._from_timestamp, self._to_timestamp)
def expose(self):
"""Expose calculated parameters for workflow use."""
exposed = {}
if self._from_timestamp:
exposed["from"] = self._from_timestamp.isoformat()
if self._to_timestamp:
exposed["to"] = self._to_timestamp.isoformat()
return exposedThis 允许 workflows to access the actual timestamps used in queries, not just the relative time strings.
Complete provider example
Here's a minimal example of a complete provider implementation:
python
from keep.providers.base.base_provider import BaseProvider
from keep.providers.models.provider_config import ProviderConfig
from keep.contextmanager.contextmanager import ContextManager
class MyProvider(BaseProvider):
PROVIDER_DISPLAY_NAME = "My Service"
PROVIDER_CATEGORY = ["Monitoring", "Incident Management"]
PROVIDER_TAGS = ["alert", "messaging"]
def __init__(
self,
context_manager: ContextManager,
provider_id: str,
config: ProviderConfig,
webhook_template: 可选[str] = None,
webhook_description: 可选[str] = None,
webhook_markdown: 可选[str] = None,
provider_description: 可选[str] = None,
):
super().__init__(
context_manager, provider_id, config,
webhook_template, webhook_description,
webhook_markdown, provider_description
)
def validate_config(self):
# Validate the provider configuration
pass
def dispose(self):
# Clean up resources
pass
def _query(self, **kwargs):
# Implement query logic
pass
def _notify(self, **kwargs):
# Implement notification logic
passFile references
- Base Provider Classes:
keep/providers/base/base_provider.py - Provider Models:
keep/providers/models/ - Provider Factory:
keep/providers/providers_factory.py - Provider Exceptions:
keep/exceptions/provider_exception.py - 示例 Providers:
- Simple:
keep/providers/slack_provider/slack_provider.py - Complex:
keep/providers/datadog_provider/datadog_provider.py - Database:
keep/providers/clickhouse_provider/clickhouse_provider.py - Incident:
keep/providers/pagerduty_provider/pagerduty_provider.py - Topology:
keep/providers/datadog_provider/datadog_provider.py
- Simple:
- Tests:
tests/test_*_provider.py - 文档:
docs/providers/documentation/ - Additional Docs:
docs/providers/adding-a-new-provider.mdxdocs/providers/provider-methods.mdxdocs/providers/linked-providers.mdx
Checklist
- [ ] Create provider directory and files
- [ ] Implement AuthConfig class with proper metadata
- [ ] Implement provider class with required methods
- [ ] Add provider to
__init__.py - [ ] Set appropriate PROVIDER_DISPLAY_NAME, PROVIDER_CATEGORY, and PROVIDER_TAGS
- [ ] Implement
validate_config()anddispose() - [ ] Add at least one capability (
_notify,_query, or_get_alerts) - [ ] Create documentation in
docs/providers/documentation/ - [ ] Write unit tests
- [ ] Test with provider factory
- [ ] Handle errors gracefully
- [ ] Add logging statements
- [ ] Validate in Keep UI
- [ ] If supporting webhooks, implement
_format_alert()static method - [ ] If supporting OAuth 2.0, set OAUTH2_URL as class attribute
- [ ] Consider implementing
validate_scopes()for scope validation - [ ] Consider implementing
get_provider_metadata()for provider versioning
Getting help
- Review existing providers for examples
- Check the base provider classes for available methods
- Look at test files for testing patterns
- Ask in Keep's GitHub discussions or issues
- Review the Provider Methods documentation for advanced capabilities
- Understand Linked vs 连接ed Providers