Neomega

外观

Python SDK 入门 ​

本页给出常见调用的起点。完整能力见接口总览，精确签名与类型字段见完整 API 目录。

从开发工具包获取 SDK。聊天命令、事件路由、玩家句柄等常用写法见作者便利接口，类型化数据包见协议与数据包。

在由 Host 启动的插件中使用 ctx。完整项目从第一个插件生成；以下代码块是回调内的片段，不能脱离 Host 单独执行。

python
from neomega_runtime import Plugin, PluginContext, Event

class MyPlugin(Plugin):
    async def on_start(self, ctx: PluginContext):
        ctx.log.info("插件已启动")

if __name__ == "__main__":
    MyPlugin().run()

Plugin.run() 是阻塞入口。标准输出由 IPC 协议占用，插件日志使用 ctx.log。配置在验证阶段可用，游戏接口从 on_start 开始可用，详见生命周期。

需要连续操作一组共享资源时，使用资源会话 Scope的 acquire/close 或 run。通过网络连接 Host 见远程 EP；两者均要求 1.1.0 配套版本。

玩家与身份 ​

text
await ctx.players() -> list[PlayerIdentity]
await ctx.resolve_player(*, name=None, uuid=None, xuid=None) -> PlayerIdentity

以上表示异步方法的调用签名。返回值是普通字典，含 name: str、uuid: str、xuid: str；XUID 可能为空。两种方法都需要 permissions.players: ["list"] 及管理员授权，读取的是当前缓存名单，不会主动请求服务端刷新。

python
from neomega_runtime import player_target

player = await ctx.resolve_player(name="ExamplePlayer")
target = player_target(player["name"])

多个查询条件必须同时精确匹配，姓名区分大小写。没有匹配抛出 PlayerNotFound，多条匹配抛出 AmbiguousPlayer；非空 XUID 不匹配时不会退回姓名查询。player_target(name) 生成姓名选择器，拒绝前导 !、引号、反斜杠与控制字符，不能传入任意选择器。读取成功不保证玩家下一刻仍在线，也不赋予玩家执行管理命令的权限。

需要连续为同一玩家查询位置、通知或操作计分板时，可取得玩家句柄：

python
player = await ctx.player(name="ExamplePlayer")
await player.refresh()

player 提供 name、uuid、xuid 及只读 identity。refresh() 核对缓存中的原身份；身份变化时需要重新解析。句柄上的动作方法创建意图，仍需提交，具体用法见玩家与聊天。

命令与动作意图 ​

text
ctx.commands.prepare(
    command: str, *, idempotency_key: str, deadline: str,
    identity: str = "ws", wait: bool = True, timeout: float = 30
) -> OperationIntent

ctx.operations.prepare(
    operation: str, arguments: dict, *, idempotency_key: str, deadline: str
) -> OperationIntent

prepare 只创建意图，没有游戏效果。deadline 使用带时区的绝对时间字符串；业务为每个独立动作分配一个唯一 idempotency_key，并保留原意图用于核对。命令执行需要 framework.command.execute 权限。

python
from datetime import datetime, timedelta, timezone

# business_key 必须由调用方为本次独立业务动作分配并保留。
deadline = (datetime.now(timezone.utc) + timedelta(seconds=30)).isoformat()
intent = ctx.commands.prepare(
    "say hello", idempotency_key=business_key, deadline=deadline
)
operation_id = await ctx.operations.submit(intent)
receipt = await ctx.operations.wait(operation_id, timeout=35)
ctx.log.info("动作状态=%s", receipt["state"],
             extra={"operation_id": operation_id})

保存动作意图后再提交；示例中的内存变量不会自动跨进程恢复。可靠事件业务优先用下文的事务方式将状态、事件确认和动作意图一起交给 Host。等待超时、取消或断线后的处理见错误与结果核对。

方法返回与含义
await ctx.operations.submit(intent, timeout=10)返回 operation_id: str，仅代表接受提交。
await ctx.operations.get(operation_id, timeout=10)查询原操作的 OperationReceipt。
await ctx.operations.wait(operation_id, timeout=30, poll_interval=0.05)等到任一终态，包括 failed、cancelled、unknown。不自动取消或重试。
await ctx.operations.cancel(operation_id, timeout=10)返回含 cancellation_requested 的回执；不保证已发出的效果停止。

