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.
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:
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.
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.
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.
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