WorkflowsClient¶

Client for the Kibana Workflows API.

Workflows automate sequences of steps (connector calls, Elasticsearch requests, Kibana actions, …) defined in a YAML document. The Workflows APIs are generally available since Kibana 9.4.0.

Workflows are space-scoped resources: a workflow created in one space is not visible from another space. Every method accepts an optional space_id to target a specific space.

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

Bases: NamespaceClient

Client for the Kibana Workflows API.

Workflows automate sequences of steps (connector calls, Elasticsearch requests, Kibana actions, …) defined in a YAML document. The Workflows APIs are generally available since Kibana 9.4.0.

Workflows are space-scoped resources: a workflow created in one space is not visible from another space. 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).

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>>
>>> yaml_definition = '''
... name: my-workflow
... enabled: true
... triggers:
...   - type: manual
... steps:
...   - name: log_step
...     type: console
...     with:
...       message: "hello world"
... '''
>>> created = client.workflows.create(yaml=yaml_definition)
>>> workflow_id = created.body["id"]
>>>
>>> run = client.workflows.run(id=workflow_id, inputs={})
>>> execution_id = run.body["workflowExecutionId"]
>>> execution = client.workflows.get_execution(execution_id=execution_id)
>>> client.workflows.delete(id=workflow_id)

Creating and Running Workflows

Workflows are defined in YAML:

from kibana import Kibana

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

yaml_definition = """
name: my-workflow
enabled: true
triggers:
  - type: manual
steps:
  - name: log_step
    type: console
    with:
      message: "hello world"
"""

# Create the workflow
created = client.workflows.create(yaml=yaml_definition)
workflow_id = created.body["id"]

# Run it manually
run = client.workflows.run(id=workflow_id, inputs={})
execution_id = run.body["workflowExecutionId"]

Inspecting Executions

# Get a single execution
execution = client.workflows.get_execution(execution_id=execution_id)
print(execution.body["status"])

# List executions of a workflow
executions = client.workflows.get_executions(workflow_id=workflow_id)

# Step-level details and logs
steps = client.workflows.get_step_executions(workflow_id=workflow_id)
logs = client.workflows.get_execution_logs(execution_id=execution_id)

Searching and Managing Workflows

# Search workflows
found = client.workflows.get_all(query="my-workflow", size=10)
for workflow in found.body["results"]:
    print(workflow["id"], workflow["name"])

# Update the YAML definition
client.workflows.update(id=workflow_id, yaml=yaml_definition)

# Clone a workflow
clone = client.workflows.clone(id=workflow_id)

# Delete workflows
client.workflows.delete(id=workflow_id)

Testing Workflows

Test a workflow definition without persisting it:

test_run = client.workflows.test(
    workflow_yaml=yaml_definition, inputs={}
)

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

Initialize the WorkflowsClient.

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

>>> workflows_client = WorkflowsClient(kibana_client)

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

Create a workflow.

Creates a new workflow from a YAML definition. The definition describes the workflow name, enabled flag, triggers and steps (see get_schema() for the full JSON schema).

Parameters:

yaml (str) – The YAML definition of the workflow (max 1 MiB).
id (str | None) – Optional custom workflow ID (3-255 chars, lowercase alphanumerics and hyphens, e.g. "my-workflow-1"). If omitted, Kibana generates one.
space_id (str | None) – Optional space ID to create the workflow in.
validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse containing the created workflow, including id, name, enabled, yaml, the parsed definition, valid and audit fields (createdBy, createdAt, …).

Raises:

BadRequestError – If the YAML definition is invalid.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
SpaceNotFoundError – If the space doesn’t exist and validation is enabled.

Return type:

WorkflowsClient¶

AsyncWorkflowsClient¶