Skip to content

Building xApps

Every xApp follows the same lifecycle: connect, register SM plugins, subscribe when a node appears, handle indications, act, shut down cleanly. This page covers that skeleton and the callback surface in Python, Go, and C. For what to subscribe to on each SM, see the Service Model Guides.

Enforcing A1 policies?

If your xApp exists to enforce O-RAN A1 policies, use the higher-level A1 Policy Enforcement surface instead of wiring the lifecycle by hand — you implement an A1Enforcer and the SDK owns the policy seam, the subscription lifecycle, and status reporting.

The minimal xApp

BaseXApp eliminates the lifecycle boilerplate. Subclass it, declare your plugins, override the events you care about, and run:

from airpuls_ric_sdk import kpm_plugin_get, KpmSubscriptionBuilder, KpmStyle, KpmConditionExpr
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):
        builder = KpmSubscriptionBuilder(KpmStyle.STYLE_4)
        builder.set_report_period(1000)
        builder.add_measurement("DRB.UEThpDl")
        builder.add_condition_snssai_sst(KpmConditionExpr.EQUAL, 1)
        client.subscribe_kpm(node, builder)

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

if __name__ == "__main__":
    setup_logging()
    MyXApp(config_path="/etc/airpuls/xapp.yml").run_sync()

Implement ric.EventHandler (or embed ric.DefaultEventHandler and override only what you need), then drive client.Run():

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/airpuls/ric-client-go/ric"
)

type myHandler struct{ ric.DefaultEventHandler }

func (h *myHandler) OnE2NodeAvailable(client *ric.Client, node *ric.E2Node) {
    b := ric.NewKpmSubscriptionBuilder(ric.KpmStyle4)
    b.SetReportPeriod(1000).
        AddMeasurement("DRB.UEThpDl").
        AddConditionSnssaiSST(ric.KpmExprEqual, 1)
    if _, err := client.SubscribeKPM(node, b); err != nil {
        log.Printf("subscribe failed: %v", err)
    }
}

func (h *myHandler) OnIndication(client *ric.Client, subID uint32,
    node *ric.E2Node, ranFuncID uint16,
    header *ric.IndicationHeader, data *ric.IndicationData) {
    for _, block := range data.Blocks() {
        for _, m := range block.Measurements() {
            fmt.Printf("%s = %d\n", m.Name(), m.Integer())
        }
    }
}

func main() {
    cfg := ric.NewXappConfig("/etc/airpuls/xapp.yml")
    defer cfg.Close()
    if err := cfg.Load(); err != nil {
        log.Fatal(err)
    }
    client, err := ric.Connect(cfg)
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    client.RegisterSM(ric.KpmPluginGet())
    client.SetEventHandler(&myHandler{})

    ctx, cancel := signal.NotifyContext(context.Background(),
        os.Interrupt, syscall.SIGTERM)
    defer cancel()
    go func() { <-ctx.Done(); client.InitiateShutdown() }()

    if err := client.Run(); err != nil {
        log.Fatal(err)
    }
}

The C surface is the primitive every binding wraps: create a config, connect, register plugins, install callbacks, run.

#include "ric-client.h"
#include "xapp-config.h"
#include "e2sm-kpm.h"

static void on_node(ric_client_t *client, const ric_e2_node_t *node, void *ud) {
    kpm_subscription_builder_t *b = kpm_subscription_builder_new(KPM_STYLE_4);
    kpm_sub_set_report_period(b, 1000);
    kpm_sub_add_measurement(b, "DRB.UEThpDl");
    kpm_sub_add_condition_snssai_sst(b, KPM_EXPR_EQUAL, 1);
    uint32_t sub_id = 0;
    ric_client_subscribe_kpm(client, node, b, &sub_id);   /* consumes b */
}

static void on_ind(ric_client_t *client, uint32_t sub_id,
                   const ric_e2_node_t *node, uint16_t ran_func_id,
                   const sm_indication_header_t *hdr,
                   const sm_indication_data_t *data, void *ud) {
    for (size_t bi = 0; bi < sm_indication_data_block_count(data); bi++) {
        const sm_measurement_block_t *block = sm_indication_data_block_at(data, bi);
        for (size_t j = 0; j < sm_measurement_block_count(block); j++) {
            const sm_measurement_t *m = sm_measurement_block_at(block, j);
            printf("%s = %lld\n", sm_measurement_name(m),
                   (long long)sm_measurement_integer(m));
        }
    }
}

