| from app.plugins.models import PluginStatus |
| from app.plugins.tool_catalog import ToolCatalogService |
|
|
|
|
| class _Registry: |
| def get_tool_docs(self): |
| return { |
| "demo_plugin_run": { |
| "plugin": "demo_plugin", |
| "tool": "run", |
| "title": "运行任务", |
| "description": "执行演示任务", |
| "availability": "available", |
| "unavailable_reason": None, |
| "incomplete_reasons": None, |
| } |
| } |
|
|
|
|
| class _PluginManager: |
| def get_plugin_status(self, plugin_name: str): |
| return PluginStatus.ENABLED |
|
|
| def get_unavailable_reason(self, plugin_name: str): |
| return None |
|
|
|
|
| def test_sync_mcp_tools_adds_registered_tools_to_catalog(): |
| service = ToolCatalogService() |
|
|
| service.sync_mcp_tools(_Registry(), _PluginManager()) |
|
|
| tools = service.get_all_tools() |
| assert len(tools) == 1 |
| assert tools[0].name == "demo_plugin_run" |
| assert tools[0].source == "mcp" |
| assert tools[0].plugin_name == "demo_plugin" |
|
|