命令默认请求实际响应。使用无响应身份或关闭等待时，不能把提交成功当作服务端执行成功。不要把玩家聊天文字直接拼进任意命令；先验证业务参数，再使用精确玩家目标和对应 SDK 方法。

通知与计分板 ​

三种单条通知都返回 OperationIntent，需要 framework.command.execute：

text
ctx.notifications.title(name: str, text: str, *, idempotency_key: str, deadline: str)
ctx.notifications.subtitle(name: str, text: str, *, idempotency_key: str, deadline: str)
ctx.notifications.actionbar(name: str, text: str, *, idempotency_key: str, deadline: str)

name 来自已解析的玩家记录。文本由 SDK 编码为 rawtext；subtitle 设置副标题，需要标题动作触发显示。

计分板使用精确 holder，不接受选择器。每种方法分别申请对应的完整操作权限：

方法（均需关键字参数 idempotency_key、deadline）操作权限
ctx.scoreboard.prepare_get(holder, objective)framework.scoreboard.get
ctx.scoreboard.prepare_set(holder, objective, value)framework.scoreboard.set
ctx.scoreboard.prepare_add(holder, objective, delta)framework.scoreboard.add
ctx.scoreboard.prepare_reset(holder, objective)framework.scoreboard.reset
ctx.scoreboard.prepare_objective_add(objective, display_name=None)framework.scoreboard.objectives.add

display_name 是关键字参数。objective 为 1–16 个 ASCII 字母、数字、下划线、点或连字符。value 为有符号 32 位整数；delta 范围为 -2147483647 至 2147483647，负值扣分。加减分前确保计分项存在。读取结果的 found=false 表示该次查询没有分数，不应直接索引缺失的 value。

持久状态 ​

python
await ctx.storage.set("settings", {"enabled": True})
settings = await ctx.storage.get("settings", {})
count = await ctx.storage.increment("count", 1, default=0)
changed = await ctx.storage.compare_and_set("phase", "waiting", "running")
await ctx.storage.delete("temporary")

数据按安装身份隔离，停止、重载和 Host 重启后保留。compare_and_set 只替换存在且等于期望值的键，缺失与 JSON null 不相同。数据必须可编码为有限 JSON 值，不能存放 Python 对象或非有限浮点数。

python
tx = await ctx.storage.transaction(keys=["count", "phase"])
tx.increment("count")
tx.set("phase", "ready")
receipt = await tx.save()

transaction(*, event=None, commit_id=None, keys=None) 读取快照；修改留在本地，直到 save()。keys=None 读取全量，keys=[] 只读取状态 revision，显式键最多 256 个且唯一，每键 1–256 UTF-8 字节。storage.get 使用单键快照。

事务以整个安装的 revision 检查冲突，不同键上的并发修改也可能产生 revision_conflict。set/delete/increment/compare_and_set 便捷方法仅对明确的 revision 冲突最多尝试 4 次；任意业务事务不会自动重跑。

状态、事件确认与动作共同提交 ​

python
# 位于 on_event(ctx, event) 或聊天 handler 中。
tx = await event.transaction()
tx.increment("handled_events")
tx.action(intent)  # 使用前面 prepare 得到的 OperationIntent。
commit = await tx.save()
for operation_id in commit["operation_ids"]:
    receipt = await ctx.operations.get(operation_id)
    ctx.log.info("动作当前状态=%s", receipt["state"],
                 extra={"operation_id": operation_id})

同一次数据库事务保存状态、推进该事件确认，并登记动作意图。save() 返回含 commit_id、revision、duplicate 和 operation_ids 的回执；游戏动作随后执行，数据库提交成功不能替代游戏效果。

tx.prepare() 封存请求，之后不能修改该事务。若提交响应不确定，保留 CommitUncertain.intent，用 await ctx.storage.receipt(commit_id) 查询原提交。不要重新执行发奖逻辑或生成新 ID，完整处理见错误页。

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