AlertingClient¶

Client for the Kibana Alerting API.

The Alerting API manages rules that run on schedules, detect conditions in your data, and trigger actions (through connectors) when those conditions are met. The client exposes rule operations via client.alerting.rule, backfill operations via client.alerting.backfill, and framework-level endpoints (health, rule types) directly on client.alerting.

class kibana._sync.client.alerting.AlertingClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: NamespaceClient

Client for Kibana Alerting API operations.

Exposes rule operations via rule, backfill operations via backfill, and framework-level endpoints (health(), rule_types()) directly.

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>> client.alerting.health()["alerting_framework_health"]
>>> client.alerting.rule.find(search="cpu")

Framework Endpoints

from kibana import Kibana

client = Kibana("http://localhost:5601", api_key="your_api_key")

# Alerting framework health
health = client.alerting.health()
print(health["alerting_framework_health"])

# List the rule types available in this Kibana instance
types = client.alerting.rule_types()
for rule_type in types.body:
    print(rule_type["id"], rule_type["name"])

__init__(client, default_space_id=None, validate_spaces=True)[source]¶

Initialize the AlertingClient.

Parameters:

client (Kibana) – The parent Kibana client instance.
default_space_id (str | None) – Optional default space ID for all operations.
validate_spaces (bool) – Whether to validate space existence.

property rule: RulesClient¶: Access rule operations.

property backfill: BackfillClient¶: Access backfill operations.

health(*, space_id=None, validate_spaces=None)[source]¶

Get the alerting framework health.

Calls GET /api/alerting/_health.

Parameters:

space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

is_sufficiently_secure, has_permanent_encryption_key and alerting_framework_health.

Return type:

Health details

Example

>>> health = client.alerting.health()
>>> health["alerting_framework_health"]["execution_health"]

rule_types(*, space_id=None, validate_spaces=None)[source]¶

Get the rule types available in the current space.

Calls GET /api/alerting/rule_types. The response body is a JSON array of rule type descriptors.

Parameters:

space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

List of rule type descriptors (id, name, action_groups, authorized_consumers, …).

Return type:

ObjectApiResponse[Any]

Example

>>> types = client.alerting.rule_types()
>>> [t["id"] for t in types]

perform_request(method, path, *, params=None, headers=None, body=None)¶

Perform an HTTP request via the parent client with space context enhancement.

Parameters:

method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)
path (str) – API endpoint path
params (dict[str, Any] | None) – Query parameters
headers (dict[str, str] | None) – Request headers
body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

RulesClient¶

Client for Kibana Alerting rule operations, available as client.alerting.rule.

Provides CRUD plus lifecycle operations (enable/disable, mute/unmute, snooze/unsnooze, API-key rotation) for alerting rules.

class kibana._sync.client.alerting.RulesClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: NamespaceClient

Client for Kibana Alerting rule operations.

Provides CRUD plus lifecycle operations (enable/disable, mute/unmute, snooze/unsnooze, API-key rotation) for alerting rules.

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>> rule = client.alerting.rule.create(
...     name="CPU Alert",
...     consumer="alerts",
...     rule_type_id=".index-threshold",
...     schedule={"interval": "1m"},
...     params={"index": ["logs-*"], "timeField": "@timestamp"},
... )

Creating Rules

# Create an index threshold rule
rule = client.alerting.rule.create(
    name="CPU Alert",
    consumer="alerts",
    rule_type_id=".index-threshold",
    schedule={"interval": "1m"},
    params={
        "index": ["logs-*"],
        "timeField": "@timestamp",
        "aggType": "count",
        "groupBy": "all",
        "threshold": [1000],
        "thresholdComparator": ">",
        "timeWindowSize": 5,
        "timeWindowUnit": "m",
    },
)

rule_id = rule.body["id"]

Finding and Retrieving Rules

# Find rules by search term
results = client.alerting.rule.find(search="cpu")
for rule in results.body["data"]:
    print(rule["id"], rule["name"], rule["enabled"])

# Get a single rule
rule = client.alerting.rule.get(id=rule_id)

Rule Lifecycle

# Disable and re-enable a rule
client.alerting.rule.disable(id=rule_id)
client.alerting.rule.enable(id=rule_id)

# Mute/unmute all alerts for a rule
client.alerting.rule.mute_all(id=rule_id)
client.alerting.rule.unmute_all(id=rule_id)