int main(void) {
    ric_xapp_config_t *cfg = ric_xapp_config_new("/etc/airpuls/xapp.yml");
    ric_xapp_config_load(cfg);

    ric_client_t *client = ric_client_connect(cfg);
    ric_client_register_sm(client, kpm_plugin_get());
    ric_client_set_on_e2_node_available(client, on_node, NULL);
    ric_client_set_on_indication(client, on_ind, NULL);
    ric_client_run(client);

    ric_client_destroy(client);
    ric_xapp_config_free(cfg);
    return 0;
}

What BaseXApp handles for you (Python)

Subclassing BaseXApp gives you, for free: loading + validating xapp.yml, the RIC connection from ric.*, SM plugin registration from plugins(), callback wiring to overridable methods, SIGINT/SIGTERM handlers, the event loop, graceful shutdown, and optional auto-stop after N indications (xapp.max_indications). You override only the events you need.

The callback surface

The events every binding delivers (Python BaseXApp method / Go EventHandler method / C ric_client_set_* callback):

Event When called
on_connected / OnConnected After initial connect or a reconnect
on_disconnected / OnDisconnected Transport dropped
on_e2_node_available / OnE2NodeAvailable An E2 node completed E2 Setup
on_e2_node_lost / OnE2NodeLost An E2 node disconnected
on_e2_node_capabilities_changed / OnE2NodeCapabilitiesChanged A node's RAN function list changed
on_indication / OnIndication A REPORT indication arrived
on_insert_indication / OnInsertIndication An INSERT indication arrived — answer it with a CONTROL echoing the Call Process ID (E2SM-RC)
on_subscription_deleted / OnSubscriptionDeleted The server tore down a subscription
on_notification / OnNotification A RIC advisory notification (e.g. MEASUREMENT_NOT_ADVERTISED)
on_control_result_async / OnControlResultAsync A CONTROL issued with control_async completed
on_ping_result / OnPingResult A periodic ping completed (enabled by xapp.ping_interval_ms)

Python's BaseXApp adds three hooks with no wire event behind them, driven by the loop and by construction:

Hook When called
register_params(config) Once during __init__, before config.load() — register app-specific config keys here
on_tick(client) Every xapp.tick_interval_ms from run_sync (0 = never). Runs between run_iteration calls, so blocking SDK calls — control, query_rc — are safe here: their internal wait pumps the loop, so indications keep flowing. Use it for a recurring RIC Query
on_shutdown() After the loop exits

Indication callback signatures

The indication callbacks carry more than the payload: sub_id is the subscription the indication belongs to, and node is the E2 node it came from (None / nil if that node disconnected between RIC-side dispatch and delivery). Route on sub_id, not on ran_func_id — one xApp can hold several subscriptions on the same SM.

def on_indication(self, client, sub_id, node, ran_func_id, header, data): ...
def on_insert_indication(self, client, sub_id, node, ran_func_id,
                         header_bytes, message_bytes, cpi_bytes): ...
OnIndication(client *Client, subID uint32, node *E2Node,
    ranFuncID uint16, header *IndicationHeader, data *IndicationData)
OnInsertIndication(client *Client, subID uint32, node *E2Node,
    ranFuncID uint16, headerBytes, messageBytes, cpiBytes []byte)
void (*ric_client_on_indication_fn)(
    ric_client_t *client, uint32_t subscription_id,
    const ric_e2_node_t *node, uint16_t ran_func_id,
    const sm_indication_header_t *header,
    const sm_indication_data_t *data, void *user_data);

Callbacks run on the dispatch thread; handles are borrowed

All callbacks fire on the thread that called run() — so your code needs no locking. The node / header / data handles are valid only for the callback duration; the SDK frees them on return. Copy out anything you need to keep.

The INSERT callback is the exception: its three byte buffers are heap-owned copies, safe to retain past the return.

Reading indication data

REPORT indications from ASN.1 SMs (KPM, RC, LLC Style 2) lower into a generic measurement-block tree: blocks (one per UE or cell), each holding named measurements.

def on_indication(self, client, sub_id, node, ran_func_id, header, data):
    # header.timestamp_ms — indication timestamp
    for block in data.blocks:
        ue = block.ue_id.display        # e.g. "ue:amf_ue_ngap_id:12345"
        for m in block.measurements:
            # m.name  e.g. "DRB.UEThpDl"
            # m.type  one of the seven SmMeasurementType kinds below
            # m.value int | float | bool | bytes | str | None
            print(ue, m.name, m.value)

A measurement's value is a tagged union — m.type discriminates, and it covers every primitive branch of the ASN.1 RANParameter-Value, not just numbers:

SmMeasurementType Python m.value Typed accessor
NO_VALUE None
BOOL bool m.integer (0/1)
INTEGER int m.integer
REAL float m.real
BIT_STRING bytes m.bits, m.bit_length
OCTET_STRING bytes m.octet
PRINTABLE_STRING str m.string

