Mcp Manager

#!/bin/bash
# MCP Manager - 命令行入口

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

# 健康检测
case "$1" in
    health)
        case "$2" in
            check)
                python3 "$SCRIPT_DIR/scripts/health_check.py" "${@:3}"
                ;;
            monitor)
                echo "🔍 MCP Health Monitor (Press Ctrl+C to stop)"
                echo "="
                while true; do
                    clear
                    python3 "$SCRIPT_DIR/scripts/health_check.py"
                    echo ""
                    echo "[Last check: $(date '+%Y-%m-%d %H:%M:%S')]"
                    echo "[Next check in 60 seconds...]"
                    sleep 60
                done
                ;;
            *)
                echo "Usage: mcp health <check|monitor> [options]"
                ;;
        esac
        ;;
    list)
        python3 "$SCRIPT_DIR/scripts/mcp_manager.py" list
        ;;
    show)
        python3 "$SCRIPT_DIR/scripts/mcp_manager.py" show "$2"
        ;;
    enable)
        python3 "$SCRIPT_DIR/scripts/mcp_manager.py" enable "$2"
        ;;
    disable)
        python3 "$SCRIPT_DIR/scripts/mcp_manager.py" disable "$2"
        ;;
    recommend)
        shift
        python3 "$SCRIPT_DIR/scripts/mcp_manager.py" recommend "$*"
        ;;
    help)
        cat << EOF
MCP Manager - MCP 服务器智能管理工具

命令：
  mcp list                          列出所有 MCP
  mcp show <name>                   显示 MCP 功能
  mcp recommend <task>              为任务推荐 MCP

  mcp health check [name]           健康检查
  mcp health monitor               持续监控

  mcp enable <name>                 启用 MCP
  mcp disable <name>                禁用 MCP

示例：
  mcp list                          查看所有 MCP
  mcp show github                   查看 GitHub MCP 功能
  mcp recommend 网页自动化          推荐适合的 MCP
  mcp health check                  检查所有 MCP 状态
  mcp health monitor               持续监控

文档：~/clawd/skills/mcp-manager/SKILL.md
EOF
        ;;
    *)
        python3 "$SCRIPT_DIR/scripts/mcp_manager.py" "$@"
        ;;
esac

#!/usr/bin/env python3
"""
MCP 健康检测脚本
"""
import json
import subprocess
import time
from datetime import datetime
from pathlib import Path

class MCPHealthChecker:
    def __init__(self, config_path=None):
        self.config_path = Path.home() / ".claude.json"
        self.config = self.load_config()
        self.results = {}

    def load_config(self):
        """加载 MCP 配置"""
        if self.config_path.exists():
            with open(self.config_path, 'r') as f:
                config = json.load(f)
                return config.get('mcpServers', {})
        return {}

    def test_mcp_connection(self, name, config):
        """测试 MCP 连接"""
        start_time = time.time()

        try:
            # 尝试运行 MCP 命令
            command = config.get('command', '')
            args = config.get('args', [])

            if not command:
                return {
                    'status': 'error',
                    'error': 'No command specified'
                }

            # 测试命令是否可用
            result = subprocess.run(
                ['which', command] if command != 'npx' else ['which', 'npx'],
                capture_output=True,
                timeout=5
            )

            if result.returncode != 0:
                return {
                    'status': 'error',
                    'error': f'Command not found: {command}'
                }

            # 计算响应时间
            elapsed = (time.time() - start_time) * 1000

            return {
                'status': 'ok',
                'response_time': round(elapsed, 2),
                'command': command
            }

        except subprocess.TimeoutExpired:
            return {
                'status': 'timeout',
                'error': 'Connection timeout'
            }
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e)
            }

    def check_all(self):
        """检查所有 MCP"""
        print("🔍 MCP Health Check")
        print("=" * 50)
        print()

        results = {}
        for name, config in self.config.items():
            result = self.test_mcp_connection(name, config)
            results[name] = result

            # 显示结果
            status_icon = {
                'ok': '✅',
                'timeout': '⏱️',
                'error': '❌'
            }.get(result['status'], '❓')

            print(f"{status_icon} {name}")

            if result['status'] == 'ok':
                print(f"   Status: OK ({result['response_time']}ms)")
                print(f"   Command: {result['command']}")
            else:
                print(f"   Status: {result['status'].upper()}")
                print(f"   Error: {result['error']}")

            print()

        self.results = results
        return results

    def check_single(self, name):
        """检查单个 MCP"""
        if name not in self.config:
            print(f"❌ MCP not found: {name}")
            return None

        print(f"🔍 Checking MCP: {name}")
        print("=" * 50)
        print()

        result = self.test_mcp_connection(name, config=self.config[name])

        status_icon = {
            'ok': '✅',
            'timeout': '⏱️',
            'error': '❌'
        }.get(result['status'], '❓')

        print(f"{status_icon} {name}")

        if result['status'] == 'ok':
            print(f"   Status: OK ({result['response_time']}ms)")
            print(f"   Command: {result['command']}")
            print(f"   Args: {' '.join(result.get('args', []))}")
        else:
            print(f"   Status: {result['status'].upper()}")
            print(f"   Error: {result['error']}")

        return result

    def get_summary(self):
        """获取摘要"""
        if not self.results:
            return "No results available"

        ok_count = sum(1 for r in self.results.values() if r['status'] == 'ok')
        error_count = len(self.results) - ok_count

        return f"OK: {ok_count}, Errors: {error_count}, Total: {len(self.results)}"

