Plugin API

Plugin SDK - Client and server helpers for the AIPC plugin system.

DEPRECATED: the platform does not ship the /run/aipc/plugins discovery mechanism. This module is kept importable for compatibility, emits a DeprecationWarning on use, and will be removed in v0.8.0.

For plugin providers:

server = PluginServer(“my-plugin”) server.start(my_grpc_servicer)

For plugin consumers:

discovery = PluginDiscovery() endpoint = discovery.get(“rtsp-server”) channel = endpoint.connect()

class neoruntime_ipc_sdk.plugin.PluginEndpoint(app_id, capability_id, version, transport, socket_path=None, grpc_service=None, event_publish=<factory>, event_subscribe=<factory>, state='unknown')

Bases: object

Resolved endpoint for a discovered plugin capability.

app_id: str

capability_id: str

version: str

transport: str

socket_path: str | None = None

grpc_service: str | None = None

event_publish: list[str]

event_subscribe: list[str]

state: str = 'unknown'

connect(**kwargs)

Create a gRPC channel to this plugin.

property is_available: bool

__init__(app_id, capability_id, version, transport, socket_path=None, grpc_service=None, event_publish=<factory>, event_subscribe=<factory>, state='unknown')

class neoruntime_ipc_sdk.plugin.PluginDiscovery(discovery_dir='/run/aipc/plugins')

Bases: object

Client-side plugin discovery using discovery.json + optional file watcher.

__init__(discovery_dir='/run/aipc/plugins')

reload()

Reload discovery data from file.

get(capability_id)

Find the first running plugin providing a capability.

require(capability_id, timeout=30.0)

Wait for a capability to become available. Raises TimeoutError.

list_plugins()

Return all known plugins.

list_capabilities()

Return all known capability IDs.

watch(callback)

Register a callback invoked when discovery.json changes.

close()

class neoruntime_ipc_sdk.plugin.PluginServer(plugin_id, socket_dir='/run/aipc/plugins')

Bases: object

Helper for plugin containers to set up a gRPC server on the standard socket path.

__init__(plugin_id, socket_dir='/run/aipc/plugins')

create_server(max_workers=4, **kwargs)

Create a gRPC server bound to the plugin socket.

start()

Start the gRPC server.

wait()

Block until server terminates.

stop(grace=5.0)

Gracefully stop the server.

PluginDiscovery

class neoruntime_ipc_sdk.PluginDiscovery(discovery_dir='/run/aipc/plugins')

Client-side plugin discovery using discovery.json + optional file watcher.

__init__(discovery_dir='/run/aipc/plugins')

reload()

Reload discovery data from file.

get(capability_id)

Find the first running plugin providing a capability.

require(capability_id, timeout=30.0)

Wait for a capability to become available. Raises TimeoutError.

list_plugins()

Return all known plugins.

list_capabilities()

Return all known capability IDs.

watch(callback)

Register a callback invoked when discovery.json changes.

close()

PluginEndpoint

class neoruntime_ipc_sdk.PluginEndpoint(app_id, capability_id, version, transport, socket_path=None, grpc_service=None, event_publish=<factory>, event_subscribe=<factory>, state='unknown')

Resolved endpoint for a discovered plugin capability.

app_id: str

capability_id: str

version: str

transport: str

socket_path: str | None = None

grpc_service: str | None = None

event_publish: list[str]

event_subscribe: list[str]

state: str = 'unknown'

connect(**kwargs)

Create a gRPC channel to this plugin.

property is_available: bool

__init__(app_id, capability_id, version, transport, socket_path=None, grpc_service=None, event_publish=<factory>, event_subscribe=<factory>, state='unknown')

PluginServer

class neoruntime_ipc_sdk.PluginServer(plugin_id, socket_dir='/run/aipc/plugins')

Helper for plugin containers to set up a gRPC server on the standard socket path.

__init__(plugin_id, socket_dir='/run/aipc/plugins')

create_server(max_workers=4, **kwargs)

Create a gRPC server bound to the plugin socket.

start()

Start the gRPC server.

wait()

Block until server terminates.

stop(grace=5.0)

Gracefully stop the server.

Usage Examples

App side: discover and call a plugin

from neoruntime_ipc_sdk import PluginDiscovery

discovery = PluginDiscovery()  # defaults to /run/aipc/plugins

# List available capabilities
for cap in discovery.list_capabilities():
    print(f"{cap}")

# Get an endpoint providing a capability (optional)
endpoint = discovery.get("lpr.postprocess")

# Or block until the plugin appears (consumer starts before provider)
endpoint = discovery.require("lpr.postprocess", timeout=30.0)

# Connect to the plugin's gRPC service
channel = endpoint.connect()

discovery.close()

Watch for plugin changes

def on_change(event):
    print(f"plugin event: {event}")

discovery.watch(on_change)

Plugin side: advertise a capability

import grpc
from neoruntime_ipc_sdk import PluginServer

server = PluginServer("lpr.postprocess")  # writes a discovery file
                                          # under /run/aipc/plugins

grpc_server = server.create_server(max_workers=4)
my_service.add_to_grpc_server(grpc_server)  # register your servicer
grpc_server.start()

server.start()        # announce the capability
grpc_server.wait_for_termination()
server.stop()

Hot reload

# Rescan the discovery directory after plugin updates
discovery.reload()