Automate Java Applications Using the NSP JVM Probe

NSP’s JVM probe attaches to any running Java process using the JVMTI (JVM Tool Interface) and JNI reflection APIs. It loads a native JVMTI agent into the target process, enumerates classes and fields, converts Java objects to the Substrate Schema Format (SSF), and makes the resulting state — and the application’s methods — available to your agent over the standard NSP API. This guide walks you through detecting a Java app, reading its state, discovering actions, executing JVM method calls, and interpreting results.

Prerequisites

Confirm the following before you begin:

NSP daemon installed and running on port 7842. Download from platform.nelieo.com/download.
JDK 11 or later installed on the host machine. The JVM probe requires the AttachAPI from tools.jar (JDK 11+). A JRE alone is not sufficient.
The target Java application is running. The daemon scans for processes every 5 seconds, so start your app before querying for it. This guide uses a generic Java GCS application as its primary example; the same steps apply to IntelliJ IDEA, Jenkins, SAP ERP, Eclipse, Elasticsearch, and any other JVM-based app.
Python 3.11+ and the NSP SDK installed.

pip install nelieo-nsp

An NSP API key from platform.nelieo.com, stored in the NSP_API_KEY environment variable.

# Windows (Command Prompt)
set NSP_API_KEY=nsp_live_sk_your_key_here

Detect the Java App

Call /substrate/v1/apps to verify the daemon has classified your application as a JVM app. Look for "runtime": "jvm" in the response.

curl -H "X-Substrate-Key: nsp_live_sk_..." http://localhost:7842/substrate/v1/apps

Example response for a Java GCS application:

{
  "total": 2,
  "apps": [
    {
      "pid": 8833,
      "app_id": "javagcs-8833",
      "app_name": "Java GCS",
      "runtime": "jvm",
      "runtime_name": "JVM (Java)",
      "probe_status": "active",
      "confidence": 0.94,
      "object_count": 38,
      "action_count": 9,
      "last_captured_at": "2026-07-20T13:00:05Z",
      "has_state": true,
      "current_view": null,
      "current_url": null
    }
  ]
}

You can also list apps with the Python SDK:

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        apps = await client.list_apps()
        for app in apps:
            print(f"{app.app_name:<25}  runtime={app.runtime:<5}  "
                  f"conf={app.confidence:.2f}  status={app.probe_status}")

asyncio.run(main())

Expected output:

Java GCS                   runtime=jvm    conf=0.94  status=active

The first JVMTI attachment (cold probe) can take up to 60–90 seconds as the probe enumerates all loaded classes and builds the schema. Subsequent runs use the cached schema and attach in under two seconds. Watch probe_status progress from attaching → cold_probe → active.

Read Java App State

Attach to the app by name and read state using dot-notation keys. The JVM probe normalises Java object fields into SSF, giving you typed access with get_int(), get_float(), get_str(), and get_bool().

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        session = await client.attach("Java GCS")
        print(f"Confidence: {session.confidence:.2f}")

        # Normalised SSF state keys (depends on the app's schema)
        alt     = session.get_float("flight.altitude_m")
        lat     = session.get_float("flight.gps.latitude")
        lon     = session.get_float("flight.gps.longitude")
        mode    = session.get_str("flight.mode")
        armed   = session.get_bool("flight.is_armed")
        sats    = session.get_int("flight.gps.satellites")

        print(f"Position   : {lat:.6f}, {lon:.6f}")
        print(f"Altitude   : {alt:.1f} m")
        print(f"Mode       : {mode}")
        print(f"Armed      : {armed}")
        print(f"Satellites : {sats}")

        # Print all discovered state keys
        print("\nAll state keys:")
        for key, value in session.state.objects.items():
            print(f"  {key} = {value!r}")

asyncio.run(main())

Expected output:

Confidence: 0.94
Position   : 34.052200, -118.243700
Altitude   : 38.7 m
Mode       : GUIDED
Armed      : True
Satellites : 12

All state keys:
  flight.altitude_m = 38.7
  flight.gps.latitude = 34.0522
  flight.gps.longitude = -118.2437
  flight.mode = 'GUIDED'
  flight.is_armed = True
  flight.gps.satellites = 12
  ...

Explore Available Actions

