3.5 环境验证

🎯 学习目标：通过全面的验证测试确保MCP开发环境完全就绪
⏱️ 预计时间：25分钟
📊 难度等级：⭐⭐

🎪 验证概览

环境验证是确保一切就绪的最后一步。我们将通过一系列测试来验证：

🔧 系统环境全面检查

🖥️ 综合系统验证脚本

创建一个全面的验证脚本：

#!/bin/bash
# comprehensive_check.sh - 综合环境验证脚本

set -e  # 遇到错误立即退出

# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

# 日志函数
log_info() { echo -e "${BLUE}ℹ️  $1${NC}"; }
log_success() { echo -e "${GREEN}✅ $1${NC}"; }
log_warning() { echo -e "${YELLOW}⚠️  $1${NC}"; }
log_error() { echo -e "${RED}❌ $1${NC}"; }

# 检查结果统计
TOTAL_CHECKS=0
PASSED_CHECKS=0
FAILED_CHECKS=0

check_result() {
    TOTAL_CHECKS=$((TOTAL_CHECKS + 1))
    if [ $1 -eq 0 ]; then
        PASSED_CHECKS=$((PASSED_CHECKS + 1))
        log_success "$2"
    else
        FAILED_CHECKS=$((FAILED_CHECKS + 1))
        log_error "$2"
    fi
}

echo "🔍 MCP开发环境综合验证"
echo "================================="
echo

# 1. 系统基础检查
log_info "1. 系统基础环境检查"
echo "-------------------"

# 操作系统检查
if [[ "$OSTYPE" == "linux-gnu"* ]]; then
    OS_INFO=$(lsb_release -d 2>/dev/null | cut -f2 || echo "Linux (未知版本)")
    check_result 0 "操作系统: $OS_INFO"
elif [[ "$OSTYPE" == "darwin"* ]]; then
    OS_INFO="macOS $(sw_vers -productVersion)"
    check_result 0 "操作系统: $OS_INFO"
elif [[ "$OSTYPE" == "msys" ]] || [[ "$OSTYPE" == "cygwin" ]]; then
    OS_INFO="Windows (通过Git Bash/WSL)"
    check_result 0 "操作系统: $OS_INFO"
else
    check_result 1 "操作系统: 未识别的系统 ($OSTYPE)"
fi

# 内存检查
if command -v free &> /dev/null; then
    TOTAL_MEM=$(free -m | awk 'NR==2{printf "%.1f", $2/1024}')
    if (( $(echo "$TOTAL_MEM >= 4.0" | bc -l) )); then
        check_result 0 "系统内存: ${TOTAL_MEM}GB (符合要求)"
    else
        check_result 1 "系统内存: ${TOTAL_MEM}GB (建议4GB以上)"
    fi
elif command -v vm_stat &> /dev/null; then
    # macOS内存检查
    check_result 0 "系统内存: macOS系统 (请手动确认 >= 4GB)"
else
    check_result 1 "系统内存: 无法检测"
fi

# 磁盘空间检查
AVAILABLE_SPACE=$(df -h . | awk 'NR==2 {print $4}')
check_result 0 "可用磁盘空间: $AVAILABLE_SPACE"

echo

# 2. Node.js环境检查
log_info "2. Node.js环境验证"
echo "-------------------"

if command -v node &> /dev/null; then
    NODE_VERSION=$(node --version)
    NODE_MAJOR=$(echo $NODE_VERSION | cut -d'.' -f1 | sed 's/v//')
    
    if [ $NODE_MAJOR -ge 18 ]; then
        check_result 0 "Node.js版本: $NODE_VERSION"
    else
        check_result 1 "Node.js版本: $NODE_VERSION (需要 >= 18.0)"
    fi
else
    check_result 1 "Node.js: 未安装"
fi

if command -v npm &> /dev/null; then
    NPM_VERSION=$(npm --version)
    check_result 0 "npm版本: v$NPM_VERSION"
    
    # 测试npm权限
    if npm list -g --depth=0 &> /dev/null; then
        check_result 0 "npm权限: 正常"
    else
        check_result 1 "npm权限: 存在问题"
    fi
else
    check_result 1 "npm: 未安装"
fi

echo

# 3. Python环境检查
log_info "3. Python环境验证"
echo "-------------------"

