FleetEnrollmentClient

Client for the Kibana Fleet enrollment keys and tokens API.

This namespace covers the Fleet APIs used to enroll Elastic Agents and Fleet Servers: enrollment API keys, Fleet Server service tokens, Logstash API keys, uninstall tokens for tamper-protected agents, the message signing service key pair, and the Kubernetes agent manifest.

All operations are space-aware: every method accepts an optional space_id to target a specific space.

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

Bases: NamespaceClient

Client for the Kibana Fleet enrollment keys and tokens API.

Covers the Fleet APIs used to enroll Elastic Agents and Fleet Servers:

  • Enrollment API keys: tokens that Elastic Agents present to Fleet Server to enroll into an agent policy (list, create, get, revoke).

  • Service tokens: Elasticsearch service tokens used to enroll Fleet Server instances with Kibana.

  • Logstash API keys: API keys for Logstash to receive data from Elastic Agents via a Fleet Logstash output.

  • Uninstall tokens: per-policy tokens required to uninstall tamper-protected Elastic Agents.

  • Message signing service: rotate the key pair Fleet uses to sign messages sent to Elastic Agents.

  • Kubernetes manifests: fetch or download the manifest for deploying Elastic Agent on Kubernetes.

All operations 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 an enrollment API key for an agent policy
>>> created = client.fleet_enrollment.create_key(
...     policy_id="agent-policy-id", name="my-enrollment-key"
... )
>>> key_id = created.body["item"]["id"]
>>>
>>> # List keys, then revoke the one we created
>>> keys = client.fleet_enrollment.get_keys(per_page=50)
>>> client.fleet_enrollment.delete_key(key_id=key_id)

Managing Enrollment API Keys

Enrollment API keys are tied to an agent policy; agents present the key to Fleet Server to enroll into that policy:

from kibana import Kibana

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

# Create an enrollment API key for an agent policy
created = client.fleet_enrollment.create_key(
    policy_id="agent-policy-id", name="my-enrollment-key"
)
key_id = created.body["item"]["id"]
enrollment_token = created.body["item"]["api_key"]

# List keys (optionally filtered with KQL) and fetch one by ID
keys = client.fleet_enrollment.get_keys(
    per_page=50, kuery='policy_id:"agent-policy-id"'
)
key = client.fleet_enrollment.get_key(key_id=key_id)

# Revoke the key (it is marked inactive, not removed)
client.fleet_enrollment.delete_key(key_id=key_id)

Service Tokens and Logstash API Keys

# Service token for enrolling a Fleet Server (remote=True for the
# elastic/fleet-server-remote service account)
token = client.fleet_enrollment.create_service_token()
remote_token = client.fleet_enrollment.create_service_token(remote=True)

# API key for a Logstash output. Note: requires basic (user)
# authentication - Elasticsearch cannot derive an API key from
# another API key.
logstash_key = client.fleet_enrollment.create_logstash_api_key()

Uninstall Tokens

Uninstall tokens are generated per agent policy and are required to uninstall tamper-protected Elastic Agents:

# List token metadata (no decrypted values)
tokens = client.fleet_enrollment.get_uninstall_tokens(
    policy_id="agent-policy-id"
)
token_id = tokens.body["items"][0]["id"]

# Fetch one decrypted token
decrypted = client.fleet_enrollment.get_uninstall_token(
    uninstall_token_id=token_id
)
print(decrypted.body["item"]["token"])

Message Signing and Kubernetes Manifests

# Rotate the Fleet message signing key pair (irreversible; all agents
# must be re-enrolled afterwards, hence the explicit acknowledge)
client.fleet_enrollment.rotate_message_signing_key_pair(acknowledge=True)

# Kubernetes manifest as JSON ({"item": "<yaml>"}) with the Fleet URL
# and enrollment token substituted in
manifest = client.fleet_enrollment.get_kubernetes_manifest(
    fleet_server="https://fleet.example.com:8220",
    enrol_token=enrollment_token,
)

# ... or as a raw YAML download (response body is the YAML string)
yaml_text = client.fleet_enrollment.download_kubernetes_manifest().body
__init__(client, default_space_id=None, validate_spaces=True)[source]

