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.
Beyond the tutorial
O-RAN RIC and xApps covers the architecture this code
sits in. For the part a tutorial cannot teach — where inference can run given a
control-loop deadline, and what each placement costs — the
AI-RAN specialist track works it with
numbers rather than reference designs.