if command -v python3 &> /dev/null; then
    PYTHON_VERSION=$(python3 --version | cut -d' ' -f2)
    PYTHON_MAJOR=$(echo $PYTHON_VERSION | cut -d'.' -f1)
    PYTHON_MINOR=$(echo $PYTHON_VERSION | cut -d'.' -f2)
    
    if [ $PYTHON_MAJOR -eq 3 ] && [ $PYTHON_MINOR -ge 9 ]; then
        check_result 0 "Python版本: $PYTHON_VERSION"
    else
        check_result 1 "Python版本: $PYTHON_VERSION (需要 >= 3.9)"
    fi
    
    # 检查是否在虚拟环境中
    if [[ "$VIRTUAL_ENV" != "" ]]; then
        check_result 0 "虚拟环境: 激活 ($VIRTUAL_ENV)"
    else
        if [ -d "venv" ] || [ -d "env" ] || [ -d ".venv" ]; then
            check_result 1 "虚拟环境: 已创建但未激活"
        else
            check_result 1 "虚拟环境: 未创建"
        fi
    fi
else
    check_result 1 "Python3: 未安装"
fi

if command -v pip3 &> /dev/null; then
    PIP_VERSION=$(pip3 --version | cut -d' ' -f2)
    check_result 0 "pip版本: $PIP_VERSION"
else
    check_result 1 "pip3: 未安装"
fi

echo

# 4. Git版本控制检查
log_info "4. Git版本控制验证"
echo "-------------------"

if command -v git &> /dev/null; then
    GIT_VERSION=$(git --version | cut -d' ' -f3)
    check_result 0 "Git版本: $GIT_VERSION"
    
    # 检查Git配置
    if git config --global user.name &> /dev/null && git config --global user.email &> /dev/null; then
        GIT_USER=$(git config --global user.name)
        GIT_EMAIL=$(git config --global user.email)
        check_result 0 "Git配置: $GIT_USER <$GIT_EMAIL>"
    else
        check_result 1 "Git配置: 用户信息未设置"
    fi
else
    check_result 1 "Git: 未安装"
fi

echo

# 5. 网络连通性检查
log_info "5. 网络连通性验证"
echo "-------------------"

# 检查基础网络服务
SERVICES=("github.com" "npmjs.com" "pypi.org")
for service in "${SERVICES[@]}"; do
    if ping -c 1 "$service" &> /dev/null; then
        check_result 0 "网络连接: $service"
    else
        check_result 1 "网络连接: $service 无法访问"
    fi
done

# 检查HTTP(S)访问
if command -v curl &> /dev/null; then
    if curl -I https://api.github.com &> /dev/null; then
        check_result 0 "HTTPS访问: GitHub API"
    else
        check_result 1 "HTTPS访问: GitHub API 失败"
    fi
else
    check_result 1 "curl命令: 未安装"
fi

echo

# 6. 开发工具检查
log_info "6. 开发工具验证"
echo "-------------------"

# VS Code检查
if command -v code &> /dev/null; then
    check_result 0 "VS Code: 已安装"
else
    check_result 1 "VS Code: 未安装或未添加到PATH"
fi

# Docker检查（可选）
if command -v docker &> /dev/null; then
    if docker --version &> /dev/null; then
        DOCKER_VERSION=$(docker --version | cut -d' ' -f3 | sed 's/,//')
        check_result 0 "Docker: $DOCKER_VERSION (可选)"
    else
        check_result 1 "Docker: 已安装但无法运行"
    fi
else
    log_warning "Docker: 未安装 (可选工具)"
fi

echo

# 统计结果
echo "================================="
echo "📊 验证结果统计"
echo "================================="
echo "总检查项目: $TOTAL_CHECKS"
echo "通过检查: $PASSED_CHECKS"
echo "失败检查: $FAILED_CHECKS"

SUCCESS_RATE=$(( PASSED_CHECKS * 100 / TOTAL_CHECKS ))
echo "成功率: ${SUCCESS_RATE}%"

if [ $FAILED_CHECKS -eq 0 ]; then
    echo
    log_success "🎉 所有检查项目都通过了！环境配置完美！"
    exit 0
elif [ $SUCCESS_RATE -ge 80 ]; then
    echo
    log_warning "⚠️  环境基本就绪，但还有一些小问题需要解决"
    exit 1
else
    echo
    log_error "❌ 环境配置存在较多问题，需要重新配置"
    exit 2
fi

🪟 Windows PowerShell版本

# comprehensive_check.ps1 - Windows环境验证脚本

param(
    [switch]$Detailed = $false
)