Initialize the FleetEnrollmentClient.

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

>>> fleet_enrollment_client = FleetEnrollmentClient(kibana_client)
get_keys(*, page=None, per_page=None, kuery=None, space_id=None, validate_spaces=None)[source]

Get enrollment API keys.

Lists all enrollment API keys. Requires the fleet-agents-all or fleet-setup privilege.

Parameters:

  • page (int | None) – Page number (default: 1).

  • per_page (int | None) – Number of results per page (default: 20).

  • kuery (str | None) – A KQL query string to filter results (for example, 'policy_id:"my-policy-id"').

  • space_id (str | None) – Optional space ID to list enrollment API keys from.

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

Returns:

  • items: List of enrollment API keys (id, api_key_id, api_key, name, policy_id, active, created_at, hidden)

  • list: Deprecated duplicate of items

  • total / page / perPage: pagination metadata

Return type:

ObjectApiResponse containing

Raises:

Example

>>> keys = client.fleet_enrollment.get_keys(
...     per_page=50, kuery='policy_id:"my-policy-id"'
... )
>>> for key in keys.body["items"]:
...     print(key["id"], key["name"], key["active"])
create_key(*, policy_id, name=None, expiration=None, space_id=None, validate_spaces=None)[source]

Create an enrollment API key.

Creates an enrollment API key for a given agent policy. Elastic Agents use the returned api_key to enroll into the policy. Requires the fleet-agents-all privilege.

Note: Kibana appends the generated key ID to the provided name (the stored name looks like "my-name (<key-id>)").

Parameters:

  • policy_id (str) – The ID of the agent policy the Elastic Agent will be enrolled in.

  • name (str | None) – The name of the enrollment API key.

  • expiration (str | None) – Expiration timestamp for the key (for example, "2025-01-01T00:00:00.000Z").

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

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

Returns:

  • item: The created enrollment API key (id, api_key_id, api_key, name, policy_id, active, created_at)

  • action: "created"

Return type:

ObjectApiResponse containing

Raises:

Example

>>> created = client.fleet_enrollment.create_key(
...     policy_id="agent-policy-id", name="my-enrollment-key"
... )
>>> print(created.body["item"]["api_key"])
get_key(*, key_id, space_id=None, validate_spaces=None)[source]

Get an enrollment API key.

Gets an enrollment API key by ID. Requires the fleet-agents-all or fleet-setup privilege.

Parameters:

  • key_id (str) – The ID of the enrollment API key.

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

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

Returns:

ObjectApiResponse containing the enrollment API key under item (id, api_key_id, api_key, name, policy_id, active, created_at, hidden).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> key = client.fleet_enrollment.get_key(key_id="key-id-1")
>>> print(key.body["item"]["policy_id"])
delete_key(*, key_id, space_id=None, validate_spaces=None)[source]

Revoke an enrollment API key.

Revokes an enrollment API key by ID by marking it as inactive. The key is not removed, but it can no longer be used to enroll Elastic Agents. Requires the fleet-agents-all privilege.

Parameters:

  • key_id (str) – The ID of the enrollment API key.

  • space_id (str | None) – Optional space ID to revoke the enrollment API key in.

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

Returns:

ObjectApiResponse containing {"action": "deleted"} on success.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> result = client.fleet_enrollment.delete_key(key_id="key-id-1")
>>> print(result.body["action"])
deleted
create_service_token(*, remote=None, space_id=None, validate_spaces=None)[source]

Create a service token.

Creates a Fleet Server service token. The token is used to enroll Fleet Server instances with Kibana. Requires the fleet-agents-all privilege.

Parameters:

  • remote (bool | None) – When True, generates a token for a remote Fleet Server (service account elastic/fleet-server-remote) instead of the default elastic/fleet-server account.

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

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

Returns:

  • name: The name of the generated service token

  • value: The service token secret value

Return type:

ObjectApiResponse containing

Raises:

Example

>>> token = client.fleet_enrollment.create_service_token()
>>> print(token.body["name"], token.body["value"])
create_logstash_api_key(*, space_id=None, validate_spaces=None)[source]

Generate a Logstash API key.

Generates an Elasticsearch API key for Logstash to use with a Fleet Logstash output (the logstash-output permission set). Requires the fleet-settings-all privilege.

