Neomega

外观

plugin API ​

完整接口目录 · 插件运行时用法 · 工具包下载

模块说明 ​

text
Small native plugin authoring facade over the existing managed Worker.

Event ​

python
class Event

SDK 说明：

text
One durable delivery. Explicit transaction/save or ack advances its cursor.

Event.__init__ ​

python
def __init__(self, context, payload: EventPayload)

Event.kind ​

python
@property
def kind(self) -> str

Event.transaction ​

python
async def transaction(self, *, commit_id=None)

Event.ack ​

python
async def ack(self)

Event.context ​

python
self.context

公开实例属性，在 __init__ 中设置。

Event.payload ​

python
self.payload

公开实例属性，在 __init__ 中设置。

PluginContext ​

python
class PluginContext

PluginContext.__init__ ​

python
def __init__(self, worker, data_dir, *, bus=None)

PluginContext.scope ​

python
@property
def scope(self)

SDK 说明：

text
Current opaque scope ID; ownership is always verified by Host.

PluginContext.data_dir ​

python
@property
def data_dir(self) -> Path

PluginContext.queries ​

python
@property
def queries(self) -> Queries

PluginContext.inventory ​

python
@property
def inventory(self) -> Inventory

PluginContext.actors ​

python
@property
def actors(self) -> Actors

PluginContext.environment ​

python
@property
def environment(self) -> Environment

PluginContext.digging ​

python
@property
def digging(self) -> BlockDigging

PluginContext.crafting ​

python
@property
def crafting(self) -> Crafting

PluginContext.recipes ​

python
@property
def recipes(self) -> Recipes

PluginContext.artifacts ​

python
@property
def artifacts(self) -> Artifacts

PluginContext.containers ​

python
@property
def containers(self) -> Containers

PluginContext.item_actions ​

python
@property
def item_actions(self) -> ItemActions

PluginContext.notifications ​

python
@property
def notifications(self) -> Notifications

PluginContext.secrets ​

python
@property
def secrets(self) -> Secrets

PluginContext.log ​

python
@property
def log(self)

PluginContext.installation ​

python
@property
def installation(self)

PluginContext.operations ​

python
@property
def operations(self) -> Operations

PluginContext.scoreboard ​

python
@property
def scoreboard(self) -> Scoreboard

PluginContext.commands ​

python
@property
def commands(self) -> Commands

PluginContext.control ​

python
@property
def control(self) -> RobotControl

PluginContext.player_abilities ​

python
@property
def player_abilities(self) -> PlayerAbilities

PluginContext.player_actions ​

python
@property
def player_actions(self) -> PlayerActions

PluginContext.command_queries ​

python
@property
def command_queries(self) -> CommandQueries

PluginContext.tickingareas ​

python
@property
def tickingareas(self) -> TickingAreas

PluginContext.world ​

python
@property
def world(self) -> World

PluginContext.observe_players ​

python
async def observe_players(self, **options)

SDK 说明：

text
Return (stream, initial snapshot); caller closes stream in finally.

PluginContext.player_events ​

python
async def player_events(self, **options)

PluginContext.spawn ​

python
def spawn(self, coroutine, *, name=None)

PluginContext.later ​

python
def later(self, seconds: float, callback, *, name=None)

SDK 说明：

text
Run one async callback after a delay, owned and drained by Worker.

PluginContext.every ​

python
def every(self, seconds: float, callback, *, immediate=False, name=None)

SDK 说明：

text
Bounded Worker-owned serial timer; drain cancels and joins callbacks.

Interval follows completion, so slow callbacks never accumulate tasks.
A callback failure fails this plugin and is logged by Worker.

PluginContext.robot ​

python
async def robot(self)

PluginContext.resolve_player ​

python
async def resolve_player(
    self,
    *,
    name: str | None = None,
    uuid: str | None = None,
    xuid: str | None = None,
) -> PlayerIdentity

PluginContext.players ​

python
async def players(self) -> list[PlayerIdentity]

PluginContext.player ​

python
async def player(
    self,
    *,
    name: str | None = None,
    uuid: str | None = None,
    xuid: str | None = None,
) -> PlayerHandle

SDK 说明：

text
Resolve a bound player handle from the current Host roster cache.

PluginContext.peer ​

python
self.peer

公开实例属性，在 __init__ 中设置。

PluginContext.storage ​

python
self.storage

公开实例属性，在 __init__ 中设置。

PluginContext.business ​

python
self.business

公开实例属性，在 __init__ 中设置。

PluginContext.services ​

python
self.services

公开实例属性，在 __init__ 中设置。

PluginContext.bus ​

python
self.bus

公开实例属性，在 __init__ 中设置。

PluginContext.scopes ​

python
self.scopes

公开实例属性，在 __init__ 中设置。

PluginContext.positions ​

python
self.positions

公开实例属性，在 __init__ 中设置。

PluginContext.packets ​

python
self.packets

公开实例属性，在 __init__ 中设置。

PluginContext.raw_packets ​

python
self.raw_packets

公开实例属性，在 __init__ 中设置。

PluginContext.chat ​

python
self.chat

公开实例属性，在 __init__ 中设置。

PluginContext.events ​

python
self.events

公开实例属性，在 __init__ 中设置。

PluginContext.chat_waiters ​

python
self.chat_waiters

公开实例属性，在 __init__ 中设置。

PluginContext.config ​

python
self.config

公开实例属性，在 __init__ 中设置。

PluginContext.capabilities ​

python
self.capabilities

公开实例属性，在 __init__ 中设置。

Plugin ​

python
class Plugin

SDK 说明：

text
Subclass lifecycle hooks and call run(). No detached tasks or automatic ACK.

Set config_type to a dataclass. Configuration lives in data_dir/config.json
and uses dataclass defaults when absent. Advanced callers may use ctx.peer.
on_validate is read-only; start tasks only from on_start or active handlers.

Plugin.config_type ​

python
config_type = None

Plugin.on_init ​

python
async def on_init(self, ctx: PluginContext)

SDK 说明：

text
Register typed read/service members once per Worker during validation.

Plugin.on_validate ​

python
async def on_validate(self, ctx: PluginContext)

Plugin.on_start ​

python
async def on_start(self, ctx: PluginContext)

Plugin.on_stop ​

python
async def on_stop(self, ctx: PluginContext)

Plugin.on_maintenance ​

python
async def on_maintenance(self, ctx: PluginContext, request)

SDK 说明：

text
Optionally seal/release business admission under the plugin's own lock.

Echo request token with status sealed/busy/unsupported (seal), or
released/busy/unsupported (release). Release must be idempotent. This
hook does not drain callbacks or cancel Worker-owned background tasks.

Plugin.on_event ​

python
async def on_event(self, ctx: PluginContext, event: Event)

Plugin.serve ​

python
async def serve(self, peer=None, *, data_dir=None)

SDK 说明：

text
Own a fresh IPC connection until Host stops the plugin.

Plugin.run ​

python
def run(self)

SDK 说明：

text
Blocking executable entry point. stdout belongs to the IPC protocol.

Neomega · 使用、开发与参考文档