# 检查统计
$Global:TotalChecks = 0
$Global:PassedChecks = 0
$Global:FailedChecks = 0

function Write-ColoredOutput($Message, $Color) {
    Write-Host $Message -ForegroundColor $Color
}

function Test-Requirement($TestName, $TestBlock) {
    $Global:TotalChecks++
    try {
        $result = & $TestBlock
        if ($result) {
            $Global:PassedChecks++
            Write-ColoredOutput "✅ $TestName" Green
            return $true
        } else {
            $Global:FailedChecks++
            Write-ColoredOutput "❌ $TestName" Red
            return $false
        }
    } catch {
        $Global:FailedChecks++
        Write-ColoredOutput "❌ $TestName - 错误: $($_.Exception.Message)" Red
        return $false
    }
}

Write-ColoredOutput "🔍 MCP开发环境综合验证 (Windows)" Cyan
Write-ColoredOutput "=================================" Cyan
Write-Host

# 1. 系统环境检查
Write-ColoredOutput "1. 系统环境检查" Yellow
Write-ColoredOutput "---------------" Yellow

Test-Requirement "Windows版本检查" {
    $version = [System.Environment]::OSVersion.Version
    if ($version.Major -ge 10) {
        Write-Host "   Windows $($version.Major).$($version.Minor)" -ForegroundColor Gray
        return $true
    }
    return $false
}

Test-Requirement "内存检查" {
    $memory = Get-CimInstance -ClassName Win32_ComputerSystem
    $totalMemGB = [math]::Round($memory.TotalPhysicalMemory / 1GB, 1)
    Write-Host "   总内存: $totalMemGB GB" -ForegroundColor Gray
    return $totalMemGB -ge 4
}

Test-Requirement "磁盘空间检查" {
    $disk = Get-CimInstance -ClassName Win32_LogicalDisk -Filter "DeviceID='C:'"
    $freeSpaceGB = [math]::Round($disk.FreeSpace / 1GB, 1)
    Write-Host "   C盘可用: $freeSpaceGB GB" -ForegroundColor Gray
    return $freeSpaceGB -ge 5
}

Write-Host

# 2. Node.js环境检查
Write-ColoredOutput "2. Node.js环境验证" Yellow
Write-ColoredOutput "------------------" Yellow

Test-Requirement "Node.js版本" {
    try {
        $nodeVersion = node --version
        $majorVersion = [int]($nodeVersion -replace 'v(\d+)\..*', '$1')
        Write-Host "   版本: $nodeVersion" -ForegroundColor Gray
        return $majorVersion -ge 18
    } catch {
        return $false
    }
}

Test-Requirement "npm版本" {
    try {
        $npmVersion = npm --version
        Write-Host "   版本: v$npmVersion" -ForegroundColor Gray
        return $true
    } catch {
        return $false
    }
}

Test-Requirement "npm权限测试" {
    try {
        npm list -g --depth=0 | Out-Null
        return $true
    } catch {
        return $false
    }
}

Write-Host

# 3. Python环境检查
Write-ColoredOutput "3. Python环境验证" Yellow
Write-ColoredOutput "------------------" Yellow

Test-Requirement "Python版本" {
    try {
        $pythonVersion = python --version 2>&1
        if ($pythonVersion -match "Python (\d+)\.(\d+)") {
            $major = [int]$matches[1]
            $minor = [int]$matches[2]
            Write-Host "   版本: $pythonVersion" -ForegroundColor Gray
            return ($major -eq 3 -and $minor -ge 9)
        }
        return $false
    } catch {
        return $false
    }
}

Test-Requirement "虚拟环境检查" {
    if ($env:VIRTUAL_ENV) {
        Write-Host "   虚拟环境已激活: $env:VIRTUAL_ENV" -ForegroundColor Gray
        return $true
    } elseif (Test-Path "venv" -or Test-Path "env" -or Test-Path ".venv") {
        Write-Host "   虚拟环境已创建但未激活" -ForegroundColor Yellow
        return $false
    } else {
        return $false
    }
}

Test-Requirement "pip版本" {
    try {
        $pipVersion = pip --version
        Write-Host "   $pipVersion" -ForegroundColor Gray
        return $true
    } catch {
        return $false
    }
}

Write-Host

# 4. Git检查
Write-ColoredOutput "4. Git版本控制验证" Yellow
Write-ColoredOutput "-------------------" Yellow

