mcp-manager

name	mcp-manager
description	管理 MCP（Model Context Protocol）服务的列出、添加、删除和启停操作
version	2.0.0
author	AstrBot Community
requires_admin	true
requires_computer_use	true

MCP Manager

管理 MCP（Model Context Protocol）服务连接。

使用场景

• "有哪些 MCP 服务？" / "列出 MCP 服务"
• "查看 database 服务的详情"
• "添加一个文件系统 MCP 服务"
• "删除 XXX 服务"
• "启用/禁用 XXX 服务"

可用工具

1. 列出 MCP 服务

async def list_mcp_servers(context):
    try:
        servers = await context.mcp_mgr.list_servers()
        if not servers:
            return "没有配置任何 MCP 服务。"
        
        lines = ["🔌 已配置的 MCP 服务：\n"]
        for srv in servers:
            name = srv.get("name", "")
            type_ = srv.get("type", "unknown")
            enabled = srv.get("enabled", True)
            status = "✅启用" if enabled else "❌禁用"
            lines.append(f"- [{status}] **{name}** ({type_})")
        
        return "\n".join(lines)
    except Exception as e:
        return f"❌ 列出 MCP 服务失败：{e}"

event = globals().get('event')
if not event.is_admin():
    print("❌ 错误：仅管理员可用")
    import sys; sys.exit(1)

context = globals().get('context')
result = await list_mcp_servers(context)
print(result)

2. 查看 MCP 服务详情

async def get_mcp_server(context, server_name: str):
    try:
        server = await context.mcp_mgr.get_server(server_name)
        if not server:
            return f"❌ MCP 服务「{server_name}」不存在。"
        
        name = server.get("name", "")
        type_ = server.get("type", "")
        command = server.get("command", "")
        args = server.get("args", [])
        env = server.get("env", {})
        enabled = server.get("enabled", True)
        
        lines = [
            f"🔌 MCP 服务详情：**{name}**\n",
            f"类型: {type_}",
            f"状态: {'✅启用' if enabled else '❌禁用'}",
            f"命令: {command}",
            f"参数: {' '.join(args)}" if args else "参数: 无",
        ]
        
        if env:
            lines.append("环境变量:")
            for k, v in env.items():
                lines.append(f"  - {k}={v}")
        
        return "\n".join(lines)
     Exception  e:
         

result =  get_mcp_server(context, server_name=)
(result)

3. 添加 MCP 服务

async def add_mcp_server(context, name: str, type_: str, command: str, args: list = None, env: dict = None):
    try:
        config = {
            "name": name,
            "type": type_,
            "command": command,
            "args": args or [],
            "env": env or {},
            "enabled": True
        }
        
        # 测试连接
        test_result = await context.mcp_mgr.test_connection(config)
        if not test_result.get("success"):
            return f"❌ 连接测试失败：{test_result.get('error')}"
        
        # 添加服务
        await context.mcp_mgr.add_server(config)
        return f"✅ MCP 服务「{name}」已添加。"
    except Exception as e:
        return f"❌ 添加失败：{e}"

# 示例：添加文件系统服务
result = await add_mcp_server(
    context,
    name="filesystem",
    type_="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"]
)
(result)

4. 删除 MCP 服务

async def remove_mcp_server(context, server_name: str):
    try:
        await context.mcp_mgr.remove_server(server_name)
        return f"✅ MCP 服务「{server_name}」已删除。"
    except Exception as e:
        return f"❌ 删除失败：{e}"

result = await remove_mcp_server(context, server_name="<服务名>")
print(result)

5. 启用/禁用 MCP 服务

async def toggle_mcp_server(context, server_name: str, enable: bool):
    try:
        await context.mcp_mgr.set_server_enabled(server_name, enable)
        action = "启用" if enable else "禁用"
        return f"✅ 已{action} MCP 服务「{server_name}」。"
    except Exception as e:
        return f"❌ 操作失败：{e}"

result = await toggle_mcp_server(context, server_name="<服务名>", enable=False)
print(result)

注意事项

• 添加 MCP 服务前会自动测试连接
• 确保命令和路径正确，否则服务无法启动
• 禁用的服务不会在 Agent 中加载其工具
• MCP 服务重启需要重载 AstrBot