Iterate session.actions to see every action the JVM probe discovered for this application. The probe discovers both NSP-native actions defined in the schema registry and raw Java method invocation actions exposed via JNI reflection.

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        session = await client.attach("Java GCS")

        print(f"{'Action':<28} {'Reversibility':<22} Signature")
        print("-" * 90)
        for name, schema in session.actions.items():
            print(f"{name:<28} {schema.reversibility:<22} {schema.signature}")

asyncio.run(main())

Example output for a Java GCS:

Action                       Reversibility          Signature
----------------------------  ---------------------  -------------------------------------------
arm_motors                   reversible_write       fn() -> bool
disarm_motors                reversible_write       fn() -> bool
set_mode                     reversible_write       fn(mode: string) -> bool
take_off                     irreversible_write     fn(altitude_m: float) -> bool
add_waypoint                 reversible_write       fn(lat: float, lon: float, alt_m: float) -> bool
clear_waypoints              reversible_write       fn() -> bool
get_waypoint_count           read                   fn() -> int
get_status_text              read                   fn() -> string

Execute a JVM Action

Use session.execute() to run actions on the Java app. Reversible actions (like set_mode) do not require a verify_expression. Irreversible actions (like take_off) require verify=True and a verify_expression — NSP enforces this safety gate automatically.

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        session = await client.attach("Java GCS")

        # Reversible: set flight mode
        result = await session.execute(
            "set_mode",
            parameters={"mode": "GUIDED"},
        )
        print(f"Set mode result : {result.success}  latency={result.latency_ms}ms")

        # Read-only: get status text
        status_result = await session.execute(
            "get_status_text",
            parameters={},
        )
        print(f"Status text     : {status_result.raw_return!r}")

        # Irreversible: take off to 40 m — requires verify
        takeoff_result = await session.execute(
            "take_off",
            parameters={"altitude_m": 40.0},
            verify=True,
            verify_expression="flight.altitude_m > 35.0",
            verify_timeout_ms=30000,
        )
        print(f"Takeoff         : {takeoff_result.success}")
        if takeoff_result.verification:
            print(f"Verified        : {takeoff_result.verification.passed}  "
                  f"({takeoff_result.verification.explanation})")

asyncio.run(main())

Expected output:

Set mode result : True  latency=185ms
Status text     : 'EKF3 IMU0 initialised'
Takeoff         : True
Verified        : True  (flight.altitude_m changed from 0.0 to 40.4)

Handle the jvm_invoke_snapshot Field

When NSP executes a JVM action that invokes a Java method directly, the action response includes a jvm_invoke_snapshot field. This field contains a serialised snapshot of the Java object returned by the method call — useful when the method returns a complex object rather than a primitive.

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        session = await client.attach("Java GCS")

        # Execute an action that returns a Java object
        result = await session.execute(
            "get_telemetry_snapshot",
            parameters={},
        )

        print(f"Success           : {result.success}")
        print(f"jvm_invoke_snapshot: {result.jvm_invoke_snapshot}")

        # jvm_invoke_snapshot is a dict of field names → values
        # for Java objects, or a primitive value for simple types
        snap = result.jvm_invoke_snapshot
        if isinstance(snap, dict):
            print(f"\nSnapshot fields:")
            for field, value in snap.items():
                print(f"  {field} = {value!r}")

asyncio.run(main())

Example jvm_invoke_snapshot output:

Success           : True
jvm_invoke_snapshot: {'altitudeM': 38.7, 'groundspeedMs': 4.2, 'mode': 'GUIDED', 'isArmed': True}

Snapshot fields:
  altitudeM = 38.7
  groundspeedMs = 4.2
  mode = 'GUIDED'
  isArmed = True

jvm_invoke_snapshot is null for actions that do not invoke a Java method directly. Always check result.success first — a True result with a null snapshot simply means the action completed without returning a value (i.e. the Java method returns void).

Cold Probe Timing and Confidence

JVMTI attachment involves loading a native agent into the JVM and performing class enumeration — this takes time on first run. Understanding the lifecycle helps you write robust agents.

import asyncio
import nelieo_nsp as nsp