Test-Requirement "Git版本" {
    try {
        $gitVersion = git --version
        Write-Host "   $gitVersion" -ForegroundColor Gray
        return $true
    } catch {
        return $false
    }
}

Test-Requirement "Git配置" {
    try {
        $userName = git config --global user.name
        $userEmail = git config --global user.email
        if ($userName -and $userEmail) {
            Write-Host "   用户: $userName <$userEmail>" -ForegroundColor Gray
            return $true
        }
        return $false
    } catch {
        return $false
    }
}

Write-Host

# 5. 网络连通性检查
Write-ColoredOutput "5. 网络连通性验证" Yellow
Write-ColoredOutput "-------------------" Yellow

$services = @("github.com", "npmjs.com", "pypi.org")
foreach ($service in $services) {
    Test-Requirement "网络连接: $service" {
        Test-Connection -ComputerName $service -Count 1 -Quiet
    }
}

Write-Host

# 6. 开发工具检查
Write-ColoredOutput "6. 开发工具验证" Yellow
Write-ColoredOutput "-----------------" Yellow

Test-Requirement "VS Code" {
    try {
        code --version | Out-Null
        return $true
    } catch {
        return $false
    }
}

Test-Requirement "PowerShell版本" {
    $psVersion = $PSVersionTable.PSVersion
    Write-Host "   PowerShell $psVersion" -ForegroundColor Gray
    return $psVersion.Major -ge 5
}

Write-Host
Write-ColoredOutput "=================================" Cyan
Write-ColoredOutput "📊 验证结果统计" Cyan
Write-ColoredOutput "=================================" Cyan
Write-Host "总检查项目: $Global:TotalChecks" 
Write-Host "通过检查: $Global:PassedChecks"
Write-Host "失败检查: $Global:FailedChecks"

$successRate = [math]::Round(($Global:PassedChecks / $Global:TotalChecks) * 100, 1)
Write-Host "成功率: $successRate%"

if ($Global:FailedChecks -eq 0) {
    Write-Host
    Write-ColoredOutput "🎉 所有检查项目都通过了！环境配置完美！" Green
    exit 0
} elseif ($successRate -ge 80) {
    Write-Host
    Write-ColoredOutput "⚠️  环境基本就绪，但还有一些小问题需要解决" Yellow
    exit 1
} else {
    Write-Host
    Write-ColoredOutput "❌ 环境配置存在较多问题，需要重新配置" Red
    exit 2
}

🧪 MCP功能测试

📡 创建MCP功能验证器

#!/usr/bin/env python3
"""
mcp_functionality_test.py - MCP功能全面测试
"""

import asyncio
import json
import time
import sys
from typing import Dict, Any, List
import aiohttp
import websockets
from dataclasses import dataclass

@dataclass
class TestResult:
    name: str
    passed: bool
    duration: float
    error: str = ""

