UE5 UDP Listener Setup

2026-01-22 08:45 UTC gpt-5-2-thinking Open in ChatGPT ↗

If you want UDP messages coming into a UE 5.5 project with 0 external plugins, the “optimal” approach is to use UE’s built-in socket stack (engine modules), not marketplace plugins.

Below are two good setups—both no external dependencies. Pick based on how much control you want.

Option A (recommended): C++ UDP listener using UE’s `FUdpSocketReceiver`

Why this is optimal:

Uses UE’s own networking modules (Sockets, Networking)
Runs receiving on its own thread (via FUdpSocketReceiver)
Minimal code, solid performance, easy to dispatch to game thread

1) Enable modules (no plugins)

In your .Build.cs:

PublicDependencyModuleNames.AddRange(new string[]
{
    "Core", "CoreUObject", "Engine",
    "Sockets", "Networking"
});

2) Create a receiver component (cleanest architecture)

Make a UActorComponent like UUdpListenerComponent and attach it to a manager Actor (or your GameInstance subsystem).

Header (simplified):

#pragma once

#include "CoreMinimal.h"
#include "Components/ActorComponent.h"
#include "UdpListenerComponent.generated.h"

UCLASS(ClassGroup=(Custom), meta=(BlueprintSpawnableComponent))
class YOURPROJECT_API UUdpListenerComponent : public UActorComponent
{
    GENERATED_BODY()

public:
    UPROPERTY(EditAnywhere, BlueprintReadWrite, Category="UDP")
    int32 ListenPort = 7777;

    UPROPERTY(BlueprintAssignable, Category="UDP")
    FSimpleMulticastDelegate OnPacketReceived; // Replace with a custom delegate w/ payload

    virtual void BeginPlay() override;
    virtual void EndPlay(const EEndPlayReason::Type EndPlayReason) override;

private:
    class FSocket* ListenSocket = nullptr;
    TUniquePtr<class FUdpSocketReceiver> Receiver;

    void HandleUdpData(const FArrayReaderPtr& Data, const FIPv4Endpoint& Endpoint);
};

CPP (core idea):

#include "UdpListenerComponent.h"
#include "Sockets.h"
#include "SocketSubsystem.h"
#include "IPAddress.h"
#include "Common/UdpSocketBuilder.h"
#include "UdpSocketReceiver.h"
#include "HAL/RunnableThread.h"

void UUdpListenerComponent::BeginPlay()
{
    Super::BeginPlay();

    const FIPv4Endpoint Endpoint(FIPv4Address::Any, ListenPort);

    ListenSocket = FUdpSocketBuilder(TEXT("UE_UDP_ListenSocket"))
        .AsNonBlocking()
        .AsReusable()
        .BoundToEndpoint(Endpoint)
        .WithReceiveBufferSize(2 * 1024 * 1024);

    if (!ListenSocket)
    {
        UE_LOG(LogTemp, Error, TEXT("Failed to create UDP listen socket on port %d"), ListenPort);
        return;
    }

    Receiver = MakeUnique<FUdpSocketReceiver>(ListenSocket, FTimespan::FromMilliseconds(1), TEXT("UE_UDP_Receiver"));
    Receiver->OnDataReceived().BindUObject(this, &UUdpListenerComponent::HandleUdpData);
    Receiver->Start();

    UE_LOG(LogTemp, Log, TEXT("UDP listening on %s"), *Endpoint.ToString());
}

void UUdpListenerComponent::EndPlay(const EEndPlayReason::Type EndPlayReason)
{
    if (Receiver)
    {
        Receiver->Stop();
        Receiver.Reset();
    }

    if (ListenSocket)
    {
        ListenSocket->Close();
        ISocketSubsystem::Get(PLATFORM_SOCKETSUBSYSTEM)->DestroySocket(ListenSocket);
        ListenSocket = nullptr;
    }

    Super::EndPlay(EndPlayReason);
}

void UUdpListenerComponent::HandleUdpData(const FArrayReaderPtr& Data, const FIPv4Endpoint& Endpoint)
{
    // IMPORTANT: this callback is typically on the receiver thread.
    // Move work to game thread if you touch UObjects / gameplay state.

    TArray<uint8> Bytes;
    Bytes.Append(Data->GetData(), Data->Num());

    AsyncTask(ENamedThreads::GameThread, [this, Bytes = MoveTemp(Bytes), Endpoint]()
    {
        // Parse Bytes here (binary, JSON, etc.)
        // Example: log as string (only if you know it’s UTF-8)
        // FString Msg = FString(UTF8_TO_TCHAR(reinterpret_cast<const char*>(Bytes.GetData())));
        // UE_LOG(LogTemp, Log, TEXT("UDP from %s: %s"), *Endpoint.ToString(), *Msg);

        OnPacketReceived.Broadcast();
    });
}

Notes for production quality

Don’t parse heavy JSON on the game thread. If parsing is costly, do it on the receiver thread, then only dispatch the final struct to the game thread.
UDP packets can arrive out of order, duplicated, or dropped. If you need ordering, include a sequence number in your payload.
If you send strings, standardize on UTF-8 and include a terminator or a length prefix.

Option B: Manual `FSocket::RecvFrom` loop (max control)

Why you’d use this:

You want total control over threads/timing
You want to integrate with your own FRunnable or tick-based polling

Downside: more code and easier to accidentally block the game thread.

Typical pattern:

Create a socket (FUdpSocketBuilder)
In a worker thread (FRunnable), call RecvFrom in a loop
Use a lock-free queue or TQueue to pass packets to game thread

If you want, I can give you a small FRunnable receiver that pushes packets into a TQueue<TArray<uint8>>.

What I would not do (given your “0 external dependencies” goal)

Marketplace UDP plugins: unnecessary.
Blueprint-only UDP: UE doesn’t expose raw UDP receive nicely in BP without plugins or extra glue.
UDP Messaging plugin: it’s built-in, but it’s still a plugin and is designed around UE’s message bus patterns (great for some workflows, not “raw UDP listener” minimalism).

Recommended “optimal” setup summary

C++ component or subsystem that creates a UDP socket
FUdpSocketReceiver for threaded receive
Async dispatch to game thread for gameplay interaction
A well-defined packet format (binary with a header is best; JSON is fine for debugging)

If you tell me:

your sender’s packet format (raw bytes / JSON / protobuf-like / OSC), and
expected message rate (e.g. 10/sec vs 5000/sec),
…I can suggest the best parsing/queueing approach and a clean Blueprint-facing API (events + decoded structs).