Calling the wrong accessor returns a zero, not an error

m.integer on an OCTET_STRING measurement returns 0 — it does not raise. A loop that assumes integers reads plausible-looking zeros off every non-numeric measurement, which is why E2SM-RC's RRC_Message looks like a decode bug when read that way. Prefer m.value, which dispatches on m.type; reach for a typed accessor only when you already know the kind.

func (h *myHandler) OnIndication(client *ric.Client, subID uint32,
    node *ric.E2Node, ranFuncID uint16,
    header *ric.IndicationHeader, data *ric.IndicationData) {
    for _, block := range data.Blocks() {
        for _, m := range block.Measurements() {
            switch m.Type() {
            case ric.MeasTypeInteger:
                fmt.Printf("%s = %d\n", m.Name(), m.Integer())
            case ric.MeasTypeReal:
                fmt.Printf("%s = %.2f\n", m.Name(), m.Real())
            }
        }
    }
}
for (size_t bi = 0; bi < sm_indication_data_block_count(data); bi++) {
    const sm_measurement_block_t *block = sm_indication_data_block_at(data, bi);
    const char *ue = sm_ue_id_display(sm_measurement_block_ue_id(block));
    for (size_t j = 0; j < sm_measurement_block_count(block); j++) {
        const sm_measurement_t *m = sm_measurement_block_at(block, j);
        if (sm_measurement_type(m) == SM_MEAS_TYPE_INTEGER)
            printf("%s %s = %lld\n", ue, sm_measurement_name(m),
                   (long long)sm_measurement_integer(m));
    }
}

Shutdown

BaseXApp installs SIGINT/SIGTERM handlers and drives a graceful shutdown — you generally do nothing. Override on_shutdown() to run cleanup after the loop exits.

Go's signal handling integrates cleanly — no polling loop needed:

ctx, cancel := signal.NotifyContext(context.Background(),
    os.Interrupt, syscall.SIGTERM)
defer cancel()
go func() { <-ctx.Done(); client.InitiateShutdown() }()
err := client.Run()

Call ric_client_initiate_shutdown(client) from your signal handler; ric_client_run returns, then ric_client_destroy frees the client.

Measuring RTT to the RIC

Periodic pings are driven by xapp.ping_interval_ms. Observe the result to alert on latency regressions.

class MyXApp(BaseXApp):
    def on_ping_result(self, client, rtt_ms):
        if rtt_ms > 50:
            log.warning("RTT degradation: %.3f ms", rtt_ms)

# Or one-shot / low-level:
rtt = client.ping(timeout_ms=5000)
client.set_ping_interval(1000)
client.set_on_ping_result(lambda c, rtt: print(f"RTT: {rtt:.3f} ms"))
func (h *myHandler) OnPingResult(client *ric.Client, rttMs float64) {
    if rttMs > 50 {
        log.Printf("RTT degradation: %.3f ms", rttMs)
    }
}
double rtt_ms = 0.0;
ric_client_ping(client, /*timeout_ms=*/5000, &rtt_ms);
/* Or install a periodic callback via ric_client_set_on_ping_result. */

EPI probes give a richer latency breakdown

A ping measures round-trip to the RIC. The default-on E2SM-EPI probe driver measures full-trip RTT all the way to the agent and back, decomposed by leg — with no code in your xApp.

Managing subscriptions

Every subscribe_<sm> returns a subscription id. Keep it: it is how you route indications (sub_id on the callback) and how you tear the subscription down.

sub_id = client.subscribe_kpm(node, builder)
self._subs[sub_id] = node.display
...
client.unsubscribe(sub_id)      # idempotent
subID, err := client.SubscribeKPM(node, b)
...
err = client.Unsubscribe(subID)
uint32_t sub_id = 0;
ric_client_subscribe_kpm(client, node, b, &sub_id);
...
ric_client_unsubscribe(client, sub_id);

unsubscribe is synchronous — it returns when the UnsubscribeAck arrives — and idempotent: calling it for an id the RIC already tore down (one you saw on on_subscription_deleted) returns normally.

An xApp that exits without unsubscribing is not leaking: the RIC removes every subscription an xApp held when its connection drops. Unsubscribe when the subscription should end but the xApp should keep running — a policy was deleted, a node went out of scope, a one-shot read completed.

A subscription that sets no period picks up the configured default

When the deployment sets xapp.default_report_period_ms and the builder carries no period of its own, that value is used — for KPM on every REPORT style, for LLC and AIR on Style 2 only, which are the periodic ones. A builder that sets its own period always keeps it. See Configuration → The default reporting period.

