feat: add dynamic registration probe plugin with capabilities for skill and MCP management

- Implemented DynamicRegistrationProbe class with methods for registering, listing, and unregistering skills.
- Added capabilities for managing global MCP servers including registration, listing, enabling, disabling, and unregistration.
- Created plugin.yaml for dynamic_registration_probe with necessary metadata.
- Added runtime probe skill documentation.
- Developed unit tests for context API round trip, including file, platform, provider, and session management.
- Implemented tests for dynamic registration and lifecycle of skills and MCP servers.
- Ensured proper handling of global MCP risk acknowledgment in plugin capabilities.

Author: whatevertogo

tests/fixtures/sdk_plugins/dynamic_registration_probe/main.py

@@ -0,0 +1,133 @@
+from pathlib import Path
+
+from astrbot_sdk import Context, Star, acknowledge_global_mcp_risk
+from astrbot_sdk.decorators import provide_capability
+
+
+@acknowledge_global_mcp_risk
+class DynamicRegistrationProbe(Star):
+    @staticmethod
+    def _skill_dir() -> Path:
+        return Path(__file__).resolve().parent / "skills" / "runtime_probe"
+
+    @staticmethod
+    def _skill_payload(record) -> dict:
+        return {
+            "name": record.name,
+            "description": record.description,
+            "path": record.path,
+            "skill_dir": record.skill_dir,
+        }
+
+    @staticmethod
+    def _mcp_payload(record) -> dict | None:
+        if record is None:
+            return None
+        return {
+            "name": record.name,
+            "scope": record.scope.value,
+            "active": record.active,
+            "running": record.running,
+            "config": dict(record.config),
+            "tools": list(record.tools),
+            "errlogs": list(record.errlogs),
+            "last_error": record.last_error,
+        }
+
+    @provide_capability(
+        "dynamic_probe.skill.register",
+        description="Register the probe skill through ctx.skills",
+    )
+    async def register_skill_capability(self, payload: dict, ctx: Context) -> dict:
+        description = str(payload.get("description", "Runtime probe skill"))
+        record = await ctx.skills.register(
+            name=str(payload.get("name", "dynamic_probe.runtime_probe")),
+            path=str(self._skill_dir()),
+            description=description,
+        )
+        return self._skill_payload(record)
+
+    @provide_capability(
+        "dynamic_probe.skill.list",
+        description="List registered probe skills through ctx.skills",
+    )
+    async def list_skill_capability(self, payload: dict, ctx: Context) -> dict:
+        del payload
+        items = await ctx.skills.list()
+        return {"skills": [self._skill_payload(item) for item in items]}
+
+    @provide_capability(
+        "dynamic_probe.skill.unregister",
+        description="Unregister the probe skill through ctx.skills",
+    )
+    async def unregister_skill_capability(self, payload: dict, ctx: Context) -> dict:
+        removed = await ctx.skills.unregister(
+            str(payload.get("name", "dynamic_probe.runtime_probe"))
+        )
+        return {"removed": bool(removed)}
+
+    @provide_capability(
+        "dynamic_probe.mcp.global.register",
+        description="Register a global MCP server through ctx.mcp",
+    )
+    async def register_global_mcp_capability(self, payload: dict, ctx: Context) -> dict:
+        record = await ctx.mcp.register_global_server(
+            str(payload.get("name", "probe-global")),
+            dict(payload.get("config", {"mock_tools": ["inspect"]})),
+            timeout=float(payload.get("timeout", 0.2)),
+        )
+        return {"server": self._mcp_payload(record)}
+
+    @provide_capability(
+        "dynamic_probe.mcp.global.get",
+        description="Get a global MCP server through ctx.mcp",
+    )
+    async def get_global_mcp_capability(self, payload: dict, ctx: Context) -> dict:
+        record = await ctx.mcp.get_global_server(
+            str(payload.get("name", "probe-global"))
+        )
+        return {"server": self._mcp_payload(record)}
+
+    @provide_capability(
+        "dynamic_probe.mcp.global.list",
+        description="List global MCP servers through ctx.mcp",
+    )
+    async def list_global_mcp_capability(self, payload: dict, ctx: Context) -> dict:
+        del payload
+        records = await ctx.mcp.list_global_servers()
+        return {"servers": [self._mcp_payload(record) for record in records]}
+
+    @provide_capability(
+        "dynamic_probe.mcp.global.disable",
+        description="Disable a global MCP server through ctx.mcp",
+    )
+    async def disable_global_mcp_capability(self, payload: dict, ctx: Context) -> dict:
+        record = await ctx.mcp.disable_global_server(
+            str(payload.get("name", "probe-global"))
+        )
+        return {"server": self._mcp_payload(record)}
+
+    @provide_capability(
+        "dynamic_probe.mcp.global.enable",
+        description="Enable a global MCP server through ctx.mcp",
+    )
+    async def enable_global_mcp_capability(self, payload: dict, ctx: Context) -> dict:
+        record = await ctx.mcp.enable_global_server(
+            str(payload.get("name", "probe-global")),
+            timeout=float(payload.get("timeout", 0.2)),
+        )
+        return {"server": self._mcp_payload(record)}
+
+    @provide_capability(
+        "dynamic_probe.mcp.global.unregister",
+        description="Unregister a global MCP server through ctx.mcp",
+    )
+    async def unregister_global_mcp_capability(
+        self,
+        payload: dict,
+        ctx: Context,
+    ) -> dict:
+        record = await ctx.mcp.unregister_global_server(
+            str(payload.get("name", "probe-global"))
+        )
+        return {"server": self._mcp_payload(record)}

