Kibana Client¶

The main synchronous client for interacting with Kibana’s REST API.

class kibana.Kibana(hosts=None, *, cloud_id=None, api_key=None, basic_auth=None, bearer_auth=None, headers=<kibana._sync.client._base.DefaultType object>, request_timeout=<kibana._sync.client._base.DefaultType object>, verify_certs=<kibana._sync.client._base.DefaultType object>, ca_certs=<kibana._sync.client._base.DefaultType object>, client_cert=<kibana._sync.client._base.DefaultType object>, client_key=<kibana._sync.client._base.DefaultType object>, ssl_assert_hostname=<kibana._sync.client._base.DefaultType object>, ssl_assert_fingerprint=<kibana._sync.client._base.DefaultType object>, ssl_version=<kibana._sync.client._base.DefaultType object>, ssl_context=<kibana._sync.client._base.DefaultType object>, ssl_show_warn=<kibana._sync.client._base.DefaultType object>, max_retries=<kibana._sync.client._base.DefaultType object>, retry_on_status=<kibana._sync.client._base.DefaultType object>, retry_on_timeout=<kibana._sync.client._base.DefaultType object>, connections_per_node=<kibana._sync.client._base.DefaultType object>, dead_node_backoff_factor=<kibana._sync.client._base.DefaultType object>, max_dead_node_backoff=<kibana._sync.client._base.DefaultType object>, node_class=<kibana._sync.client._base.DefaultType object>, node_pool_class=<kibana._sync.client._base.DefaultType object>, randomize_nodes_in_pool=<kibana._sync.client._base.DefaultType object>, max_requests_per_second=None, _transport=None)[source]¶

Bases: BaseClient

Synchronous client for Kibana.

Provides a Pythonic interface to interact with Kibana’s REST APIs. Each API group is exposed as a namespace attribute (client.dashboards, client.spaces, client.alerting, …), mirroring the structure of the official Kibana API reference.

Example usage:

>>> from kibana import Kibana
>>> client = Kibana(
...     hosts=["http://localhost:5601"],
...     api_key="your_api_key"
... )
>>> # Use the client
>>> client.close()

Or use as a context manager:

>>> with Kibana(hosts=["http://localhost:5601"]) as client:
...     # Use the client
...     pass

Initialization

The Kibana client can be initialized with various connection and authentication options:

from kibana import Kibana

# Basic initialization with URL
client = Kibana("http://localhost:5601")

# With API key authentication
client = Kibana(
    "http://localhost:5601",
    api_key="your_api_key"
)

# With basic authentication
client = Kibana(
    "http://localhost:5601",
    basic_auth=("username", "password")
)

# With multiple hosts
client = Kibana([
    "http://localhost:5601",
    "http://localhost:5602"
])

Context Manager Usage

The client can be used as a context manager to ensure proper resource cleanup:

with Kibana("http://localhost:5601") as client:
    # Use the client
    status = client.status.get_status()
    print(status.body["status"]["overall"]["level"])
# Client is automatically closed

Namespace Clients

The Kibana client provides access to various API namespaces through properties:

actions - Actions API for managing connectors
spaces - Spaces API for managing Kibana Spaces
saved_objects - Saved Objects API for managing saved objects
status - Status API for monitoring server health

Space-Scoped Operations

Create a space-scoped client for operations within a specific space:

# Create space-scoped client with validation
marketing_client = client.space("marketing")

# All operations are automatically scoped to the "marketing" space
connector = marketing_client.actions.create(
    name="Marketing Webhook",
    connector_type_id=".webhook",
    config={"url": "https://example.com/webhook"}
)

# Create space-scoped client without validation (for performance)
fast_client = client.space("marketing", validate=False)

actions: ActionsClient¶

agent_builder: AgentBuilderClient¶

alerting: AlertingClient¶

apm: ApmClient¶

attack_discovery: AttackDiscoveryClient¶

cases: CasesClient¶

connectors: ConnectorsClient¶