Parameters:

  • space_id (str | None) – Optional space ID to generate the API key in.

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

Returns:

ObjectApiResponse containing the generated key under api_key in the id:secret format expected by the Logstash elastic_agent input.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> key = client.fleet_enrollment.create_logstash_api_key()
>>> print(key.body["api_key"])
get_uninstall_tokens(*, policy_id=None, search=None, per_page=None, page=None, space_id=None, validate_spaces=None)[source]

Get metadata for the latest uninstall tokens.

Lists the metadata for the latest uninstall tokens per agent policy. Uninstall tokens are required to uninstall tamper-protected Elastic Agents. Token values are not included; use get_uninstall_token() to get a decrypted token. Requires the fleet-agents-all privilege.

Note: policy_id and search cannot be used at the same time.

Parameters:

  • policy_id (str | None) – Partial match filtering for policy IDs (max length 50).

  • search (str | None) – Partial match filtering for uninstall token values (max length 50).

  • per_page (int | None) – The number of items to return (minimum 5).

  • page (int | None) – Page number (minimum 1).

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

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

Returns:

  • items: List of token metadata (id, policy_id, policy_name, created_at, namespaces)

  • total / page / perPage: pagination metadata

Return type:

ObjectApiResponse containing

Raises:

Example

>>> tokens = client.fleet_enrollment.get_uninstall_tokens()
>>> for token in tokens.body["items"]:
...     print(token["id"], token["policy_id"])
get_uninstall_token(*, uninstall_token_id, space_id=None, validate_spaces=None)[source]

Get a decrypted uninstall token.

Gets one decrypted uninstall token by its ID. The returned token value can be used to uninstall a tamper-protected Elastic Agent enrolled in the token’s agent policy. Requires the fleet-agents-all privilege.

Parameters:

  • uninstall_token_id (str) – The ID of the uninstall token.

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

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

Returns:

ObjectApiResponse containing the decrypted token under item (id, policy_id, policy_name, token, created_at, namespaces).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> token = client.fleet_enrollment.get_uninstall_token(
...     uninstall_token_id="token-id-1"
... )
>>> print(token.body["item"]["token"])
rotate_message_signing_key_pair(*, acknowledge=None, space_id=None, validate_spaces=None)[source]

Rotate the Fleet message signing key pair.

Rotates the key pair used by Fleet to sign messages sent to Elastic Agents. This operation is irreversible and requires all agents in the Fleet to be re-enrolled after rotation. You must explicitly acknowledge the risk by passing acknowledge=True; the server rejects the request with a 400 warning otherwise. Requires the fleet-agents-all, fleet-agent-policies-all and fleet-settings-all privileges.

Parameters:

  • acknowledge (bool | None) – Set to True to confirm you understand the risks of rotating the key pair (default: False, which is rejected).

  • space_id (str | None) – Optional space ID to perform the rotation in.

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

Returns:

ObjectApiResponse containing a confirmation message ("Key pair rotated successfully.").

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> result = client.fleet_enrollment.rotate_message_signing_key_pair(
...     acknowledge=True
... )
>>> print(result.body["message"])
Key pair rotated successfully.
get_kubernetes_manifest(*, download=None, fleet_server=None, enrol_token=None, space_id=None, validate_spaces=None)[source]

Get a full Kubernetes agent manifest.

Gets the Kubernetes manifest for deploying Elastic Agent as a DaemonSet, as a JSON object with the YAML document under item. Requires the fleet-agent-policies-read or fleet-setup privilege.

Parameters:

  • download (bool | None) – If True, returns the manifest as a downloadable file.

  • fleet_server (str | None) – Fleet Server host URL to include in the manifest (substituted for the FLEET_URL placeholder).

  • enrol_token (str | None) – Enrollment token to include in the manifest (substituted for the FLEET_ENROLLMENT_TOKEN placeholder).

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

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

Returns:

ObjectApiResponse containing the manifest YAML string under item.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> manifest = client.fleet_enrollment.get_kubernetes_manifest(
...     fleet_server="https://fleet.example.com:8220",
...     enrol_token="my-enrollment-token",
... )
>>> print(manifest.body["item"][:22])
download_kubernetes_manifest(*, download=None, fleet_server=None, enrol_token=None, space_id=None, validate_spaces=None)[source]