tests/fixtures/sdk_plugins/dynamic_registration_probe/plugin.yaml

@@ -0,0 +1,11 @@
+_schema_version: 2
+name: dynamic_registration_probe
+author: tests
+version: 1.0.0
+desc: Dynamic registration probe plugin
+
+runtime:
+  python: "3.12"
+
+components:
+  - class: main:DynamicRegistrationProbe

tests/fixtures/sdk_plugins/dynamic_registration_probe/requirements.txt

@@ -0,0 +1 @@
+

tests/fixtures/sdk_plugins/dynamic_registration_probe/skills/runtime_probe/SKILL.md

@@ -0,0 +1,3 @@
+# Runtime Probe Skill
+
+This skill exists to validate runtime registration through the SDK context.

tests/test_sdk/unit/_context_api_roundtrip.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import sys
 import types
 import uuid
@@ -245,6 +246,9 @@ def upsert_plugin(
             "version": str(metadata.get("version", "1.0.0")),
             "enabled": bool(metadata.get("enabled", True)),
             "reserved": bool(metadata.get("reserved", False)),
+            "acknowledge_global_mcp_risk": bool(
+                metadata.get("acknowledge_global_mcp_risk", False)
+            ),
             "support_platforms": list(metadata.get("support_platforms", [])),
         }
         self._plugin_configs.setdefault(plugin_id, dict(config or {}))
@@ -532,6 +536,70 @@ def unregister_skill(self, *, plugin_id: str, name: str) -> bool:
     def list_registered_skills(self, plugin_id: str) -> list[dict[str, str]]:
         return [dict(item) for item in self._skill_records.get(plugin_id, [])]
 
+    def acknowledges_global_mcp_risk(self, plugin_id: str) -> bool:
+        metadata = self._plugin_metadata.get(str(plugin_id), {})
+        return bool(metadata.get("acknowledge_global_mcp_risk", False))
+
+    def remove_plugin(self, plugin_id: str) -> None:
+        normalized_plugin_id = str(plugin_id)
+        self._plugin_metadata.pop(normalized_plugin_id, None)
+        self._plugin_configs.pop(normalized_plugin_id, None)
+        self._skill_records.pop(normalized_plugin_id, None)
+        self._handlers_by_plugin.pop(normalized_plugin_id, None)
+        self.http_routes.pop(normalized_plugin_id, None)
+        self._latest_request_context_by_plugin.pop(normalized_plugin_id, None)
+        request_ids = [
+            request_id
+            for request_id in self._request_contexts
+            if self.resolve_request_plugin_id(request_id) == normalized_plugin_id
+        ]
+        for request_id in request_ids:
+            request_context = self._request_contexts.pop(request_id, None)
+            self._request_overlays.pop(request_id, None)
+            if request_context is None:
+                continue
+            self._request_contexts_by_token.pop(request_context.dispatch_token, None)
+
+
+class FakeFunctionToolManager:
+    def __init__(self) -> None:
+        self.func_list: list[object] = []
+        self._config: dict[str, Any] = {"mcpServers": {}}
+        self.mcp_server_runtime_view: dict[str, Any] = {}
+
+    def load_mcp_config(self) -> dict[str, Any]:
+        return json.loads(json.dumps(self._config))
+
+    def save_mcp_config(self, config: dict[str, Any]) -> bool:
+        self._config = json.loads(json.dumps(config))
+        return True
+
+    async def enable_mcp_server(
+        self,
+        name: str,
+        config: dict[str, Any],
+        *_args,
+        **_kwargs,
+    ) -> None:
+        tools = [
+            SimpleNamespace(name=str(tool_name))
+            for tool_name in config.get("mock_tools", [f"{name}_tool"])
+            if str(tool_name).strip()
+        ]
+        self.mcp_server_runtime_view[str(name)] = SimpleNamespace(
+            client=SimpleNamespace(tools=tools, server_errlogs=[]),
+        )
+
+    async def disable_mcp_server(
+        self,
+        name: str | None = None,
+        **_kwargs,
+    ) -> None:
+        if name is None:
+            self.mcp_server_runtime_view.clear()
+            return
+        self.mcp_server_runtime_view.pop(str(name), None)
+
 
 @dataclass(slots=True)
 class FakeProviderMeta:
@@ -892,12 +960,14 @@ def __init__(
         self,
         *,
         plugin_bridge: FakePluginBridge,
+        func_tool_manager: FakeFunctionToolManager,
         provider_manager: FakeProviderManager,
         platforms: list[FakePlatform],
         config: FakeConfig,
         message_history_manager: FakeMessageHistoryManager,
     ) -> None:
         self._plugin_bridge = plugin_bridge