# Snooze notifications for one hour
snoozed = client.alerting.rule.snooze(
    id=rule_id,
    schedule={
        "custom": {
            "duration": "1h",
            "start": "2026-07-03T00:00:00.000Z",
        }
    },
)
client.alerting.rule.unsnooze(
    rule_id=rule_id,
    schedule_id=snoozed["schedule"]["id"],
)

# Delete the rule
client.alerting.rule.delete(id=rule_id)

create(*, name, consumer, rule_type_id, schedule, params, id=None, actions=None, tags=None, notify_when=None, enabled=True, throttle=None, alert_delay=None, flapping=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶

Create a new rule.

Calls POST /api/alerting/rule (or POST /api/alerting/rule/{id} when a caller-chosen id is given).

Parameters:

name (str) – Human-readable name for the rule.
consumer (str) – The application or feature that owns the rule (e.g. alerts, siem, stackAlerts).
rule_type_id (str) – The rule type ID (e.g. .es-query).
schedule (dict[str, Any]) – Check interval, e.g. {"interval": "1m"}.
params (dict[str, Any]) – Rule-type specific parameters.
id (str | None) – Optional caller-chosen rule ID. If omitted, an ID is randomly generated by Kibana.
actions (list[dict[str, Any]] | None) – List of actions to run when the rule triggers.
tags (list[str] | None) – List of tags to organize the rule.
notify_when (str | None) – When to notify (e.g. onActionGroupChange, onActiveAlert, onThrottleInterval).
enabled (bool) – Whether the rule is enabled on creation.
throttle (str | None) – Deprecated in favor of the action frequency object. How long to wait between actions (e.g. 1m).
alert_delay (dict[str, Any] | None) – Number of consecutive runs that must meet the rule conditions before an alert occurs, e.g. {"active": 3}.
flapping (dict[str, Any] | None) – Flapping detection settings, e.g. {"look_back_window": 10, "status_change_threshold": 3}.
artifacts (dict[str, Any] | None) – Rule artifacts (e.g. dashboards and investigation guide) attached to the rule.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The created rule.

Raises:

ValueError – If a required parameter is missing.
BadRequestError – If the rule configuration is invalid.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.create(
...     name="ES query rule",
...     consumer="alerts",
...     rule_type_id=".es-query",
...     schedule={"interval": "1m"},
...     params={"searchType": "esQuery", "size": 1},
...     id="my-rule-id",
... )

get(*, id, space_id=None, validate_spaces=None)[source]¶

Retrieve a rule by ID.

Calls GET /api/alerting/rule/{id}.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The rule object.

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> rule = client.alerting.rule.get(id="my-rule-id")
>>> rule["name"]

update(*, id, name, schedule, params=None, actions=None, tags=None, notify_when=None, throttle=None, alert_delay=None, flapping=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶

Update an existing rule.

Calls PUT /api/alerting/rule/{id}. Note: rule_type_id, consumer and enabled are immutable after creation; use the enable/disable API to change the enabled state.

Parameters:

id (str) – Rule ID to update.
name (str) – Human-readable name for the rule.
schedule (dict[str, Any]) – Check interval, e.g. {"interval": "1m"}.
params (dict[str, Any] | None) – Rule-type specific parameters.
actions (list[dict[str, Any]] | None) – List of actions to run when the rule triggers.
tags (list[str] | None) – List of tags to organize the rule.
notify_when (str | None) – When to notify.
throttle (str | None) – Deprecated in favor of the action frequency object. How long to wait between actions.
alert_delay (dict[str, Any] | None) – Number of consecutive runs that must meet the rule conditions before an alert occurs, e.g. {"active": 3}.
flapping (dict[str, Any] | None) – Flapping detection settings.
artifacts (dict[str, Any] | None) – Rule artifacts attached to the rule.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The updated rule.

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.update(
...     id="my-rule-id",
...     name="Renamed rule",
...     schedule={"interval": "5m"},
...     params={"searchType": "esQuery", "size": 1},
... )

delete(*, id, space_id=None, validate_spaces=None)[source]¶

Delete a rule.

Calls DELETE /api/alerting/rule/{id}.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.delete(id="my-rule-id")

find(*, search=None, page=None, per_page=None, sort_field=None, sort_order=None, fields=None, filter=None, default_search_operator=None, search_fields=None, has_reference=None, filter_consumers=None, space_id=None, validate_spaces=None)[source]¶

Find rules matching the given criteria.

Calls GET /api/alerting/rules/_find. Only parameters explicitly provided are sent; server defaults apply otherwise (page=1, per_page=10). Note that Kibana rejects sort_order when sort_field is not set.

Parameters:

search (str | None) – Search string applied to search_fields.
page (int | None) – Page number (server default 1).
per_page (int | None) – Items per page (server default 10).
sort_field (str | None) – Rule attribute to sort by (e.g. name).
sort_order (str | None) – Sort order, asc or desc (requires sort_field).
fields (list[str] | None) – Rule attributes to return (repeated query key).
filter (str | None) – KQL filter string, e.g. alert.attributes.tags:production.
default_search_operator (str | None) – Operator for the search query, OR (default) or AND.
search_fields (list[str] | None) – Rule attributes the search applies to (repeated query key).
has_reference (dict[str, str] | None) – Filter to rules referencing a saved object, e.g. {"type": "tag", "id": "tag-1"} (sent as a JSON string).
filter_consumers (list[str] | None) – List of consumers to filter by (sent as a JSON array string).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Paginated rules matching the criteria (total, data, …).

Return type:

ObjectApiResponse[Any]

Example

>>> found = client.alerting.rule.find(
...     search="cpu", sort_field="name", sort_order="asc"
... )
>>> found["total"]

enable(*, id, space_id=None, validate_spaces=None)[source]¶

Enable a rule.

Calls POST /api/alerting/rule/{id}/_enable.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.enable(id="my-rule-id")

disable(*, id, untrack=None, space_id=None, validate_spaces=None)[source]¶

Disable a rule.

Calls POST /api/alerting/rule/{id}/_disable.

Parameters:

id (str) – Rule ID.
untrack (bool | None) – Whether the rule’s alerts should be untracked when the rule is disabled.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.disable(id="my-rule-id", untrack=True)

mute_all(*, id, space_id=None, validate_spaces=None)[source]¶

Mute all alerts for a rule.

Calls POST /api/alerting/rule/{id}/_mute_all.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.mute_all(id="my-rule-id")

unmute_all(*, id, space_id=None, validate_spaces=None)[source]¶

Unmute all alerts for a rule.

Calls POST /api/alerting/rule/{id}/_unmute_all.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.unmute_all(id="my-rule-id")

update_api_key(*, id, space_id=None, validate_spaces=None)[source]¶

Update the API key for a rule.

Calls POST /api/alerting/rule/{id}/_update_api_key. The rule’s API key is regenerated using the credentials of the caller.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.
ConflictError – If the rule is being updated concurrently.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.update_api_key(id="my-rule-id")

snooze(*, id, schedule, space_id=None, validate_spaces=None)[source]¶

Schedule a snooze for a rule.

Calls POST /api/alerting/rule/{id}/snooze_schedule. Generally available; added in 8.19.0.

Parameters:

id (str) – Rule ID.
schedule (dict[str, Any]) – Snooze schedule object, e.g. {"custom": {"duration": "1h", "start": "2026-01-01T00:00:00.000Z"}}.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The created snooze schedule (includes the schedule id needed to delete it via unsnooze()).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> snoozed = client.alerting.rule.snooze(
...     id="my-rule-id",
...     schedule={"custom": {"duration": "1h",
...               "start": "2026-01-01T00:00:00.000Z"}},
... )
>>> schedule_id = snoozed["schedule"]["id"]

unsnooze(*, rule_id, schedule_id, space_id=None, validate_spaces=None)[source]¶

Delete a snooze schedule for a rule.

Calls DELETE /api/alerting/rule/{ruleId}/snooze_schedule/{scheduleId}. Generally available; added in 8.19.0.

Parameters:

rule_id (str) – Rule ID.
schedule_id (str) – Snooze schedule ID (returned by snooze()).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule or schedule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.unsnooze(
...     rule_id="my-rule-id", schedule_id="schedule-uuid"
... )

mute_alert(*, rule_id, alert_id, validate_alerts_existence=None, space_id=None, validate_spaces=None)[source]¶

Mute a specific alert of a rule.

Calls POST /api/alerting/rule/{rule_id}/alert/{alert_id}/_mute.

Parameters:

rule_id (str) – Rule ID.
alert_id (str) – Alert (instance) ID to mute.
validate_alerts_existence (bool | None) – Whether to validate that the alert instance exists before muting (live 9.4.3 default: true, returning 404 for unknown alerts; pass False to mute an alert that has not fired yet).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule or the alert instance does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.mute_alert(
...     rule_id="my-rule-id", alert_id="server-1"
... )

unmute_alert(*, rule_id, alert_id, space_id=None, validate_spaces=None)[source]¶

Unmute a specific alert of a rule.

Calls POST /api/alerting/rule/{rule_id}/alert/{alert_id}/_unmute.

Parameters:

rule_id (str) – Rule ID.
alert_id (str) – Alert (instance) ID to unmute.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.rule.unmute_alert(
...     rule_id="my-rule-id", alert_id="server-1"
... )

__init__(client, default_space_id=None, validate_spaces=True)¶

Initialize NamespaceClient with optional space support.

Parameters:

client (BaseClient) – Parent BaseClient instance to delegate requests to
default_space_id (str | None) – Optional default space ID for all operations
validate_spaces (bool) – Whether to validate space existence (default: True)

perform_request(method, path, *, params=None, headers=None, body=None)¶

Perform an HTTP request via the parent client with space context enhancement.

Parameters:

method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)
path (str) – API endpoint path
params (dict[str, Any] | None) – Query parameters
headers (dict[str, str] | None) – Request headers
body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

BackfillClient¶

Client for Kibana Alerting backfill operations, available as client.alerting.backfill.

Backfills run a rule over a historical time range. Only certain rule types support backfills (e.g. detection rules); scheduling a backfill for an unsupported rule type returns a per-item error in the response body.

class kibana._sync.client.alerting.BackfillClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: NamespaceClient

Client for Kibana Alerting backfill operations.

Backfills run a rule over a historical time range. Only certain rule types support backfills (e.g. detection rules); scheduling a backfill for an unsupported rule type returns a per-item error in the response body.

Example

>>> client.alerting.backfill.schedule(
...     backfills=[{
...         "rule_id": "my-rule-id",
...         "ranges": [{"start": "2026-01-01T00:00:00.000Z",
...                     "end": "2026-01-01T12:00:00.000Z"}],
...     }]
... )

Scheduling Backfills

# Schedule a backfill over a historical range
result = client.alerting.backfill.schedule(
    backfills=[
        {
            "rule_id": "my-rule-id",
            "ranges": [
                {
                    "start": "2026-01-01T00:00:00.000Z",
                    "end": "2026-01-01T12:00:00.000Z",
                }
            ],
        }
    ]
)

# Find and inspect backfills
backfills = client.alerting.backfill.find(rule_ids="my-rule-id")

schedule(*, backfills, space_id=None, validate_spaces=None)[source]¶

Schedule backfill runs for one or more rules.

Calls POST /api/alerting/rules/backfill/_schedule.

Parameters:

backfills (list[dict[str, Any]]) – List of 1-100 backfill request objects. Each requires rule_id and ranges (list of {"start", "end"} ISO 8601 timestamps) and accepts an optional run_actions boolean.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

A list with one result (or per-item error object) per requested backfill.

Raises:

ValueError – If backfills is missing or empty.
BadRequestError – If the request payload is invalid.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.backfill.schedule(
...     backfills=[{
...         "rule_id": "my-rule-id",
...         "ranges": [{"start": "2026-01-01T00:00:00.000Z",
...                     "end": "2026-01-01T12:00:00.000Z"}],
...         "run_actions": False,
...     }]
... )

find(*, rule_ids=None, start=None, end=None, page=None, per_page=None, sort_field=None, sort_order=None, initiator=None, space_id=None, validate_spaces=None)[source]¶

Find backfills for rules.

Calls POST /api/alerting/rules/backfill/_find. All filters are query parameters; there is no request body.

Parameters:

rule_ids (str | None) – Comma-separated list of rule IDs to filter by.
start (str | None) – Backfills with a start time after this ISO 8601 timestamp.
end (str | None) – Backfills with an end time before this ISO 8601 timestamp.
page (int | None) – Page number (server default 1).
per_page (int | None) – Items per page (server default 10).
sort_field (str | None) – Field to sort by (createdAt or start).
sort_order (str | None) – Sort order, asc or desc.
initiator (str | None) – Filter by who started the backfill (user or system).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Paginated backfills matching the criteria.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.backfill.find(rule_ids="my-rule-id")

get(*, id, space_id=None, validate_spaces=None)[source]¶

Get a backfill by ID.

Calls GET /api/alerting/rules/backfill/{id}.

Parameters:

id (str) – Backfill ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The backfill object.

Raises:

ValueError – If id is missing.
NotFoundError – If the backfill does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.backfill.get(id="backfill-id")

delete(*, id, space_id=None, validate_spaces=None)[source]¶

Delete a backfill by ID.

Calls DELETE /api/alerting/rules/backfill/{id}.

Parameters:

id (str) – Backfill ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the backfill does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> client.alerting.backfill.delete(id="backfill-id")

__init__(client, default_space_id=None, validate_spaces=True)¶

Initialize NamespaceClient with optional space support.

Parameters:

client (BaseClient) – Parent BaseClient instance to delegate requests to
default_space_id (str | None) – Optional default space ID for all operations
validate_spaces (bool) – Whether to validate space existence (default: True)

perform_request(method, path, *, params=None, headers=None, body=None)¶

Perform an HTTP request via the parent client with space context enhancement.

Parameters:

method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)
path (str) – API endpoint path
params (dict[str, Any] | None) – Query parameters
headers (dict[str, str] | None) – Request headers
body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