If you truly want no C++ at all, then raw UDP receive inside UE5.5 in pure Blueprints is basically not available out of the box. UE exposes sockets/networking at the C++ level; Blueprint doesn’t get a “UDP socket receive” node set by default.

So you’ve got three realistic “no C++” paths:

1) Use a built-in Engine plugin (still “no external”, Blueprint-friendly)

A) UDP Messaging plugin (recommended for no-C++)

This is shipped with Unreal (not marketplace), and it gives you a way to get messages into UE without writing socket code.

How it works (high level):

UE listens on a UDP multicast/unicast endpoint configured in Project Settings.
Messages are sent/received via UE’s messaging system (it’s not “arbitrary UDP bytes”, it’s UE message transport).

Pros

No C++
No third-party dependencies
Pretty robust for “tool ↔ UE” communication in a LAN/dev setup

Cons

Your sender must speak the expected transport/message format (not just “send any bytes to port X”).
Not ideal if you need to ingest arbitrary UDP datagrams from existing hardware/software.

Setup

Enable UDP Messaging plugin:
Edit → Plugins → Messaging → UDP Messaging
Configure:
Project Settings → Plugins → UDP Messaging
- Set Unicast Endpoint (e.g. 0.0.0.0:7777 to listen on all interfaces)
- Optionally configure Multicast Endpoint if you’re using multicast.
In Blueprints, you typically interact through the Messaging framework patterns (depends on your use case; often you end up using Remote Control / Live Link / other built-in systems that internally rely on messaging).

This is the only “no-C++” route that stays inside Epic’s shipped ecosystem.

B) OSC plugin (if your sender can do OSC)

UE has an OSC plugin in-engine (again, not marketplace). If your external app can send OSC over UDP, this is a very practical no-C++ solution because it comes with Blueprint nodes for receiving OSC messages.

Pros

No C++
Works with “real UDP packets” (OSC formatted)
Mature workflows (TouchDesigner, Max/MSP, Ableton, many libs)

Cons

Your sender must use OSC (or you add a small translator outside UE)

Setup

Enable OSC plugin: Edit → Plugins → OSC
Configure OSC server port in Project Settings / OSC settings.
Use the OSC Blueprint nodes/events to receive messages.

If you can control the sender, OSC is often the cleanest “no-C++ but still UDP” option.

2) Use an external bridge app (still no UE plugin, no C++)

If you can’t enable engine plugins and you also can’t write C++, then you need a bridge that converts UDP into something Blueprints can consume natively.

Common bridges:

UDP → HTTP (UE Blueprints can do HTTP requests easily, but note: UE can’t be an HTTP server in BP; it’s mainly a client)
UDP → WebSocket (UE has WebSocket support via engine modules/plugins; BP exposure varies)
UDP → File/Shared memory (usually worse / higher latency)
UDP → Live Link (if it fits your data type; more specialized)

In practice, if you want “incoming to UE” without C++, you usually end up enabling a built-in plugin anyway (OSC, UDP Messaging, Live Link, Remote Control).

3) Accept a tiny “glue” layer (C++ or plugin) — but you said no C++

Just stating the reality: a ~100-line C++ component (like the one we discussed) is the normal minimal dependency-free solution. Without that, Blueprint has no raw UDP receive.

If you control the sender format:

✅ Use the OSC plugin and send OSC messages over UDP. This is the most straightforward “UDP into BP”.

If you want “UE-native messaging” and can adapt to it:

✅ Use UDP Messaging plugin (built-in), but be aware it’s not “generic UDP datagrams”.

Quick decision guide

“I need to receive arbitrary UDP datagrams from an existing device/app” → You will need C++ or a third-party plugin/bridge.
“I can change what I send” → Send OSC → no C++ and stays in UE ecosystem.
“I just need UE↔tool comms and can use UE’s message bus” → UDP Messaging.

If you tell me what’s sending the UDP (Python script? another game? hardware? TouchDesigner? etc.) and what the payload looks like (string/JSON/binary), I’ll point you to the best no-C++ route and the exact Project Settings + Blueprint nodes to use.

OK, now take this spec, and build a python tester that allows me to send all thes jsonrpc commands to unreal

iii. UDP Input

The application should provide a UDP Control API. This API is used to interact with and monitor the application. The format will be newline delimited JSONRPC 2.0. The specification should be respected including responding to RPC calls containing an id with an appropriate response message. Notifications missing the id field do not need to be responded to. All methods implemented by the application must be idempotent unless otherwise stated as multiple duplicate messages may be dispatched for a single intent. The application may receive messages not intended for it, to determine if a message is relevant there are two things to inspect.

The RPC method parameter. If the method is not referenced by this document then no action is needed. If verbose logging is enabled the application should log the method and id if available and mark it as ignored.

JavaScript 2. This application should respond to any message that satisfies point 1 unless the optional targets value is present in the JSONRPC params. In this case the application should check to see if its appId is listed in the targets array. The appId is supplied as part of the static config. Under some circumstances messages may be broadcast/ multicast necessitating this step.

// example UDP RPC call with targets property { “method”: “ic.prepareChangeMode”, “params”: { “nextMode”: “theme”, “targets”: [“ic1”, “ic2”] // ic3 should ignore this rpc } }

General Messages From Pixera Message Structure (excluding “jsonrpc”: “2.0” boilerplate)

Effect

{“method”: “ic.prepareChangeMode”, “params”: {“nextMode”: “theme”}}

Prepare/ Transition to Theme mode

{“method”: “ic.prepareChangeMode”, “params”: {“nextMode”: “brand”}}

Prepare/ Transition to Brand mode

{“method”: “ic.prepareChangeMode”, “params”: {“nextMode”: “slot1”}}

Prepare/ Transition to Slot1 mode

{“method”: “ic.prepareChangeMode”, “params”: {“nextMode”: “slot2”}}

Prepare/ Transition to Slot1 mode

{“method”: “ic.prepareChangeMode”, “params”: {“nextMode”: “slot3”}}

Prepare/ Transition to Slot1 mode

{“method”: “ic.prepareChangeMode”, “params”: {“nextMode”: “slot4”}}

Prepare/ Transition to Slot1 mode

{“method”: “ic.changeMode”, “params”: {“mode”: “theme”}}

Go to Theme mode when ready

General Messages From Pixera {“method”: “ic.changeMode”, “params”: {“mode”: “brand”}}

Go to Brand mode when ready

{“method”: “ic.changeMode”, “params”: {“mode”: “slot1”}}

Go to Takeover Slot1 mode when ready

{“method”: “ic.changeMode”, “params”: {“mode”: “slot2”}}

Go to Takeover Slot2 mode when ready