dashboards: DashboardsClient¶

data_views: DataViewsClient¶

detection_engine: DetectionEngineClient¶

endpoint: EndpointClient¶

entity_analytics: EntityAnalyticsClient¶

exception_lists: ExceptionListsClient¶

fleet: FleetClient¶

fleet_agents: FleetAgentsClient¶

fleet_enrollment: FleetEnrollmentClient¶

fleet_epm: FleetEpmClient¶

fleet_outputs: FleetOutputsClient¶

fleet_policies: FleetPoliciesClient¶

lists: ListsClient¶

logstash: LogstashClient¶

maintenance_windows: MaintenanceWindowsClient¶

ml: MlClient¶

observability_ai_assistant: ObservabilityAiAssistantClient¶

osquery: OsqueryClient¶

saved_objects: SavedObjectsClient¶

security: SecurityClient¶

security_ai_assistant: SecurityAiAssistantClient¶

short_urls: ShortUrlsClient¶

slos: SlosClient¶

spaces: SpacesClient¶

status: StatusClient¶

streams: StreamsClient¶

synthetics: SyntheticsClient¶

task_manager: TaskManagerClient¶

timeline: TimelineClient¶

upgrade_assistant: UpgradeAssistantClient¶

uptime: UptimeClient¶

visualizations: VisualizationsClient¶

workflows: WorkflowsClient¶

__init__(hosts=None, *, cloud_id=None, api_key=None, basic_auth=None, bearer_auth=None, headers=<kibana._sync.client._base.DefaultType object>, request_timeout=<kibana._sync.client._base.DefaultType object>, verify_certs=<kibana._sync.client._base.DefaultType object>, ca_certs=<kibana._sync.client._base.DefaultType object>, client_cert=<kibana._sync.client._base.DefaultType object>, client_key=<kibana._sync.client._base.DefaultType object>, ssl_assert_hostname=<kibana._sync.client._base.DefaultType object>, ssl_assert_fingerprint=<kibana._sync.client._base.DefaultType object>, ssl_version=<kibana._sync.client._base.DefaultType object>, ssl_context=<kibana._sync.client._base.DefaultType object>, ssl_show_warn=<kibana._sync.client._base.DefaultType object>, max_retries=<kibana._sync.client._base.DefaultType object>, retry_on_status=<kibana._sync.client._base.DefaultType object>, retry_on_timeout=<kibana._sync.client._base.DefaultType object>, connections_per_node=<kibana._sync.client._base.DefaultType object>, dead_node_backoff_factor=<kibana._sync.client._base.DefaultType object>, max_dead_node_backoff=<kibana._sync.client._base.DefaultType object>, node_class=<kibana._sync.client._base.DefaultType object>, node_pool_class=<kibana._sync.client._base.DefaultType object>, randomize_nodes_in_pool=<kibana._sync.client._base.DefaultType object>, max_requests_per_second=None, _transport=None)[source]¶

Initialize Kibana client.

Parameters:

hosts (str | list[str | dict[str, Any]] | None) – List of Kibana nodes to connect to. Can be a single string or a list of strings/dicts. Examples: - “http://localhost:5601” - [”http://localhost:5601”, “http://localhost:5602”] - [{“host”: “localhost”, “port”: 5601, “scheme”: “http”}]
cloud_id (str | None) – Cloud ID for Elastic Cloud deployments
api_key (str | tuple[str, str] | None) – API key for authentication. Can be: - Base64-encoded string - Tuple of (id, api_key)
basic_auth (tuple[str, str] | None) – Basic authentication credentials as (username, password)
bearer_auth (str | None) – Bearer token for authentication
headers (DefaultType | Mapping[str, str]) – Custom headers to include in all requests
request_timeout (DefaultType | None | float) – Request timeout in seconds
verify_certs (DefaultType | bool) – Whether to verify SSL certificates
ca_certs (DefaultType | str) – Path to CA certificate bundle
client_cert (DefaultType | str) – Path to client certificate
client_key (DefaultType | str) – Path to client private key
ssl_assert_hostname (DefaultType | str) – Hostname to verify in SSL certificate
ssl_assert_fingerprint (DefaultType | str) – SSL certificate fingerprint to verify
ssl_version (DefaultType | int) – SSL/TLS version to use
ssl_context (DefaultType | Any) – Custom SSL context
ssl_show_warn (DefaultType | bool) – Whether to show SSL warnings
max_retries (DefaultType | int) – Maximum number of retries for failed requests
retry_on_status (DefaultType | list[int]) – HTTP status codes to retry on
retry_on_timeout (DefaultType | bool) – Whether to retry on timeout
connections_per_node (DefaultType | int) – Number of connections per node
dead_node_backoff_factor (DefaultType | float) – Backoff factor for dead nodes
max_dead_node_backoff (DefaultType | float) – Maximum backoff time for dead nodes
node_class (DefaultType | Any) – Custom node class
node_pool_class (DefaultType | Any) – Custom node pool class
randomize_nodes_in_pool (DefaultType | bool) – Whether to randomize node order
max_requests_per_second (float | None) – Optional rate limit (requests/sec). When set, outgoing requests are throttled using a token-bucket algorithm to prevent overwhelming the Kibana cluster.
_transport (Transport | None) – Pre-configured Transport instance (for testing)

close()[source]¶

Close the client and release resources.

This closes all connections in the connection pool. After calling close(), the client should not be used.

__enter__()[source]¶

Enter context manager.

__exit__(*args)[source]¶

Exit context manager and close client.

__repr__()[source]¶

Return string representation of client.

space(space_id, validate=True)[source]¶

Create a space-scoped client instance.

This method creates a new client instance that automatically operates within the specified space context. All operations performed through the returned client will be scoped to the specified space.

Parameters:

space_id (str) – The ID of the space to scope operations to
validate (bool) – Whether to validate that the space exists (default: True)

Returns:

SpaceScopedKibana instance scoped to the specified space

Raises:

SpaceNotFoundError – If validate=True and the space doesn’t exist
InvalidSpaceIdError – If the space_id format is invalid

Return type:

SpaceScopedKibana

Example

>>> # Create a space-scoped client with validation
>>> marketing_client = client.space("marketing")
>>>
>>> # Create a dashboard in the marketing space
>>> dashboard = marketing_client.dashboards.create(
...     title="Marketing KPIs"
... )
>>>
>>> # Create space-scoped client without validation (for performance)
>>> fast_client = client.space("marketing", validate=False)

options(*, api_key=<kibana._sync.client._base.DefaultType object>, basic_auth=<kibana._sync.client._base.DefaultType object>, bearer_auth=<kibana._sync.client._base.DefaultType object>, headers=<kibana._sync.client._base.DefaultType object>, request_timeout=<kibana._sync.client._base.DefaultType object>)¶

Create a new client instance with modified options.

This method allows per-request configuration without modifying the original client instance. It creates a shallow copy with updated settings, enabling different authentication or configuration for specific requests.

Parameters:

api_key (DefaultType | str | tuple[str, str]) – API key for authentication. Can be: - String: Base64-encoded API key - Tuple: (id, api_key) to be encoded automatically If provided, takes precedence over other auth methods.
basic_auth (DefaultType | tuple[str, str]) – Basic authentication credentials as (username, password) tuple. Used if api_key is not provided.
bearer_auth (DefaultType | str) – Bearer token string for authentication. Used if neither api_key nor basic_auth is provided.
headers (DefaultType | Mapping[str, str]) – Custom HTTP headers to include in requests. These will be merged with default headers.
request_timeout (DefaultType | float) – Request timeout in seconds. Overrides the default timeout for this client instance.

Returns:

A new BaseClient instance with the specified options applied. The original client remains unchanged.

Return type:

Self

Example