class MCPFunctionalityTester:
    def __init__(self):
        self.results: List[TestResult] = []
    
    async def test_json_rpc_parsing(self) -> TestResult:
        """测试JSON-RPC解析功能"""
        start_time = time.time()
        
        try:
            # 测试有效请求解析
            valid_request = {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "test_method",
                "params": {"key": "value"}
            }
            
            json_str = json.dumps(valid_request)
            parsed = json.loads(json_str)
            
            # 验证解析结果
            assert parsed["jsonrpc"] == "2.0"
            assert parsed["id"] == 1
            assert parsed["method"] == "test_method"
            assert parsed["params"]["key"] == "value"
            
            # 测试错误请求处理
            invalid_requests = [
                '{"jsonrpc": "1.0"}',  # 错误版本
                '{"id": 1}',           # 缺少必需字段
                '{"malformed": json}', # 格式错误
            ]
            
            for invalid in invalid_requests:
                try:
                    json.loads(invalid)
                except json.JSONDecodeError:
                    pass  # 预期的错误
            
            duration = time.time() - start_time
            return TestResult("JSON-RPC解析", True, duration)
            
        except Exception as e:
            duration = time.time() - start_time
            return TestResult("JSON-RPC解析", False, duration, str(e))
    
    async def test_async_capabilities(self) -> TestResult:
        """测试异步处理能力"""
        start_time = time.time()
        
        try:
            # 模拟异步任务
            async def async_task(delay: float, result: str):
                await asyncio.sleep(delay)
                return result
            
            # 并发执行多个任务
            tasks = [
                async_task(0.1, "task1"),
                async_task(0.2, "task2"),
                async_task(0.15, "task3")
            ]
            
            results = await asyncio.gather(*tasks)
            
            # 验证结果
            assert results == ["task1", "task2", "task3"]
            
            duration = time.time() - start_time
            return TestResult("异步处理能力", True, duration)
            
        except Exception as e:
            duration = time.time() - start_time
            return TestResult("异步处理能力", False, duration, str(e))
    
    async def test_http_client_capability(self) -> TestResult:
        """测试HTTP客户端能力"""
        start_time = time.time()
        
        try:
            # 测试HTTP GET请求
            async with aiohttp.ClientSession() as session:
                async with session.get('https://httpbin.org/json') as response:
                    assert response.status == 200
                    data = await response.json()
                    assert 'slideshow' in data
            
            duration = time.time() - start_time
            return TestResult("HTTP客户端", True, duration)
            
        except Exception as e:
            duration = time.time() - start_time
            return TestResult("HTTP客户端", False, duration, str(e))
    
    async def test_websocket_capability(self) -> TestResult:
        """测试WebSocket能力"""
        start_time = time.time()
        
        try:
            # 使用公共WebSocket测试服务
            uri = "wss://echo.websocket.org"
            
            async with websockets.connect(uri) as websocket:
                # 发送测试消息
                test_message = json.dumps({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "echo",
                    "params": {"message": "Hello MCP!"}
                })
                
                await websocket.send(test_message)
                response = await websocket.recv()
                
                # 验证回显
                assert response == test_message
            
            duration = time.time() - start_time
            return TestResult("WebSocket连接", True, duration)
            
        except Exception as e:
            duration = time.time() - start_time
            return TestResult("WebSocket连接", False, duration, str(e))
    
    async def test_data_processing(self) -> TestResult:
        """测试数据处理能力"""
        start_time = time.time()
        
        try:
            # 模拟MCP数据转换
            input_data = {
                "tools": [
                    {"name": "calculator", "description": "Basic calculator"},
                    {"name": "weather", "description": "Weather information"}
                ]
            }
            
            # 数据转换和验证
            processed_data = {
                "tool_count": len(input_data["tools"]),
                "tool_names": [tool["name"] for tool in input_data["tools"]],
                "has_calculator": any(tool["name"] == "calculator" for tool in input_data["tools"])
            }
            
            # 验证处理结果
            assert processed_data["tool_count"] == 2
            assert "calculator" in processed_data["tool_names"]
            assert processed_data["has_calculator"] is True
            
            duration = time.time() - start_time
            return TestResult("数据处理", True, duration)
            
        except Exception as e:
            duration = time.time() - start_time
            return TestResult("数据处理", False, duration, str(e))
    
    async def test_error_handling(self) -> TestResult:
        """测试错误处理能力"""
        start_time = time.time()
        
        try:
            # 测试各种错误情况
            def create_error_response(code: int, message: str, request_id: int = None):
                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {
                        "code": code,
                        "message": message
                    }
                }
            
            # 常见MCP错误代码
            errors = [
                create_error_response(-32700, "Parse error", 1),
                create_error_response(-32600, "Invalid Request", 2),
                create_error_response(-32601, "Method not found", 3),
                create_error_response(-32602, "Invalid params", 4),
                create_error_response(-32603, "Internal error", 5),
            ]
            
            # 验证错误响应格式
            for error in errors:
                assert error["jsonrpc"] == "2.0"
                assert "error" in error
                assert "code" in error["error"]
                assert "message" in error["error"]
            
            duration = time.time() - start_time
            return TestResult("错误处理", True, duration)
            
        except Exception as e:
            duration = time.time() - start_time
            return TestResult("错误处理", False, duration, str(e))
    
    async def run_all_tests(self) -> Dict[str, Any]:
        """运行所有测试"""
        print("🧪 开始MCP功能测试")
        print("=" * 50)
        
        # 定义所有测试
        tests = [
            self.test_json_rpc_parsing,
            self.test_async_capabilities,
            self.test_http_client_capability,
            self.test_websocket_capability,
            self.test_data_processing,
            self.test_error_handling
        ]
        
        # 执行测试
        for test in tests:
            print(f"🔍 执行测试: {test.__name__.replace('test_', '').replace('_', ' ').title()}")
            result = await test()
            self.results.append(result)
            
            if result.passed:
                print(f"  ✅ 通过 ({result.duration:.3f}s)")
            else:
                print(f"  ❌ 失败 ({result.duration:.3f}s) - {result.error}")
            print()
        
        # 统计结果
        total_tests = len(self.results)
        passed_tests = sum(1 for r in self.results if r.passed)
        failed_tests = total_tests - passed_tests
        total_duration = sum(r.duration for r in self.results)
        
        print("=" * 50)
        print("📊 测试结果统计")
        print("=" * 50)
        print(f"总测试数: {total_tests}")
        print(f"通过测试: {passed_tests}")
        print(f"失败测试: {failed_tests}")
        print(f"总耗时: {total_duration:.3f}s")
        print(f"成功率: {(passed_tests/total_tests)*100:.1f}%")
        
        if failed_tests == 0:
            print("\n🎉 所有功能测试都通过了！MCP功能完全就绪！")
        else:
            print(f"\n⚠️  有 {failed_tests} 项测试失败，请检查相关功能")
        
        return {
            "total": total_tests,
            "passed": passed_tests,
            "failed": failed_tests,
            "success_rate": (passed_tests/total_tests)*100,
            "total_duration": total_duration,
            "results": [
                {
                    "name": r.name,
                    "passed": r.passed,
                    "duration": r.duration,
                    "error": r.error
                }
                for r in self.results
            ]
        }