{“method”: “ic.changeMode”, “params”: {“mode”: “slot3”}}

Go to Takeover Slot3 mode when ready

{“method”: “ic.changeMode”, “params”: {“mode”: “slot4”}}

Go to Takeover Slot4 mode when ready

{“method”: “ic.ready”, “params”: {“mode”: }}

Sent when a level is ready to be used. Used by pixera to disable hold states and loading transitions.

{“method”: “ic.svmPrepareStart”} The SVM will start shortly {“method”: “ic.svmStart”} The SVM has started (touch should be

disabled)

{“method”: “ic.svmPrepareStop”} The SVM will stop shortly {“method”: “ic.svmStop”} The SVM has stopped (touch should be

enabled)

{“id”: , “method”: “tp.ping”} Respond to RPC with {“id”: ,

”result”:”pong”}

The software should handle all the requests mentioned above. If the software is not meant to render slot3 - 4 for example then the request should be gracefully logged and ignored.

Below is a self-contained Python “UDP JSON-RPC tester” tailored to your spec:

Sends newline-delimited JSON-RPC 2.0 over UDP
Supports requests (with id) and notifications (no id)
Optionally adds params.targets: [...]
Can listen for responses and match them back to id
Comes with built-in presets for all methods you listed (prepareChangeMode, changeMode, ready, svm*, ping)

This is a sender/tester. Unreal should be listening on host:port and (for id requests) replying to the sender’s ephemeral port.

`udp_jsonrpc_tester.py`

#!/usr/bin/env python3
"""
UDP JSON-RPC 2.0 tester (newline-delimited) for Unreal / Pixera-style control APIs.

Features:
- Sends newline-delimited JSON-RPC 2.0 messages over UDP.
- Supports requests (with id) + waits for responses.
- Supports notifications (no id) with no response expected.
- Optional params.targets filtering.
- Presets for the spec methods: ic.prepareChangeMode, ic.changeMode, ic.ready,
  ic.svmPrepareStart, ic.svmStart, ic.svmPrepareStop, ic.svmStop, tp.ping.
"""

from __future__ import annotations

import argparse
import json
import socket
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Dict, Optional, Tuple

def now_ms() -> int:
    return int(time.time() * 1000)

@dataclass
class PendingRequest:
    req_id: int
    sent_at_ms: int = field(default_factory=now_ms)
    response: Optional[Dict[str, Any]] = None
    event: threading.Event = field(default_factory=threading.Event)

class UdpJsonRpcClient:
    def __init__(
        self,
        remote_host: str,
        remote_port: int,
        bind_host: str = "0.0.0.0",
        bind_port: int = 0,
        recv_buf: int = 2 * 1024 * 1024,
        verbose: bool = False,
    ) -> None:
        self.remote_addr = (remote_host, remote_port)
        self.verbose = verbose

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, recv_buf)
        self.sock.bind((bind_host, bind_port))
        self.local_addr = self.sock.getsockname()

        self._pending: Dict[int, PendingRequest] = {}
        self._pending_lock = threading.Lock()

        self._stop = threading.Event()
        self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True)

    def start(self) -> None:
        self._rx_thread.start()
        if self.verbose:
            print(f"[info] UDP bound to {self.local_addr[0]}:{self.local_addr[1]}")
            print(f"[info] Remote is {self.remote_addr[0]}:{self.remote_addr[1]}")

    def close(self) -> None:
        self._stop.set()
        try:
            self.sock.close()
        except OSError:
            pass

    def _rx_loop(self) -> None:
        """
        Receives datagrams and splits on newline to support newline-delimited JSON.
        In practice, most senders put exactly one JSON per datagram, but we handle both.
        """
        while not self._stop.is_set():
            try:
                data, addr = self.sock.recvfrom(65535)
            except OSError:
                break

            text = data.decode("utf-8", errors="replace")
            # A datagram may contain multiple newline-delimited JSON objects
            chunks = [c for c in text.splitlines() if c.strip()]

            for chunk in chunks:
                try:
                    msg = json.loads(chunk)
                except json.JSONDecodeError:
                    if self.verbose:
                        print(f"[rx] Non-JSON from {addr}: {chunk!r}")
                    continue

                if self.verbose:
                    print(f"[rx] from {addr}: {json.dumps(msg, ensure_ascii=False)}")

                # JSON-RPC response matching: {"jsonrpc":"2.0","id":X,"result":...} or {"error":...}
                if isinstance(msg, dict) and "id" in msg and ("result" in msg or "error" in msg):
                    req_id = msg.get("id")
                    if isinstance(req_id, int):
                        with self._pending_lock:
                            pending = self._pending.get(req_id)
                        if pending:
                            pending.response = msg
                            pending.event.set()

    def send_notification(self, method: str, params: Optional[Dict[str, Any]] = None) -> None:
        payload: Dict[str, Any] = {"jsonrpc": "2.0", "method": method}
        if params is not None:
            payload["params"] = params

        wire = json.dumps(payload, ensure_ascii=False) + "\n"
        self.sock.sendto(wire.encode("utf-8"), self.remote_addr)

        if self.verbose:
            print(f"[tx] notif: {wire.strip()}")

    def send_request(
        self,
        method: str,
        params: Optional[Dict[str, Any]] = None,
        req_id: Optional[int] = None,
        timeout_s: float = 1.5,
    ) -> Tuple[int, Optional[Dict[str, Any]]]:
        if req_id is None:
            # simplistic increasing-ish id
            req_id = int(time.time() * 1000) % 2_000_000_000

        payload: Dict[str, Any] = {"jsonrpc": "2.0", "id": req_id, "method": method}
        if params is not None:
            payload["params"] = params

        pending = PendingRequest(req_id=req_id)
        with self._pending_lock:
            self._pending[req_id] = pending

        wire = json.dumps(payload, ensure_ascii=False) + "\n"
        self.sock.sendto(wire.encode("utf-8"), self.remote_addr)

        if self.verbose:
            print(f"[tx] req: {wire.strip()}")

        ok = pending.event.wait(timeout_s)

        with self._pending_lock:
            self._pending.pop(req_id, None)

        return req_id, pending.response if ok else None

def add_targets(params: Optional[Dict[str, Any]], targets: Optional[str]) -> Optional[Dict[str, Any]]:
    if targets is None:
        return params
    targets_list = [t.strip() for t in targets.split(",") if t.strip()]
    if not targets_list:
        return params

    if params is None:
        params = {}
    # spec: params.targets is an array of appIds
    params["targets"] = targets_list
    return params

# ---- Presets per your spec ----

VALID_MODES = ["theme", "brand", "slot1", "slot2", "slot3", "slot4"]