Download a Kubernetes agent manifest.

Downloads the Kubernetes manifest for deploying Elastic Agent as a raw YAML document (content type text/x-yaml); the response body is the YAML string itself. Requires the fleet-agent-policies-read or fleet-setup privilege.

Parameters:

  • download (bool | None) – If True, returns the manifest as a downloadable file.

  • fleet_server (str | None) – Fleet Server host URL to include in the manifest (substituted for the FLEET_URL placeholder).

  • enrol_token (str | None) – Enrollment token to include in the manifest (substituted for the FLEET_ENROLLMENT_TOKEN placeholder).

  • space_id (str | None) – Optional space ID to download the manifest from.

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

Returns:

Response whose body is the manifest YAML document as a string.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> manifest = client.fleet_enrollment.download_kubernetes_manifest()
>>> yaml_text = manifest.body
>>> print(yaml_text.splitlines()[0])
---
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]

AsyncFleetEnrollmentClient

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

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

Bases: AsyncNamespaceClient

Async client for the Kibana Fleet enrollment keys and tokens API.

Covers the Fleet APIs used to enroll Elastic Agents and Fleet Servers:

  • Enrollment API keys: tokens that Elastic Agents present to Fleet Server to enroll into an agent policy (list, create, get, revoke).

  • Service tokens: Elasticsearch service tokens used to enroll Fleet Server instances with Kibana.

  • Logstash API keys: API keys for Logstash to receive data from Elastic Agents via a Fleet Logstash output.

  • Uninstall tokens: per-policy tokens required to uninstall tamper-protected Elastic Agents.

  • Message signing service: rotate the key pair Fleet uses to sign messages sent to Elastic Agents.

  • Kubernetes manifests: fetch or download the manifest for deploying Elastic Agent on Kubernetes.

All operations 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 AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>>
>>> # Create an enrollment API key for an agent policy
>>> created = await client.fleet_enrollment.create_key(
...     policy_id="agent-policy-id", name="my-enrollment-key"
... )
>>> key_id = created.body["item"]["id"]
>>>
>>> # List keys, then revoke the one we created
>>> keys = await client.fleet_enrollment.get_keys(per_page=50)
>>> await client.fleet_enrollment.delete_key(key_id=key_id)

Usage

The AsyncFleetEnrollmentClient provides the same methods as FleetEnrollmentClient 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:
        # Create and revoke an enrollment API key (async)
        created = await client.fleet_enrollment.create_key(
            policy_id="agent-policy-id", name="async-enrollment-key"
        )
        key_id = created.body["item"]["id"]

        keys = await client.fleet_enrollment.get_keys(per_page=50)
        await client.fleet_enrollment.delete_key(key_id=key_id)

        # Tokens and manifests (async)
        tokens = await client.fleet_enrollment.get_uninstall_tokens()
        manifest = await client.fleet_enrollment.get_kubernetes_manifest()

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

Initialize the AsyncFleetEnrollmentClient.

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

>>> fleet_enrollment_client = AsyncFleetEnrollmentClient(kibana_client)
async get_keys(*, page=None, per_page=None, kuery=None, space_id=None, validate_spaces=None)[source]

Get enrollment API keys.

Lists all enrollment API keys. Requires the fleet-agents-all or fleet-setup privilege.

Parameters:

  • page (int | None) – Page number (default: 1).

  • per_page (int | None) – Number of results per page (default: 20).

  • kuery (str | None) – A KQL query string to filter results (for example, 'policy_id:"my-policy-id"').

  • space_id (str | None) – Optional space ID to list enrollment API keys from.

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

Returns:

  • items: List of enrollment API keys (id, api_key_id, api_key, name, policy_id, active, created_at, hidden)

  • list: Deprecated duplicate of items

  • total / page / perPage: pagination metadata

Return type:

ObjectApiResponse containing

Raises:

Example

>>> keys = await client.fleet_enrollment.get_keys(
...     per_page=50, kuery='policy_id:"my-policy-id"'
... )
>>> for key in keys.body["items"]:
...     print(key["id"], key["name"], key["active"])
async create_key(*, policy_id, name=None, expiration=None, space_id=None, validate_spaces=None)[source]

