Elastic Transport for Python

Common errors

Warnings

Install

Install compatibility verified last tested: 2026-05-12

Imports

• Transport

  from elastic_transport import Transport

  The core class for managing nodes and performing HTTP requests.

• RequestsHttpNode

  from elastic_transport import RequestsHttpNode

  To use 'requests' as the underlying HTTP client.

• AiohttpHttpNode

  from elastic_transport import AiohttpHttpNode

  To use 'aiohttp' as the underlying asynchronous HTTP client.

• HttpxHttpNode

  from elastic_transport import HttpxHttpNode

  To use 'httpx' as the underlying synchronous HTTP client.

• HttpxAsyncHttpNode

  from elastic_transport import HttpxAsyncHttpNode

  To use 'httpx' as the underlying asynchronous HTTP client.

• ApiError

  from elastic_transport import ApiError

  For handling API-specific errors, introduced in version 8.x.

• TransportError

  from elastic_transport import TransportError

  For handling transport-level errors (e.g., connection issues).

Quickstart stale last tested: 2026-04-24

This quickstart demonstrates how to initialize the `Transport` class, either connecting to a local Elasticsearch instance or an Elastic Cloud deployment using environment variables. It then performs a basic GET request to the cluster root and prints connection details. The `transport.close()` method should be called to close HTTP connections when the transport is no longer needed.

import os
from elastic_transport import Transport, NodeConfig

# Configure nodes - replace with your Elasticsearch host(s)
# For local testing, default to 'http://localhost:9200'
# For cloud, use CLOUD_ID and API_KEY environment variables
cloud_id = os.environ.get('ELASTIC_CLOUD_ID', '')
api_key = os.environ.get('ELASTIC_API_KEY', '')

node_configs = [
    NodeConfig("http://localhost:9200")
]

if cloud_id and api_key:
    # Use Cloud ID and API Key for Elastic Cloud
    transport = Transport(
        node_configs=[], # node_configs are derived from cloud_id
        cloud_id=cloud_id,
        api_key=(api_key,)
    )
else:
    transport = Transport(node_configs=node_configs)

try:
    # Perform a GET request to the root ('/') which often returns cluster info
    # The response is a tuple of (ApiResponseMeta, deserialized_body)
    meta, body = transport.perform_request("GET", "/")
    print("Successfully connected to Elasticsearch!")
    print(f"Cluster Name: {body.get('cluster_name')}")
    print(f"Status Code: {meta.status}")
except Exception as e:
    print(f"Failed to connect or perform request: {e}")
    print("Ensure Elasticsearch is running and accessible at the specified host(s)")
    print("Or check your ELASTIC_CLOUD_ID and ELASTIC_API_KEY environment variables.")

transport.close()