def preset_prepare_change_mode(mode: str, targets: Optional[str]) -> Tuple[str, Dict[str, Any], bool]:
    if mode not in VALID_MODES:
        raise ValueError(f"mode must be one of {VALID_MODES}")
    method = "ic.prepareChangeMode"
    params = add_targets({"nextMode": mode}, targets)
    return method, params or {"nextMode": mode}, False  # notification by default

def preset_change_mode(mode: str, targets: Optional[str]) -> Tuple[str, Dict[str, Any], bool]:
    if mode not in VALID_MODES:
        raise ValueError(f"mode must be one of {VALID_MODES}")
    method = "ic.changeMode"
    params = add_targets({"mode": mode}, targets)
    return method, params or {"mode": mode}, False  # notification by default

def preset_ready(mode: str, targets: Optional[str]) -> Tuple[str, Dict[str, Any], bool]:
    # mode-id could be anything per spec; we allow any string but suggest VALID_MODES
    method = "ic.ready"
    params = add_targets({"mode": mode}, targets)
    return method, params or {"mode": mode}, False  # notification by default

def preset_simple(method: str, targets: Optional[str]) -> Tuple[str, Optional[Dict[str, Any]], bool]:
    # SVM messages have no params
    params = add_targets(None, targets)
    return method, params, False  # notification by default

def preset_ping(targets: Optional[str]) -> Tuple[str, Optional[Dict[str, Any]], bool]:
    method = "tp.ping"
    params = add_targets(None, targets)
    return method, params, True  # request (expects response)

PRESETS = {
    "prepare": preset_prepare_change_mode,     # needs mode
    "change": preset_change_mode,              # needs mode
    "ready": preset_ready,                     # needs mode
    "svm_prepare_start": lambda targets: preset_simple("ic.svmPrepareStart", targets),
    "svm_start": lambda targets: preset_simple("ic.svmStart", targets),
    "svm_prepare_stop": lambda targets: preset_simple("ic.svmPrepareStop", targets),
    "svm_stop": lambda targets: preset_simple("ic.svmStop", targets),
    "ping": preset_ping,
}

def main() -> int:
    ap = argparse.ArgumentParser(description="UDP newline-delimited JSON-RPC 2.0 tester for Unreal.")
    ap.add_argument("--host", required=True, help="Unreal listen host/IP")
    ap.add_argument("--port", required=True, type=int, help="Unreal listen UDP port")
    ap.add_argument("--bind-host", default="0.0.0.0", help="Local bind host (default 0.0.0.0)")
    ap.add_argument("--bind-port", default=0, type=int, help="Local bind port (0 = ephemeral)")
    ap.add_argument("--timeout", default=1.5, type=float, help="Request timeout seconds (default 1.5)")
    ap.add_argument("--verbose", action="store_true", help="Verbose rx/tx logging")

    sub = ap.add_subparsers(dest="cmd", required=True)

    # presets
    sp = sub.add_parser("prepare", help="ic.prepareChangeMode (notification)")
    sp.add_argument("mode", choices=VALID_MODES)
    sp.add_argument("--targets", help="Comma-separated appIds to put in params.targets")

    sp = sub.add_parser("change", help="ic.changeMode (notification)")
    sp.add_argument("mode", choices=VALID_MODES)
    sp.add_argument("--targets", help="Comma-separated appIds to put in params.targets")

    sp = sub.add_parser("ready", help="ic.ready (notification)")
    sp.add_argument("mode", help="Mode id (e.g. theme/brand/slot1...)")
    sp.add_argument("--targets", help="Comma-separated appIds to put in params.targets")

    for name, help_text in [
        ("svm_prepare_start", "ic.svmPrepareStart (notification)"),
        ("svm_start", "ic.svmStart (notification)"),
        ("svm_prepare_stop", "ic.svmPrepareStop (notification)"),
        ("svm_stop", "ic.svmStop (notification)"),
    ]:
        sp = sub.add_parser(name, help=help_text)
        sp.add_argument("--targets", help="Comma-separated appIds to put in params.targets")

    sp = sub.add_parser("ping", help="tp.ping (request; expects result pong)")
    sp.add_argument("--targets", help="Comma-separated appIds to put in params.targets")
    sp.add_argument("--id", type=int, help="Explicit request id (optional)")

    # raw method support
    sp = sub.add_parser("raw", help="Send an arbitrary JSON-RPC method (request or notification).")
    sp.add_argument("method", help="JSON-RPC method name")
    sp.add_argument("--params", default=None, help='Params as JSON string, e.g. \'{"a":1}\'')
    sp.add_argument("--targets", help="Comma-separated appIds to put in params.targets")
    sp.add_argument("--id", type=int, default=None, help="If provided, send as request and wait for response")
    sp.add_argument("--no-wait", action="store_true", help="If set, do not wait even if id is provided")

    args = ap.parse_args()

    client = UdpJsonRpcClient(
        remote_host=args.host,
        remote_port=args.port,
        bind_host=args.bind_host,
        bind_port=args.bind_port,
        verbose=args.verbose,
    )
    client.start()

    try:
        if args.cmd in PRESETS:
            if args.cmd in ("prepare", "change", "ready"):
                method, params, is_request = PRESETS[args.cmd](args.mode, getattr(args, "targets", None))
                if is_request:
                    req_id, resp = client.send_request(method, params=params, timeout_s=args.timeout)
                    print(json.dumps({"sent_id": req_id, "response": resp}, ensure_ascii=False, indent=2))
                else:
                    client.send_notification(method, params=params)
                    print(json.dumps({"sent": {"method": method, "params": params}}, ensure_ascii=False, indent=2))

            elif args.cmd == "ping":
                method, params, _is_request = PRESETS["ping"](getattr(args, "targets", None))
                req_id, resp = client.send_request(method, params=params, req_id=getattr(args, "id", None), timeout_s=args.timeout)
                if resp is None:
                    print(json.dumps({"sent_id": req_id, "error": f"timeout after {args.timeout}s"}, ensure_ascii=False, indent=2))
                else:
                    print(json.dumps({"sent_id": req_id, "response": resp}, ensure_ascii=False, indent=2))

            else:
                # SVM messages
                method, params, is_request = PRESETS[args.cmd](getattr(args, "targets", None))
                if is_request:
                    req_id, resp = client.send_request(method, params=params, timeout_s=args.timeout)
                    print(json.dumps({"sent_id": req_id, "response": resp}, ensure_ascii=False, indent=2))
                else:
                    client.send_notification(method, params=params)
                    print(json.dumps({"sent": {"method": method, "params": params}}, ensure_ascii=False, indent=2))

        elif args.cmd == "raw":
            params = None
            if args.params is not None:
                try:
                    params = json.loads(args.params)
                except json.JSONDecodeError as e:
                    raise SystemExit(f"--params must be valid JSON. Error: {e}")

            params = add_targets(params, args.targets)

            if args.id is not None and not args.no_wait:
                req_id, resp = client.send_request(args.method, params=params, req_id=args.id, timeout_s=args.timeout)
                if resp is None:
                    print(json.dumps({"sent_id": req_id, "error": f"timeout after {args.timeout}s"}, ensure_ascii=False, indent=2))
                else:
                    print(json.dumps({"sent_id": req_id, "response": resp}, ensure_ascii=False, indent=2))
            else:
                # If id provided but --no-wait, still send as request; just don't wait
                if args.id is not None:
                    payload = {"jsonrpc": "2.0", "id": args.id, "method": args.method}
                    if params is not None:
                        payload["params"] = params
                    wire = json.dumps(payload, ensure_ascii=False) + "\n"
                    client.sock.sendto(wire.encode("utf-8"), client.remote_addr)
                    print(json.dumps({"sent": payload, "note": "sent request without waiting"}, ensure_ascii=False, indent=2))
                else:
                    client.send_notification(args.method, params=params)
                    print(json.dumps({"sent": {"method": args.method, "params": params}}, ensure_ascii=False, indent=2))

        else:
            raise SystemExit("Unknown command.")

        return 0

    finally:
        client.close()

