高级主题

Router 组合

Extension 插件类型和 plugin.sdk.extension 门面均已移除。PluginRouter 仍可作为普通 Plugin 内部的代码组织工具；Router 应与所属 Plugin 放在同一源码树并显式挂载：

from plugin.sdk.plugin import PluginRouter, plugin_entry, Ok


class ExtraRouter(PluginRouter):
    @plugin_entry(id="extra_command", description="额外命令")
    async def extra_command(self, param: str = "", **_):
        return Ok({"param": param})

在所属 NekoPluginBase 的构造函数中调用 self.include_router(ExtraRouter(name="extra"))。原 Extension 必须合并进该 Plugin 的源码树，或改造成独立的普通 Plugin；type = "extension"、[plugin.host] 和 plugin.sdk.extension 导入都会被拒绝。参见 v0.9 迁移指南。

---

适配器（Adapter）

适配器将外部协议（MCP、NoneBot 等）桥接到内部插件调用。它们实现了一个网关管线模式。

何时使用适配器

• 你想通过 MCP（模型上下文协议）暴露 N.E.K.O 插件
• 你想接受 NoneBot 消息并将其路由到插件
• 你想将任何外部协议桥接到插件系统

适配器网关管线

External Request → Normalizer → PolicyEngine → RouteEngine → PluginInvoker → ResponseSerializer → External Response

阶段	职责
Normalizer	将外部协议格式转换为 GatewayRequest
PolicyEngine	访问控制、速率限制、验证
RouteEngine	决定调用哪个插件/入口
PluginInvoker	执行实际的插件调用
ResponseSerializer	将结果转换回外部协议格式

创建适配器

from plugin.sdk.plugin import neko_plugin, plugin_entry, lifecycle, Ok, Err, SdkError
from plugin.sdk.adapter import (
    AdapterGatewayCore, DefaultPolicyEngine, NekoAdapterPlugin,
)
from plugin.sdk.adapter.gateway_models import ExternalRequest

@neko_plugin
class MyProtocolAdapter(NekoAdapterPlugin):
    def __init__(self, ctx):
        super().__init__(ctx)
        self.gateway = None

    @lifecycle(id="startup")
    async def startup(self, **_):
        self.gateway = AdapterGatewayCore(
            normalizer=MyNormalizer(),
            policy_engine=DefaultPolicyEngine(),
            route_engine=MyRouteEngine(),
            invoker=MyInvoker(self.ctx),
            serializer=MySerializer(),
            logger=self.logger,
        )
        return Ok({"status": "ready"})

    @plugin_entry(id="handle_request")
    async def handle_request(self, raw_data: dict, **_):
        external = ExternalRequest(protocol="my_protocol", raw=raw_data)
        response = await self.gateway.process(external)
        return Ok(response.to_dict())

适配器模式

模式	说明
GATEWAY	完整管线处理
ROUTER	仅路由（跳过策略）
BRIDGE	直接透传
HYBRID	按请求选择模式

内置参考：MCP 适配器

参见 plugin/plugins/mcp_adapter/ 获取完整的适配器实现，它将 MCP 协议桥接到 N.E.K.O 插件。其中演示了：

• 自定义规范化器（MCPRequestNormalizer）
• 自定义路由引擎（MCPRouteEngine）
• 自定义调用器（MCPPluginInvoker）
• 自定义序列化器（MCPResponseSerializer）
• 自定义传输层（MCPTransportAdapter）

---

跨插件通信

直接入口调用

# 调用另一个插件的入口点
result = await self.plugins.call_entry("target_plugin:entry_id", {"arg": "value"})

if isinstance(result, Ok):
    data = result.value
else:
    self.logger.error(f"Call failed: {result.error}")

发现

# 列出所有可用的插件
plugins = await self.plugins.list(enabled=True)

# 检查依赖是否存在
exists = await self.plugins.exists("required_plugin")

# 要求某个插件存在（如果缺失则快速失败）
dep = await self.plugins.require_enabled("required_plugin")

Bus 读取与监听

self.bus 暴露五个可读命名空间快照：messages、events、lifecycle、conversations、memory，且没有 emit() 或 on() 方法。只有 messages、events、lifecycle 支持 watch()；conversations 与 memory 是只读快照。

# 在异步入口中必须 await get()
events = await self.bus.events.get(plugin_id=self.plugin_id, max_count=50)
recent = events.filter(priority_min=1).sort(by="timestamp", reverse=True).limit(20)

# subscribe() 仅接受 "add"、"del"、"change"
watcher = recent.watch(self.ctx)

@watcher.subscribe(on="add")
def _handle_event(delta):
    for event in delta.added:
        self.logger.info(f"new event: {event.type}")

watcher.start()

可调用形式 filter(predicate)、where(predicate) 与 sort(key=callable) 只处理当前已经物化的本地快照，不能由 watch() 重放。需要监听的链必须像上例一样使用结构化 filter(field=value, ...) 与 sort(by=...)。

最近记忆记录使用 await self.bus.memory.get(bucket_id="default", limit=...)，语义检索使用 await self.ctx.query_memory("default", query)。旧的高层 self.memory / MemoryClient 已不存在。

---

异步编程

入口点可以是同步或异步的：

# 同步入口（在线程池中运行）
@plugin_entry(id="sync_task")
def sync_task(self, **_):
    return Ok({"result": "done"})

# 异步入口（在事件循环中运行）
@plugin_entry(id="async_task")
async def async_task(self, url: str, **_):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return Ok({"data": await response.json()})

---

线程安全

定时任务在独立线程中运行。请保护共享状态：

import threading

@neko_plugin
class ThreadSafePlugin(NekoPluginBase):
    def __init__(self, ctx):
        super().__init__(ctx)
        self._lock = threading.Lock()
        self._counter = 0

    @plugin_entry(id="increment")
    def increment(self, **_):
        with self._lock:
            self._counter += 1
            return Ok({"count": self._counter})

    @timer_interval(id="report", seconds=60, auto_start=True)
    def report(self, **_):
        with self._lock:
            count = self._counter
        self.report_status({"count": count})

---

自定义配置

import json

class ConfigurablePlugin(NekoPluginBase):
    def __init__(self, ctx):
        super().__init__(ctx)
        config_file = self.config_dir / "config.json"
        if config_file.exists():
            self.config = json.loads(config_file.read_text())
        else:
            self.config = {"timeout": 30}

或使用 PluginConfig 进行带配置文件的结构化配置：

from plugin.sdk.plugin import PluginConfig

config = PluginConfig(self.ctx)
timeout = config.get("timeout", default=30)

---

使用 SQLite 进行数据持久化

import sqlite3

class PersistentPlugin(NekoPluginBase):
    def __init__(self, ctx):
        super().__init__(ctx)
        self.db_path = self.data_path("records.db")
        self.data_path().mkdir(parents=True, exist_ok=True)
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                key TEXT UNIQUE,
                value TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()
        conn.close()