AttackDiscoveryClient

Client for the Kibana Security Attack Discovery API.

Attack discovery analyzes security alerts with AI (through a Kibana Gen AI connector) and surfaces potential attack chains as “Attack discoveries”. The client covers ad-hoc generation, discovery search and bulk updates, generation-run tracking, and Attack Discovery schedules (Kibana 9.4).

Attack Discovery resources are space-scoped: every method accepts an optional space_id to target a specific space. The securitySolutionAttackDiscovery feature must be enabled in the target space – spaces using the pure Elasticsearch solution view disable it (there, find_schedules reports total: 0 even though schedules exist). Prefer a space created with solution="security".

class kibana._sync.client.attack_discovery.AttackDiscoveryClient(client, default_space_id=None, validate_spaces=True)[source]

Bases: NamespaceClient

Client for the Kibana Security Attack Discovery API.

Attack discovery uses AI (via a Kibana Gen AI connector) to analyze security alerts and surface potential attack chains as “Attack discoveries”. This client covers the full 9.4 public surface:

  • ad-hoc discovery generation (generate), retrieval (find) and bulk workflow-status/visibility updates (bulk_update),

  • generation runs metadata (get_generations, get_generation, dismiss_generation),

  • Attack Discovery schedules CRUD plus enable/disable and search (create_schedule, get_schedule, update_schedule, delete_schedule, enable_schedule, disable_schedule, find_schedules).

All Attack Discovery resources are space-scoped: every method accepts an optional space_id to target a specific space (None targets the default space or the space the client is scoped to).

Note

The Attack Discovery Kibana feature must be enabled in the target space. Spaces using the pure Elasticsearch solution view (for example, a default space created with solution: "es") disable the securitySolutionAttackDiscovery feature: in such spaces find_schedules returns total: 0 even though schedule create/get/update/delete/enable/disable still work. Use a space with the security solution view (or the classic view) for full functionality.

Note