if __name__ == "__main__":
    raise SystemExit(main())

Examples

1) Prepare for Theme mode (notification)

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 prepare theme

2) Change to Slot2 (notification)

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 change slot2

3) Mark level ready (notification)

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 ready theme

4) SVM start/stop signals (notifications)

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 svm_prepare_start
python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 svm_start
python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 svm_prepare_stop
python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 svm_stop

5) Ping as a request (expects response with `result:"pong"`)

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 ping --timeout 2.0

6) Send with targets (Unreal should ignore unless its appId is in targets)

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 prepare brand --targets ic1,ic2

7) Send a raw method + params JSON

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 raw ic.prepareChangeMode --params '{"nextMode":"theme"}'

8) Raw request with explicit id

python udp_jsonrpc_tester.py --host 127.0.0.1 --port 7777 raw tp.ping --id 123

Notes that matter for Unreal

This tester binds to an ephemeral local UDP port by default (--bind-port 0). Unreal’s responses will go back to that source port.
Messages are sent as one JSON + newline per datagram, which matches “newline delimited JSONRPC”.
If your Unreal receiver expects a specific encoding: this uses UTF-8.

If you want, I can also add:

a REPL mode (interactive prompt: prepare theme, ping, etc.)
a batch mode to replay a script file of newline-delimited JSONRPC messages at a given rate (useful for stress testing / duplicate dispatch simulation).

You can absolutely enable and configure UE’s UDP Messaging plugin in UE5.5 with Blueprints-only — but you cannot use it as a “raw UDP socket that receives newline-delimited JSON strings”.

UDP Messaging is a transport for UE’s Message Bus (structured UE messages), not a “give me arbitrary UDP datagrams” interface. citeturn0search14turn0search0

So: a clean Blueprint-only implementation of JSON-RPC request/response over UDP is not realistically achievable using UDP Messaging alone.

That said, here are the clean options, depending on what you mean by “using UDP Messaging”.

What UDP Messaging can do cleanly (Blueprint-friendly conceptually)

Best fit: UE ↔ UE communication (Message Bus)

If both sides speak UE’s messaging system (i.e., another Unreal tool/app, or a custom native app implementing the UE Message Bus protocol), then UDP Messaging works great as the network transport. citeturn0search14turn1search11

Blueprint architecture (clean pattern):

Create a single “Control API Router” Blueprint Actor placed in the persistent level (or a GameInstance Subsystem).
It receives typed messages (not JSON) like:
- FIC_PrepareChangeMode { NextMode, Targets }
- FIC_ChangeMode { Mode, Targets }
- FIC_Ready { Mode, Targets }
- FIC_SvmStart etc.
- FTP_Ping { Id } (and you respond with FTP_Pong { Id, Result="pong" })
It then:
1. checks Targets contains AppId (if provided)
2. routes to Blueprint Events (e.g. OnPrepareChangeMode, OnChangeMode, …)
3. emits a typed response message only for request-types (those with id)

Problem: Blueprints don’t natively create/consume FMessageEndpoint/message-bus endpoints without C++ glue in most setups; the messaging APIs are primarily C++-driven. (That’s why most “UDP in BP” solutions use plugins or C++.)

So this is “clean” as an architecture, but not cleanly doable purely in BP in practice.

If you must stay Blueprint-only and still do UDP + responses

Recommendation: Use the built-in OSC plugin, not UDP Messaging

OSC is UDP, and Epic provides a Blueprint API for receiving messages (type-safe). citeturn1search12

You can map your JSON-RPC methods to OSC addresses like:

ic.prepareChangeMode → OSC address /ic/prepareChangeMode with string arg nextMode
ic.changeMode → /ic/changeMode with string arg mode
ic.ready → /ic/ready with string arg mode
ic.svmPrepareStart → /ic/svmPrepareStart no args
tp.ping → /tp/ping with int arg id and respond via /tp/pong with same id + string pong

Blueprint implementation (clean and practical):

Enable OSC plugin
Create BP_ControlApiRouter
Add an OSC Server (BP nodes) set to listen on your port
Bind event OnOscMessageReceived
Switch on Address:
- Parse args
- Apply targets filter (you can include targets as repeated string args, or one JSON string arg if you want)
- Fire Blueprint events
For ping: send OSC message back to sender (OSC nodes support sending)

This gives you what you asked for (Blueprint-only, UDP, request/response), just not “UDP Messaging”.

If you insist on JSON-RPC over UDP and insist on UDP Messaging

Here’s the honest situation:

Your Python tester sends newline-delimited JSON strings.
UDP Messaging expects UE message-bus packets, not arbitrary JSON.
Therefore your Unreal project will not see your JSON-RPC via UDP Messaging.

The only “clean” way to reconcile that is to add a tiny bridge somewhere:

either small C++ shim inside UE: raw UDP socket receive → parse JSON-RPC → emit Blueprint events + send responses
or external bridge app: JSON-RPC UDP ↔ UE Message Bus UDP (complex)

If your “0 external plugin dependencies” goal is strong, the small C++ shim is still the best engineering answer (it’s not a plugin dependency; it’s your code).