Re-subscribe on on_e2_node_available, not once at startup

on_e2_node_available fires for every node that completes E2 Setup — including after a node reconnects, and after the xApp reconnects to the RIC. Subscribing there (rather than once, at startup) is what makes an xApp survive both. Drop the per-node state in on_e2_node_lost.

Handling subscribe failures

A failed subscribe carries a structured cause so an xApp can tell a transient failure (retry) from a permanent one (disable). Read it after the failure return: Python enriches RicError with .origin, .cause_category, .cause_value, .reason; Go returns a *SubscribeError with the same fields; C exposes ric_client_last_subscribe_result(client). The origin selects how to read the cause pair — a RIC-local rejection (bad request, timeout) never collides with an E2AP cause from the node.

Issuing control

CONTROL comes in three shapes, and which one you use depends on the SM:

Shape Call SMs
Typed builder control_rc(node, builder, timeout_ms) RC — the builder encodes the E2SM payload
JSON control_ccc(node, header, message) CCC — native dicts / maps, carried verbatim
Generic bytes control(node, ran_func_id, request, timeout_ms) any SM, including EPI: you supply the encoded header / message

All three are synchronous: they block the dispatch thread until the node's ControlAcknowledge or ControlFailure arrives, or the per-request timeout expires (default 5 s; there is no infinite wait). Blocking is safe from a callback or from on_tick — the internal wait pumps the event loop, so indications keep flowing.

A missing Control Outcome is success

The RIC Control Outcome IE is OPTIONAL in E2AP. control_rc returns None and control returns an outcome with had_outcome = False when the node acknowledged without one — that is an accepted control, not an error. Always NULL-check before reading the outcome.

Not blocking on the ack

control_async submits the request and returns immediately with a wire request id; the outcome is delivered later to on_control_result_async / OnControlResultAsync, correlated by that id. Use it when you issue many controls at once — the EPI probe driver and the benchmark xApps do — and read pending_count() to see how many are outstanding.

The number of simultaneously pending requests is capped by xapp.max_pending_requests in xapp.yml (default 16384; 0 disables the cap). A fire attempted past the cap is rejected locally with E_OVERLOADED (Python) / EOverloaded (Go) without reaching the wire — back off and retry once outstanding requests complete. set_max_pending_requests() / SetMaxPendingRequests() retunes the cap at runtime, and pending_overflows_total() / PendingOverflowsTotal() counts rejected fires, so the cap can be monitored as a backpressure signal alongside pending_count().

The inbound side has its own bound: xapp.max_pending_indications (default 1024; 0 disables) caps how many decoded REPORT indications may wait for the dispatch loop at once. When the xApp processes indications slower than they arrive, an indication arriving at the cap is dropped before its decode instead of growing the queue without limit — INSERT indications and control-plane events are never dropped. set_max_pending_indications() / SetMaxPendingIndications() retunes it at runtime, and indications_dropped_total() / IndicationsDroppedTotal() counts the drops: a non-zero value means the xApp is not keeping up with its subscription rate.

request_id = client.control_async(node, ran_func_id,
                                  header_bytes, message_bytes,
                                  ack_required=True)

class MyXApp(BaseXApp):
    def on_control_result_async(self, client, result):
        # result.request_id, result.status, result.outcome_bytes,
        # result.epi_timing (EPI only)
        ...
reqID, err := client.ControlAsync(node, ranFuncID,
    headerBytes, messageBytes, nil /*callProcessID*/,
    true /*ackRequired*/, 0 /*timeoutMs*/)

func (h *myHandler) OnControlResultAsync(client *ric.Client,
    requestID uint32, outcome ric.ControlOutcome) { ... }
uint32_t request_id = 0;
ric_client_control_async(client, node, ran_func_id,
                         hdr, hdr_len, msg, msg_len,
                         /*cpid=*/NULL, 0,
                         /*ack_required=*/true, /*timeout_ms=*/0,
                         &request_id);
/* Install the completion callback with
 * ric_client_set_on_control_result_async(). */

Low-level control over the event loop (Python)

For asyncio or a custom loop, use RicClient directly instead of BaseXApp:

from airpuls_ric_sdk import RicClient, RicXappConfig, kpm_plugin_get

cfg = RicXappConfig("/etc/airpuls/xapp.yml"); cfg.load()
client = RicClient(cfg)
client.register_sm(kpm_plugin_get())
client.set_on_indication(my_callback, None)

# Option A: blocking run
client.run()

# Option B: integrate with your own loop
fd = client.poll_fd            # add to selectors / asyncio
# ... call client.dispatch() when readable