def main():
    import argparse

    parser = argparse.ArgumentParser(description="MCP Health Checker")
    parser.add_argument('name', nargs='?', help='MCP name to check (optional)')
    parser.add_argument('--json', action='store_true', help='Output as JSON')

    args = parser.parse_args()

    checker = MCPHealthChecker()

    if args.name:
        result = checker.check_single(args.name)
        if args.json and result:
            print(json.dumps(result, indent=2))
    else:
        results = checker.check_all()
        if args.json:
            print(json.dumps(results, indent=2))
        else:
            print()
            print("📊 Summary:", checker.get_summary())

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
MCP 管理脚本 - 开关、列表、功能说明
"""
import json
import subprocess
from pathlib import Path
from datetime import datetime

# MCP 功能库
MCP_CAPABILITIES = {
    "chrome-devtools": {
        "description": "Chrome DevTools MCP - 浏览器自动化工具",
        "can_do": [
            "🌐 自动化浏览器操作（点击、输入、导航）",
            "📸 截图和快照",
            "🔍 网络请求监控",
            "🐛 控制台日志查看",
            "⚡ 性能分析",
            "🖱️ 元素悬停和拖拽"
        ],
        "cannot_do": [
            "❌ 需要 API key 的外部服务调用",
            "❌ 代码执行（仅 JavaScript 评估）",
            "❌ 文件系统访问",
            "❌ 跨浏览器支持（仅 Chrome）"
        ],
        "use_cases": ["网页测试", "数据抓取", "UI 自动化", "性能分析"],
        "resource_usage": "High (~200MB)",
        "priority": "low"
    },
    "github": {
        "description": "GitHub MCP - GitHub 仓库操作工具",
        "can_do": [
            "📂 搜索仓库和代码",
            "🔍 查看 Issue 和 PR",
            "📊 获取仓库统计信息",
            "🌿 分支和标签管理",
            "👥 用户和仓库信息查询",
            "📝 查看 README 和文档"
        ],
        "cannot_do": [
            "❌ 修改代码（只读操作）",
            "❌ 创建/删除仓库",
            "❌ 管理 Issues（需要额外权限）",
            "❌ 执行 Git 命令"
        ],
        "use_cases": ["代码搜索", "仓库分析", "协作信息查询", "开源项目调研"],
        "resource_usage": "Low (~50MB)",
        "priority": "medium"
    },
    "context7": {
        "description": "Context7 MCP - 长期记忆存储",
        "can_do": [
            "🧠 长期记忆存储",
            "💾 保存和检索上下文",
            "🔗 跨会话信息共享",
            "📚 知识库管理",
            "🔍 语义搜索"
        ],
        "cannot_do": [
            "❌ 实时数据处理",
            "❌ 复杂数值计算",
            "❌ 图像/视频处理",
            "❌ 本地存储（需要网络连接）"
        ],
        "use_cases": ["长期记忆", "上下文保持", "知识管理", "个性化助手机"],
        "resource_usage": "Low (~50MB)",
        "priority": "medium"
    },
    "filesystem": {
        "description": "Filesystem MCP - 文件系统操作",
        "can_do": [
            "📁 读取和写入文件",
            "🔍 搜索文件内容",
            "📋 列出目录结构",
            "📝 创建和删除文件",
            "📊 统计文件信息"
        ],
        "cannot_do": [
            "❌ 执行系统命令",
            "❌ 访问受限目录",
            "❌ 修改系统配置",
            "❌ 符号链接操作"
        ],
        "use_cases": ["文件操作", "代码生成", "文档处理", "项目分析"],
        "resource_usage": "Low (~20MB)",
        "priority": "high"
    },
    "browser": {
        "description": "Browser MCP - 基础浏览器操作",
        "can_do": [
            "🌐 导航到网页",
            "📸 页面截图",
            "🔍 查看页面内容",
            "🖱️ 基本点击操作"
        ],
        "cannot_do": [
            "❌ 复杂表单填写",
            "❌ 多标签页管理",
            "❌ JavaScript 执行",
            "❌ Cookie 管理"
        ],
        "use_cases": ["简单网页访问", "内容抓取", "页面预览"],
        "resource_usage": "Medium (~100MB)",
        "priority": "low"
    }
}

class MCPManager:
    def __init__(self):
        self.config_path = Path.home() / ".claude.json"
        self.config = self.load_config()

    def load_config(self):
        """加载配置"""
        if self.config_path.exists():
            with open(self.config_path, 'r') as f:
                return json.load(f)
        return {}

    def save_config(self):
        """保存配置"""
        with open(self.config_path, 'w') as f:
            json.dump(self.config, f, indent=2)

    def list_mcps(self):
        """列出所有 MCP"""
        print("📦 MCP List")
        print("=" * 60)
        print()

        mcps = self.config.get('mcpServers', {})

        if not mcps:
            print("❌ No MCP servers configured")
            return

        for name, config in mcps.items():
            cap = MCP_CAPABILITIES.get(name, {})

            # 优先级图标
            priority_icon = {
                'high': '🔴',
                'medium': '🟡',
                'low': '🟢'
            }.get(cap.get('priority', 'low'), '⚪')

            print(f"{priority_icon} {name}")
            print(f"   {cap.get('description', 'No description')}")
            print(f"   资源占用: {cap.get('resource_usage', 'Unknown')}")
            print(f"   适用场景: {', '.join(cap.get('use_cases', []))}")
            print()

    def show_capabilities(self, name):
        """显示 MCP 功能"""
        cap = MCP_CAPABILITIES.get(name)

        if not cap:
            print(f"❌ MCP not found: {name}")
            return

        print(f"🎯 {name}")
        print("=" * 60)
        print()
        print(f"{cap.get('description', 'No description')}")
        print()

        print("✅ 能做什么：")
        for item in cap.get('can_do', []):
            print(f"  {item}")
        print()

        print("❌ 不能做什么：")
        for item in cap.get('cannot_do', []):
            print(f"  {item}")
        print()

        print("💡 适用场景：")
        for item in cap.get('use_cases', []):
            print(f"  • {item}")
        print()

        print(f"📊 资源占用: {cap.get('resource_usage', 'Unknown')}")
        print(f"⚡ 优先级: {cap.get('priority', 'unknown')}")

    def enable_mcp(self, name):
        """启用 MCP"""
        print(f"✅ 启用 MCP: {name}")
        # 实际实现：从 disabled 列表中移除
        # 这里只是示例
        print(f"   {name} 已启用")

    def disable_mcp(self, name):
        """禁用 MCP"""
        print(f"❌ 禁用 MCP: {name}")
        # 实际实现：添加到 disabled 列表
        # 这里只是示例
        print(f"   {name} 已禁用")

    def recommend(self, task):
        """推荐 MCP"""
        print(f"🤔 任务: {task}")
        print("=" * 60)
        print()

        recommendations = []

        for name, cap in MCP_CAPABILITIES.items():
            use_cases = cap.get('use_cases', [])
            for use_case in use_cases:
                if use_case in task or task in use_case:
                    recommendations.append((name, cap))
                    break

        if recommendations:
            print("💡 推荐使用的 MCP：")
            for name, cap in recommendations:
                print(f"  • {name} - {cap.get('description', '')}")
        else:
            print("❓ 没有找到匹配的 MCP")
            print()
            print("可用的 MCP：")
            for name in MCP_CAPABILITIES.keys():
                print(f"  • {name}")

def main():
    import argparse

    parser = argparse.ArgumentParser(description="MCP Manager")
    subparsers = parser.add_subparsers(dest='command', help='Commands')

    # list command
    subparsers.add_parser('list', help='List all MCPs')

    # show command
    show_parser = subparsers.add_parser('show', help='Show MCP capabilities')
    show_parser.add_argument('name', help='MCP name')

    # enable command
    enable_parser = subparsers.add_parser('enable', help='Enable MCP')
    enable_parser.add_argument('name', help='MCP name')

    # disable command
    disable_parser = subparsers.add_parser('disable', help='Disable MCP')
    disable_parser.add_argument('name', help='MCP name')

    # recommend command
    recommend_parser = subparsers.add_parser('recommend', help='Recommend MCP for task')
    recommend_parser.add_argument('task', help='Task description')

    args = parser.parse_args()

    manager = MCPManager()

    if args.command == 'list':
        manager.list_mcps()
    elif args.command == 'show':
        manager.show_capabilities(args.name)
    elif args.command == 'enable':
        manager.enable_mcp(args.name)
    elif args.command == 'disable':
        manager.disable_mcp(args.name)
    elif args.command == 'recommend':
        manager.recommend(args.task)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()