Minimal “UDP Messaging” setup anyway (so your project is ready)

Even if you switch to OSC/raw UDP later, here’s the correct UDP Messaging config:

Enable UDP Messaging plugin
Project Settings → Plugins → UDP Messaging:
- Enable Transport = true
- Unicast Endpoint = 0.0.0.0:<port> (listen on all adapters) or a specific NIC IP
- Multicast settings only if you need discovery/broadcast
  Docs: citeturn0search0

Bottom line

Blueprint-only + JSON-RPC over UDP: use OSC plugin (built-in) or accept you need C++.
UDP Messaging plugin: great for UE Message Bus, not for arbitrary UDP JSON strings. citeturn0search14turn0search0

If you tell me which constraint is least strict:

“must be JSON-RPC exactly” or
“must be Blueprint-only” or
“must use UDP Messaging” …I’ll give you the cleanest concrete implementation path for that choice (with exact Blueprint graphs / message mappings).

In getnamo/UDP-Unreal, you don’t “subscribe” by writing code — you bind to the UDPComponent’s Blueprint Event Dispatcher.

The README’s key line is: once the receive socket is open, data arrives on OnReceivedBytes. citeturn3search0

Here’s the clean Blueprint workflow:

1) Add the component

Create (or pick) an Actor Blueprint, e.g. BP_UdpReceiver.
Add Component → UDPComponent (from the plugin). citeturn3search0
Place BP_UdpReceiver into your level (or spawn it from GameInstance / persistent level).

2) Configure it to actually listen

Select the UDPComponent and set:

Should Auto Open Receive = ✅ true (so it opens on BeginPlay) citeturn3search0
Receive IP = 0.0.0.0 (listen on all interfaces) citeturn3search0
Receive Port = your port (e.g. 7777)

If you turn off “auto open”, you must manually call the component’s “open receive” function on BeginPlay.

Open the Actor Blueprint’s Event Graph and do either of these:

Method A (most obvious)

Click the UDPComponent in the Components panel
In the Details panel, scroll to Events
Click the + next to OnReceivedBytes

That creates an event node bound to the component.

Method B (right-click in graph)

Right-click in the Event Graph
Look for something like: Add Event for UDPComponent → OnReceivedBytes

Either way, you end up with an event node:

Event OnReceivedBytes
- gives you a Byte Array (and depending on plugin version may include sender info)

That’s your “subscription”. citeturn3search0

4) Turn bytes into your JSON-RPC line and parse it

Inside OnReceivedBytes:

Convert TArray<uint8> → String (UTF-8)
Split by \n (because your protocol is newline-delimited JSON)
For each non-empty line:
- JSON-decode
- route by method
- if id exists, send a JSON-RPC response back (using the component’s send/emit)

(Exact node names for byte→string vary by UE version / what other plugins you have enabled, but the structure above is the same.)

5) Common gotchas that look like “event never fires”

Actor isn’t in the level / never spawned.
Receive socket never opened (auto-open off and you didn’t call Open).
Wrong port, or you’re sending to the wrong machine/interface.
OS firewall blocking UDP.
In PIE with multiple instances: both trying to bind the same receive port.

6) One important packaging note

The repo notes that for Blueprint-only projects, you may need to “convert to mixed (BP + C++)” for packaging (shipping/build), even if it works in-editor. citeturn3search0

If you paste (or describe) what you see on the UDPComponent in the Details panel (especially the Events section and the receive/open options), I can tell you exactly which node names you should be looking for in your specific UE5.5 + plugin version.

Here’s a clean single-file Python “hotkey sender” that:

Opens a UDP socket once
Runs a while loop
Lets you press keys 0–9 to send preset JSON-RPC messages (newline-delimited)
Supports both notifications (no id) and requests (with id, waits for response)
Works cross-platform without external deps by using:
- Windows: msvcrt.getwch()
- macOS/Linux: termios + tty + select

Unreal will reply to the sender’s UDP source port for requests (tp.ping). This script listens on that same socket and prints responses.

`udp_hotkeys_jsonrpc.py`

#!/usr/bin/env python3
from __future__ import annotations

import json
import socket
import sys
import time
import threading
from typing import Any, Dict, Optional, Tuple

# ----------------------------
# Cross-platform single-key input
# ----------------------------
def get_key_nonblocking(timeout_s: float = 0.05) -> Optional[str]:
    """
    Returns a single key pressed (as a string) or None if no key.
    Non-blocking with timeout.
    """
    if sys.platform.startswith("win"):
        import msvcrt  # type: ignore
        start = time.time()
        while (time.time() - start) < timeout_s:
            if msvcrt.kbhit():
                ch = msvcrt.getwch()
                return ch
            time.sleep(0.005)
        return None
    else:
        import select
        import termios
        import tty

        fd = sys.stdin.fileno()
        old = termios.tcgetattr(fd)
        try:
            tty.setcbreak(fd)
            r, _, _ = select.select([sys.stdin], [], [], timeout_s)
            if r:
                ch = sys.stdin.read(1)
                return ch
            return None
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old)

