OsqueryClient¶

Client for the Kibana Security Osquery API.

Osquery lets you query hosts like a database using SQL. Kibana’s Osquery Manager integration runs queries on Elastic Agents and stores the results in Elasticsearch. The API manages three resource types: packs (named sets of queries scheduled on agent policies), saved queries (reusable single queries), and live queries (one-off queries dispatched to a selection of agents).

Osquery resources are space-aware: a pack or saved query 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.osquery.OsqueryClient(client, default_space_id=None, validate_spaces=True)[source]¶

Bases: NamespaceClient

Client for the Kibana Security Osquery API.

Run live queries against Elastic Agents with Osquery Manager, and manage reusable saved queries and scheduled query packs. The Osquery integration must be added to an agent policy for queries to actually execute on hosts.

The API manages three resource types:

Packs (/api/osquery/packs): named sets of queries scheduled to run at an interval on the agent policies the pack is assigned to.
Saved queries (/api/osquery/saved_queries): reusable single queries that can be run as live queries or included in packs.
Live queries (/api/osquery/live_queries): one-off queries dispatched to a selection of agents, with per-action results.

All Osquery resources are space-aware: 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="...")
>>>
>>> # Create a pack with one scheduled query
>>> pack = client.osquery.create_pack(
...     name="my_pack",
...     queries={
...         "uptime": {"query": "select * from uptime;", "interval": 3600}
...     },
... )
>>> pack_id = pack.body["data"]["saved_object_id"]
>>>
>>> # Run a live query on all agents and fetch its results
>>> live = client.osquery.create_live_query(
...     query="select * from uptime;", agent_all=True
... )
>>> action_id = live.body["data"]["action_id"]
>>> details = client.osquery.get_live_query(id=action_id)

Managing Packs

A pack groups queries and schedules them on the agent policies it is assigned to:

from kibana import Kibana

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

# Create a pack with one scheduled query
pack = client.osquery.create_pack(
    name="my_pack",
    description="Track host uptime",
    enabled=False,
    queries={
        "uptime": {"query": "select * from uptime;", "interval": 3600}
    },
)
pack_id = pack.body["data"]["saved_object_id"]

# List, fetch, update and delete packs
packs = client.osquery.find_packs(page=1, page_size=20)
details = client.osquery.get_pack(id=pack_id)

# Note: the server treats updates as replacements — send every
# field you want to keep.
client.osquery.update_pack(
    id=pack_id,
    name="my_pack",
    description="Track host uptime (updated)",
    enabled=False,
    queries={
        "uptime": {"query": "select * from uptime;", "interval": 600}
    },
)
client.osquery.delete_pack(id=pack_id)

Managing Saved Queries

Saved queries are reusable SQL queries that can be run as live queries or added to packs:

saved = client.osquery.create_saved_query(
    id="my_saved_query",
    query="select * from uptime;",
    interval="60",  # the server requires a string, not an integer
    description="Host uptime",
    ecs_mapping={"host.uptime": {"field": "total_seconds"}},
)
saved_object_id = saved.body["data"]["saved_object_id"]

queries = client.osquery.find_saved_queries(page_size=100)
details = client.osquery.get_saved_query(id=saved_object_id)

# Updates are replacements too; `new_id` (the saved query name)
# is always required by the server.
client.osquery.update_saved_query(
    id=saved_object_id,
    new_id="my_saved_query",
    query="select * from uptime;",
    interval="120",
)
client.osquery.delete_saved_query(id=saved_object_id)

Running Live Queries

Live queries are dispatched to agents immediately. They require at least one enrolled Elastic Agent running the Osquery Manager integration:

live = client.osquery.create_live_query(
    query="select * from uptime;",
    agent_all=True,
    ecs_mapping={"host.uptime": {"field": "total_seconds"}},
)
live_query_id = live.body["data"]["action_id"]
action_id = live.body["data"]["queries"][0]["action_id"]

# Check the status and fetch per-query results
details = client.osquery.get_live_query(id=live_query_id)
results = client.osquery.get_live_query_results(
    id=live_query_id, action_id=action_id, page_size=100
)

# List past live queries
history = client.osquery.find_live_queries(kuery="user_id:elastic")

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

Initialize the OsqueryClient.

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

>>> osquery_client = OsqueryClient(kibana_client)

create_pack(*, name, queries, description=None, enabled=None, policy_ids=None, shards=None, space_id=None, validate_spaces=None)[source]¶

Create a query pack.

Creates an Osquery pack: a named set of queries that are scheduled to run on the agent policies the pack is assigned to.

Parameters:

name (str) – The pack name.
queries (dict[str, Any]) – An object of queries keyed by query ID. Each value may contain query (the SQL to run), interval (seconds between runs), platform, version, ecs_mapping, snapshot and removed. Kibana requires query and interval for each entry.
description (str | None) – The pack description.
enabled (bool | None) – Enables the pack. Enabled packs run on the agent policies listed in policy_ids.
policy_ids (list[str] | None) – A list of agent policy IDs to schedule the pack on.
shards (dict[str, Any] | None) – An object with shard configuration for policies included in the pack. For each policy, set the shard configuration to a percentage (1-100) of target hosts.
space_id (str | None) – Optional space ID to create the pack in.
validate_spaces (bool | None) – Override space validation setting for this operation.

Returns:

ObjectApiResponse whose data object contains the created pack, including saved_object_id (use it for get/update/delete), name, queries, enabled and audit metadata.

Raises:

BadRequestError – If required fields are missing or invalid.
AuthenticationException – If authentication fails.
AuthorizationException – If insufficient privileges.
SpaceNotFoundError – If the space doesn’t exist and validation is enabled.

Return type:

OsqueryClient¶

AsyncOsqueryClient¶