For OpenAI-compatible connectors (.gen-ai), the connector’s apiUrl must be the full chat-completions endpoint (for example http://localhost:1234/v1/chat/completions), not just the API base URL.

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>>
>>> # Kick off an ad-hoc generation with a Gen AI connector
>>> resp = client.attack_discovery.generate(
...     alerts_index_pattern=".alerts-security.alerts-default",
...     anonymization_fields=[
...         {"id": "f1", "field": "_id", "allowed": True, "anonymized": False},
...         {"id": "f2", "field": "host.name", "allowed": True, "anonymized": True},
...     ],
...     api_config={"actionTypeId": ".gen-ai", "connectorId": "my-connector"},
...     size=25,
... )
>>> execution_uuid = resp.body["execution_uuid"]
>>>
>>> # Poll the generation until it completes
>>> gen = client.attack_discovery.get_generation(execution_uuid=execution_uuid)
>>> print(gen.body["generation"]["status"])

Generating Attack Discoveries

Generation needs a Gen AI connector. For OpenAI-compatible backends the connector’s apiUrl must be the full /chat/completions endpoint. The _id field must be allowed in the anonymization fields:

import time

from kibana import Kibana

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

connector = client.connectors.create(
    name="My GenAI connector",
    connector_type_id=".gen-ai",
    config={
        "apiProvider": "OpenAI",
        "apiUrl": "http://localhost:1234/v1/chat/completions",
        "defaultModel": "qwen/qwen3.5-9b",
    },
    secrets={"apiKey": "dummy-key"},
)

started = client.attack_discovery.generate(
    alerts_index_pattern=".alerts-security.alerts-default",
    anonymization_fields=[
        {"id": "f0", "field": "_id", "allowed": True, "anonymized": False},
        {"id": "f1", "field": "host.name", "allowed": True, "anonymized": True},
        {"id": "f2", "field": "user.name", "allowed": True, "anonymized": True},
    ],
    api_config={
        "actionTypeId": ".gen-ai",
        "connectorId": connector.body["id"],
    },
    size=100,
    start="now-24h",
    end="now",
)
execution_uuid = started.body["execution_uuid"]

# Poll the generation until it completes (an LLM run can take minutes)
while True:
    generation = client.attack_discovery.get_generation(
        execution_uuid=execution_uuid
    ).body["generation"]
    if generation["status"] in ("succeeded", "failed", "canceled"):
        break
    time.sleep(10)

Searching and Updating Discoveries

# Search discoveries from the last day
found = client.attack_discovery.find(
    start="now-24h", end="now", status=["open"], per_page=25
)
for discovery in found.body["data"]:
    print(discovery["id"], discovery.get("title"))

# Acknowledge them in bulk
ids = [d["id"] for d in found.body["data"]]
if ids:
    client.attack_discovery.bulk_update(
        ids=ids, kibana_alert_workflow_status="acknowledged"
    )

# List generation runs; dismiss one
generations = client.attack_discovery.get_generations(size=20)
client.attack_discovery.dismiss_generation(execution_uuid=execution_uuid)

Managing Schedules

Schedules run attack discovery generation periodically. Their params carry the alerts index pattern, the LLM connector configuration and the maximum number of alerts:

created = client.attack_discovery.create_schedule(
    name="Daily attack discovery",
    params={
        "alerts_index_pattern": ".alerts-security.alerts-default",
        "api_config": {
            "connectorId": connector.body["id"],
            "actionTypeId": ".gen-ai",
            "name": "My GenAI connector",
        },
        "size": 100,
    },
    schedule={"interval": "24h"},
)
schedule_id = created.body["id"]

client.attack_discovery.enable_schedule(id=schedule_id)
client.attack_discovery.disable_schedule(id=schedule_id)

# Full update (all properties are required)
client.attack_discovery.update_schedule(
    id=schedule_id,
    name="Daily attack discovery (100 alerts)",
    params=created.body["params"],
    schedule={"interval": "12h"},
    actions=[],
)

found = client.attack_discovery.find_schedules(per_page=100)
client.attack_discovery.delete_schedule(id=schedule_id)
__init__(client, default_space_id=None, validate_spaces=True)[source]

Initialize the AttackDiscoveryClient.

Parameters:

  • client (Kibana) – The parent Kibana client 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 before space-scoped operations (default: True).

Example

>>> attack_discovery_client = AttackDiscoveryClient(kibana_client)
find(*, alert_ids=None, connector_names=None, enable_field_rendering=None, end=None, ids=None, include_unique_alert_ids=None, page=None, per_page=None, search=None, scheduled=None, shared=None, sort_field=None, sort_order=None, start=None, status=None, with_replacements=None, space_id=None, validate_spaces=None)[source]

Find Attack discoveries that match the search criteria.

Supports free text search, filtering, pagination, and sorting (GET /api/attack_discovery/_find).

Parameters:

  • alert_ids (list[str] | None) – Filter results to Attack discoveries that include any of the provided alert IDs.

  • connector_names (list[str] | None) – Filter results to Attack discoveries created by any of the provided human readable connector names (the connector_name property, e.g. "GPT-5 Chat"), which are distinct from connector_id values.

  • enable_field_rendering (bool | None) – Enables a markdown syntax used to render pivot fields, for example {{ user.name james }}. Defaults to False on the server.

  • end (str | None) – End of the time range for the search. Accepts absolute timestamps (ISO 8601) or relative date math (e.g. "now").

  • ids (list[str] | None) – Filter results to the Attack discoveries with the specified IDs.

  • include_unique_alert_ids (bool | None) – If True, the response includes unique_alert_ids and unique_alert_ids_count aggregated across the matched Attack discoveries.

  • page (int | None) – Page number to return. Defaults to 1 on the server.

  • per_page (int | None) – Number of Attack discoveries per page. Defaults to 10 on the server.

  • search (str | None) – Free-text search query applied to relevant text fields (title, description, tags, etc.).

  • scheduled (bool | None) – Use True to return only scheduled discoveries, False for only ad-hoc discoveries; omit for both.

  • shared (bool | None) – Use True to return only shared discoveries, False for only those visible to the current user; omit for both.

  • sort_field (str | None) – Field used to sort results. In 9.4 only "@timestamp" is allowed.

  • sort_order (str | None) – Sort direction, "asc" or "desc" (server default "desc").

  • start (str | None) – Start of the time range for the search (ISO 8601 or date math such as "now-24h").

  • status (list[str] | None) – Filter by alert workflow status; one or more of "open", "acknowledged", "closed".

  • with_replacements (bool | None) – When True, returns discoveries with text replacements applied to the detailsMarkdown, entitySummaryMarkdown, summaryMarkdown, and title fields. Defaults to True on the server.

  • space_id (str | None) – Optional space ID to search in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse containing connector_names, data (list of Attack discovery alerts), page, per_page, total and unique_alert_ids_count.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> found = client.attack_discovery.find(
...     start="now-24h", end="now", status=["open"], per_page=10
... )
>>> print(found.body["total"])
bulk_update(*, ids, enable_field_rendering=None, kibana_alert_workflow_status=None, visibility=None, with_replacements=None, space_id=None, validate_spaces=None)[source]

Bulk update Attack discoveries.

Performs bulk updates on multiple Attack discoveries, including workflow status changes and visibility settings, without requiring individual API calls for each alert (POST /api/attack_discovery/_bulk).

Parameters:

  • ids (list[str]) – Array of Attack Discovery IDs to update.

  • enable_field_rendering (bool | None) – Enables a markdown syntax used to render pivot fields, for example {{ user.name james }}. Defaults to False on the server.

  • kibana_alert_workflow_status (str | None) – When provided, updates the kibana.alert.workflow_status of the attack discovery alerts; one of "open", "acknowledged", "closed".

  • visibility (str | None) – When provided, updates the visibility of the alert; one of "not_shared", "shared".

  • with_replacements (bool | None) – When True, returns the updated discoveries with text replacements applied. Defaults to True on the server.

  • space_id (str | None) – Optional space ID to update discoveries in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

the array of updated Attack Discovery alert objects. IDs that match no existing discovery are silently ignored (the live server returns {"data": []} rather than an error).

Return type:

ObjectApiResponse with data

Raises:

Example

>>> updated = client.attack_discovery.bulk_update(
...     ids=["c0c8a8bb..."],
...     kibana_alert_workflow_status="acknowledged",
... )
>>> print(len(updated.body["data"]))
generate(*, alerts_index_pattern, anonymization_fields, api_config, size, sub_action='invokeAI', connector_name=None, end=None, filter=None, model=None, replacements=None, start=None, space_id=None, validate_spaces=None)[source]

Generate attack discoveries from alerts.

Initiates the generation of attack discoveries by analyzing security alerts using AI. Returns an execution UUID that can be used to track the generation progress and retrieve results (POST /api/attack_discovery/_generate). Results may also be retrieved via find().

Note

The _id field must be present (and allowed) in anonymization_fields; otherwise the generation fails with “The _id field must be allowed to generate Attack discoveries.” The alerts index is sorted on kibana.alert.risk_score, so a custom index pattern must have a mapping for that field.

Parameters:

  • alerts_index_pattern (str) – The (space specific) index pattern that contains the alerts to use as context, e.g. ".alerts-security.alerts-default".

  • anonymization_fields (list[dict[str, Any]]) – The list of fields, and whether or not they are anonymized, allowed to be sent to LLMs. Each entry has id, field, allowed and anonymized keys. Consider using the output of the /api/security_ai_assistant/anonymization_fields/_find API.

  • api_config (dict[str, Any]) – LLM API configuration with required connectorId and actionTypeId keys (optionally model, provider, defaultSystemPromptId).

  • size (int) – The maximum number of alerts to analyze.

  • sub_action (str) – LLM invocation mode, "invokeAI" (default) or "invokeStream".

  • connector_name (str | None) – Optional human readable connector name recorded on the generation.

  • end (str | None) – End of the alert time range (ISO 8601 or date math, e.g. "now").

  • filter (dict[str, Any] | None) – An Elasticsearch query DSL object used to filter alerts.

  • model (str | None) – Optional model override.

  • replacements (dict[str, str] | None) – Optional anonymization replacements mapping (anonymized value -> original value).

  • start (str | None) – Start of the alert time range (e.g. "now-24h").

  • space_id (str | None) – Optional space ID to generate discoveries in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

the identifier for the attack discovery generation process, usable with get_generation() and dismiss_generation().

Return type:

ObjectApiResponse with execution_uuid

Raises:

Example

>>> resp = client.attack_discovery.generate(
...     alerts_index_pattern=".alerts-security.alerts-default",
...     anonymization_fields=[
...         {"id": "f1", "field": "_id", "allowed": True,
...          "anonymized": False},
...     ],
...     api_config={
...         "actionTypeId": ".gen-ai",
...         "connectorId": "my-connector-id",
...     },
...     size=25,
...     start="now-24h",
...     end="now",
... )
>>> print(resp.body["execution_uuid"])
get_generations(*, end=None, size=None, start=None, space_id=None, validate_spaces=None)[source]

Get the latest Attack Discovery generations metadata.

Retrieves generation metadata for the current user, including execution status and statistics (GET /api/attack_discovery/generations).

Parameters:

  • end (str | None) – End of the time range for filtering generations (ISO 8601 or date math, e.g. "now").

  • size (int | None) – The maximum number of generations to retrieve (server default 50).

  • start (str | None) – Start of the time range for filtering generations (e.g. "now-24h").

  • space_id (str | None) – Optional space ID to list generations from.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

a list of generation metadata objects (execution_uuid, status, connector_id, discoveries, timing fields, …). Note: despite the API description, on a live 9.4.3 server dismissed generations are still listed, with status: "dismissed".

Return type:

ObjectApiResponse with generations

Raises:

Example

>>> gens = client.attack_discovery.get_generations(size=10)
>>> for g in gens.body["generations"]:
...     print(g["execution_uuid"], g["status"])
get_generation(*, execution_uuid, enable_field_rendering=None, with_replacements=None, space_id=None, validate_spaces=None)[source]

Get a single Attack Discovery generation.

Returns a specific generation, including all generated Attack discoveries and associated metadata such as execution status and statistics (GET /api/attack_discovery/generations/{execution_uuid}).

Parameters:

  • execution_uuid (str) – The unique identifier for the generation execution, as returned by generate().

  • enable_field_rendering (bool | None) – Enables a markdown syntax used to render pivot fields, for example {{ user.name james }}. Defaults to False on the server.

  • with_replacements (bool | None) – When True, returns discoveries with text replacements applied. Defaults to True on the server.

  • space_id (str | None) – Optional space ID to get the generation from.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with generation (metadata including status: one of started, succeeded, failed, dismissed, canceled) and data (the generated Attack discoveries).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> gen = client.attack_discovery.get_generation(
...     execution_uuid="edd26039-0990-4d9f-9829-2a1fcacb77b5"
... )
>>> print(gen.body["generation"]["status"])
dismiss_generation(*, execution_uuid, space_id=None, validate_spaces=None)[source]

Dismiss an Attack Discovery generation.

Marks a generation as dismissed for the current user (POST /api/attack_discovery/generations/{execution_uuid}/_dismiss).

Parameters:

  • execution_uuid (str) – The unique identifier for the generation execution, as returned by generate().

  • space_id (str | None) – Optional space ID the generation belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the (dismissed) generation metadata object. The updated dismissed status becomes visible in get_generations() shortly after (event-log refresh).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.attack_discovery.dismiss_generation(
...     execution_uuid="edd26039-0990-4d9f-9829-2a1fcacb77b5"
... )
create_schedule(*, name, params, schedule, actions=None, enabled=None, space_id=None, validate_spaces=None)[source]

Create an Attack Discovery schedule.

Creates a schedule that periodically generates attack discoveries (POST /api/attack_discovery/schedules). A schedule is backed by an alerting rule of type attack-discovery.

Parameters:

  • name (str) – The name of the schedule.

  • params (dict[str, Any]) – The schedule configuration parameters. Required keys: alerts_index_pattern (index pattern to get alerts from), api_config (LLM configuration with connectorId, actionTypeId and the connector name) and size (max number of alerts). Optional: query, filters, combined_filter, start, end.

  • schedule (dict[str, Any]) – The schedule interval, e.g. {"interval": "24h"}.

  • actions (list[dict[str, Any]] | None) – Optional schedule actions (connector notifications).

  • enabled (bool | None) – Whether the schedule is enabled (server default False).

  • space_id (str | None) – Optional space ID to create the schedule in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the created schedule (id, name, created_by, updated_by, created_at, updated_at, enabled, params, schedule, actions).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> created = client.attack_discovery.create_schedule(
...     name="Daily attack discovery",
...     params={
...         "alerts_index_pattern": ".alerts-security.alerts-default",
...         "api_config": {
...             "connectorId": "my-connector-id",
...             "actionTypeId": ".gen-ai",
...             "name": "My GenAI connector",
...         },
...         "size": 100,
...     },
...     schedule={"interval": "24h"},
... )
>>> schedule_id = created.body["id"]
get_schedule(*, id, space_id=None, validate_spaces=None)[source]

Get an Attack Discovery schedule by ID.

Retrieves a single schedule (GET /api/attack_discovery/schedules/{id}).

Parameters:

  • id (str) – The identifier of the schedule.

  • space_id (str | None) – Optional space ID to get the schedule from.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the schedule details (id, name, enabled, params, schedule, actions, audit fields).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> schedule = client.attack_discovery.get_schedule(id="b0ab787f-...")
>>> print(schedule.body["name"])
update_schedule(*, id, name, params, schedule, actions, space_id=None, validate_spaces=None)[source]

Update an Attack Discovery schedule.

Replaces the schedule’s properties (PUT /api/attack_discovery/schedules/{id}). All properties are required by the 9.4 API (this is a full update, not a patch).

Parameters:

  • id (str) – The identifier of the schedule to update.

  • name (str) – The name of the schedule.

  • params (dict[str, Any]) – The schedule configuration parameters (see create_schedule() for the required keys).

  • schedule (dict[str, Any]) – The schedule interval, e.g. {"interval": "12h"}.

  • actions (list[dict[str, Any]]) – The schedule actions. Pass [] to keep no actions.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the updated schedule details.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> updated = client.attack_discovery.update_schedule(
...     id="b0ab787f-...",
...     name="Renamed schedule",
...     params={
...         "alerts_index_pattern": ".alerts-security.alerts-default",
...         "api_config": {
...             "connectorId": "my-connector-id",
...             "actionTypeId": ".gen-ai",
...             "name": "My GenAI connector",
...         },
...         "size": 50,
...     },
...     schedule={"interval": "12h"},
...     actions=[],
... )
delete_schedule(*, id, space_id=None, validate_spaces=None)[source]

Delete an Attack Discovery schedule.

Deletes the schedule and its backing alerting rule (DELETE /api/attack_discovery/schedules/{id}).

Parameters:

  • id (str) – The identifier of the schedule to delete.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the deleted schedule’s id.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.attack_discovery.delete_schedule(id="b0ab787f-...")
enable_schedule(*, id, space_id=None, validate_spaces=None)[source]

Enable an Attack Discovery schedule.

Starts periodic attack discovery generation for the schedule (POST /api/attack_discovery/schedules/{id}/_enable).

Parameters:

  • id (str) – The identifier of the schedule to enable.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the schedule’s id.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.attack_discovery.enable_schedule(id="b0ab787f-...")
disable_schedule(*, id, space_id=None, validate_spaces=None)[source]

Disable an Attack Discovery schedule.

Stops periodic attack discovery generation for the schedule (POST /api/attack_discovery/schedules/{id}/_disable).

Parameters:

  • id (str) – The identifier of the schedule to disable.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the schedule’s id.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.attack_discovery.disable_schedule(id="b0ab787f-...")
find_schedules(*, page=None, per_page=None, sort_direction=None, sort_field=None, space_id=None, validate_spaces=None)[source]

Find Attack Discovery schedules.

Lists schedules with pagination and sorting (GET /api/attack_discovery/schedules/_find).

Note

On a live 9.4.3 server the page parameter is off by one: the server treats an explicit page=N as page N + 1 (page=1 returns the second page). Omit page (or pass page=0) to get the first page. Also, in spaces where the securitySolutionAttackDiscovery feature is disabled (e.g. spaces with the Elasticsearch solution view), this endpoint returns total: 0 even though schedules exist.

Parameters:

  • page (int | None) – Page number (see the note above about the live off-by-one behavior).

  • per_page (int | None) – Number of schedules per page (server default 10).

  • sort_direction (str | None) – Sort direction, "asc" or "desc" (server default "asc").

  • sort_field (str | None) – Field to sort by. Common fields: "name", "created_at", "updated_at", "enabled".

  • space_id (str | None) – Optional space ID to search in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with page, per_page, total and data (the list of schedules).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> found = client.attack_discovery.find_schedules(per_page=100)
>>> for s in found.body["data"]:
...     print(s["id"], s["name"], s["enabled"])
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]