+        self._func_tool_manager = func_tool_manager
         self.provider_manager = provider_manager
         self.platform_manager = SimpleNamespace(get_insts=lambda: list(platforms))
         self._config = config
@@ -922,6 +992,9 @@ async def send_message(self, session: str, message_chain: MessageChain) -> None:
     def get_config(self) -> FakeConfig:
         return self._config
 
+    def get_llm_tool_manager(self) -> FakeFunctionToolManager:
+        return self._func_tool_manager
+
     def get_all_stars(self) -> list[Any]:
         return [
             SimpleNamespace(
@@ -1028,6 +1101,7 @@ class RoundTripRuntime:
     bridge: CoreCapabilityBridge
     peer: BridgeBackedPeer
     plugin_bridge: FakePluginBridge
+    func_tool_manager: FakeFunctionToolManager
     runtime_sp: FakeRuntimeSP
     star_context: FakeStarContext
     provider_manager: FakeProviderManager
@@ -1114,11 +1188,13 @@ def build_roundtrip_runtime(
     file_token_service = FakeFileTokenService()
     config = FakeConfig()
     plugin_bridge = FakePluginBridge()
+    func_tool_manager = FakeFunctionToolManager()
     chat_provider = FakeChatProvider("chat-provider-a", model="gpt-roundtrip")
     provider_manager = FakeProviderManager(chat_provider)
     message_history_manager = FakeMessageHistoryManager()
     star_context = FakeStarContext(
         plugin_bridge=plugin_bridge,
+        func_tool_manager=func_tool_manager,
         provider_manager=provider_manager,
         platforms=[FakePlatform()],
         config=config,
@@ -1156,6 +1232,7 @@ def build_roundtrip_runtime(
         bridge=bridge,
         peer=peer,
         plugin_bridge=plugin_bridge,
+        func_tool_manager=func_tool_manager,
         runtime_sp=runtime_sp,
         star_context=star_context,
         provider_manager=provider_manager,

tests/test_sdk/unit/test_context_api_files_round_trip.py

@@ -0,0 +1,91 @@
+# ruff: noqa: E402
+"""Files 客户端 Core Bridge 集成测试。
+
+测试覆盖 01_context_api.md 中 ctx.files 的所有方法：
+- register_file(): 注册文件并获取令牌
+- handle_file(): 通过令牌解析文件路径
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.test_sdk.unit._context_api_roundtrip import build_roundtrip_runtime
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_context_files_register_file_returns_token(tmp_path, monkeypatch):
+    """register_file 注册文件并返回 token。"""
+    runtime = build_roundtrip_runtime(monkeypatch, tmp_path=tmp_path)
+    ctx = runtime.make_context("plugin-a")
+
+    # 创建测试文件
+    test_file = tmp_path / "test_image.jpg"
+    test_file.write_text("fake image content")
+
+    token = await ctx.files.register_file(str(test_file))
+
+    assert token is not None
+    assert token.startswith("file-token-")
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_context_files_register_file_with_timeout(tmp_path, monkeypatch):
+    """register_file 支持 timeout 参数。"""
+    runtime = build_roundtrip_runtime(monkeypatch, tmp_path=tmp_path)
+    ctx = runtime.make_context("plugin-a")
+
+    test_file = tmp_path / "timeout_test.png"
+    test_file.write_text("content")
+
+    token = await ctx.files.register_file(str(test_file), timeout=3600)
+
+    assert token is not None
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_context_files_handle_file_resolves_token(tmp_path, monkeypatch):
+    """handle_file 通过 token 解析回原始文件路径。"""
+    runtime = build_roundtrip_runtime(monkeypatch, tmp_path=tmp_path)
+    ctx = runtime.make_context("plugin-a")
+
+    test_file = tmp_path / "resolve_test.txt"
+    test_file.write_text("test content")
+
+    # 先注册
+    token = await ctx.files.register_file(str(test_file))
+
+    # 再解析
+    resolved_path = await ctx.files.handle_file(token)
+
+    assert Path(resolved_path) == test_file
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_context_files_round_trip_workflow(tmp_path, monkeypatch):
+    """完整的文件注册和解析工作流。"""
+    runtime = build_roundtrip_runtime(monkeypatch, tmp_path=tmp_path)
+    ctx = runtime.make_context("plugin-a")
+
+    # 创建多个测试文件
+    files = []
+    for i in range(3):
+        file_path = tmp_path / f"file_{i}.dat"
+        file_path.write_text(f"content {i}")
+        files.append(file_path)
+
+    # 注册所有文件
+    tokens = []
+    for file_path in files:
+        token = await ctx.files.register_file(str(file_path))
+        tokens.append(token)
+
+    # 验证每个 token 都能解析回正确的路径
+    for token, expected_path in zip(tokens, files):
+        resolved = await ctx.files.handle_file(token)
+        assert Path(resolved) == expected_path