async def main():
    """主函数"""
    tester = MCPFunctionalityTester()
    
    try:
        results = await tester.run_all_tests()
        
        # 根据结果设置退出码
        if results["failed"] == 0:
            sys.exit(0)  # 成功
        else:
            sys.exit(1)  # 有失败
            
    except KeyboardInterrupt:
        print("\n⚠️  测试被用户中断")
        sys.exit(130)
    except Exception as e:
        print(f"\n❌ 测试执行出错: {e}")
        sys.exit(1)

if __name__ == "__main__":
    # 检查必需的依赖
    required_packages = ["aiohttp", "websockets"]
    missing_packages = []
    
    for package in required_packages:
        try:
            __import__(package)
        except ImportError:
            missing_packages.append(package)
    
    if missing_packages:
        print(f"❌ 缺少必需包: {', '.join(missing_packages)}")
        print(f"请运行: pip install {' '.join(missing_packages)}")
        sys.exit(1)
    
    asyncio.run(main())

🚀 性能基准测试

⚡ 创建性能测试脚本

#!/usr/bin/env python3
"""
performance_benchmark.py - MCP性能基准测试
"""

import asyncio
import time
import psutil
import json
import sys
from typing import Dict, List, Any
from dataclasses import dataclass
import statistics

@dataclass
class BenchmarkResult:
    name: str
    operations: int
    duration: float
    throughput: float
    memory_usage: float
    cpu_usage: float

