PyCentral provides native support for Streaming APIs, allowing consumption of real-time event data using a simple, programmatic interface. The SDK manages authentication, connection lifecycle, and message decoding enabling automated processing of streamed events.

📘

Version Availability

Streaming API support is available starting from SDK version v2.0a13 .

Features

• Native WebSocket support for Streaming APIs
• Automated management of authentication tokens
• Handling of long-lived streaming connections
• Event-based streaming support for filters
• Decoded event payloads ready for application-level processing

Setup

No additional configuration is required beyond what is already used for REST APIs in the SDK. Please reference PyCentral's prerequisites guide for more information on initial setup.

Decoding a bytestream from Central's Streaming API requires a protocol buffer file. Each Streaming API event has an associated Protocol Buffer definition file. PyCentral includes the required decoder files internally. Users do not need to manage proto files or decoding logic manually. Proto files can be found in the Supported Events section of the CloudEvents guide.

Basic Usage

Streaming APIs are initialized using the SDK's streaming module by specifying an event and providing valid credentials.
Once connected, PyCentral automatically decodes incoming messages and delivers them as structured event payloads. By default, the SDK print's the decoded messages on the terminal.

Usage steps:

1. Initialize the SDK with credentials and base URL
2. Specify the Streaming API event
3. Start the stream

If the connection is successful, events for the selected event are streamed continuously and made available for processing.

Examples

Authentication Configuration (token.yaml)

Both examples below require a token.yaml file that contains the credentials needed to authenticate with Central.

new_central:
  base_url: <api-base-url>
  client_id: <client-id>
  client_secret: <client-secret>

📘

Central Credential File

This file is reused across all streaming examples below.
Refer to the PyCentral QuickStart Guide for details on generating and populating these values.

Example 1: Basic Streaming

This example opens a streaming connection to the Audit Trail event stream and continuously prints decoded events.

Audit Trail Events is used because it is one of the easiest streams to validate a successful connection.

from pycentral import NewCentralBase
from pycentral.streaming import Streaming

central_conn = NewCentralBase(
    token_info="token.yaml",
)

# Create a streaming instance for audit event
# The event parameter specifies which event stream to subscribe to
audit_stream = Streaming(
    central_conn=central_conn,
    event="audit-trail-events",  # Subscribe to audit trail events stream
)

# Start streaming events - this will continuously listen for and process events
audit_stream.stream()

On successful execution the decoded Audit Trail events will output to the terminal as they are generated from Central.

To stream a different event, update the event parameter with the desired streaming endpoint. Available events can be found in the Supported Events section of CloudEvents Guide.

Example 2: Streaming with Filters

This example builds on the previous one by applying an event-type filter. Only matching events are streamed and printed.

from pycentral import NewCentralBase
from pycentral.streaming import Streaming

central_conn = NewCentralBase(
    token_info="token.yaml",
)

# Create a streaming instance for audit event
# The event parameter specifies which event stream to subscribe to
audit_stream = Streaming(
    central_conn=central_conn,
    event="audit-trail-events",  # Subscribe to audit trail events stream
    filters="com.hpe.greenlake.network-services.v1alpha.audit-trail.floorplan-manager", # Optional: Apply filters to get specific events
)

# Start streaming events - this will continuously listen for and process events
audit_stream.stream()

In the above example, the filter com.hpe.greenlake.network-services.v1alpha.audit-trail.floorplan-manager corresponds to Floorplan Manager, so only Audit Events related to Floorplan manager is send to this Streaming API connection.

This example applies event-type filtering using the filters parameter of the Streaming instance.
Options include:

• A single event type as a string
• Multiple event types as a list of strings

Supported filters is event-specific. Supported filters for an event is documented in the specific event's page in the Supported Events section of CloudEvents Guide.

Event Processing

All streamed messages are delivered as decoded event objects by the Streaming module based on the built-in event Protocol Buffer schema. The data is ready to be consumed in a human readable format.

Common uses include:

• Inspect event attributes and metadata
• Apply custom logic or transformations
• Persist events to external systems
• Trigger downstream automation or workflows

The SDK does not restrict how events are handled, allowing full flexibility for different use cases.