Async Alerting Clients¶

Asynchronous versions of the alerting clients for use with async/await syntax. They provide the same methods as their synchronous counterparts, but all methods are async and must be awaited:

from kibana import AsyncKibana
import asyncio

async def main():
    async with AsyncKibana("http://localhost:5601") as client:
        # Framework health (async)
        health = await client.alerting.health()

        # Create and enable a rule (async)
        rule = await client.alerting.rule.create(
            name="Async CPU Alert",
            consumer="alerts",
            rule_type_id=".index-threshold",
            schedule={"interval": "1m"},
            params={"index": ["logs-*"], "timeField": "@timestamp"},
        )
        await client.alerting.rule.disable(id=rule.body["id"])
        await client.alerting.rule.delete(id=rule.body["id"])

asyncio.run(main())

class kibana._async.client.alerting.AsyncAlertingClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: AsyncNamespaceClient

Async client for Kibana Alerting API operations.

Exposes rule operations via rule, backfill operations via backfill, and framework-level endpoints (health(), rule_types()) directly.

Example

>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>> (await client.alerting.health())["alerting_framework_health"]
>>> await client.alerting.rule.find(search="cpu")

__init__(client, default_space_id=None, validate_spaces=True)[source]¶

Initialize the AsyncAlertingClient.