# ----------------------------
# UDP JSON-RPC sender/receiver
# ----------------------------
class UdpJsonRpcHotkeyClient:
    def __init__(
        self,
        remote_host: str,
        remote_port: int,
        bind_host: str = "0.0.0.0",
        bind_port: int = 0,
        verbose: bool = True,
    ) -> None:
        self.remote_addr = (remote_host, remote_port)
        self.verbose = verbose

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((bind_host, bind_port))
        self.local_addr = self.sock.getsockname()

        self._stop = threading.Event()
        self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True)

        # request/response tracking
        self._pending: Dict[int, threading.Event] = {}
        self._pending_resp: Dict[int, Dict[str, Any]] = {}
        self._lock = threading.Lock()

    def start(self) -> None:
        self._rx_thread.start()
        print(f"[info] Bound UDP: {self.local_addr[0]}:{self.local_addr[1]}")
        print(f"[info] Remote UDP: {self.remote_addr[0]}:{self.remote_addr[1]}")

    def close(self) -> None:
        self._stop.set()
        try:
            self.sock.close()
        except OSError:
            pass

    def _rx_loop(self) -> None:
        while not self._stop.is_set():
            try:
                data, addr = self.sock.recvfrom(65535)
            except OSError:
                break

            text = data.decode("utf-8", errors="replace")
            for line in [l for l in text.splitlines() if l.strip()]:
                try:
                    msg = json.loads(line)
                except json.JSONDecodeError:
                    if self.verbose:
                        print(f"[rx] non-json from {addr}: {line!r}")
                    continue

                print(f"[rx] from {addr}: {json.dumps(msg, ensure_ascii=False)}")

                # Match JSON-RPC responses by id
                if isinstance(msg, dict) and "id" in msg and ("result" in msg or "error" in msg):
                    req_id = msg.get("id")
                    if isinstance(req_id, int):
                        with self._lock:
                            ev = self._pending.get(req_id)
                            if ev:
                                self._pending_resp[req_id] = msg
                                ev.set()

    def _send_obj(self, obj: Dict[str, Any]) -> None:
        wire = json.dumps(obj, ensure_ascii=False) + "\n"
        self.sock.sendto(wire.encode("utf-8"), self.remote_addr)
        if self.verbose:
            print(f"[tx] {wire.strip()}")

    def send_notification(self, method: str, params: Optional[Dict[str, Any]] = None) -> None:
        obj: Dict[str, Any] = {"jsonrpc": "2.0", "method": method}
        if params is not None:
            obj["params"] = params
        self._send_obj(obj)

    def send_request(self, method: str, params: Optional[Dict[str, Any]] = None, timeout_s: float = 1.5) -> Tuple[int, Optional[Dict[str, Any]]]:
        req_id = int(time.time() * 1000) % 2_000_000_000
        obj: Dict[str, Any] = {"jsonrpc": "2.0", "id": req_id, "method": method}
        if params is not None:
            obj["params"] = params

        ev = threading.Event()
        with self._lock:
            self._pending[req_id] = ev
            self._pending_resp.pop(req_id, None)

        self._send_obj(obj)

        ok = ev.wait(timeout_s)
        with self._lock:
            self._pending.pop(req_id, None)
            resp = self._pending_resp.pop(req_id, None)

        return req_id, resp if ok else None

def make_targets(targets_csv: Optional[str]) -> Optional[list[str]]:
    if not targets_csv:
        return None
    targets = [t.strip() for t in targets_csv.split(",") if t.strip()]
    return targets or None

def main() -> int:
    if len(sys.argv) < 3:
        print("Usage: python udp_hotkeys_jsonrpc.py <host> <port> [targets_csv]")
        print("Example: python udp_hotkeys_jsonrpc.py 127.0.0.1 7777 ic1,ic2")
        return 2

    host = sys.argv[1]
    port = int(sys.argv[2])
    targets_csv = sys.argv[3] if len(sys.argv) >= 4 else None
    targets = make_targets(targets_csv)

    def maybe_targets(params: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
        if targets is None:
            return params
        if params is None:
            params = {}
        params["targets"] = targets
        return params

    client = UdpJsonRpcHotkeyClient(host, port, verbose=True)
    client.start()

    # Map keys 0-9 to messages.
    # Adjust as you like.
    def send_key(k: str) -> None:
        if k == "0":
            client.send_notification("ic.svmPrepareStart", params=maybe_targets(None))
        elif k == "1":
            client.send_notification("ic.svmStart", params=maybe_targets(None))
        elif k == "2":
            client.send_notification("ic.svmPrepareStop", params=maybe_targets(None))
        elif k == "3":
            client.send_notification("ic.svmStop", params=maybe_targets(None))
        elif k == "4":
            client.send_notification("ic.prepareChangeMode", params=maybe_targets({"nextMode": "theme"}))
        elif k == "5":
            client.send_notification("ic.prepareChangeMode", params=maybe_targets({"nextMode": "brand"}))
        elif k == "6":
            client.send_notification("ic.changeMode", params=maybe_targets({"mode": "theme"}))
        elif k == "7":
            client.send_notification("ic.changeMode", params=maybe_targets({"mode": "brand"}))
        elif k == "8":
            # ready theme (example)
            client.send_notification("ic.ready", params=maybe_targets({"mode": "theme"}))
        elif k == "9":
            # tp.ping is a request; expects {"result":"pong"}
            req_id, resp = client.send_request("tp.ping", params=maybe_targets(None), timeout_s=2.0)
            if resp is None:
                print(f"[ping] id={req_id} timeout")
            else:
                print(f"[ping] id={req_id} response={json.dumps(resp, ensure_ascii=False)}")

    print("\nHotkeys:")
    print("  0: ic.svmPrepareStart")
    print("  1: ic.svmStart")
    print("  2: ic.svmPrepareStop")
    print("  3: ic.svmStop")
    print("  4: ic.prepareChangeMode theme")
    print("  5: ic.prepareChangeMode brand")
    print("  6: ic.changeMode theme")
    print("  7: ic.changeMode brand")
    print("  8: ic.ready theme")
    print("  9: tp.ping (request)")
    print("  q: quit\n")

    try:
        while True:
            key = get_key_nonblocking(0.05)
            if not key:
                continue

            if key.lower() == "q":
                print("[info] quitting")
                break

            if key in "0123456789":
                send_key(key)
            else:
                # ignore other keys quietly
                pass

    finally:
        client.close()
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

Run it

Send to Unreal listening on 127.0.0.1:7777:

python udp_hotkeys_jsonrpc.py 127.0.0.1 7777

If you want every message to include a targets list:

python udp_hotkeys_jsonrpc.py 127.0.0.1 7777 ic1,ic2

Then press 0..9 to send messages, q to quit.

If you want, I can change the mapping so 4–9 are slot1..slot4 transitions (prepare + change), or make it a two-step “prepare then change” macro per key.

Yeah — that’s normal. PyCharm’s Run console isn’t a real TTY, so low-level “press a key without Enter” input often doesn’t reach msvcrt / termios the way it does in a real terminal.

Here are clean fixes (pick one):

Option 1 (best): run it in a real terminal

Windows: run in Command Prompt, PowerShell, or Windows Terminal
macOS/Linux: run in Terminal/iTerm

Example:

python udp_hotkeys_jsonrpc.py 127.0.0.1 7777

That should make single-key hotkeys work immediately.

Option 2: keep PyCharm, but use “input + Enter” instead of raw keypresses

This works reliably everywhere (including PyCharm). Below is a version that:

still runs a while loop
you type 0..9 then press Enter
also supports q to quit

Replace the loop with this “line input” loop

Here’s a complete minimal adaptation (drop-in idea):

print("Type 0-9 then Enter. Type q then Enter to quit.\n")
while True:
    try:
        s = input("> ").strip()
    except (EOFError, KeyboardInterrupt):
        break

    if not s:
        continue

    if s.lower() == "q":
        break

    # allow people to paste multiple digits, e.g. "469"
    for ch in s:
        if ch in "0123456789":
            send_key(ch)

If you want, I can paste the full script with this mode built-in and a --mode hotkeys|line flag.

Option 3: PyCharm setting (sometimes works): “Emulate terminal in output console”

In Run/Debug Configuration for your script:

check “Emulate terminal in output console”

This sometimes allows single-key reading to work, depending on OS/PyCharm version, but it’s less reliable than Option 1.

Option 4: use `pynput` / `keyboard` (not recommended for your constraints)

These libraries can capture global keypresses even inside IDE consoles, but they’re external deps and can require elevated permissions.

My recommendation

Use Option 1 for true hotkeys.
If you need to stay inside PyCharm, switch to Option 2 (digits + Enter).

If you tell me your OS (Windows/macOS/Linux), I’ll paste a polished script that supports both modes automatically (detects if stdin is a TTY; if not, falls back to “type a number then Enter”).

From the spec you pasted, Unreal only must send a JSON-RPC response for RPC calls that include an id (i.e., requests). Notifications (missing id) do not require any response.

1) For `tp.ping` (request)