>>> # Create client with default auth
>>> client = Kibana("http://localhost:5601", api_key="default_key")
>>>
>>> # Make a request with different auth
>>> special_client = client.options(api_key="special_key")
>>> response = special_client.actions.get_all()
>>>
>>> # Original client still uses default auth
>>> response = client.actions.get_all()
>>>
>>> # Use custom headers for a single request
>>> custom_client = client.options(
...     headers={"X-Custom-Header": "value"},
...     request_timeout=30.0
... )

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

Perform an HTTP request to Kibana.

This method handles the complete request lifecycle including: - Building authentication headers - Adding required Kibana headers (kbn-xsrf, content-type) - Creating OpenTelemetry spans for observability - Executing the request via elastic-transport - Processing the response and handling errors

Parameters:

method (str) – HTTP method to use (GET, POST, PUT, DELETE, PATCH, etc.).
path (str) – API endpoint path starting with / (e.g., “/api/status”).
params (dict[str, Any] | None) – Optional query parameters as a dictionary. Will be URL-encoded and appended to the path (booleans become true/false, lists become repeated keys, dicts become JSON strings).
headers (dict[str, str] | None) – Optional HTTP headers to include in the request. These will be merged with authentication and default headers.
body (Any | None) – Optional request body. Dicts and lists are JSON-serialized automatically; str/bytes bodies are sent as-is (set an explicit content-type header for NDJSON or multipart payloads).

Returns:

ObjectApiResponse containing the parsed JSON response body and metadata about the request/response.

Raises:

BadRequestError – If the request is malformed (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
NotFoundError – If the resource is not found (404).
ConflictError – If there’s a conflict (409).
ApiError – For other API errors (4xx, 5xx status codes).
TransportError – For connection or transport-level errors.

Return type:

ObjectApiResponse[Any]

Example

>>> # Simple GET request
>>> response = client.perform_request("GET", "/api/status")
>>> print(response.body["status"]["overall"]["level"])
>>>
>>> # POST request with body
>>> response = client.perform_request(
...     "POST",
...     "/api/actions/connector",
...     body={
...         "name": "My Webhook",
...         "connector_type_id": ".webhook",
...         "config": {"url": "https://example.com"}
...     }
... )
>>>
>>> # Request with query parameters
>>> response = client.perform_request(
...     "GET",
...     "/api/saved_objects/_find",
...     params={"type": "dashboard", "per_page": 10}
... )

SpaceScopedKibana¶

A space-scoped client that automatically operates within a specific space context.

class kibana.SpaceScopedKibana(client, space_id, validate=True)[source]¶

Bases: object

Space-scoped client that delegates to main client with space context.

This class provides the same API surface as the main Kibana client but automatically scopes all operations to a specific space. All child clients (dashboards, saved_objects, alerting, etc.) created through this instance inherit the space context and validation settings. Namespaces that are not space-aware (spaces, status, security, task_manager, upgrade_assistant, logstash) delegate to the parent client unscoped.

Example

>>> # Create space-scoped client with validation
>>> marketing_client = client.space("marketing")
>>>
>>> # All operations are automatically scoped to "marketing" space
>>> dashboard = marketing_client.dashboards.create(
...     title="Marketing KPIs"
... )
>>>
>>> # Create space-scoped client without validation for performance
>>> fast_client = client.space("marketing", validate=False)

Usage

Space-scoped clients are created using the Kibana.space() method:

# Create space-scoped client
marketing_client = client.space("marketing")

# All operations inherit the space context
connector = marketing_client.actions.create(
    name="Test Connector",
    connector_type_id=".index",
    config={"index": "test"}
)

# The connector is created in the "marketing" space
# No need to pass space_id parameter

Validation

By default, space-scoped clients validate that the space exists when created. This can be disabled for performance-critical scenarios:

# With validation (default)
client_with_validation = client.space("marketing")

# Without validation (faster, but may fail on operations if space doesn't exist)
client_without_validation = client.space("marketing", validate=False)