插件 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.

使用示例

App 侧：发现并调用插件

from neoruntime_ipc_sdk import PluginDiscovery

discovery = PluginDiscovery()  # 默认 /run/aipc/plugins

# 列出可用能力
for cap in discovery.list_capabilities():
    print(f"{cap}")

# 取提供某能力的插件端点（可选）
endpoint = discovery.get("lpr.postprocess")

# 或阻塞等待插件上线（依赖方先于提供方启动的场景）
endpoint = discovery.require("lpr.postprocess", timeout=30.0)

# 连接到插件的 gRPC 服务
channel = endpoint.connect()

discovery.close()

监听插件上下线

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

discovery.watch(on_change)

插件侧：发布能力

import grpc
from neoruntime_ipc_sdk import PluginServer

server = PluginServer("lpr.postprocess")   # 在 /run/aipc/plugins 落盘发现文件

grpc_server = server.create_server(max_workers=4)
my_service.add_to_grpc_server(grpc_server)  # 注册你的 servicer
grpc_server.start()

server.start()        # 向外宣告能力可用
grpc_server.wait_for_termination()
server.stop()

热加载

# 插件更新后重新扫描发现目录
discovery.reload()