async def wait_for_jvm_app(app_name: str, timeout: float = 120.0):
    """Wait for a JVM app to reach active status, then return the session."""
    async with nsp.NSPClient() as client:
        # wait_for_app polls until probe_status == "active"
        session = await client.wait_for_app(app_name, timeout=timeout)
        print(f"Ready : {session.app_name}")
        print(f"  confidence = {session.confidence:.2f}")
        print(f"  objects    = {len(session.state.objects)}")
        print(f"  actions    = {len(session.actions)}")
        return session

asyncio.run(wait_for_jvm_app("Java GCS"))

The probe_status field progresses through these stages:

`probe_status`	Meaning
`detecting`	Daemon spotted the JVM process
`attaching`	JVMTI agent loading via `AttachAPI`
`cold_probe`	First class/field enumeration in progress
`active`	Probe is live; state is fresh and actions are available
`error`	Probe attempt failed; daemon retries automatically

JVM confidence typically stabilises at 0.93–0.97 once active. Actions are gated by confidence level: reversible_write requires >= 0.80; irreversible_write requires >= 0.95.

Full Workflow Example

The script below covers every step: detecting the app, reading state, listing actions, executing both reversible and irreversible actions, and streaming live telemetry.

import asyncio
import nelieo_nsp as nsp

async def java_agent():
    async with nsp.NSPClient() as client:
        # ── Wait for the app to be ready ─────────────────────────────────
        print("Waiting for Java GCS (cold probe may take ~90s)…")
        session = await client.wait_for_app("Java GCS", timeout=120.0)
        print(f"Connected: conf={session.confidence:.2f}  "
              f"objects={len(session.state.objects)}  "
              f"actions={len(session.actions)}")

        # ── Read normalised state ────────────────────────────────────────
        alt  = session.get_float("flight.altitude_m")
        mode = session.get_str("flight.mode")
        lat  = session.get_float("flight.gps.latitude")
        lon  = session.get_float("flight.gps.longitude")
        print(f"\nPosition : {lat:.4f}, {lon:.4f}  Altitude={alt:.1f}m  Mode={mode}")

        # ── List available actions ───────────────────────────────────────
        print("\nAvailable actions:")
        for name, schema in session.actions.items():
            print(f"  {name:<28} [{schema.reversibility}]")

        # ── Set mode (reversible) ────────────────────────────────────────
        mode_result = await session.execute(
            "set_mode",
            parameters={"mode": "GUIDED"},
        )
        print(f"\nSet GUIDED: {mode_result.success}")

        # ── Add a waypoint (reversible) ──────────────────────────────────
        wp_result = await session.execute(
            "add_waypoint",
            parameters={"lat": 34.0600, "lon": -118.2400, "alt_m": 50.0},
        )
        print(f"Waypoint added: {wp_result.success}")

        # ── Arm (reversible) ─────────────────────────────────────────────
        arm_result = await session.execute("arm_motors", parameters={})
        print(f"Armed: {arm_result.success}")

        # ── Take off (irreversible — requires verify) ────────────────────
        takeoff = await session.execute(
            "take_off",
            parameters={"altitude_m": 50.0},
            verify=True,
            verify_expression="flight.altitude_m > 45.0",
            verify_timeout_ms=30000,
        )
        print(f"Airborne: {takeoff.success}")
        if takeoff.jvm_invoke_snapshot:
            print(f"JVM snapshot: {takeoff.jvm_invoke_snapshot}")

        # ── Stream telemetry ─────────────────────────────────────────────
        print("\nStreaming telemetry (Ctrl+C to stop)…")
        async with session.watch(auto_refresh_on_event=True) as stream:
            async for event in stream:
                if not any(k.startswith("flight.") for k in event.changed_keys):
                    continue
                alt  = session.get_float("flight.altitude_m")
                mode = session.get_str("flight.mode")
                batt = session.get_float("flight.battery_remaining")
                print(f"  [seq={event.sequence}] "
                      f"Alt={alt:.1f}m  Mode={mode}  Batt={batt:.0f}%")

                if batt < 15.0:
                    print("⚠  Critical battery — commanding RTL")
                    await session.execute(
                        "set_mode",
                        parameters={"mode": "RTL"},
                        verify=True,
                        verify_expression="flight.mode == 'RTL'",
                    )

asyncio.run(java_agent())

Get Started

The Daemon

Runtime Probes

Security

Guides

Full Workflow Example

​Full Workflow Example

Full Workflow Example