class MCPPerformanceBenchmark:
    def __init__(self):
        self.results: List[BenchmarkResult] = []
        self.process = psutil.Process()
    
    def measure_performance(self, func, *args, **kwargs):
        """性能测量装饰器"""
        # 记录初始状态
        initial_memory = self.process.memory_info().rss / 1024 / 1024  # MB
        initial_cpu = self.process.cpu_percent()
        
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        
        # 记录结束状态
        final_memory = self.process.memory_info().rss / 1024 / 1024  # MB
        final_cpu = self.process.cpu_percent()
        
        duration = end_time - start_time
        memory_usage = final_memory - initial_memory
        cpu_usage = final_cpu
        
        return result, duration, memory_usage, cpu_usage
    
    def benchmark_json_processing(self, operations: int = 10000) -> BenchmarkResult:
        """JSON处理性能测试"""
        print(f"🔍 JSON处理性能测试 ({operations} 次操作)")
        
        def json_operations():
            for i in range(operations):
                # 创建复杂JSON对象
                data = {
                    "jsonrpc": "2.0",
                    "id": i,
                    "method": "tools/call",
                    "params": {
                        "name": f"tool_{i}",
                        "arguments": {
                            "data": list(range(100)),
                            "nested": {
                                "key": f"value_{i}",
                                "numbers": [j * 2 for j in range(50)]
                            }
                        }
                    }
                }
                
                # 序列化和反序列化
                json_str = json.dumps(data)
                parsed = json.loads(json_str)
                
                # 简单验证
                assert parsed["id"] == i
        
        _, duration, memory_usage, cpu_usage = self.measure_performance(json_operations)
        throughput = operations / duration
        
        result = BenchmarkResult(
            "JSON处理",
            operations,
            duration,
            throughput,
            memory_usage,
            cpu_usage
        )
        
        print(f"  ⏱️  耗时: {duration:.3f}s")
        print(f"  🚀 吞吐量: {throughput:.0f} ops/s")
        print(f"  💾 内存使用: {memory_usage:.2f}MB")
        print()
        
        return result
    
    async def benchmark_async_operations(self, operations: int = 1000) -> BenchmarkResult:
        """异步操作性能测试"""
        print(f"🔍 异步操作性能测试 ({operations} 次操作)")
        
        async def async_operations():
            async def single_operation(i: int):
                # 模拟I/O操作
                await asyncio.sleep(0.001)
                return {"result": i * 2, "processed": True}
            
            # 并发执行
            tasks = [single_operation(i) for i in range(operations)]
            results = await asyncio.gather(*tasks)
            
            # 验证结果
            assert len(results) == operations
            for i, result in enumerate(results):
                assert result["result"] == i * 2
        
        start_time = time.time()
        initial_memory = self.process.memory_info().rss / 1024 / 1024
        
        await async_operations()
        
        end_time = time.time()
        final_memory = self.process.memory_info().rss / 1024 / 1024
        
        duration = end_time - start_time
        memory_usage = final_memory - initial_memory
        throughput = operations / duration
        cpu_usage = self.process.cpu_percent()
        
        result = BenchmarkResult(
            "异步操作",
            operations,
            duration,
            throughput,
            memory_usage,
            cpu_usage
        )
        
        print(f"  ⏱️  耗时: {duration:.3f}s")
        print(f"  🚀 吞吐量: {throughput:.0f} ops/s")  
        print(f"  💾 内存使用: {memory_usage:.2f}MB")
        print()
        
        return result
    
    def benchmark_memory_usage(self, data_size: int = 10000) -> BenchmarkResult:
        """内存使用测试"""
        print(f"🔍 内存使用测试 (数据大小: {data_size})")
        
        def memory_operations():
            # 创建大量数据
            data_store = []
            for i in range(data_size):
                item = {
                    "id": i,
                    "data": f"data_item_{i}" * 10,  # 增加数据大小
                    "numbers": list(range(100)),
                    "metadata": {
                        "created": time.time(),
                        "processed": False,
                        "tags": [f"tag_{j}" for j in range(10)]
                    }
                }
                data_store.append(item)
            
            # 处理数据
            processed_count = 0
            for item in data_store:
                item["metadata"]["processed"] = True
                processed_count += 1
            
            return processed_count
        
        _, duration, memory_usage, cpu_usage = self.measure_performance(memory_operations)
        throughput = data_size / duration
        
        result = BenchmarkResult(
            "内存使用",
            data_size,
            duration,
            throughput,
            memory_usage,
            cpu_usage
        )
        
        print(f"  ⏱️  耗时: {duration:.3f}s")
        print(f"  🚀 处理速度: {throughput:.0f} items/s")
        print(f"  💾 内存增长: {memory_usage:.2f}MB")
        print()
        
        return result
    
    async def run_all_benchmarks(self) -> Dict[str, Any]:
        """运行所有性能测试"""
        print("🚀 开始MCP性能基准测试")
        print("=" * 50)
        
        # JSON处理测试
        json_result = self.benchmark_json_processing(10000)
        self.results.append(json_result)
        
        # 异步操作测试
        async_result = await self.benchmark_async_operations(1000)
        self.results.append(async_result)
        
        # 内存使用测试
        memory_result = self.benchmark_memory_usage(5000)
        self.results.append(memory_result)
        
        # 生成报告
        print("=" * 50)
        print("📊 性能测试报告")
        print("=" * 50)
        
        for result in self.results:
            print(f"\n{result.name}:")
            print(f"  操作数量: {result.operations}")
            print(f"  总耗时: {result.duration:.3f}s")
            print(f"  吞吐量: {result.throughput:.0f} ops/s")
            print(f"  内存使用: {result.memory_usage:.2f}MB")
            print(f"  CPU使用: {result.cpu_usage:.1f}%")
        
        # 性能评估
        avg_throughput = statistics.mean([r.throughput for r in self.results])
        total_memory = sum([r.memory_usage for r in self.results])
        avg_cpu = statistics.mean([r.cpu_usage for r in self.results])
        
        print(f"\n🎯 综合性能指标:")
        print(f"  平均吞吐量: {avg_throughput:.0f} ops/s")
        print(f"  总内存使用: {total_memory:.2f}MB")
        print(f"  平均CPU使用: {avg_cpu:.1f}%")
        
        # 性能等级评估
        if avg_throughput > 5000 and total_memory < 100:
            performance_grade = "优秀"
            grade_emoji = "🏆"
        elif avg_throughput > 2000 and total_memory < 200:
            performance_grade = "良好"
            grade_emoji = "✅"
        elif avg_throughput > 1000:
            performance_grade = "一般"
            grade_emoji = "⚠️"
        else:
            performance_grade = "较差"
            grade_emoji = "❌"
        
        print(f"\n{grade_emoji} 性能等级: {performance_grade}")
        
        return {
            "results": [
                {
                    "name": r.name,
                    "operations": r.operations,
                    "duration": r.duration,
                    "throughput": r.throughput,
                    "memory_usage": r.memory_usage,
                    "cpu_usage": r.cpu_usage
                }
                for r in self.results
            ],
            "summary": {
                "avg_throughput": avg_throughput,
                "total_memory": total_memory,
                "avg_cpu": avg_cpu,
                "performance_grade": performance_grade
            }
        }