AsyncAttackDiscoveryClient

Asynchronous version of the AttackDiscoveryClient for use with async/await syntax.

class kibana._async.client.attack_discovery.AsyncAttackDiscoveryClient(client, default_space_id=None, validate_spaces=True)[source]

Bases: AsyncNamespaceClient

Async client for the Kibana Security Attack Discovery API.

Attack discovery uses AI (via a Kibana Gen AI connector) to analyze security alerts and surface potential attack chains as “Attack discoveries”. This client covers the full 9.4 public surface:

  • ad-hoc discovery generation (generate), retrieval (find) and bulk workflow-status/visibility updates (bulk_update),

  • generation runs metadata (get_generations, get_generation, dismiss_generation),

  • Attack Discovery schedules CRUD plus enable/disable and search (create_schedule, get_schedule, update_schedule, delete_schedule, enable_schedule, disable_schedule, find_schedules).

All Attack Discovery resources are space-scoped: every method accepts an optional space_id to target a specific space (None targets the default space or the space the client is scoped to).

Note

The Attack Discovery Kibana feature must be enabled in the target space. Spaces using the pure Elasticsearch solution view (for example, a default space created with solution: "es") disable the securitySolutionAttackDiscovery feature: in such spaces find_schedules returns total: 0 even though schedule create/get/update/delete/enable/disable still work. Use a space with the security solution view (or the classic view) for full functionality.

