Working against the Wazuh API in Python involves a predictable set of chores: authenticate, store the token, track its age, refresh before it expires, stay within the rate limit, and unpack JSON into something typed. wazuh-api-kit handles all of that so application code does not have to.
Before and After
A concrete example makes the difference tangible. The task: fetch all active agents across a deployment that has more than 500 of them, so pagination is required.
Without wazuh-api-kit — everything has to be written by hand:
import time
import requests
from base64 import b64encode
BASE_URL = "https://localhost:55000"
USERNAME = "wazuh"
PASSWORD = "wazuh"
TOKEN_TTL = 840
def authenticate() -> tuple[str, float]:
credentials = b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode()
response = requests.post(
url=f"{BASE_URL}/security/user/authenticate",
headers={"Authorization": f"Basic {credentials}"},
verify=False,
timeout=10,
)
response.raise_for_status()
token = response.json()["data"]["token"]
return token, time.time()
token, token_time = authenticate()
agents = []
offset = 0
page_size = 500
while True:
# Refresh token if close to expiry
if time.time() - token_time >= TOKEN_TTL:
token, token_time = authenticate()
response = requests.get(
url=f"{BASE_URL}/agents",
headers={"Authorization": f"Bearer {token}"},
params={"status": "active", "limit": page_size, "offset": offset, "wait_for_complete": True},
verify=False,
)
response.raise_for_status()
page = response.json()["data"]["affected_items"]
agents.extend(page)
offset += len(page)
if len(page) < page_size:
break
for agent in agents:
print(agent["id"], agent["name"])
That is ~40 lines before any rate limiting, 401 handling, or error parsing is added.
With wazuh-api-kit:
from wazuh_api_kit import WazuhApiConnection, WazuhApiClient
connection = WazuhApiConnection(
wazuh_api_url="https://localhost:55000",
wazuh_api_username="wazuh",
wazuh_api_password="wazuh",
wazuh_api_insecure=True,
)
client = WazuhApiClient(wazuh_api_connection=connection)
for agent in client.agents_list(status="active"):
print(agent.id, agent.name)
Authentication, token refresh, pagination, and rate limiting are handled internally. agent is a WazuhAgent dataclass, not a dict.
Two Classes, Two Levels of Abstraction
The library exposes two classes. Pick whichever fits the use case.
WazuhApiConnection is the low-level entry point. It owns the HTTP session, manages authentication and token refresh, and enforces rate limiting. Every call goes through connection.request(), which returns a WazuhResponse containing the raw data dict from the API JSON body. For paginated list endpoints, get_stream() handles pagination automatically and yields individual items.
WazuhApiClient wraps WazuhApiConnection and adds a typed layer on top. Each method maps to a specific API endpoint, and responses come back as dataclasses (WazuhAgent, WazuhApiInfo, WazuhActiveResponseCommand, etc.) rather than dictionaries. The client also carries version-constraint decorators that raise RuntimeError at call time if the connected Wazuh version does not support an endpoint, and log a warning for deprecated ones.
Both classes are in the same package, so it is straightforward to use WazuhApiConnection directly for endpoints not yet covered by WazuhApiClient.
Authentication and Token Lifecycle
On the first request, WazuhApiConnection authenticates against POST /security/user/authenticate (or GET for pre-4.4.x deployments via legacy_auth=True) and stores the resulting JWT in a requests.Session. The token TTL defaults to 840 seconds. Before every request the connection checks whether time.time() - token_creation_time >= token_ttl and refreshes the session proactively if needed.
If the API returns HTTP 401 despite a seemingly valid token — for example because the server-side TTL is shorter than the configured client-side TTL — the connection re-authenticates once and retries the request automatically. A warning is logged to help diagnose mismatched TTL configurations.
Rate Limiting
The default RateLimiter allows 12 requests per 3-second window, yielding ~240 requests/minute. Wazuh’s default ceiling is 300/minute, so this leaves headroom without changes to the server configuration.
RateLimiter uses a thread-safe sliding window (a deque protected by a threading.Lock). When the limit is reached, the calling thread sleeps for a configurable sleep_time (default: 1 s) and retries until a slot is free. A custom limiter can be passed to the constructor:
from wazuh_api_kit.util import RateLimiter
connection = WazuhApiConnection(
...,
rate_limiter=RateLimiter(max_actions=20, period=3, sleep_time=1),
)
Note that increasing beyond 300 actions/minute requires a matching change in the Wazuh server configuration.
Response Types
WazuhResponse wraps the standard Wazuh API envelope. It exposes data (the payload dict), and optional message and error fields. If the API returns a non-standard body without a data key, the full dict is stored in data.
WazuhBulkResponse is returned by endpoints that operate on multiple resources simultaneously (agent deletion, active response, group assignment, etc.). It separates results into affected_items / total_affected_items and failed_items / total_failed_items. Both lists accept an optional mapping callable at construction time, so items can be deserialized inline:
bulk: WazuhBulkResponse = client.agent_delete(agent_list=["001", "002"])
print(bulk.total_affected_items) # number of deleted agents
print(bulk.total_failed_items) # number that could not be deleted
WazuhApiError is raised for any non-200 response and carries the HTTP status code alongside the Wazuh error body.
Streaming Paginated Results
WazuhApiConnection.get_stream() iterates through any paginated list endpoint by issuing sequential GET requests with incrementing offset values until a page smaller than page_size is returned. It yields one item at a time, keeping memory usage flat regardless of result count. An optional mapping callable transforms each raw dict on the fly:
from wazuh_api_kit.model import WazuhAgent
agents = connection.get_stream(
path="/agents",
parameters={"status": "active"},
page_size=500,
mapping=lambda d: WazuhAgent(**d),
)
for agent in agents:
print(agent.id, agent.version)
WazuhApiClient.agents_list() wraps this with the WazuhAgent mapping already applied.
Model Coverage
WazuhApiClient currently covers the following endpoint groups:
- API info —
GET / - Active response —
PUT /active-response - Agents — List, Add, Delete, Get active configuration
For anything outside this set, fall through to connection.request() directly.
Installation
pip install wazuh-api-kit
Basic Usage
from wazuh_api_kit import WazuhApiConnection, WazuhApiClient
connection = WazuhApiConnection(
wazuh_api_url="https://localhost:55000",
wazuh_api_username="wazuh",
wazuh_api_password="wazuh",
wazuh_api_insecure=True, # skip TLS verification for self-signed certs
token_lifetime=840, # seconds; must not exceed server-side JWT TTL
)
client = WazuhApiClient(wazuh_api_connection=connection)
# Typed response
info = client.api_info_get()
print(info.version)
# Stream all active agents without manual pagination
for agent in client.agents_list(status="active"):
print(agent.id, agent.name)
Kontaktanfrage
Terminanfrage