async def main():
    """主函数"""
    benchmark = MCPPerformanceBenchmark()
    
    try:
        results = await benchmark.run_all_benchmarks()
        
        # 保存结果到文件
        with open("performance_report.json", "w", encoding="utf-8") as f:
            json.dump(results, f, indent=2, ensure_ascii=False)
        
        print(f"\n📄 详细报告已保存到: performance_report.json")
        
    except KeyboardInterrupt:
        print("\n⚠️  测试被用户中断")
        sys.exit(130)
    except Exception as e:
        print(f"\n❌ 测试执行出错: {e}")
        sys.exit(1)

if __name__ == "__main__":
    # 检查psutil依赖
    try:
        import psutil
    except ImportError:
        print("❌ 缺少psutil包")
        print("请运行: pip install psutil")
        sys.exit(1)
    
    asyncio.run(main())

🎯 完整验证流程

📋 创建一键验证脚本

#!/bin/bash
# one_click_verification.sh - 一键完整验证

echo "🎪 MCP开发环境一键验证"
echo "=========================="
echo

# 1. 系统环境检查
echo "🔧 1. 运行系统环境检查..."
if bash comprehensive_check.sh; then
    echo "✅ 系统环境检查通过"
else
    echo "❌ 系统环境检查失败，请先解决基础环境问题"
    exit 1
fi

echo

# 2. MCP功能测试
echo "🧪 2. 运行MCP功能测试..."
if python3 mcp_functionality_test.py; then
    echo "✅ MCP功能测试通过"
else
    echo "❌ MCP功能测试失败，请检查功能实现"
    # 不退出，继续性能测试
fi

echo

# 3. 性能基准测试
echo "🚀 3. 运行性能基准测试..."
if python3 performance_benchmark.py; then
    echo "✅ 性能基准测试完成"
else
    echo "⚠️  性能基准测试出现问题"
fi

echo
echo "🎉 完整验证流程执行完毕！"
echo "详细结果请查看各个测试的输出和生成的报告文件。"

🎯 本节小结

通过这一小节，你应该已经完成了：

✅ 系统环境验证：全面检查硬件、软件、网络环境
✅ 功能测试：验证MCP相关功能的正确性
✅ 性能基准：测试环境的性能指标
✅ 一键验证：整合所有验证步骤的自动化脚本

🏆 验证成功标准

优秀环境（推荐生产使用）：

• ✅ 系统检查成功率 100%
• ✅ 功能测试通过率 100%
• ✅ 性能等级"优秀"或"良好"

合格环境（适合学习开发）：

• ✅ 系统检查成功率 ≥ 80%
• ✅ 功能测试通过率 ≥ 80%
• ✅ 性能等级"一般"以上

需要改进：

• ❌ 系统检查成功率 < 80%
• ❌ 功能测试通过率 < 80%
• ❌ 性能等级"较差"

🔗 最终检查清单

恭喜你！如果通过了所有验证，你的MCP开发环境已经完全就绪！

• [ ] 系统环境检查通过
• [ ] Node.js和Python环境正常
• [ ] 开发工具配置完成
• [ ] 网络连接正常
• [ ] MCP功能测试通过
• [ ] 性能基准测试完成

🚀 现在你可以开始第一个MCP项目的开发了！

---

👉 下一章：4.1 第一个MCP服务器