Create an enrollment API key.

Creates an enrollment API key for a given agent policy. Elastic Agents use the returned api_key to enroll into the policy. Requires the fleet-agents-all privilege.

Note: Kibana appends the generated key ID to the provided name (the stored name looks like "my-name (<key-id>)").

Parameters:

  • policy_id (str) – The ID of the agent policy the Elastic Agent will be enrolled in.

  • name (str | None) – The name of the enrollment API key.

  • expiration (str | None) – Expiration timestamp for the key (for example, "2025-01-01T00:00:00.000Z").

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

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

Returns:

  • item: The created enrollment API key (id, api_key_id, api_key, name, policy_id, active, created_at)

  • action: "created"

Return type:

ObjectApiResponse containing

Raises:

Example

>>> created = await client.fleet_enrollment.create_key(
...     policy_id="agent-policy-id", name="my-enrollment-key"
... )
>>> print(created.body["item"]["api_key"])
async get_key(*, key_id, space_id=None, validate_spaces=None)[source]

Get an enrollment API key.

Gets an enrollment API key by ID. Requires the fleet-agents-all or fleet-setup privilege.

Parameters:

  • key_id (str) – The ID of the enrollment API key.

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

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

Returns:

ObjectApiResponse containing the enrollment API key under item (id, api_key_id, api_key, name, policy_id, active, created_at, hidden).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> key = await client.fleet_enrollment.get_key(key_id="key-id-1")
>>> print(key.body["item"]["policy_id"])
async delete_key(*, key_id, space_id=None, validate_spaces=None)[source]

Revoke an enrollment API key.

Revokes an enrollment API key by ID by marking it as inactive. The key is not removed, but it can no longer be used to enroll Elastic Agents. Requires the fleet-agents-all privilege.

Parameters:

  • key_id (str) – The ID of the enrollment API key.

  • space_id (str | None) – Optional space ID to revoke the enrollment API key in.

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

Returns:

ObjectApiResponse containing {"action": "deleted"} on success.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> result = await client.fleet_enrollment.delete_key(key_id="key-id-1")
>>> print(result.body["action"])
deleted
async create_service_token(*, remote=None, space_id=None, validate_spaces=None)[source]

Create a service token.

Creates a Fleet Server service token. The token is used to enroll Fleet Server instances with Kibana. Requires the fleet-agents-all privilege.

Parameters:

  • remote (bool | None) – When True, generates a token for a remote Fleet Server (service account elastic/fleet-server-remote) instead of the default elastic/fleet-server account.

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

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

Returns:

  • name: The name of the generated service token

  • value: The service token secret value

Return type:

ObjectApiResponse containing

Raises:

Example

>>> token = await client.fleet_enrollment.create_service_token()
>>> print(token.body["name"], token.body["value"])
async create_logstash_api_key(*, space_id=None, validate_spaces=None)[source]

Generate a Logstash API key.

Generates an Elasticsearch API key for Logstash to use with a Fleet Logstash output (the logstash-output permission set). Requires the fleet-settings-all privilege.

Parameters:

  • space_id (str | None) – Optional space ID to generate the API key in.

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

Returns:

ObjectApiResponse containing the generated key under api_key in the id:secret format expected by the Logstash elastic_agent input.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> key = await client.fleet_enrollment.create_logstash_api_key()
>>> print(key.body["api_key"])
async get_uninstall_tokens(*, policy_id=None, search=None, per_page=None, page=None, space_id=None, validate_spaces=None)[source]

Get metadata for the latest uninstall tokens.

Lists the metadata for the latest uninstall tokens per agent policy. Uninstall tokens are required to uninstall tamper-protected Elastic Agents. Token values are not included; use get_uninstall_token() to get a decrypted token. Requires the fleet-agents-all privilege.

Note: policy_id and search cannot be used at the same time.

Parameters:

  • policy_id (str | None) – Partial match filtering for policy IDs (max length 50).

  • search (str | None) – Partial match filtering for uninstall token values (max length 50).

  • per_page (int | None) – The number of items to return (minimum 5).

  • page (int | None) – Page number (minimum 1).

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

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

