An xApp is an application running on the Near-RT RIC that observes and influences the RAN through the E2 interface. This tutorial builds one — a simple traffic-steering xApp that subscribes to KPM (Key Performance Measurement) reports, identifies overloaded cells, and issues a control action to move UEs.

We use the O-RAN Software Community (O-RAN SC) Near-RT RIC, J-release. The patterns apply to other RICs (Mavenir, Juniper, AT&T's RIC platform) with naming differences.

Prerequisites

What you need running:

O-RAN SC Near-RT RIC, deployed via the official it/dep Helm charts on a small Kubernetes cluster (3 nodes is enough).
An E2 simulator (e2sim from O-RAN SC) or a real DU with E2 termination.
Python 3.11+ for the xApp code, or C++ if you want lower latency.

Budget a day to stand up the RIC. The first deployment is rarely the fast one.

E2 service models you will use

E2 is the interface between RIC and RAN. On top of E2AP (the application protocol), service models define what you can monitor and control:

Service Model	Purpose
E2SM-KPM	Key Performance Measurement — pull metrics from RAN
E2SM-RC	RAN Control — issue control actions (handover, QoS, slicing)
E2SM-CCC	Cell Configuration and Control
E2SM-NI	Network Interface — observe X2/Xn/F1/E1 messages

For traffic steering: subscribe via KPM (read load), act via RC (move UEs).

xApp structure

Using the Python xApp framework (ricxappframe):

traffic-steering-xapp/
  src/
    main.py
    handler.py
    config.py
  config/
    config-file.json       # xApp descriptor
    schema.json
  helm/
    Chart.yaml
    values.yaml
    templates/
  Dockerfile
  requirements.txt

Step 1: the xApp descriptor

config-file.json tells the App Manager what your xApp does, what messages it sends and receives, and what its config schema is.

{

"xapp_name": "traffic-steering",

"version": "0.1.0",

"containers": [

{

"name": "traffic-steering",

"image": {

"registry": "registry.example.com",

"name": "traffic-steering-xapp",

"tag": "0.1.0"

}

"messaging": {

"ports": [

{ "name": "http", "container": "traffic-steering", "port": 8080, "description": "http service" },

{ "name": "rmr-data", "container": "traffic-steering", "port": 4560, "rmrRoutingKey": "true" },

{ "name": "rmr-route", "container": "traffic-steering", "port": 4561, "description": "rmr route port" }

"rmrMessageTypes": {

"sends": [ "RIC_SUB_REQ", "RIC_CONTROL_REQ" ],

"receives": [ "RIC_SUB_RESP", "RIC_INDICATION", "RIC_CONTROL_ACK" ]

}

"controls": { "$ref": "#/schema" }

}


RMR is the messaging library used inside the RIC. Messages are routed by message type, not by destination. The `rmrMessageTypes` block declares which types your xApp sends and receives — the RIC builds the routing table from this.

## Step 2: the xApp logic

python

from ricxappframe.xapp_frame import RMRXapp, rmr

from ricxappframe.xapp_sdl import SDLWrapper

import json, logging

log = logging.getLogger("traffic-steering")

class TrafficSteeringXapp:

def __init__(self):

self.rmr_xapp = RMRXapp(

self._default_handler,

config_handler=self._config_handler,

rmr_port=4560,

post_init=self._post_init,

use_fake_sdl=False,

)

self.rmr_xapp.register_callback(self._handle_indication, 12050) # RIC_INDICATION

self.rmr_xapp.register_callback(self._handle_sub_resp, 12011) # RIC_SUB_RESP

self.sdl = SDLWrapper(use_fake_sdl=False)

def _post_init(self, rmr_xapp):

log.info("sending E2 subscription request")

sub_req = self._build_kpm_subscription()

rmr_xapp.rmr_send(sub_req, 12010) # RIC_SUB_REQ

def _build_kpm_subscription(self):

# ASN.1-encoded E2SM-KPM subscription:

# report on DRB.UEThpDl, RRU.PrbUsedDl every 1000ms, all cells

return encode_kpm_subscription(

ran_function_id=2, # KPM

event_trigger={"period_ms": 1000},

actions=[

{

"action_id": 1,

"action_type": "report",

"measurements": [

"DRB.UEThpDl",

"RRU.PrbUsedDl",

"DRB.RlcSduTransmittedVolumeDL",

}

)

def _handle_indication(self, rmr_xapp, summary, sbuf):

payload = json.loads(summary["payload"])

cell_id = payload["cell_id"]

prb_used = payload["RRU.PrbUsedDl"]

if prb_used > 80:

log.warning(f"cell {cell_id} overloaded at {prb_used}% PRB")

self._issue_handover(cell_id, payload["ue_list"])

rmr.rmr_free_msg(sbuf)

def _issue_handover(self, source_cell_id, ue_list):

target_cell_id = self._select_target_cell(source_cell_id)

if not target_cell_id:

return

for ue_id in ue_list[:5]: # move at most 5 UEs at a time

ctrl = encode_rc_control(

ran_function_id=3, # RC

style=3, # connected mode mobility

action="handover",

ue_id=ue_id,

target_cell=target_cell_id,

)

self.rmr_xapp.rmr_send(ctrl, 12040) # RIC_CONTROL_REQ

def _select_target_cell(self, source):

neighbors = self.sdl.get("ts-neighbors", source) or []

for n in neighbors:

load = self.sdl.get("cell-load", n)

if load and load < 60:

return n

return None

def run(self):

self.rmr_xapp.run()

if __name__ == "__main__":

TrafficSteeringXapp().run()


The ASN.1 encoding helpers (`encode_kpm_subscription`, `encode_rc_control`) wrap the painful parts. Use the O-RAN SC libraries; do not hand-encode ASN.1 unless you enjoy suffering.

## Step 3: shared data layer (SDL)

The RIC platform provides SDL — Redis underneath, with a wrapper. Use it for:

- xApp state that needs to survive restart
- Data shared between xApps (e.g., UE-level data published by a UE Manager xApp, consumed by yours)
- Cell topology (neighbor lists)

Do not use SDL as a metrics store. Push metrics to Prometheus.

## Step 4: containerize

dockerfile

FROM python:3.11-slim

RUN apt-get update && apt-get install -y libssl-dev && rm -rf /var/lib/apt/lists/*

COPY requirements.txt /app/

RUN pip install --no-cache-dir -r /app/requirements.txt

COPY src/ /app/src/

COPY config/ /app/config/

WORKDIR /app

ENV CONFIG_FILE=/app/config/config-file.json

CMD ["python", "-u", "src/main.py"]


## Step 5: Helm chart for deployment

The O-RAN SC RIC deploys xApps as Kubernetes Deployments under the `ricxapp` namespace, with specific labels and ConfigMaps the App Manager expects.

yaml

apiVersion: apps/v1

kind: Deployment

metadata:

name: traffic-steering

namespace: ricxapp

labels:

app: ricxapp-traffic-steering

spec:

replicas: 1

selector:

matchLabels: { app: ricxapp-traffic-steering }

template:

metadata:

labels: { app: ricxapp-traffic-steering }

annotations:

prometheus.io/scrape: "true"

prometheus.io/port: "8080"

spec:

containers:

- name: traffic-steering

image: registry.example.com/traffic-steering-xapp:0.1.0

ports:

- { name: http, containerPort: 8080 }

- { name: rmr-data, containerPort: 4560 }

- { name: rmr-route, containerPort: 4561 }

envFrom:

- configMapRef: { name: configmap-ricxapp-traffic-steering-appconfig }


Deploy via the App Manager (`appmgr`) API rather than `kubectl apply` directly — App Manager handles RMR route table updates:

bash

curl -X POST http://appmgr-service:8080/ric/v1/xapps \

-H "Content-Type: application/json" \

-d @config-file.json

Step 6: validate

Watch xApp logs: kubectl logs -n ricxapp -l app=ricxapp-traffic-steering -f
Verify subscription: RIC_SUB_RESP should arrive within seconds.
Confirm KPM indications flowing: log lines per second matching the report period.
With e2sim, inject a high-PRB scenario and check that a RIC_CONTROL_REQ is emitted.

Common failures

No subscription response. Usually E2 termination cannot route to the E2 node, or the RAN function ID does not match what the simulator advertises. Check e2term logs.
Subscription response says "no matching action." Your event trigger or measurement names do not match the RAN's advertised KPM service model version. Pin the version.
Control request rejected. RIC_CONTROL_FAILURE with a cause code. Most common: the target cell is not in the same RAN function, or the UE ID format is wrong (gNB UE NGAP ID vs CU UE F1AP ID — they are not interchangeable).
xApp not receiving messages despite RMR routes. RMR routing relies on the App Manager updating the route table. Restart the App Manager subscription manager.

Where to go from here

Add an A1 policy interface so the SMO can configure thresholds without redeploying.
Persist learned neighbor relations in SDL.
Replace the hardcoded threshold with a model — published via AI/ML framework.
Write a Conflict Mitigation hook so two xApps do not issue conflicting handovers.

> The first xApp is mostly plumbing. The interesting engineering starts at the second xApp, when you discover what coordination between them costs.

Build one end-to-end before reading any more O-RAN architecture documents — the architecture only makes sense once you have shipped a message through it.

Building Your First O-RAN xApp: A Step-by-Step Tutorial