Parameters:

client (AsyncKibana) – The parent Kibana client instance.
default_space_id (str | None) – Optional default space ID for all operations.
validate_spaces (bool) – Whether to validate space existence.

property rule: AsyncRulesClient¶: Access rule operations.

property backfill: AsyncBackfillClient¶: Access backfill operations.

async health(*, space_id=None, validate_spaces=None)[source]¶

Get the alerting framework health.

Calls GET /api/alerting/_health.

Parameters:

space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

is_sufficiently_secure, has_permanent_encryption_key and alerting_framework_health.

Return type:

Health details

Example

>>> health = await client.alerting.health()
>>> health["alerting_framework_health"]["execution_health"]

async rule_types(*, space_id=None, validate_spaces=None)[source]¶

Get the rule types available in the current space.

Calls GET /api/alerting/rule_types. The response body is a JSON array of rule type descriptors.

Parameters:

space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

List of rule type descriptors (id, name, action_groups, authorized_consumers, …).

Return type:

ObjectApiResponse[Any]

Example

>>> types = await client.alerting.rule_types()
>>> [t["id"] for t in types]

async perform_request(method, path, *, params=None, headers=None, body=None)¶

Perform an async HTTP request via the parent client with space context enhancement.

Parameters:

method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)
path (str) – API endpoint path
params (dict[str, Any] | None) – Query parameters
headers (dict[str, str] | None) – Request headers
body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

