StreamsClient¶
Client for the Kibana Streams API.
Streams provide a single-pane experience to manage log (and other) data in Elasticsearch: routing (wired streams), processing pipelines, field mappings, lifecycle, significant-events queries and linked attachments (dashboards, rules, SLOs).
Note
All Streams APIs are marked as technical preview in Kibana 9.4 and may change in future releases.
Streams must be enabled first (see enable()); once
enabled, Kibana manages wired root streams (in 9.4 these are logs.ecs and
logs.otel) from which child streams can be forked.
Streams are space-scoped: every method accepts an optional space_id to
target a specific space.
- class kibana._sync.client.streams.StreamsClient(client, default_space_id=None, validate_spaces=True)[source]¶
Bases:
NamespaceClientClient for the Kibana Streams API.
Streams provide a single-pane experience to manage log (and other) data in Elasticsearch: routing (wired streams), processing pipelines, field mappings, lifecycle, significant-events queries and linked attachments (dashboards, rules, SLOs).
All Streams APIs are marked as Technical Preview in Kibana 9.4 and may change in future releases. Streams must be enabled first (see
enable()); once enabled, Kibana manages wired root streams (in 9.4 these arelogs.ecsandlogs.otel) from which child streams can be forked.Streams are space-scoped: every method accepts an optional
space_idto target a specific space (Nonetargets the default space or the space the client is scoped to).Example
>>> from kibana import Kibana >>> client = Kibana("http://localhost:5601", api_key="...") >>> >>> # Enable streams, then fork a child stream from the ECS root >>> client.streams.enable() >>> client.streams.fork( ... name="logs.ecs", ... stream_name="logs.ecs.myapp", ... where={"field": "service.name", "eq": "myapp"}, ... ) >>> streams = client.streams.get_all() >>> print([s["name"] for s in streams.body["streams"]])
Enabling Streams and Forking
from kibana import Kibana client = Kibana("http://localhost:5601", api_key="your_api_key") # Enable streams, then fork a child stream from the ECS root client.streams.enable() client.streams.fork( name="logs.ecs", stream_name="logs.ecs.myapp", where={"field": "service.name", "eq": "myapp"}, ) # List all streams streams = client.streams.get_all() print([s["name"] for s in streams.body["streams"]]) # Disable streams again (removes stream management) client.streams.disable()
Reading and Updating Stream Definitions
# Get a stream definition (routing, processing, lifecycle, ...) stream = client.streams.get(name="logs.ecs.myapp") # Get only the ingest configuration ingest = client.streams.get_ingest(name="logs.ecs.myapp") # Delete a stream client.streams.delete(name="logs.ecs.myapp")
Significant Events Queries
# Add a significant-events query (ES|QL) to a stream client.streams.upsert_query( name="logs.ecs.myapp", query_id="failed-logins", title="Failed logins", esql='FROM logs.ecs.myapp | WHERE event.outcome == "failure"', ) # List queries attached to a stream queries = client.streams.get_queries(name="logs.ecs.myapp") # Read significant events events = client.streams.get_significant_events( name="logs.ecs.myapp", from_="2026-07-01T00:00:00.000Z", to="2026-07-03T00:00:00.000Z", bucket_size="1h", )
Attachments
Link Kibana assets (dashboards, rules, SLOs) to a stream:
# Link a dashboard to a stream client.streams.link_attachment( name="logs.ecs.myapp", attachment_id="my-dashboard-id", attachment_type="dashboard", ) # List attachments attachments = client.streams.get_attachments(name="logs.ecs.myapp") # Unlink it again client.streams.unlink_attachment( name="logs.ecs.myapp", attachment_id="my-dashboard-id", attachment_type="dashboard", )
- __init__(client, default_space_id=None, validate_spaces=True)[source]¶
Initialize the StreamsClient.
- Parameters:
Example
>>> streams_client = StreamsClient(kibana_client)
- enable(*, space_id=None, validate_spaces=None)[source]¶
Enable streams.
Technical preview in 9.4. Enables the wired streams framework: Kibana creates the root streams (
logs.ecsandlogs.otelin 9.4) and the Elasticsearch resources backing them. The call is idempotent — enabling an already-enabled deployment is a no-op.- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult("created"when newly enabled,"noop"when already enabled).- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> resp = client.streams.enable() >>> print(resp.body["acknowledged"]) True
- disable(*, space_id=None, validate_spaces=None)[source]¶
Disable streams.
Technical preview in 9.4. Disables the wired streams framework and deletes the stream definitions. Use with care: wired child streams and their configuration are removed.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult.- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.disable()
- resync(*, space_id=None, validate_spaces=None)[source]¶
Resync streams.
Technical preview in 9.4. Rebuilds the Elasticsearch assets (index templates, ingest pipelines, component templates) backing all streams from the stored stream definitions. Useful when the assets have drifted from the definitions.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult.- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.resync()
- get_all(*, space_id=None, validate_spaces=None)[source]¶
Get the stream list.
Technical preview in 9.4. Fetches the list of all streams (wired, classic and query streams) with their effective configuration.
- Parameters:
- Returns:
ObjectApiResponse with a
streamslist; every entry containsname,type(wired/classic/query),descriptionand the type-specific configuration.- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> streams = client.streams.get_all() >>> print([s["name"] for s in streams.body["streams"]]) ['logs.ecs', 'logs.otel']
- get(*, name, space_id=None, validate_spaces=None)[source]¶
Get a stream.
Technical preview in 9.4. Fetches a stream definition and its related objects (dashboards, rules, queries) plus the caller’s effective privileges on the stream.
- Parameters:
- Returns:
ObjectApiResponse with
stream(definition),dashboards,rules,queries,privilegesand, for wired streams,inherited_fieldsandeffective_*settings.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> stream = client.streams.get(name="logs.ecs") >>> print(stream.body["stream"]["type"]) wired
- upsert(*, name, stream, dashboards=None, queries=None, rules=None, space_id=None, validate_spaces=None)[source]¶
Create or update a stream.
Technical preview in 9.4. Upserts the full stream definition together with its linked dashboards, rules and queries. The
streamobject must carry atypediscriminator:"wired"(with aningestobject containing awiredsection),"classic"(ingestwith aclassicsection) or"query"(with aqueryobject containingviewandesql).- Parameters:
name (str) – The stream name (e.g.
"logs.ecs.myapp").stream (dict[str, Any]) – The stream definition, e.g.
{"type": "wired", "description": "...", "ingest": { "lifecycle": {"inherit": {}}, "processing": {"steps": []}, "settings": {}, "failure_store": {"inherit": {}}, "wired": {"fields": {}, "routing": []}}}.dashboards (list[str] | None) – Dashboard IDs to link to the stream (defaults to an empty list; the API requires the field).
queries (list[dict[str, Any]] | None) – Significant-events queries to store on the stream; each item needs
id,title,descriptionandesql: {"query": ...}(defaults to an empty list).rules (list[str] | None) – Rule IDs to link to the stream (defaults to an empty list).
space_id (str | None) – Optional space ID to upsert the stream in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult("created"or"updated").- Raises:
BadRequestError – If the definition fails validation.
NotFoundError – If a parent stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.upsert( ... name="logs.ecs.myapp", ... stream={ ... "type": "wired", ... "description": "My app logs", ... "ingest": { ... "lifecycle": {"inherit": {}}, ... "processing": {"steps": []}, ... "settings": {}, ... "failure_store": {"inherit": {}}, ... "wired": {"fields": {}, "routing": []}, ... }, ... }, ... )
- delete(*, name, space_id=None, validate_spaces=None)[source]¶
Delete a stream.
Technical preview in 9.4. Deletes a stream definition and, for wired streams, the routing rule pointing at it from the parent stream. Root streams cannot be deleted.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult("deleted").- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.delete(name="logs.ecs.myapp")
- fork(*, name, stream_name, where, status=None, space_id=None, validate_spaces=None)[source]¶
Fork a stream.
Technical preview in 9.4. Forks a wired stream: creates a child stream and adds a routing rule to the parent so that documents matching the
wherecondition are routed to the child.- Parameters:
name (str) – The parent stream name (e.g.
"logs.ecs").stream_name (str) – The name of the child stream to create; it must be prefixed by the parent name (e.g.
"logs.ecs.myapp").where (dict[str, Any]) – Routing condition, e.g.
{"field": "service.name", "eq": "myapp"}. Conditions can be combined withand/or/notand the special{"always": {}}/{"never": {}}values.status (str | None) – Optional routing rule status:
"enabled"or"disabled".space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult("created").- Raises:
BadRequestError – If the condition or child name is invalid.
NotFoundError – If the parent stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.fork( ... name="logs.ecs", ... stream_name="logs.ecs.myapp", ... where={"field": "service.name", "eq": "myapp"}, ... )
- get_ingest(*, name, space_id=None, validate_spaces=None)[source]¶
Get ingest stream settings.
Technical preview in 9.4. Fetches only the ingest configuration of a stream: lifecycle, processing steps, index settings, failure store and the
wired/classicspecific section.- Parameters:
- Returns:
ObjectApiResponse with an
ingestobject.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> ingest = client.streams.get_ingest(name="logs.ecs") >>> print(ingest.body["ingest"]["lifecycle"]) {'dsl': {}}
- update_ingest(*, name, ingest, space_id=None, validate_spaces=None)[source]¶
Update ingest stream settings.
Technical preview in 9.4. Replaces the ingest configuration of a stream (lifecycle, processing steps, index settings, failure store and field mappings/routing for wired streams).
- Parameters:
name (str) – The stream name.
ingest (dict[str, Any]) – The full ingest object, e.g.
{"lifecycle": {"inherit": {}}, "processing": {"steps": []}, "settings": {}, "failure_store": {"inherit": {}}, "wired": {"fields": {...}, "routing": []}}. Classic streams use a"classic"section instead of"wired". Note: when replaying a document returned byget_ingest(), strip the read-onlyprocessing.updated_atfield first — Kibana 9.4.3 rejects it on write.space_id (str | None) – Optional space ID to update in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult.- Raises:
BadRequestError – If the ingest object fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> ingest = client.streams.get_ingest(name="logs.ecs.myapp") >>> body = ingest.body["ingest"] >>> body["wired"]["fields"]["attributes.env"] = {"type": "keyword"} >>> client.streams.update_ingest(name="logs.ecs.myapp", ingest=body)
- get_query_settings(*, name, space_id=None, validate_spaces=None)[source]¶
Get query stream settings.
Technical preview, added in 9.4. Fetches the query configuration (the ES|QL query and optional field descriptions) of a query stream. Fails with a 400 error when the target is not a query stream.
- Parameters:
- Returns:
ObjectApiResponse with the query stream settings.
- Raises:
BadRequestError – If the stream is not a query stream.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> settings = client.streams.get_query_settings( ... name="myquerystream" ... )
- update_query_settings(*, name, esql, field_descriptions=None, space_id=None, validate_spaces=None)[source]¶
Upsert query stream settings.
Technical preview, added in 9.4. Creates or updates a query stream — a stream backed by an ES|QL query instead of ingested data.
- Parameters:
name (str) – The query stream name.
esql (str) – The ES|QL query backing the stream, e.g.
"FROM logs.ecs, logs.ecs.* METADATA _id, _source | LIMIT 10".field_descriptions (dict[str, str] | None) – Optional mapping of field name to human-readable description.
space_id (str | None) – Optional space ID to upsert in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult("created"or"updated").- Raises:
BadRequestError – If the query fails validation.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.update_query_settings( ... name="myquerystream", ... esql="FROM logs.ecs, logs.ecs.* METADATA _id, _source", ... )
- get_queries(*, name, space_id=None, validate_spaces=None)[source]¶
Get stream queries.
Technical preview in 9.4. Fetches the significant-events queries linked to a stream.
- Parameters:
- Returns:
ObjectApiResponse with a
querieslist; each entry hasid,title,description,esqland optionalseverity_score/evidence.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> queries = client.streams.get_queries(name="logs.ecs") >>> print(len(queries.body["queries"])) 0
- bulk_queries(*, name, operations, space_id=None, validate_spaces=None)[source]¶
Bulk update queries of a stream.
Technical preview in 9.4. Executes multiple query upsert/delete operations on a stream in a single request.
- Parameters:
name (str) – The stream name.
operations (list[dict[str, Any]]) – List of operations. Each item is either
{"index": {"id": ..., "title": ..., "description": ..., "esql": {"query": ...}}}to upsert a query or{"delete": {"id": ...}}to remove one.space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
BadRequestError – If an operation fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.bulk_queries( ... name="logs.ecs.myapp", ... operations=[{"delete": {"id": "old-query"}}], ... )
- upsert_query(*, name, query_id, title, esql, description=None, severity_score=None, evidence=None, space_id=None, validate_spaces=None)[source]¶
Upsert a query to a stream.
Technical preview in 9.4. Creates or updates a significant-events query on a stream. Kibana validates the ES|QL shape: the query must read
FROM <stream>, <stream>.*and includeMETADATA _id, _source.- Parameters:
name (str) – The stream name.
query_id (str) – The identifier of the query to upsert.
title (str) – Query title.
esql (str) – The ES|QL query string, e.g.
"FROM logs.ecs.myapp, logs.ecs.myapp.* METADATA _id, _source | WHERE message LIKE \"*error*\"".description (str | None) – Optional query description.
severity_score (float | None) – Optional severity score for matching events.
evidence (list[str] | None) – Optional list of evidence strings explaining why the query is significant.
space_id (str | None) – Optional space ID to upsert in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
BadRequestError – If the ES|QL query fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.upsert_query( ... name="logs.ecs.myapp", ... query_id="errors", ... title="Error spike", ... esql=( ... "FROM logs.ecs.myapp, logs.ecs.myapp.* " ... 'METADATA _id, _source | WHERE message LIKE "*error*"' ... ), ... )
- delete_query(*, name, query_id, space_id=None, validate_spaces=None)[source]¶
Remove a query from a stream.
Technical preview in 9.4. Deletes a significant-events query from a stream.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
NotFoundError – If the stream or query does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.delete_query( ... name="logs.ecs.myapp", query_id="errors" ... )
- get_significant_events(*, name, from_, to, bucket_size, query=None, search_mode=None, space_id=None, validate_spaces=None)[source]¶
Read the significant events of a stream.
Technical preview in 9.4. Evaluates the stream’s significant-events queries over a time range and returns the occurrences bucketed by
bucket_sizetogether with change-point detection results.- Parameters:
name (str) – The stream name.
from – Start of the time range (ISO 8601 date string, sent as the
fromquery parameter).to (str) – End of the time range (ISO 8601 date string).
bucket_size (str) – Bucket size for occurrence aggregation (e.g.
"1h").query (str | None) – Optional text to filter/search the queries with.
search_mode (str | None) – Optional search mode:
"keyword","semantic"or"hybrid".space_id (str | None) – Optional space ID to read from.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
significant_events(per-query occurrences and change points) andaggregated_occurrences.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> events = client.streams.get_significant_events( ... name="logs.ecs.myapp", ... from_="2026-07-01T00:00:00.000Z", ... to="2026-07-02T00:00:00.000Z", ... bucket_size="1h", ... )
- generate_significant_events(*, name, from_, to, connector_id=None, sample_docs_size=None, space_id=None, validate_spaces=None)[source]¶
Generate significant-events queries for a stream.
Technical preview, added in 9.2. Uses an AI connector to analyze sample documents from the stream and generate candidate significant-events queries. Requires an AI connector: pass
connector_idor configure a default AI connector in Kibana, otherwise the API responds with a 400 error.- Parameters:
name (str) – The stream name.
from – Start of the sampling time range (ISO 8601 date string, sent as the
fromquery parameter).to (str) – End of the sampling time range (ISO 8601 date string).
connector_id (str | None) – Optional AI connector ID to use for generation.
sample_docs_size (float | None) – Optional number of sample documents to feed to the model.
space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with the generated queries.
- Raises:
BadRequestError – If no connector ID is provided and no default AI connector is configured.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> generated = client.streams.generate_significant_events( ... name="logs.ecs.myapp", ... from_="2026-07-01T00:00:00.000Z", ... to="2026-07-02T00:00:00.000Z", ... connector_id="my-ai-connector", ... )
- preview_significant_events(*, name, from_, to, bucket_size, esql, space_id=None, validate_spaces=None)[source]¶
Preview significant events for an ad-hoc query.
Technical preview in 9.4. Evaluates a candidate significant-events ES|QL query against a stream without saving it, returning bucketed occurrences and change-point analysis.
- Parameters:
name (str) – The stream name.
from – Start of the time range (ISO 8601 date string, sent as the
fromquery parameter).to (str) – End of the time range (ISO 8601 date string).
bucket_size (str) – Bucket size for occurrence aggregation (e.g.
"1h").esql (str) – The ES|QL query to preview; the same shape rules apply as for
upsert_query().space_id (str | None) – Optional space ID to run the preview in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
occurrencesandchange_pointsfor the previewed query.- Raises:
BadRequestError – If the ES|QL query fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> preview = client.streams.preview_significant_events( ... name="logs.ecs.myapp", ... from_="2026-07-01T00:00:00.000Z", ... to="2026-07-02T00:00:00.000Z", ... bucket_size="1h", ... esql=( ... "FROM logs.ecs.myapp, logs.ecs.myapp.* " ... "METADATA _id, _source" ... ), ... )
- export_content(*, name, content_name, description, version, include=None, space_id=None, validate_spaces=None)[source]¶
Export stream content as a content pack.
Technical preview in 9.4. Exports the stream’s objects (queries, routing/child streams, mappings) as a content pack ZIP archive.
- Parameters:
name (str) – The stream name to export from.
content_name (str) – Name of the content pack (body field
name).description (str) – Description of the content pack.
version (str) – Semantic version of the content pack (e.g.
"1.0.0").include (dict[str, Any] | None) – Included-objects filter. Defaults to
{"objects": {"all": {}}}. A selective filter looks like{"objects": {"queries": [{"id": ...}], "mappings": True, "routing": [...]}}.space_id (str | None) – Optional space ID to export from.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
BinaryApiResponse whose
bodyholds the ZIP archive bytes (served withcontent-type: application/zip).- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> exported = client.streams.export_content( ... name="logs.ecs.myapp", ... content_name="myapp-pack", ... description="My app stream content", ... version="1.0.0", ... ) >>> with open("myapp-pack.zip", "wb") as f: ... _ = f.write(exported.body)
- import_content(*, name, content, include=None, filename='content_pack.zip', space_id=None, validate_spaces=None)[source]¶
Import content into a stream.
Technical preview in 9.4. Imports a content pack ZIP archive (uploaded as
multipart/form-data) into a stream, creating or updating the included objects.- Parameters:
name (str) – The stream name to import into.
content (bytes) – Content pack archive bytes (e.g. the body returned by
export_content()).include (dict[str, Any] | None) – Included-objects filter, JSON-encoded into the
includeform field. Defaults to{"objects": {"all": {}}}.filename (str) – Filename advertised in the multipart upload (default:
"content_pack.zip").space_id (str | None) – Optional space ID to import into.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedand aresultobject listingcreatedandupdatedstreams.- Raises:
BadRequestError – If the archive or filter is invalid.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> exported = client.streams.export_content( ... name="logs.ecs.myapp", ... content_name="myapp-pack", ... description="My app stream content", ... version="1.0.0", ... ) >>> client.streams.import_content( ... name="logs.ecs.myapp", content=exported.body ... )
- get_attachments(*, name, query=None, attachment_types=None, tags=None, space_id=None, validate_spaces=None)[source]¶
Get stream attachments.
Technical preview, added in 9.3. Fetches the attachments (dashboards, rules, SLOs) linked to a stream, optionally filtered by search text, attachment type or tags.
- Parameters:
name (str) – The stream name (path parameter
streamName).query (str | None) – Optional text to filter attachments by title.
attachment_types (list[str] | None) – Optional list of attachment types to include:
"dashboard","rule"and/or"slo".tags (list[str] | None) – Optional list of tags to filter by.
space_id (str | None) – Optional space ID to read from.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with an
attachmentslist; each entry hasid,type,title,tagsand timestamps.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> attachments = client.streams.get_attachments( ... name="logs.ecs.myapp", ... attachment_types=["dashboard"], ... )
- bulk_attachments(*, name, operations, space_id=None, validate_spaces=None)[source]¶
Bulk update attachments of a stream.
Technical preview, added in 9.3. Executes multiple attachment link/unlink operations on a stream in a single request.
- Parameters:
name (str) – The stream name (path parameter
streamName).operations (list[dict[str, Any]]) – List of operations. Each item is either
{"index": {"id": ..., "type": ...}}to link an attachment or{"delete": {"id": ..., "type": ...}}to unlink one;typeis"dashboard","rule"or"slo".space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
BadRequestError – If an operation fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.bulk_attachments( ... name="logs.ecs.myapp", ... operations=[ ... {"index": {"id": "my-dashboard", "type": "dashboard"}}, ... ], ... )
- link_attachment(*, name, attachment_type, attachment_id, space_id=None, validate_spaces=None)[source]¶
Link an attachment to a stream.
Technical preview, added in 9.3. Links an existing dashboard, rule or SLO to a stream so it shows up in the stream’s overview.
- Parameters:
name (str) – The stream name (path parameter
streamName).attachment_type (str) – The attachment type:
"dashboard","rule"or"slo".attachment_id (str) – The ID of the object to link.
space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
NotFoundError – If the stream or the object does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.link_attachment( ... name="logs.ecs.myapp", ... attachment_type="dashboard", ... attachment_id="my-dashboard", ... )
- unlink_attachment(*, name, attachment_type, attachment_id, space_id=None, validate_spaces=None)[source]¶
Unlink an attachment from a stream.
Technical preview, added in 9.3. Unlinks a dashboard, rule or SLO from a stream. The underlying object is not deleted.
- Parameters:
name (str) – The stream name (path parameter
streamName).attachment_type (str) – The attachment type:
"dashboard","rule"or"slo".attachment_id (str) – The ID of the object to unlink.
space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
NotFoundError – If the stream or the attachment does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> client.streams.unlink_attachment( ... name="logs.ecs.myapp", ... attachment_type="dashboard", ... attachment_id="my-dashboard", ... )
AsyncStreamsClient¶
Asynchronous version of the StreamsClient for use with async/await syntax.
- class kibana._async.client.streams.AsyncStreamsClient(client, default_space_id=None, validate_spaces=True)[source]¶
Bases:
AsyncNamespaceClientAsync client for the Kibana Streams API.
Streams provide a single-pane experience to manage log (and other) data in Elasticsearch: routing (wired streams), processing pipelines, field mappings, lifecycle, significant-events queries and linked attachments (dashboards, rules, SLOs).
All Streams APIs are marked as Technical Preview in Kibana 9.4 and may change in future releases. Streams must be enabled first (see
enable()); once enabled, Kibana manages wired root streams (in 9.4 these arelogs.ecsandlogs.otel) from which child streams can be forked.Streams are space-scoped: every method accepts an optional
space_idto target a specific space (Nonetargets the default space or the space the client is scoped to).Example
>>> from kibana import AsyncKibana >>> client = AsyncKibana("http://localhost:5601", api_key="...") >>> >>> # Enable streams, then fork a child stream from the ECS root >>> await client.streams.enable() >>> await client.streams.fork( ... name="logs.ecs", ... stream_name="logs.ecs.myapp", ... where={"field": "service.name", "eq": "myapp"}, ... ) >>> streams = await client.streams.get_all() >>> print([s["name"] for s in streams.body["streams"]])
Usage
The AsyncStreamsClient provides the same methods as StreamsClient 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: # Enable streams (async) await client.streams.enable() # List streams (async) streams = await client.streams.get_all() print([s["name"] for s in streams.body["streams"]]) asyncio.run(main())
- __init__(client, default_space_id=None, validate_spaces=True)[source]¶
Initialize the AsyncStreamsClient.
- 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
>>> streams_client = AsyncStreamsClient(kibana_client)
- async enable(*, space_id=None, validate_spaces=None)[source]¶
Enable streams.
Technical preview in 9.4. Enables the wired streams framework: Kibana creates the root streams (
logs.ecsandlogs.otelin 9.4) and the Elasticsearch resources backing them. The call is idempotent — enabling an already-enabled deployment is a no-op.- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult("created"when newly enabled,"noop"when already enabled).- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> resp = await client.streams.enable() >>> print(resp.body["acknowledged"]) True
- async disable(*, space_id=None, validate_spaces=None)[source]¶
Disable streams.
Technical preview in 9.4. Disables the wired streams framework and deletes the stream definitions. Use with care: wired child streams and their configuration are removed.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult.- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.disable()
- async resync(*, space_id=None, validate_spaces=None)[source]¶
Resync streams.
Technical preview in 9.4. Rebuilds the Elasticsearch assets (index templates, ingest pipelines, component templates) backing all streams from the stored stream definitions. Useful when the assets have drifted from the definitions.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult.- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.resync()
- async get_all(*, space_id=None, validate_spaces=None)[source]¶
Get the stream list.
Technical preview in 9.4. Fetches the list of all streams (wired, classic and query streams) with their effective configuration.
- Parameters:
- Returns:
ObjectApiResponse with a
streamslist; every entry containsname,type(wired/classic/query),descriptionand the type-specific configuration.- Raises:
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> streams = await client.streams.get_all() >>> print([s["name"] for s in streams.body["streams"]]) ['logs.ecs', 'logs.otel']
- async get(*, name, space_id=None, validate_spaces=None)[source]¶
Get a stream.
Technical preview in 9.4. Fetches a stream definition and its related objects (dashboards, rules, queries) plus the caller’s effective privileges on the stream.
- Parameters:
- Returns:
ObjectApiResponse with
stream(definition),dashboards,rules,queries,privilegesand, for wired streams,inherited_fieldsandeffective_*settings.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> stream = await client.streams.get(name="logs.ecs") >>> print(stream.body["stream"]["type"]) wired
- async upsert(*, name, stream, dashboards=None, queries=None, rules=None, space_id=None, validate_spaces=None)[source]¶
Create or update a stream.
Technical preview in 9.4. Upserts the full stream definition together with its linked dashboards, rules and queries. The
streamobject must carry atypediscriminator:"wired"(with aningestobject containing awiredsection),"classic"(ingestwith aclassicsection) or"query"(with aqueryobject containingviewandesql).- Parameters:
name (str) – The stream name (e.g.
"logs.ecs.myapp").stream (dict[str, Any]) – The stream definition, e.g.
{"type": "wired", "description": "...", "ingest": { "lifecycle": {"inherit": {}}, "processing": {"steps": []}, "settings": {}, "failure_store": {"inherit": {}}, "wired": {"fields": {}, "routing": []}}}.dashboards (list[str] | None) – Dashboard IDs to link to the stream (defaults to an empty list; the API requires the field).
queries (list[dict[str, Any]] | None) – Significant-events queries to store on the stream; each item needs
id,title,descriptionandesql: {"query": ...}(defaults to an empty list).rules (list[str] | None) – Rule IDs to link to the stream (defaults to an empty list).
space_id (str | None) – Optional space ID to upsert the stream in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult("created"or"updated").- Raises:
BadRequestError – If the definition fails validation.
NotFoundError – If a parent stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.upsert( ... name="logs.ecs.myapp", ... stream={ ... "type": "wired", ... "description": "My app logs", ... "ingest": { ... "lifecycle": {"inherit": {}}, ... "processing": {"steps": []}, ... "settings": {}, ... "failure_store": {"inherit": {}}, ... "wired": {"fields": {}, "routing": []}, ... }, ... }, ... )
- async delete(*, name, space_id=None, validate_spaces=None)[source]¶
Delete a stream.
Technical preview in 9.4. Deletes a stream definition and, for wired streams, the routing rule pointing at it from the parent stream. Root streams cannot be deleted.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledgedandresult("deleted").- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.delete(name="logs.ecs.myapp")
- async fork(*, name, stream_name, where, status=None, space_id=None, validate_spaces=None)[source]¶
Fork a stream.
Technical preview in 9.4. Forks a wired stream: creates a child stream and adds a routing rule to the parent so that documents matching the
wherecondition are routed to the child.- Parameters:
name (str) – The parent stream name (e.g.
"logs.ecs").stream_name (str) – The name of the child stream to create; it must be prefixed by the parent name (e.g.
"logs.ecs.myapp").where (dict[str, Any]) – Routing condition, e.g.
{"field": "service.name", "eq": "myapp"}. Conditions can be combined withand/or/notand the special{"always": {}}/{"never": {}}values.status (str | None) – Optional routing rule status:
"enabled"or"disabled".space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult("created").- Raises:
BadRequestError – If the condition or child name is invalid.
NotFoundError – If the parent stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.fork( ... name="logs.ecs", ... stream_name="logs.ecs.myapp", ... where={"field": "service.name", "eq": "myapp"}, ... )
- async get_ingest(*, name, space_id=None, validate_spaces=None)[source]¶
Get ingest stream settings.
Technical preview in 9.4. Fetches only the ingest configuration of a stream: lifecycle, processing steps, index settings, failure store and the
wired/classicspecific section.- Parameters:
- Returns:
ObjectApiResponse with an
ingestobject.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> ingest = await client.streams.get_ingest(name="logs.ecs") >>> print(ingest.body["ingest"]["lifecycle"]) {'dsl': {}}
- async update_ingest(*, name, ingest, space_id=None, validate_spaces=None)[source]¶
Update ingest stream settings.
Technical preview in 9.4. Replaces the ingest configuration of a stream (lifecycle, processing steps, index settings, failure store and field mappings/routing for wired streams).
- Parameters:
name (str) – The stream name.
ingest (dict[str, Any]) – The full ingest object, e.g.
{"lifecycle": {"inherit": {}}, "processing": {"steps": []}, "settings": {}, "failure_store": {"inherit": {}}, "wired": {"fields": {...}, "routing": []}}. Classic streams use a"classic"section instead of"wired". Note: when replaying a document returned byget_ingest(), strip the read-onlyprocessing.updated_atfield first — Kibana 9.4.3 rejects it on write.space_id (str | None) – Optional space ID to update in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult.- Raises:
BadRequestError – If the ingest object fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> ingest = await client.streams.get_ingest(name="logs.ecs.myapp") >>> body = ingest.body["ingest"] >>> body["wired"]["fields"]["attributes.env"] = {"type": "keyword"} >>> await client.streams.update_ingest(name="logs.ecs.myapp", ingest=body)
- async get_query_settings(*, name, space_id=None, validate_spaces=None)[source]¶
Get query stream settings.
Technical preview, added in 9.4. Fetches the query configuration (the ES|QL query and optional field descriptions) of a query stream. Fails with a 400 error when the target is not a query stream.
- Parameters:
- Returns:
ObjectApiResponse with the query stream settings.
- Raises:
BadRequestError – If the stream is not a query stream.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> settings = await client.streams.get_query_settings( ... name="myquerystream" ... )
- async update_query_settings(*, name, esql, field_descriptions=None, space_id=None, validate_spaces=None)[source]¶
Upsert query stream settings.
Technical preview, added in 9.4. Creates or updates a query stream — a stream backed by an ES|QL query instead of ingested data.
- Parameters:
name (str) – The query stream name.
esql (str) – The ES|QL query backing the stream, e.g.
"FROM logs.ecs, logs.ecs.* METADATA _id, _source | LIMIT 10".field_descriptions (dict[str, str] | None) – Optional mapping of field name to human-readable description.
space_id (str | None) – Optional space ID to upsert in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedandresult("created"or"updated").- Raises:
BadRequestError – If the query fails validation.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.update_query_settings( ... name="myquerystream", ... esql="FROM logs.ecs, logs.ecs.* METADATA _id, _source", ... )
- async get_queries(*, name, space_id=None, validate_spaces=None)[source]¶
Get stream queries.
Technical preview in 9.4. Fetches the significant-events queries linked to a stream.
- Parameters:
- Returns:
ObjectApiResponse with a
querieslist; each entry hasid,title,description,esqland optionalseverity_score/evidence.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> queries = await client.streams.get_queries(name="logs.ecs") >>> print(len(queries.body["queries"])) 0
- async bulk_queries(*, name, operations, space_id=None, validate_spaces=None)[source]¶
Bulk update queries of a stream.
Technical preview in 9.4. Executes multiple query upsert/delete operations on a stream in a single request.
- Parameters:
name (str) – The stream name.
operations (list[dict[str, Any]]) – List of operations. Each item is either
{"index": {"id": ..., "title": ..., "description": ..., "esql": {"query": ...}}}to upsert a query or{"delete": {"id": ...}}to remove one.space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
BadRequestError – If an operation fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.bulk_queries( ... name="logs.ecs.myapp", ... operations=[{"delete": {"id": "old-query"}}], ... )
- async upsert_query(*, name, query_id, title, esql, description=None, severity_score=None, evidence=None, space_id=None, validate_spaces=None)[source]¶
Upsert a query to a stream.
Technical preview in 9.4. Creates or updates a significant-events query on a stream. Kibana validates the ES|QL shape: the query must read
FROM <stream>, <stream>.*and includeMETADATA _id, _source.- Parameters:
name (str) – The stream name.
query_id (str) – The identifier of the query to upsert.
title (str) – Query title.
esql (str) – The ES|QL query string, e.g.
"FROM logs.ecs.myapp, logs.ecs.myapp.* METADATA _id, _source | WHERE message LIKE \"*error*\"".description (str | None) – Optional query description.
severity_score (float | None) – Optional severity score for matching events.
evidence (list[str] | None) – Optional list of evidence strings explaining why the query is significant.
space_id (str | None) – Optional space ID to upsert in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
BadRequestError – If the ES|QL query fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.upsert_query( ... name="logs.ecs.myapp", ... query_id="errors", ... title="Error spike", ... esql=( ... "FROM logs.ecs.myapp, logs.ecs.myapp.* " ... 'METADATA _id, _source | WHERE message LIKE "*error*"' ... ), ... )
- async delete_query(*, name, query_id, space_id=None, validate_spaces=None)[source]¶
Remove a query from a stream.
Technical preview in 9.4. Deletes a significant-events query from a stream.
- Parameters:
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
NotFoundError – If the stream or query does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.delete_query( ... name="logs.ecs.myapp", query_id="errors" ... )
- async get_significant_events(*, name, from_, to, bucket_size, query=None, search_mode=None, space_id=None, validate_spaces=None)[source]¶
Read the significant events of a stream.
Technical preview in 9.4. Evaluates the stream’s significant-events queries over a time range and returns the occurrences bucketed by
bucket_sizetogether with change-point detection results.- Parameters:
name (str) – The stream name.
from – Start of the time range (ISO 8601 date string, sent as the
fromquery parameter).to (str) – End of the time range (ISO 8601 date string).
bucket_size (str) – Bucket size for occurrence aggregation (e.g.
"1h").query (str | None) – Optional text to filter/search the queries with.
search_mode (str | None) – Optional search mode:
"keyword","semantic"or"hybrid".space_id (str | None) – Optional space ID to read from.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
significant_events(per-query occurrences and change points) andaggregated_occurrences.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> events = await client.streams.get_significant_events( ... name="logs.ecs.myapp", ... from_="2026-07-01T00:00:00.000Z", ... to="2026-07-02T00:00:00.000Z", ... bucket_size="1h", ... )
- async generate_significant_events(*, name, from_, to, connector_id=None, sample_docs_size=None, space_id=None, validate_spaces=None)[source]¶
Generate significant-events queries for a stream.
Technical preview, added in 9.2. Uses an AI connector to analyze sample documents from the stream and generate candidate significant-events queries. Requires an AI connector: pass
connector_idor configure a default AI connector in Kibana, otherwise the API responds with a 400 error.- Parameters:
name (str) – The stream name.
from – Start of the sampling time range (ISO 8601 date string, sent as the
fromquery parameter).to (str) – End of the sampling time range (ISO 8601 date string).
connector_id (str | None) – Optional AI connector ID to use for generation.
sample_docs_size (float | None) – Optional number of sample documents to feed to the model.
space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with the generated queries.
- Raises:
BadRequestError – If no connector ID is provided and no default AI connector is configured.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> generated = await client.streams.generate_significant_events( ... name="logs.ecs.myapp", ... from_="2026-07-01T00:00:00.000Z", ... to="2026-07-02T00:00:00.000Z", ... connector_id="my-ai-connector", ... )
- async preview_significant_events(*, name, from_, to, bucket_size, esql, space_id=None, validate_spaces=None)[source]¶
Preview significant events for an ad-hoc query.
Technical preview in 9.4. Evaluates a candidate significant-events ES|QL query against a stream without saving it, returning bucketed occurrences and change-point analysis.
- Parameters:
name (str) – The stream name.
from – Start of the time range (ISO 8601 date string, sent as the
fromquery parameter).to (str) – End of the time range (ISO 8601 date string).
bucket_size (str) – Bucket size for occurrence aggregation (e.g.
"1h").esql (str) – The ES|QL query to preview; the same shape rules apply as for
upsert_query().space_id (str | None) – Optional space ID to run the preview in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
occurrencesandchange_pointsfor the previewed query.- Raises:
BadRequestError – If the ES|QL query fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> preview = await client.streams.preview_significant_events( ... name="logs.ecs.myapp", ... from_="2026-07-01T00:00:00.000Z", ... to="2026-07-02T00:00:00.000Z", ... bucket_size="1h", ... esql=( ... "FROM logs.ecs.myapp, logs.ecs.myapp.* " ... "METADATA _id, _source" ... ), ... )
- async export_content(*, name, content_name, description, version, include=None, space_id=None, validate_spaces=None)[source]¶
Export stream content as a content pack.
Technical preview in 9.4. Exports the stream’s objects (queries, routing/child streams, mappings) as a content pack ZIP archive.
- Parameters:
name (str) – The stream name to export from.
content_name (str) – Name of the content pack (body field
name).description (str) – Description of the content pack.
version (str) – Semantic version of the content pack (e.g.
"1.0.0").include (dict[str, Any] | None) – Included-objects filter. Defaults to
{"objects": {"all": {}}}. A selective filter looks like{"objects": {"queries": [{"id": ...}], "mappings": True, "routing": [...]}}.space_id (str | None) – Optional space ID to export from.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
BinaryApiResponse whose
bodyholds the ZIP archive bytes (served withcontent-type: application/zip).- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> exported = await client.streams.export_content( ... name="logs.ecs.myapp", ... content_name="myapp-pack", ... description="My app stream content", ... version="1.0.0", ... ) >>> with open("myapp-pack.zip", "wb") as f: ... _ = f.write(exported.body)
- async import_content(*, name, content, include=None, filename='content_pack.zip', space_id=None, validate_spaces=None)[source]¶
Import content into a stream.
Technical preview in 9.4. Imports a content pack ZIP archive (uploaded as
multipart/form-data) into a stream, creating or updating the included objects.- Parameters:
name (str) – The stream name to import into.
content (bytes) – Content pack archive bytes (e.g. the body returned by
export_content()).include (dict[str, Any] | None) – Included-objects filter, JSON-encoded into the
includeform field. Defaults to{"objects": {"all": {}}}.filename (str) – Filename advertised in the multipart upload (default:
"content_pack.zip").space_id (str | None) – Optional space ID to import into.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledgedand aresultobject listingcreatedandupdatedstreams.- Raises:
BadRequestError – If the archive or filter is invalid.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> exported = await client.streams.export_content( ... name="logs.ecs.myapp", ... content_name="myapp-pack", ... description="My app stream content", ... version="1.0.0", ... ) >>> await client.streams.import_content( ... name="logs.ecs.myapp", content=exported.body ... )
- async get_attachments(*, name, query=None, attachment_types=None, tags=None, space_id=None, validate_spaces=None)[source]¶
Get stream attachments.
Technical preview, added in 9.3. Fetches the attachments (dashboards, rules, SLOs) linked to a stream, optionally filtered by search text, attachment type or tags.
- Parameters:
name (str) – The stream name (path parameter
streamName).query (str | None) – Optional text to filter attachments by title.
attachment_types (list[str] | None) – Optional list of attachment types to include:
"dashboard","rule"and/or"slo".tags (list[str] | None) – Optional list of tags to filter by.
space_id (str | None) – Optional space ID to read from.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with an
attachmentslist; each entry hasid,type,title,tagsand timestamps.- Raises:
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> attachments = await client.streams.get_attachments( ... name="logs.ecs.myapp", ... attachment_types=["dashboard"], ... )
- async bulk_attachments(*, name, operations, space_id=None, validate_spaces=None)[source]¶
Bulk update attachments of a stream.
Technical preview, added in 9.3. Executes multiple attachment link/unlink operations on a stream in a single request.
- Parameters:
name (str) – The stream name (path parameter
streamName).operations (list[dict[str, Any]]) – List of operations. Each item is either
{"index": {"id": ..., "type": ...}}to link an attachment or{"delete": {"id": ..., "type": ...}}to unlink one;typeis"dashboard","rule"or"slo".space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
BadRequestError – If an operation fails validation.
NotFoundError – If the stream does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.bulk_attachments( ... name="logs.ecs.myapp", ... operations=[ ... {"index": {"id": "my-dashboard", "type": "dashboard"}}, ... ], ... )
- async link_attachment(*, name, attachment_type, attachment_id, space_id=None, validate_spaces=None)[source]¶
Link an attachment to a stream.
Technical preview, added in 9.3. Links an existing dashboard, rule or SLO to a stream so it shows up in the stream’s overview.
- Parameters:
name (str) – The stream name (path parameter
streamName).attachment_type (str) – The attachment type:
"dashboard","rule"or"slo".attachment_id (str) – The ID of the object to link.
space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
NotFoundError – If the stream or the object does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.link_attachment( ... name="logs.ecs.myapp", ... attachment_type="dashboard", ... attachment_id="my-dashboard", ... )
- async unlink_attachment(*, name, attachment_type, attachment_id, space_id=None, validate_spaces=None)[source]¶
Unlink an attachment from a stream.
Technical preview, added in 9.3. Unlinks a dashboard, rule or SLO from a stream. The underlying object is not deleted.
- Parameters:
name (str) – The stream name (path parameter
streamName).attachment_type (str) – The attachment type:
"dashboard","rule"or"slo".attachment_id (str) – The ID of the object to unlink.
space_id (str | None) – Optional space ID to run the operation in.
validate_spaces (bool | None) – Override space validation setting for this operation.
- Returns:
ObjectApiResponse with
acknowledged.- Raises:
NotFoundError – If the stream or the attachment does not exist.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
- Return type:
Example
>>> await client.streams.unlink_attachment( ... name="logs.ecs.myapp", ... attachment_type="dashboard", ... attachment_id="my-dashboard", ... )