Note

For OpenAI-compatible connectors (.gen-ai), the connector’s apiUrl must be the full chat-completions endpoint (for example http://localhost:1234/v1/chat/completions), not just the API base URL.

Example

>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>>
>>> # Kick off an ad-hoc generation with a Gen AI connector
>>> resp = await client.attack_discovery.generate(
...     alerts_index_pattern=".alerts-security.alerts-default",
...     anonymization_fields=[
...         {"id": "f1", "field": "_id", "allowed": True, "anonymized": False},
...         {"id": "f2", "field": "host.name", "allowed": True, "anonymized": True},
...     ],
...     api_config={"actionTypeId": ".gen-ai", "connectorId": "my-connector"},
...     size=25,
... )
>>> execution_uuid = resp.body["execution_uuid"]
>>>
>>> # Poll the generation until it completes
>>> gen = await client.attack_discovery.get_generation(execution_uuid=execution_uuid)
>>> print(gen.body["generation"]["status"])

Usage

The AsyncAttackDiscoveryClient provides the same methods as AttackDiscoveryClient but all methods are async and must be awaited:

import asyncio

from kibana import AsyncKibana

async def main():
    async with AsyncKibana("http://localhost:5601") as client:
        # Kick off a generation (async)
        started = await client.attack_discovery.generate(
            alerts_index_pattern=".alerts-security.alerts-default",
            anonymization_fields=[
                {"id": "f0", "field": "_id",
                 "allowed": True, "anonymized": False},
            ],
            api_config={
                "actionTypeId": ".gen-ai",
                "connectorId": "my-connector-id",
            },
            size=100,
        )

        # Track it (async)
        generation = await client.attack_discovery.get_generation(
            execution_uuid=started.body["execution_uuid"]
        )

        # Search discoveries (async)
        found = await client.attack_discovery.find(
            start="now-24h", end="now"
        )

asyncio.run(main())
__init__(client, default_space_id=None, validate_spaces=True)[source]

Initialize the AsyncAttackDiscoveryClient.

Parameters:

  • client (AsyncKibana) – The parent AsyncKibana client 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 before space-scoped operations (default: True).

Example

>>> attack_discovery_client = AsyncAttackDiscoveryClient(kibana_client)
async find(*, alert_ids=None, connector_names=None, enable_field_rendering=None, end=None, ids=None, include_unique_alert_ids=None, page=None, per_page=None, search=None, scheduled=None, shared=None, sort_field=None, sort_order=None, start=None, status=None, with_replacements=None, space_id=None, validate_spaces=None)[source]

Find Attack discoveries that match the search criteria.

Supports free text search, filtering, pagination, and sorting (GET /api/attack_discovery/_find).

Parameters:

  • alert_ids (list[str] | None) – Filter results to Attack discoveries that include any of the provided alert IDs.

  • connector_names (list[str] | None) – Filter results to Attack discoveries created by any of the provided human readable connector names (the connector_name property, e.g. "GPT-5 Chat"), which are distinct from connector_id values.

  • enable_field_rendering (bool | None) – Enables a markdown syntax used to render pivot fields, for example {{ user.name james }}. Defaults to False on the server.

  • end (str | None) – End of the time range for the search. Accepts absolute timestamps (ISO 8601) or relative date math (e.g. "now").

  • ids (list[str] | None) – Filter results to the Attack discoveries with the specified IDs.

  • include_unique_alert_ids (bool | None) – If True, the response includes unique_alert_ids and unique_alert_ids_count aggregated across the matched Attack discoveries.

  • page (int | None) – Page number to return. Defaults to 1 on the server.

  • per_page (int | None) – Number of Attack discoveries per page. Defaults to 10 on the server.

  • search (str | None) – Free-text search query applied to relevant text fields (title, description, tags, etc.).

  • scheduled (bool | None) – Use True to return only scheduled discoveries, False for only ad-hoc discoveries; omit for both.

  • shared (bool | None) – Use True to return only shared discoveries, False for only those visible to the current user; omit for both.

  • sort_field (str | None) – Field used to sort results. In 9.4 only "@timestamp" is allowed.

  • sort_order (str | None) – Sort direction, "asc" or "desc" (server default "desc").

  • start (str | None) – Start of the time range for the search (ISO 8601 or date math such as "now-24h").

  • status (list[str] | None) – Filter by alert workflow status; one or more of "open", "acknowledged", "closed".

  • with_replacements (bool | None) – When True, returns discoveries with text replacements applied to the detailsMarkdown, entitySummaryMarkdown, summaryMarkdown, and title fields. Defaults to True on the server.

  • space_id (str | None) – Optional space ID to search in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse containing connector_names, data (list of Attack discovery alerts), page, per_page, total and unique_alert_ids_count.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> found = await client.attack_discovery.find(
...     start="now-24h", end="now", status=["open"], per_page=10
... )
>>> print(found.body["total"])
async bulk_update(*, ids, enable_field_rendering=None, kibana_alert_workflow_status=None, visibility=None, with_replacements=None, space_id=None, validate_spaces=None)[source]

Bulk update Attack discoveries.

Performs bulk updates on multiple Attack discoveries, including workflow status changes and visibility settings, without requiring individual API calls for each alert (POST /api/attack_discovery/_bulk).

Parameters:

  • ids (list[str]) – Array of Attack Discovery IDs to update.

  • enable_field_rendering (bool | None) – Enables a markdown syntax used to render pivot fields, for example {{ user.name james }}. Defaults to False on the server.

  • kibana_alert_workflow_status (str | None) – When provided, updates the kibana.alert.workflow_status of the attack discovery alerts; one of "open", "acknowledged", "closed".

  • visibility (str | None) – When provided, updates the visibility of the alert; one of "not_shared", "shared".

  • with_replacements (bool | None) – When True, returns the updated discoveries with text replacements applied. Defaults to True on the server.

  • space_id (str | None) – Optional space ID to update discoveries in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

the array of updated Attack Discovery alert objects. IDs that match no existing discovery are silently ignored (the live server returns {"data": []} rather than an error).

Return type:

ObjectApiResponse with data

Raises:

Example

>>> updated = await client.attack_discovery.bulk_update(
...     ids=["c0c8a8bb..."],
...     kibana_alert_workflow_status="acknowledged",
... )
>>> print(len(updated.body["data"]))
async generate(*, alerts_index_pattern, anonymization_fields, api_config, size, sub_action='invokeAI', connector_name=None, end=None, filter=None, model=None, replacements=None, start=None, space_id=None, validate_spaces=None)[source]

Generate attack discoveries from alerts.

Initiates the generation of attack discoveries by analyzing security alerts using AI. Returns an execution UUID that can be used to track the generation progress and retrieve results (POST /api/attack_discovery/_generate). Results may also be retrieved via find().

Note

The _id field must be present (and allowed) in anonymization_fields; otherwise the generation fails with “The _id field must be allowed to generate Attack discoveries.” The alerts index is sorted on kibana.alert.risk_score, so a custom index pattern must have a mapping for that field.

Parameters:

  • alerts_index_pattern (str) – The (space specific) index pattern that contains the alerts to use as context, e.g. ".alerts-security.alerts-default".

  • anonymization_fields (list[dict[str, Any]]) – The list of fields, and whether or not they are anonymized, allowed to be sent to LLMs. Each entry has id, field, allowed and anonymized keys. Consider using the output of the /api/security_ai_assistant/anonymization_fields/_find API.

  • api_config (dict[str, Any]) – LLM API configuration with required connectorId and actionTypeId keys (optionally model, provider, defaultSystemPromptId).

  • size (int) – The maximum number of alerts to analyze.

  • sub_action (str) – LLM invocation mode, "invokeAI" (default) or "invokeStream".

  • connector_name (str | None) – Optional human readable connector name recorded on the generation.

  • end (str | None) – End of the alert time range (ISO 8601 or date math, e.g. "now").

  • filter (dict[str, Any] | None) – An Elasticsearch query DSL object used to filter alerts.

  • model (str | None) – Optional model override.

  • replacements (dict[str, str] | None) – Optional anonymization replacements mapping (anonymized value -> original value).

  • start (str | None) – Start of the alert time range (e.g. "now-24h").

  • space_id (str | None) – Optional space ID to generate discoveries in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

the identifier for the attack discovery generation process, usable with get_generation() and dismiss_generation().

Return type:

ObjectApiResponse with execution_uuid

Raises:

Example

>>> resp = await client.attack_discovery.generate(
...     alerts_index_pattern=".alerts-security.alerts-default",
...     anonymization_fields=[
...         {"id": "f1", "field": "_id", "allowed": True,
...          "anonymized": False},
...     ],
...     api_config={
...         "actionTypeId": ".gen-ai",
...         "connectorId": "my-connector-id",
...     },
...     size=25,
...     start="now-24h",
...     end="now",
... )
>>> print(resp.body["execution_uuid"])
async get_generations(*, end=None, size=None, start=None, space_id=None, validate_spaces=None)[source]

Get the latest Attack Discovery generations metadata.

Retrieves generation metadata for the current user, including execution status and statistics (GET /api/attack_discovery/generations).

Parameters:

  • end (str | None) – End of the time range for filtering generations (ISO 8601 or date math, e.g. "now").

  • size (int | None) – The maximum number of generations to retrieve (server default 50).

  • start (str | None) – Start of the time range for filtering generations (e.g. "now-24h").

  • space_id (str | None) – Optional space ID to list generations from.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

a list of generation metadata objects (execution_uuid, status, connector_id, discoveries, timing fields, …). Note: despite the API description, on a live 9.4.3 server dismissed generations are still listed, with status: "dismissed".

Return type:

ObjectApiResponse with generations

Raises:

Example

>>> gens = client.attack_discovery.get_generations(size=10)
>>> for g in gens.body["generations"]:
...     print(g["execution_uuid"], g["status"])
async get_generation(*, execution_uuid, enable_field_rendering=None, with_replacements=None, space_id=None, validate_spaces=None)[source]

Get a single Attack Discovery generation.

Returns a specific generation, including all generated Attack discoveries and associated metadata such as execution status and statistics (GET /api/attack_discovery/generations/{execution_uuid}).

Parameters:

  • execution_uuid (str) – The unique identifier for the generation execution, as returned by generate().

  • enable_field_rendering (bool | None) – Enables a markdown syntax used to render pivot fields, for example {{ user.name james }}. Defaults to False on the server.

  • with_replacements (bool | None) – When True, returns discoveries with text replacements applied. Defaults to True on the server.

  • space_id (str | None) – Optional space ID to get the generation from.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with generation (metadata including status: one of started, succeeded, failed, dismissed, canceled) and data (the generated Attack discoveries).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> gen = await client.attack_discovery.get_generation(
...     execution_uuid="edd26039-0990-4d9f-9829-2a1fcacb77b5"
... )
>>> print(gen.body["generation"]["status"])
async dismiss_generation(*, execution_uuid, space_id=None, validate_spaces=None)[source]

Dismiss an Attack Discovery generation.

Marks a generation as dismissed for the current user (POST /api/attack_discovery/generations/{execution_uuid}/_dismiss).

Parameters:

  • execution_uuid (str) – The unique identifier for the generation execution, as returned by generate().

  • space_id (str | None) – Optional space ID the generation belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the (dismissed) generation metadata object. The updated dismissed status becomes visible in get_generations() shortly after (event-log refresh).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.attack_discovery.dismiss_generation(
...     execution_uuid="edd26039-0990-4d9f-9829-2a1fcacb77b5"
... )
async create_schedule(*, name, params, schedule, actions=None, enabled=None, space_id=None, validate_spaces=None)[source]

Create an Attack Discovery schedule.

Creates a schedule that periodically generates attack discoveries (POST /api/attack_discovery/schedules). A schedule is backed by an alerting rule of type attack-discovery.

Parameters:

  • name (str) – The name of the schedule.

  • params (dict[str, Any]) – The schedule configuration parameters. Required keys: alerts_index_pattern (index pattern to get alerts from), api_config (LLM configuration with connectorId, actionTypeId and the connector name) and size (max number of alerts). Optional: query, filters, combined_filter, start, end.

  • schedule (dict[str, Any]) – The schedule interval, e.g. {"interval": "24h"}.

  • actions (list[dict[str, Any]] | None) – Optional schedule actions (connector notifications).

  • enabled (bool | None) – Whether the schedule is enabled (server default False).

  • space_id (str | None) – Optional space ID to create the schedule in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the created schedule (id, name, created_by, updated_by, created_at, updated_at, enabled, params, schedule, actions).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> created = await client.attack_discovery.create_schedule(
...     name="Daily attack discovery",
...     params={
...         "alerts_index_pattern": ".alerts-security.alerts-default",
...         "api_config": {
...             "connectorId": "my-connector-id",
...             "actionTypeId": ".gen-ai",
...             "name": "My GenAI connector",
...         },
...         "size": 100,
...     },
...     schedule={"interval": "24h"},
... )
>>> schedule_id = created.body["id"]
async get_schedule(*, id, space_id=None, validate_spaces=None)[source]

Get an Attack Discovery schedule by ID.

Retrieves a single schedule (GET /api/attack_discovery/schedules/{id}).

Parameters:

  • id (str) – The identifier of the schedule.

  • space_id (str | None) – Optional space ID to get the schedule from.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the schedule details (id, name, enabled, params, schedule, actions, audit fields).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> schedule = await client.attack_discovery.get_schedule(id="b0ab787f-...")
>>> print(schedule.body["name"])
async update_schedule(*, id, name, params, schedule, actions, space_id=None, validate_spaces=None)[source]

Update an Attack Discovery schedule.

Replaces the schedule’s properties (PUT /api/attack_discovery/schedules/{id}). All properties are required by the 9.4 API (this is a full update, not a patch).

Parameters:

  • id (str) – The identifier of the schedule to update.

  • name (str) – The name of the schedule.

  • params (dict[str, Any]) – The schedule configuration parameters (see create_schedule() for the required keys).

  • schedule (dict[str, Any]) – The schedule interval, e.g. {"interval": "12h"}.

  • actions (list[dict[str, Any]]) – The schedule actions. Pass [] to keep no actions.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the updated schedule details.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> updated = await client.attack_discovery.update_schedule(
...     id="b0ab787f-...",
...     name="Renamed schedule",
...     params={
...         "alerts_index_pattern": ".alerts-security.alerts-default",
...         "api_config": {
...             "connectorId": "my-connector-id",
...             "actionTypeId": ".gen-ai",
...             "name": "My GenAI connector",
...         },
...         "size": 50,
...     },
...     schedule={"interval": "12h"},
...     actions=[],
... )
async delete_schedule(*, id, space_id=None, validate_spaces=None)[source]

Delete an Attack Discovery schedule.

Deletes the schedule and its backing alerting rule (DELETE /api/attack_discovery/schedules/{id}).

Parameters:

  • id (str) – The identifier of the schedule to delete.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the deleted schedule’s id.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.attack_discovery.delete_schedule(id="b0ab787f-...")
async enable_schedule(*, id, space_id=None, validate_spaces=None)[source]

Enable an Attack Discovery schedule.

Starts periodic attack discovery generation for the schedule (POST /api/attack_discovery/schedules/{id}/_enable).

Parameters:

  • id (str) – The identifier of the schedule to enable.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the schedule’s id.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.attack_discovery.enable_schedule(id="b0ab787f-...")
async disable_schedule(*, id, space_id=None, validate_spaces=None)[source]

Disable an Attack Discovery schedule.

Stops periodic attack discovery generation for the schedule (POST /api/attack_discovery/schedules/{id}/_disable).

Parameters:

  • id (str) – The identifier of the schedule to disable.

  • space_id (str | None) – Optional space ID the schedule belongs to.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with the schedule’s id.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.attack_discovery.disable_schedule(id="b0ab787f-...")
async find_schedules(*, page=None, per_page=None, sort_direction=None, sort_field=None, space_id=None, validate_spaces=None)[source]

Find Attack Discovery schedules.

Lists schedules with pagination and sorting (GET /api/attack_discovery/schedules/_find).

Note

On a live 9.4.3 server the page parameter is off by one: the server treats an explicit page=N as page N + 1 (page=1 returns the second page). Omit page (or pass page=0) to get the first page. Also, in spaces where the securitySolutionAttackDiscovery feature is disabled (e.g. spaces with the Elasticsearch solution view), this endpoint returns total: 0 even though schedules exist.

Parameters:

  • page (int | None) – Page number (see the note above about the live off-by-one behavior).

  • per_page (int | None) – Number of schedules per page (server default 10).

  • sort_direction (str | None) – Sort direction, "asc" or "desc" (server default "asc").

  • sort_field (str | None) – Field to sort by. Common fields: "name", "created_at", "updated_at", "enabled".

  • space_id (str | None) – Optional space ID to search in.

  • validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse with page, per_page, total and data (the list of schedules).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> found = await client.attack_discovery.find_schedules(per_page=100)
>>> for s in found.body["data"]:
...     print(s["id"], s["name"], s["enabled"])
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]