airpuls-ric-sdk

Python xApp SDK for the airpuls Near-RT RIC — a clean-room, spec-compliant O-RAN E2 implementation.

The package wraps the native libric-client xApp client library in a Pythonic API. The binary wheel is self-contained: the native client, the E2SM codecs, and their shared-library dependencies are bundled, so pip install needs no compiler and no system packages.

Supported platform

• OS / architecture: Linux x86_64, glibc ≥ 2.34 (Ubuntu 22.04+, Debian 12+, RHEL/Alma/Rocky 9+)
• Python: CPython ≥ 3.9 (one abi3 wheel covers all versions)

Other platforms are not supported at this time.

Install

pip install airpuls-ric-sdk

The distribution is named airpuls-ric-sdk; the import stays

import airpuls_ric_sdk

What you get

• E2 node discovery — node available/lost/capabilities-changed callbacks, RAN Function inspection by ID or OID.
• Six service models — KPM, RC, LLC, AIR, CCC, and EPI, each with typed subscription builders and indication decoding.
• Subscribe / Indication / Control / Query — periodic and event-triggered REPORT subscriptions, INSERT indications with CONTROL answers, synchronous and asynchronous RIC Control, RC QUERY.
• BaseXApp — batteries-included xApp base class: connect, reconnect with backoff, SM registration, signal handling, event loop (blocking run_sync() or asyncio run()).
• Telemetry sinks — declarative fan-out of measurements and events to Redis and InfluxDB, configured from xapp.yml.

Quick start

An xApp reads its parameters from a YAML config file:

# my-xapp.yml
ric:
  endpoint: "tcp://192.168.2.96:36422"   # xApp IPC endpoint of the RIC

xapp:
  deployment_name: "my-xapp"

from airpuls_ric_sdk import kpm_plugin_get, KPM_RAN_FUNC_ID
from airpuls_ric_sdk.xapp import BaseXApp, setup_logging

class MyXApp(BaseXApp):
    XAPP_TYPE = "my-xapp"
    XAPP_VERSION = "1.0.0"

    def plugins(self):
        return [kpm_plugin_get()]

    def on_e2_node_available(self, client, node):
        # Build and send a KPM subscription here.
        ...

    def on_indication(self, client, sub_id, node, ran_func_id, header, data):
        if ran_func_id != KPM_RAN_FUNC_ID:
            return
        for block in data.blocks:
            for m in block.measurements:
                print(f"{m.name} = {m.value}")

if __name__ == "__main__":
    setup_logging()
    app = MyXApp("my-xapp.yml")
    app.run_sync()

Lower-level entry points (RicClient, RicXappConfig, subscription builders, sync/async control) are exported from airpuls_ric_sdk directly for xApps that integrate the client into their own event loop via poll_fd + dispatch().

License

Proprietary (LicenseRef-CSSL-1.0). © airpuls GmbH.