Push an A1 policy from Non-RT RIC to your xApp

0Prerequisites

• Lab 01 complete — RIC platform up with a1mediator running
• Lab 02 complete — hello-xapp deployed (we'll enhance it to handle A1 policies)

1The A1 mediator API

The A1 mediator exposes a REST API. Port-forward it to your laptop:

kubectl -n ricplt port-forward svc/service-ricplt-a1mediator-http 10000:10000 &

2Define a Policy Type

A Policy Type is a schema (defines what fields a policy has). Create a TrafficSteering policy type:

cat > policy-type-20008.json <<'EOF'
{
  "name": "TrafficSteering",
  "description": "Bias handover decisions toward specified neighbors",
  "policy_type_id": 20008,
  "create_schema": {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "type": "object",
    "properties": {
      "scope": {"type":"object","properties":{"ueId":{"type":"string"}}},
      "statement": {"type":"object","properties":{"preferred_cells":{"type":"array","items":{"type":"string"}}}}
    },
    "required": ["scope","statement"]
  }
}
EOF

curl -X PUT http://localhost:10000/a1-p/policytypes/20008   -H "Content-Type: application/json"   -d @policy-type-20008.json

3Verify the type was registered

curl http://localhost:10000/a1-p/policytypes/20008 | jq .

You should see your schema echoed back.

4Push a Policy Instance

Now create a specific instance of TrafficSteering — for a fictional UE, prefer cells "C1" and "C2":

cat > policy-instance-3001.json <<'EOF'
{
  "scope": {"ueId":"imsi-001010123456789"},
  "statement": {"preferred_cells":["C1","C2"]}
}
EOF

curl -X PUT http://localhost:10000/a1-p/policytypes/20008/policies/3001   -H "Content-Type: application/json"   -d @policy-instance-3001.json

5The A1 mediator dispatches to subscribed xApps

Internally, A1 mediator now:

1. Looks up which xApps have registered interest in policy type 20008
2. Sends an RMR message of type A1_POLICY_REQ (mtype 20010) to those xApps
3. Waits for A1_POLICY_RESP (mtype 20011) acknowledging fulfillment

6Enhance hello-xapp to handle A1

Update hello_xapp.py from Lab 02 to handle A1 policy delivery:

from ricxappframe.xapp_frame import Xapp, rmr
import json

A1_POLICY_REQ  = 20010
A1_POLICY_RESP = 20011

class HelloXapp(Xapp):
    def __init__(self):
        super().__init__(entrypoint=self.entry, rmr_port=4560, use_fake_sdl=False)

    def entry(self, sdl):
        # Register interest in policy type 20008
        sdl.set("a1xapp","policy_types",[20008])
        self.logger.info("Registered for policy type 20008")
        while True:
            for msg in self.rmr_get_messages():
                if msg.mtype == A1_POLICY_REQ:
                    payload = json.loads(msg.payload)
                    self.logger.info(f"A1 policy received: {payload}")
                    # Ack
                    resp = {"status":"OK","instance_id":payload.get("policy_instance_id")}
                    msg.mtype = A1_POLICY_RESP
                    msg.payload = json.dumps(resp).encode()
                    self.rmr_send(msg)

if __name__ == "__main__":
    HelloXapp().run()

Rebuild and redeploy as in Lab 02.

7Watch the policy land

In one terminal:

kubectl -n ricxapp logs -f deployment/deployment-ricxapp-hello-xapp

In another:

curl -X PUT http://localhost:10000/a1-p/policytypes/20008/policies/3001   -d @policy-instance-3001.json -H "Content-Type: application/json"

The xApp log should print:

A1 policy received: {'policy_type_id': 20008, 'policy_instance_id': '3001', 'scope': {...}, 'statement': {...}}