Spec says:

{ "id": <req-int>, "method": "tp.ping" }
Respond to RPC with { "id": <req-id>, "result":"pong" }

To be JSON-RPC 2.0 compliant, Unreal should include "jsonrpc":"2.0" in the response too. So the correct response is:

{"jsonrpc":"2.0","id":123,"result":"pong"}

(Where 123 is whatever id was received.)

2) For everything else in your list (as written)

These are shown as messages like:

ic.prepareChangeMode (with params)
ic.changeMode (with params)
ic.ready (with params)
ic.svmPrepareStart, ic.svmStart, ic.svmPrepareStop, ic.svmStop (no params)

In the examples, they do not include an id, so they are notifications → no response required.

3) If Pixera (or your tester) sends any of those methods with an `id`

Then they become JSON-RPC requests. The spec doesn’t define results for those methods, but JSON-RPC 2.0 says you should respond either with:

Success

{"jsonrpc":"2.0","id":123,"result":<something>}

or Error

{"jsonrpc":"2.0","id":123,"error":{"code":-32601,"message":"Method not found"}}

Given your spec statement:

“All methods implemented … must be idempotent…”
“If method not referenced by this document then no action is needed…”
and “notifications missing the id field do not need to be responded to”

…the safest interpretation is:

Only guarantee a response for tp.ping.
For other methods, only respond if you deliberately choose to support them as requests; otherwise you can return a JSON-RPC error if they arrive with an id (or you can still treat them as notifications and ignore the id, but that’s not strictly JSON-RPC compliant).

4) Targets filtering and responses

Targets affects whether the message is “relevant”:

If params.targets is present and Unreal’s appId is not listed, Unreal should ignore the message.
If the ignored message is a notification: ignore silently (optionally log).
If the ignored message is a request (has id): the spec isn’t explicit, but for cleaner RPC behavior you typically either:
- respond with an error like “not targeted / ignored”, or
- don’t respond (but then the caller will time out).

If you want strict “only ping responds” behavior, you can keep it simple:

Respond only to tp.ping requests, regardless of targets logic (or apply targets first and only respond when targeted—your call, but be consistent).

Minimum required by your pasted text:
✅ Respond to tp.ping when it has an id with result:"pong" and the same id.
✅ Do not respond to notifications without id.

If you want, I can propose a clean response policy for “requests with id” for all methods (success vs error, and what to do when not targeted) so your Python tester gets deterministic behavior.

if some of the notifications contain params like this, what struct should I create in Unreal to handle the different types of params?

        client.send_notification("ic.prepareChangeMode", params=maybe_targets({"nextMode": "theme"}))
    elif k == "5":
        client.send_notification("ic.prepareChangeMode", params=maybe_targets({"nextMode": "brand"}))
    elif k == "6":
        client.send_notification("ic.changeMode", params=maybe_targets({"mode": "theme"}))
    elif k == "7":
        client.send_notification("ic.changeMode", params=maybe_targets({"mode": "brand"}))
    elif k == "8":
        # ready theme (example)
        client.send_notification("ic.ready", params=maybe_targets({"mode": "theme"}))

Make one “envelope” struct for the common JSON-RPC fields, and then small, typed param structs per method. Don’t try to force all params into one giant struct with unrelated fields unless you really want to.

1) Common envelope (what every message shares)

Create a Blueprintable struct like:

FJsonRpcEnvelope

Method (String)
Id (Integer, optional → you can store -1 when missing)
HasId (Bool)
ParamsJson (String) or ParamsMap (Map<String, String>) depending on how you parse
- If you’re parsing JSON in BP, keeping ParamsJson as a raw string is often simplest, then decode per-method.

Why: all of your notifications differ mainly by method + the shape of params.

2) Typed params per method (clean + future-proof)

FIC_PrepareChangeModeParams

NextMode (String or better: Enum)
Targets (Array of String, optional)

FIC_ChangeModeParams

Mode (String/Enum)
Targets (Array of String, optional)

FIC_ReadyParams

Mode (String/Enum)
Targets (Array of String, optional)

These match your examples exactly:

prepareChangeMode → nextMode
changeMode → mode
ready → mode
all may include targets

3) Use an Enum for mode (recommended)

Create EIC_Mode:

Theme
Brand
Slot1
Slot2
Slot3
Slot4
Unknown (fallback)

Then in your param structs, store:

Mode / NextMode as EIC_Mode (plus maybe the original string for logging)

4) Routing logic in BP

In your receiver event (bytes/string -> JSON):

Read method
Switch on method
For each case, decode only the fields you need into the matching struct:

"ic.prepareChangeMode" → build FIC_PrepareChangeModeParams
"ic.changeMode" → build FIC_ChangeModeParams
"ic.ready" → build FIC_ReadyParams

5) Why not one “Params” struct with both `mode` and `nextMode`?

You can do it, but it’s easy to misuse: FIC_GenericModeParams

Mode (String/Enum) optional
NextMode (String/Enum) optional
Targets (Array String)

This works, but you’ll constantly be checking “which field should be filled for this method?”, which is exactly what separate structs avoids.

Minimal answer: what structs should you create?

If you want the cleanest setup for your shown notifications, create:

FJsonRpcEnvelope (Method, HasId, Id, ParamsJson/Map)
FIC_PrepareChangeModeParams (NextMode, Targets)
FIC_ChangeModeParams (Mode, Targets)
FIC_ReadyParams (Mode, Targets)
EIC_Mode enum (Theme/Brand/Slot1… + Unknown)

If you tell me whether you’re parsing JSON in BP (e.g., VaRest/JsonBlueprintUtils) or only handling raw strings, I’ll recommend the best “ParamsJson vs Map” choice and the exact decode pattern.