Returns:

  • items: List of token metadata (id, policy_id, policy_name, created_at, namespaces)

  • total / page / perPage: pagination metadata

Return type:

ObjectApiResponse containing

Raises:

Example

>>> tokens = await client.fleet_enrollment.get_uninstall_tokens()
>>> for token in tokens.body["items"]:
...     print(token["id"], token["policy_id"])
async get_uninstall_token(*, uninstall_token_id, space_id=None, validate_spaces=None)[source]

Get a decrypted uninstall token.

Gets one decrypted uninstall token by its ID. The returned token value can be used to uninstall a tamper-protected Elastic Agent enrolled in the token’s agent policy. Requires the fleet-agents-all privilege.

Parameters:

  • uninstall_token_id (str) – The ID of the uninstall token.

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

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

Returns:

ObjectApiResponse containing the decrypted token under item (id, policy_id, policy_name, token, created_at, namespaces).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> token = await client.fleet_enrollment.get_uninstall_token(
...     uninstall_token_id="token-id-1"
... )
>>> print(token.body["item"]["token"])
async rotate_message_signing_key_pair(*, acknowledge=None, space_id=None, validate_spaces=None)[source]

Rotate the Fleet message signing key pair.

Rotates the key pair used by Fleet to sign messages sent to Elastic Agents. This operation is irreversible and requires all agents in the Fleet to be re-enrolled after rotation. You must explicitly acknowledge the risk by passing acknowledge=True; the server rejects the request with a 400 warning otherwise. Requires the fleet-agents-all, fleet-agent-policies-all and fleet-settings-all privileges.

Parameters:

  • acknowledge (bool | None) – Set to True to confirm you understand the risks of rotating the key pair (default: False, which is rejected).

  • space_id (str | None) – Optional space ID to perform the rotation in.

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

Returns:

ObjectApiResponse containing a confirmation message ("Key pair rotated successfully.").

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> result = await client.fleet_enrollment.rotate_message_signing_key_pair(
...     acknowledge=True
... )
>>> print(result.body["message"])
Key pair rotated successfully.
async get_kubernetes_manifest(*, download=None, fleet_server=None, enrol_token=None, space_id=None, validate_spaces=None)[source]

Get a full Kubernetes agent manifest.

Gets the Kubernetes manifest for deploying Elastic Agent as a DaemonSet, as a JSON object with the YAML document under item. Requires the fleet-agent-policies-read or fleet-setup privilege.

Parameters:

  • download (bool | None) – If True, returns the manifest as a downloadable file.

  • fleet_server (str | None) – Fleet Server host URL to include in the manifest (substituted for the FLEET_URL placeholder).

  • enrol_token (str | None) – Enrollment token to include in the manifest (substituted for the FLEET_ENROLLMENT_TOKEN placeholder).

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

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

Returns:

ObjectApiResponse containing the manifest YAML string under item.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> manifest = await client.fleet_enrollment.get_kubernetes_manifest(
...     fleet_server="https://fleet.example.com:8220",
...     enrol_token="my-enrollment-token",
... )
>>> print(manifest.body["item"][:22])
async download_kubernetes_manifest(*, download=None, fleet_server=None, enrol_token=None, space_id=None, validate_spaces=None)[source]

Download a Kubernetes agent manifest.

Downloads the Kubernetes manifest for deploying Elastic Agent as a raw YAML document (content type text/x-yaml); the response body is the YAML string itself. Requires the fleet-agent-policies-read or fleet-setup privilege.

Parameters:

  • download (bool | None) – If True, returns the manifest as a downloadable file.

  • fleet_server (str | None) – Fleet Server host URL to include in the manifest (substituted for the FLEET_URL placeholder).

  • enrol_token (str | None) – Enrollment token to include in the manifest (substituted for the FLEET_ENROLLMENT_TOKEN placeholder).

  • space_id (str | None) – Optional space ID to download the manifest from.

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

Returns:

Response whose body is the manifest YAML document as a string.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> manifest = await client.fleet_enrollment.download_kubernetes_manifest()
>>> yaml_text = manifest.body
>>> print(yaml_text.splitlines()[0])
---
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]