class kibana._async.client.alerting.AsyncRulesClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: AsyncNamespaceClient

Async client for Kibana Alerting rule operations.

Provides CRUD plus lifecycle operations (enable/disable, mute/unmute, snooze/unsnooze, API-key rotation) for alerting rules.

Example

>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>> rule = await client.alerting.rule.create(
...     name="CPU Alert",
...     consumer="alerts",
...     rule_type_id=".index-threshold",
...     schedule={"interval": "1m"},
...     params={"index": ["logs-*"], "timeField": "@timestamp"},
... )

async create(*, name, consumer, rule_type_id, schedule, params, id=None, actions=None, tags=None, notify_when=None, enabled=True, throttle=None, alert_delay=None, flapping=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶

Create a new rule.

Calls POST /api/alerting/rule (or POST /api/alerting/rule/{id} when a caller-chosen id is given).

Parameters:

name (str) – Human-readable name for the rule.
consumer (str) – The application or feature that owns the rule (e.g. alerts, siem, stackAlerts).
rule_type_id (str) – The rule type ID (e.g. .es-query).
schedule (dict[str, Any]) – Check interval, e.g. {"interval": "1m"}.
params (dict[str, Any]) – Rule-type specific parameters.
id (str | None) – Optional caller-chosen rule ID. If omitted, an ID is randomly generated by Kibana.
actions (list[dict[str, Any]] | None) – List of actions to run when the rule triggers.
tags (list[str] | None) – List of tags to organize the rule.
notify_when (str | None) – When to notify (e.g. onActionGroupChange, onActiveAlert, onThrottleInterval).
enabled (bool) – Whether the rule is enabled on creation.
throttle (str | None) – Deprecated in favor of the action frequency object. How long to wait between actions (e.g. 1m).
alert_delay (dict[str, Any] | None) – Number of consecutive runs that must meet the rule conditions before an alert occurs, e.g. {"active": 3}.
flapping (dict[str, Any] | None) – Flapping detection settings, e.g. {"look_back_window": 10, "status_change_threshold": 3}.
artifacts (dict[str, Any] | None) – Rule artifacts (e.g. dashboards and investigation guide) attached to the rule.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The created rule.

Raises:

ValueError – If a required parameter is missing.
BadRequestError – If the rule configuration is invalid.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.create(
...     name="ES query rule",
...     consumer="alerts",
...     rule_type_id=".es-query",
...     schedule={"interval": "1m"},
...     params={"searchType": "esQuery", "size": 1},
...     id="my-rule-id",
... )

async get(*, id, space_id=None, validate_spaces=None)[source]¶

Retrieve a rule by ID.

Calls GET /api/alerting/rule/{id}.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The rule object.

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> rule = await client.alerting.rule.get(id="my-rule-id")
>>> rule["name"]

async update(*, id, name, schedule, params=None, actions=None, tags=None, notify_when=None, throttle=None, alert_delay=None, flapping=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶

Update an existing rule.

Calls PUT /api/alerting/rule/{id}. Note: rule_type_id, consumer and enabled are immutable after creation; use the enable/disable API to change the enabled state.

Parameters:

id (str) – Rule ID to update.
name (str) – Human-readable name for the rule.
schedule (dict[str, Any]) – Check interval, e.g. {"interval": "1m"}.
params (dict[str, Any] | None) – Rule-type specific parameters.
actions (list[dict[str, Any]] | None) – List of actions to run when the rule triggers.
tags (list[str] | None) – List of tags to organize the rule.
notify_when (str | None) – When to notify.
throttle (str | None) – Deprecated in favor of the action frequency object. How long to wait between actions.
alert_delay (dict[str, Any] | None) – Number of consecutive runs that must meet the rule conditions before an alert occurs, e.g. {"active": 3}.
flapping (dict[str, Any] | None) – Flapping detection settings.
artifacts (dict[str, Any] | None) – Rule artifacts attached to the rule.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The updated rule.

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.update(
...     id="my-rule-id",
...     name="Renamed rule",
...     schedule={"interval": "5m"},
...     params={"searchType": "esQuery", "size": 1},
... )

async delete(*, id, space_id=None, validate_spaces=None)[source]¶

Delete a rule.

Calls DELETE /api/alerting/rule/{id}.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.delete(id="my-rule-id")

async find(*, search=None, page=None, per_page=None, sort_field=None, sort_order=None, fields=None, filter=None, default_search_operator=None, search_fields=None, has_reference=None, filter_consumers=None, space_id=None, validate_spaces=None)[source]¶

Find rules matching the given criteria.

Calls GET /api/alerting/rules/_find. Only parameters explicitly provided are sent; server defaults apply otherwise (page=1, per_page=10). Note that Kibana rejects sort_order when sort_field is not set.

Parameters:

search (str | None) – Search string applied to search_fields.
page (int | None) – Page number (server default 1).
per_page (int | None) – Items per page (server default 10).
sort_field (str | None) – Rule attribute to sort by (e.g. name).
sort_order (str | None) – Sort order, asc or desc (requires sort_field).
fields (list[str] | None) – Rule attributes to return (repeated query key).
filter (str | None) – KQL filter string, e.g. alert.attributes.tags:production.
default_search_operator (str | None) – Operator for the search query, OR (default) or AND.
search_fields (list[str] | None) – Rule attributes the search applies to (repeated query key).
has_reference (dict[str, str] | None) – Filter to rules referencing a saved object, e.g. {"type": "tag", "id": "tag-1"} (sent as a JSON string).
filter_consumers (list[str] | None) – List of consumers to filter by (sent as a JSON array string).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Paginated rules matching the criteria (total, data, …).

Return type:

ObjectApiResponse[Any]

Example

>>> found = await client.alerting.rule.find(
...     search="cpu", sort_field="name", sort_order="asc"
... )
>>> found["total"]

async enable(*, id, space_id=None, validate_spaces=None)[source]¶

Enable a rule.

Calls POST /api/alerting/rule/{id}/_enable.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.enable(id="my-rule-id")

async disable(*, id, untrack=None, space_id=None, validate_spaces=None)[source]¶

Disable a rule.

Calls POST /api/alerting/rule/{id}/_disable.

Parameters:

id (str) – Rule ID.
untrack (bool | None) – Whether the rule’s alerts should be untracked when the rule is disabled.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.disable(id="my-rule-id", untrack=True)

async mute_all(*, id, space_id=None, validate_spaces=None)[source]¶

Mute all alerts for a rule.

Calls POST /api/alerting/rule/{id}/_mute_all.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.mute_all(id="my-rule-id")

async unmute_all(*, id, space_id=None, validate_spaces=None)[source]¶

Unmute all alerts for a rule.

Calls POST /api/alerting/rule/{id}/_unmute_all.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.unmute_all(id="my-rule-id")

async update_api_key(*, id, space_id=None, validate_spaces=None)[source]¶

Update the API key for a rule.

Calls POST /api/alerting/rule/{id}/_update_api_key. The rule’s API key is regenerated using the credentials of the caller.

Parameters:

id (str) – Rule ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the rule does not exist.
ConflictError – If the rule is being updated concurrently.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.update_api_key(id="my-rule-id")

async snooze(*, id, schedule, space_id=None, validate_spaces=None)[source]¶

Schedule a snooze for a rule.

Calls POST /api/alerting/rule/{id}/snooze_schedule. Generally available; added in 8.19.0.

Parameters:

id (str) – Rule ID.
schedule (dict[str, Any]) – Snooze schedule object, e.g. {"custom": {"duration": "1h", "start": "2026-01-01T00:00:00.000Z"}}.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The created snooze schedule (includes the schedule id needed to delete it via unsnooze()).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> snoozed = await client.alerting.rule.snooze(
...     id="my-rule-id",
...     schedule={"custom": {"duration": "1h",
...               "start": "2026-01-01T00:00:00.000Z"}},
... )
>>> schedule_id = snoozed["schedule"]["id"]

async unsnooze(*, rule_id, schedule_id, space_id=None, validate_spaces=None)[source]¶

Delete a snooze schedule for a rule.

Calls DELETE /api/alerting/rule/{ruleId}/snooze_schedule/{scheduleId}. Generally available; added in 8.19.0.

Parameters:

rule_id (str) – Rule ID.
schedule_id (str) – Snooze schedule ID (returned by snooze()).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule or schedule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.unsnooze(
...     rule_id="my-rule-id", schedule_id="schedule-uuid"
... )

async mute_alert(*, rule_id, alert_id, validate_alerts_existence=None, space_id=None, validate_spaces=None)[source]¶

Mute a specific alert of a rule.

Calls POST /api/alerting/rule/{rule_id}/alert/{alert_id}/_mute.

Parameters:

rule_id (str) – Rule ID.
alert_id (str) – Alert (instance) ID to mute.
validate_alerts_existence (bool | None) – Whether to validate that the alert instance exists before muting (live 9.4.3 default: true, returning 404 for unknown alerts; pass False to mute an alert that has not fired yet).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule or the alert instance does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.mute_alert(
...     rule_id="my-rule-id", alert_id="server-1"
... )

async unmute_alert(*, rule_id, alert_id, space_id=None, validate_spaces=None)[source]¶

Unmute a specific alert of a rule.

Calls POST /api/alerting/rule/{rule_id}/alert/{alert_id}/_unmute.

Parameters:

rule_id (str) – Rule ID.
alert_id (str) – Alert (instance) ID to unmute.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If a required parameter is missing.
NotFoundError – If the rule does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.rule.unmute_alert(
...     rule_id="my-rule-id", alert_id="server-1"
... )

__init__(client, default_space_id=None, validate_spaces=True)¶

Initialize AsyncNamespaceClient with optional space support.

Parameters:

client (AsyncBaseClient) – Parent AsyncBaseClient instance to delegate requests to
default_space_id (str | None) – Optional default space ID for all operations
validate_spaces (bool) – Whether to validate space existence (default: True)

async perform_request(method, path, *, params=None, headers=None, body=None)¶

Perform an async HTTP request via the parent client with space context enhancement.

Parameters:

method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)
path (str) – API endpoint path
params (dict[str, Any] | None) – Query parameters
headers (dict[str, str] | None) – Request headers
body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

class kibana._async.client.alerting.AsyncBackfillClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: AsyncNamespaceClient

Async client for Kibana Alerting backfill operations.

Backfills run a rule over a historical time range. Only certain rule types support backfills (e.g. detection rules); scheduling a backfill for an unsupported rule type returns a per-item error in the response body.

Example

>>> await client.alerting.backfill.schedule(
...     backfills=[{
...         "rule_id": "my-rule-id",
...         "ranges": [{"start": "2026-01-01T00:00:00.000Z",
...                     "end": "2026-01-01T12:00:00.000Z"}],
...     }]
... )

async schedule(*, backfills, space_id=None, validate_spaces=None)[source]¶

Schedule backfill runs for one or more rules.

Calls POST /api/alerting/rules/backfill/_schedule.

Parameters:

backfills (list[dict[str, Any]]) – List of 1-100 backfill request objects. Each requires rule_id and ranges (list of {"start", "end"} ISO 8601 timestamps) and accepts an optional run_actions boolean.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

A list with one result (or per-item error object) per requested backfill.

Raises:

ValueError – If backfills is missing or empty.
BadRequestError – If the request payload is invalid.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.backfill.schedule(
...     backfills=[{
...         "rule_id": "my-rule-id",
...         "ranges": [{"start": "2026-01-01T00:00:00.000Z",
...                     "end": "2026-01-01T12:00:00.000Z"}],
...         "run_actions": False,
...     }]
... )

async find(*, rule_ids=None, start=None, end=None, page=None, per_page=None, sort_field=None, sort_order=None, initiator=None, space_id=None, validate_spaces=None)[source]¶

Find backfills for rules.

Calls POST /api/alerting/rules/backfill/_find. All filters are query parameters; there is no request body.

Parameters:

rule_ids (str | None) – Comma-separated list of rule IDs to filter by.
start (str | None) – Backfills with a start time after this ISO 8601 timestamp.
end (str | None) – Backfills with an end time before this ISO 8601 timestamp.
page (int | None) – Page number (server default 1).
per_page (int | None) – Items per page (server default 10).
sort_field (str | None) – Field to sort by (createdAt or start).
sort_order (str | None) – Sort order, asc or desc.
initiator (str | None) – Filter by who started the backfill (user or system).
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Paginated backfills matching the criteria.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.backfill.find(rule_ids="my-rule-id")

async get(*, id, space_id=None, validate_spaces=None)[source]¶

Get a backfill by ID.

Calls GET /api/alerting/rules/backfill/{id}.

Parameters:

id (str) – Backfill ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

The backfill object.

Raises:

ValueError – If id is missing.
NotFoundError – If the backfill does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.backfill.get(id="backfill-id")

async delete(*, id, space_id=None, validate_spaces=None)[source]¶

Delete a backfill by ID.

Calls DELETE /api/alerting/rules/backfill/{id}.

Parameters:

id (str) – Backfill ID.
space_id (str | None) – Optional space ID.
validate_spaces (bool | None) – Override space validation for this call.

Returns:

Empty response on success (HTTP 204).

Raises:

ValueError – If id is missing.
NotFoundError – If the backfill does not exist.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.alerting.backfill.delete(id="backfill-id")

__init__(client, default_space_id=None, validate_spaces=True)¶

Initialize AsyncNamespaceClient with optional space support.

Parameters:

client (AsyncBaseClient) – Parent AsyncBaseClient instance to delegate requests to
default_space_id (str | None) – Optional default space ID for all operations
validate_spaces (bool) – Whether to validate space existence (default: True)

async perform_request(method, path, *, params=None, headers=None, body=None)¶

Perform an async HTTP request via the parent client with space context enhancement.

Parameters:

method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)
path (str) – API endpoint path
params (dict[str, Any] | None) – Query parameters
headers (dict[str, str] | None) – Request headers
body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]