MCP 在 AI IDE 中的实践：从工具调用到智能开发助手

"""
MCP 智能代码补全引擎
将 MCP 工具调用与代码补全功能深度融合
"""

import json
import re
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class CodeContext:
    """代码上下文信息"""
    file_path: str
    file_type: str
    cursor_position: int
    current_line: str
    previous_lines: List[str]
    next_lines: List[str]
    project_root: str

@dataclass
class CompletionItem:
    """代码补全项"""
    label: str
    insert_text: str
    kind: str
    detail: str
    documentation: Optional[str] = None
    mcp_tool_call: Optional[Dict] = None

class MCPSmartCompletionEngine:
    def __init__(self, mcp_client, llm_client):
        self.mcp_client = mcp_client
        self.llm_client = llm_client
        self.tools = []
        self._load_mcp_tools()
    
    def _load_mcp_tools(self):
        """加载可用的 MCP 工具"""
        try:
            response = self.mcp_client.get_tools()
            self.tools = response.get('tools', [])
        except Exception as e:
            print(f"加载 MCP 工具失败: {e}")
            self.tools = []
    
    def _analyze_context(self, context: CodeContext) -> Dict:
        """分析代码上下文"""
        # 分析当前行的意图
        current_line = context.current_line.strip()
        
        # 检测是否需要调用 MCP 工具
        tool_intent = None
        
        # 示例：检测是否需要生成测试代码
        if 'test' in current_line.lower() or 'assert' in current_line.lower():
            tool_intent = 'generate_test'
        # 示例：检测是否需要重构代码
        elif 'refactor' in current_line.lower() or 'rename' in current_line.lower():
            tool_intent = 'refactor_code'
        # 示例：检测是否需要查询文档
        elif 'doc' in current_line.lower() or 'documentation' in current_line.lower():
            tool_intent = 'get_documentation'
        
        # 提取相关代码片段
        relevant_code = '\n'.join(context.previous_lines[-5:]) + '\n' + context.current_line
        
        return {
            'tool_intent': tool_intent,
            'relevant_code': relevant_code,
            'file_type': context.file_type,
            'project_root': context.project_root
        }
    
    def _select_mcp_tool(self, analysis_result: Dict) -> Optional[Dict]:
        """选择合适的 MCP 工具"""
        tool_intent = analysis_result.get('tool_intent')
        if not tool_intent:
            return None
        
        # 根据意图匹配工具
        for tool in self.tools:
            tool_name = tool.get('name', '').lower()
            tool_description = tool.get('description', '').lower()
            
            if tool_intent in tool_name or tool_intent in tool_description:
                return tool
        
        return None
    
    def _generate_tool_params(self, tool: Dict, analysis_result: Dict) -> Dict:
        """生成工具调用参数"""
        params = {}
        
        # 获取工具的参数定义
        parameters = tool.get('parameters', {}).get('properties', {})
        
        # 根据参数定义生成参数
        for param_name, param_schema in parameters.items():
            param_type = param_schema.get('type', 'string')
            
            if param_name == 'code' or param_name == 'source_code':
                params[param_name] = analysis_result.get('relevant_code', '')
            elif param_name == 'file_type' or param_name == 'language':
                params[param_name] = analysis_result.get('file_type', '')
            elif param_name == 'project_root':
                params[param_name] = analysis_result.get('project_root', '')
            
            # 处理必填参数
            if param_schema.get('required', False) and param_name not in params:
                # 尝试从 LLM 获取参数值
                params[param_name] = self._get_param_from_llm(param_name, param_schema, analysis_result)
        
        return params
    
    def _get_param_from_llm(self, param_name: str, param_schema: Dict, analysis_result: Dict) -> str:
        """从 LLM 获取参数值"""
        prompt = f"""
根据以下上下文，为参数 {param_name} 生成合适的值。
参数描述：{param_schema.get('description', '')}
参数类型：{param_schema.get('type', 'string')}
上下文代码：
{analysis_result.get('relevant_code', '')}
        """
        
        try:
            response = self.llm_client.generate(prompt)
            return response.strip()
        except Exception as e:
            print(f"从 LLM 获取参数失败: {e}")
            return ''
    
    def _call_mcp_tool(self, tool: Dict, params: Dict) -> Dict:
        """调用 MCP 工具"""
        try:
            response = self.mcp_client.call_tool(tool.get('name'), params)
            return response
        except Exception as e:
            print(f"调用 MCP 工具失败: {e}")
            return {'error': str(e)}
    
    def _generate_completion(self, tool: Dict, tool_response: Dict, analysis_result: Dict) -> List[CompletionItem]:
        """生成代码补全项"""
        completions = []
        
        if 'error' in tool_response:
            return completions
        
        # 从工具响应中提取补全内容
        result = tool_response.get('result', '')
        if not result:
            return completions
        
        # 生成补全项
        completion = CompletionItem(
            label=f"使用 {tool.get('name')} 生成",
            insert_text=result,
            kind="Snippet",
            detail=tool.get('description', ''),
            documentation=f"通过 MCP 工具 {tool.get('name')} 生成",
            mcp_tool_call={
                'tool_name': tool.get('name'),
                'params': tool_response.get('params', {})
            }
        )
        
        completions.append(completion)
        return completions
    
    def generate_completions(self, context: CodeContext) -> List[CompletionItem]:
        """生成代码补全项"""
        # 1. 分析代码上下文
        analysis_result = self._analyze_context(context)
        
        # 2. 选择合适的 MCP 工具
        tool = self._select_mcp_tool(analysis_result)
        if not tool:
            return []
        
        # 3. 生成工具调用参数
        params = self._generate_tool_params(tool, analysis_result)
        
        # 4. 调用 MCP 工具
        tool_response = self._call_mcp_tool(tool, params)
        
        # 5. 生成代码补全项
        return self._generate_completion(tool, tool_response, analysis_result)

# 示例用法
if __name__ == "__main__":
    # 模拟 MCP 客户端和 LLM 客户端
    class MockMCPClient:
        def get_tools(self):
            return {
                'tools': [
                    {
                        'name': 'test_generator',
                        'description': '生成测试代码',
                        'parameters': {
                            'properties': {
                                'code': {'type': 'string', 'description': '源代码', 'required': True},
                                'language': {'type': 'string', 'description': '编程语言'}
                            }
                        }
                    },
                    {
                        'name': 'code_refactor',
                        'description': '重构代码',
                        'parameters': {
                            'properties': {
                                'code': {'type': 'string', 'description': '源代码', 'required': True},
                                'refactor_type': {'type': 'string', 'description': '重构类型'}
                            }
                        }
                    }
                ]
            }
        
        def call_tool(self, tool_name, params):
            # 模拟工具调用
            if tool_name == 'test_generator':
                return {
                    'result': f"def test_function():\n    # Test for code: {params.get('code')}\n    assert True\n",
                    'params': params
                }
            elif tool_name == 'code_refactor':
                return {
                    'result': f"# Refactored code for: {params.get('code')}\nrefactored_code = 'improved'\n",
                    'params': params
                }
            return {'error': 'Tool not found'}
    
    class MockLLMClient:
        def generate(self, prompt):
            return 'python'
    
    # 创建代码上下文
    context = CodeContext(
        file_path='test.py',
        file_type='python',
        cursor_position=100,
        current_line='# Generate test for this function',
        previous_lines=['def my_function():', '    return 42'],
        next_lines=[],
        project_root='/path/to/project'
    )
    
    # 初始化 MCP 智能补全引擎
    mcp_client = MockMCPClient()
    llm_client = MockLLMClient()
    engine = MCPSmartCompletionEngine(mcp_client, llm_client)
    
    # 生成补全项
    completions = engine.generate_completions(context)
    
    # 输出结果
    for completion in completions:
        print(f"补全项: {completion.label}")
        print(f"插入文本: {completion.insert_text}")
        print(f"详情: {completion.detail}")
        print()

补全项: 使用 test_generator 生成
插入文本: def test_function():
    # Test for code: def my_function():
    return 42
    assert True

详情: 生成测试代码

// src/mcp-extension-framework.ts
import * as vscode from 'vscode';
import { MCPClient } from './mcp-client';
import { ToolRegistry } from './tool-registry';
import { PermissionManager } from './permission-manager';
import { LogAuditor } from './log-auditor';

/**
 * VS Code MCP 扩展框架主类
 */
export class MCPExtensionFramework {
    private context: vscode.ExtensionContext;
    private mcpClient: MCPClient;
    private toolRegistry: ToolRegistry;
    private permissionManager: PermissionManager;
    private logAuditor: LogAuditor;
    private disposable: vscode.Disposable[] = [];

    constructor(context: vscode.ExtensionContext) {
        this.context = context;
        
        // 初始化组件
        this.mcpClient = new MCPClient();
        this.toolRegistry = new ToolRegistry(this.mcpClient);
        this.permissionManager = new PermissionManager(context);
        this.logAuditor = new LogAuditor(context);
        
        // 注册命令
        this.registerCommands();
        
        // 加载配置
        this.loadConfiguration();
        
        // 启动 MCP 客户端
        this.startMCPClient();
    }

    private registerCommands(): void {
        // 注册 MCP 工具调用命令
        this.disposable.push(
            vscode.commands.registerCommand('mcp.callTool', async (toolName: string, params: any) => {
                return this.callTool(toolName, params);
            })
        );

        // 注册 MCP 工具列表命令
        this.disposable.push(
            vscode.commands.registerCommand('mcp.listTools', async () => {
                return this.listTools();
            })
        );

        // 注册 MCP 配置命令
        this.disposable.push(
            vscode.commands.registerCommand('mcp.configure', async () => {
                return this.configureMCP();
            })
        );
    }

    private loadConfiguration(): void {
        // 加载 MCP 配置
        const config = vscode.workspace.getConfiguration('mcp');
        const serverUrl = config.get<string>('serverUrl', 'https://localhost:8080');
        const apiKey = config.get<string>('apiKey', '');
        
        // 配置 MCP 客户端
        this.mcpClient.configure({
            serverUrl,
            apiKey,
            verifySSL: config.get<boolean>('verifySSL', true)
        });
    }

    private startMCPClient(): void {
        // 启动 MCP 客户端连接
        this.mcpClient.connect()
            .then(() => {
                vscode.window.showInformationMessage('MCP 客户端连接成功');
                // 加载工具列表
                return this.toolRegistry.loadTools();
            })
            .then(() => {
                const toolCount = this.toolRegistry.getToolCount();
                vscode.window.showInformationMessage(`已加载 ${toolCount} 个 MCP 工具`);
            })
            .catch((error) => {
                vscode.window.showErrorMessage(`MCP 客户端连接失败: ${error.message}`);
            });
    }

    /**
     * 调用 MCP 工具
     */
    public async callTool(toolName: string, params: any): Promise<any> {
        try {
            // 检查权限
            if (!await this.permissionManager.checkPermission(toolName, params)) {
                vscode.window.showErrorMessage(`没有权限调用工具: ${toolName}`);
                return null;
            }

            // 调用工具
            const result = await this.mcpClient.callTool(toolName, params);
            
            // 记录审计日志
            this.logAuditor.logToolCall(toolName, params, result);
            
            return result;
        } catch (error: any) {
            vscode.window.showErrorMessage(`调用 MCP 工具失败: ${error.message}`);
            this.logAuditor.logError(toolName, params, error.message);
            return null;
        }
    }

    /**
     * 获取 MCP 工具列表
     */
    public async listTools(): Promise<any[]> {
        try {
            return this.toolRegistry.getTools();
        } catch (error: any) {
            vscode.window.showErrorMessage(`获取 MCP 工具列表失败: ${error.message}`);
            return [];
        }
    }

    /**
     * 配置 MCP
     */
    public async configureMCP(): Promise<void> {
        // 显示 MCP 配置界面
        vscode.commands.executeCommand('workbench.action.openSettings', 'mcp');
    }

    /**
     * 销毁扩展资源
     */
    public dispose(): void {
        this.mcpClient.disconnect();
        this.disposable.forEach(d => d.dispose());
    }
}

// 扩展激活函数
export function activate(context: vscode.ExtensionContext): void {
    // 初始化 MCP 扩展框架
    const mcpFramework = new MCPExtensionFramework(context);
    
    // 注册销毁函数
    context.subscriptions.push(mcpFramework);
    
    // 显示激活信息
    vscode.window.showInformationMessage('MCP 扩展框架已激活');
}

// 扩展销毁函数
export function deactivate(): void {
    // 清理资源
}

# src/mcp-tool-market.py
import json
import requests
from typing import List, Dict, Optional
import os

class MCPToolMarket:
    def __init__(self, api_url: str = "https://mcp-tool-market.example.com/api/v1"):
        self.api_url = api_url
        self.cache_dir = ".mcp-tool-cache"
        self.cache_file = os.path.join(self.cache_dir, "tools.json")
        
        # 确保缓存目录存在
        os.makedirs(self.cache_dir, exist_ok=True)
    
    def search_tools(self, query: str, category: Optional[str] = None, language: Optional[str] = None) -> List[Dict]:
        """搜索 MCP 工具"""
        params = {
            "q": query,
        }
        
        if category:
            params["category"] = category
        if language:
            params["language"] = language
        
        try:
            response = requests.get(f"{self.api_url}/tools/search", params=params)
            response.raise_for_status()
            return response.json().get("tools", [])
        except Exception as e:
            print(f"搜索工具失败: {e}")
            return []
    
    def get_tool_details(self, tool_id: str) -> Optional[Dict]:
        """获取工具详细信息"""
        try:
            response = requests.get(f"{self.api_url}/tools/{tool_id}")
            response.raise_for_status()
            return response.json().get("tool")
        except Exception as e:
            print(f"获取工具详情失败: {e}")
            return None
    
    def install_tool(self, tool_id: str, install_path: str) -> bool:
        """安装 MCP 工具"""
        # 获取工具详细信息
        tool = self.get_tool_details(tool_id)
        if not tool:
            return False
        
        try:
            # 下载工具包
            download_url = tool.get("download_url")
            if not download_url:
                print("工具没有下载链接")
                return False
            
            response = requests.get(download_url)
            response.raise_for_status()
            
            # 保存工具包
            tool_file = os.path.join(install_path, f"{tool_id}.zip")
            with open(tool_file, "wb") as f:
                f.write(response.content)
            
            # 安装工具（示例：解压到指定目录）
            import zipfile
            with zipfile.ZipFile(tool_file, "r") as zip_ref:
                zip_ref.extractall(install_path)
            
            # 记录安装信息
            self._save_installed_tool(tool)
            
            print(f"工具 {tool.get('name')} 安装成功")
            return True
        except Exception as e:
            print(f"安装工具失败: {e}")
            return False
    
    def list_installed_tools(self) -> List[Dict]:
        """列出已安装的 MCP 工具"""
        try:
            with open(self.cache_file, "r") as f:
                return json.load(f)
        except FileNotFoundError:
            return []
        except Exception as e:
            print(f"读取已安装工具列表失败: {e}")
            return []
    
    def uninstall_tool(self, tool_id: str) -> bool:
        """卸载 MCP 工具"""
        try:
            # 获取已安装工具列表
            installed_tools = self.list_installed_tools()
            
            # 查找工具
            tool_to_uninstall = next((t for t in installed_tools if t.get("id") == tool_id), None)
            if not tool_to_uninstall:
                print(f"工具 {tool_id} 未安装")
                return False
            
            # 移除工具文件
            tool_path = os.path.join(self.cache_dir, tool_id)
            if os.path.exists(tool_path):
                import shutil
                shutil.rmtree(tool_path)
            
            # 更新已安装工具列表
            updated_tools = [t for t in installed_tools if t.get("id") != tool_id]
            with open(self.cache_file, "w") as f:
                json.dump(updated_tools, f, indent=2)
            
            print(f"工具 {tool_to_uninstall.get('name')} 卸载成功")
            return True
        except Exception as e:
            print(f"卸载工具失败: {e}")
            return False
    
    def _save_installed_tool(self, tool: Dict):
        """保存已安装工具信息到缓存"""
        try:
            # 获取当前已安装工具列表
            installed_tools = self.list_installed_tools()
            
            # 检查工具是否已安装
            existing_tool = next((t for t in installed_tools if t.get("id") == tool.get("id")), None)
            if existing_tool:
                # 更新工具信息
                existing_tool.update(tool)
            else:
                # 添加新工具
                installed_tools.append(tool)
            
            # 保存到缓存文件
            with open(self.cache_file, "w") as f:
                json.dump(installed_tools, f, indent=2)
        except Exception as e:
            print(f"保存工具信息失败: {e}")

# 示例用法
if __name__ == "__main__":
    # 初始化 MCP 工具市场
    tool_market = MCPToolMarket()
    
    # 搜索工具
    print("搜索测试生成工具:")
    tools = tool_market.search_tools("test generator", language="python")
    for tool in tools[:3]:
        print(f"- {tool.get('name')}: {tool.get('description')}")
    
    # 安装工具（示例）
    if tools:
        tool_id = tools[0].get("id")
        print(f"\n安装工具: {tool_id}")
        tool_market.install_tool(tool_id, "./mcp-tools")
    
    # 列出已安装工具
    print("\n已安装工具:")
    installed_tools = tool_market.list_installed_tools()
    for tool in installed_tools:
        print(f"- {tool.get('name')}")
    
    # 卸载工具（示例）
    if installed_tools:
        tool_id = installed_tools[0].get("id")
        print(f"\n卸载工具: {tool_id}")
        tool_market.uninstall_tool(tool_id)
    
    # 再次列出已安装工具
    print("\n卸载后已安装工具:")
    installed_tools = tool_market.list_installed_tools()
    for tool in installed_tools:
        print(f"- {tool.get('name')}")

npm install -g yo generator-code
yo code

npm install mcp-vscode-framework

{
  "name": "mcp-example-extension",
  "displayName": "MCP Example Extension",
  "description": "An example MCP extension for VS Code",
  "version": "0.0.1",
  "engines": {
    "vscode": "^1.80.0"
  },
  "categories": [
    "Other"
  ],
  "activationEvents": [
    "onCommand:mcp-example.helloWorld"
  ],
  "main": "./out/extension.js",
  "contributes": {
    "commands": [
      {
        "command": "mcp-example.helloWorld",
        "title": "Hello World"
      }
    ],
    "configuration": {
      "title": "MCP Example",
      "properties": {
        "mcp-example.enable": {
          "type": "boolean",
          "default": true,
          "description": "Enable/disable the MCP example extension"
        }
      }
    }
  },
  "scripts": {
    "vscode:prepublish": "npm run compile",
    "compile": "tsc -p ./",
    "watch": "tsc -watch -p ./",
    "pretest": "npm run compile && npm run lint",
    "lint": "eslint src --ext ts",
    "test": "node ./out/test/runTest.js"
  },
  "devDependencies": {
    "@types/vscode": "^1.80.0",
    "@types/glob": "^8.1.0",
    "@types/mocha": "^10.0.1",
    "@types/node": "16.x",
    "@typescript-eslint/eslint-plugin": "^5.59.8",
    "@typescript-eslint/parser": "^5.59.8",
    "eslint": "^8.41.0",
    "glob": "^8.1.0",
    "mocha": "^10.2.0",
    "typescript": "^5.1.3",
    "@vscode/test-electron": "^2.3.2"
  }
}

// src/extension.ts
import * as vscode from 'vscode';
import { MCPExtensionFramework } from 'mcp-vscode-framework';

export function activate(context: vscode.ExtensionContext) {
    console.log('MCP Example Extension activated');
    
    // 初始化 MCP 扩展框架
    const mcpFramework = new MCPExtensionFramework(context);
    context.subscriptions.push(mcpFramework);
    
    // 注册自定义命令
    let disposable = vscode.commands.registerCommand('mcp-example.helloWorld', async () => {
        // 调用 MCP 工具
        const result = await mcpFramework.callTool('example-tool', {
            message: 'Hello from VS Code MCP Extension'
        });
        
        // 显示结果
        vscode.window.showInformationMessage(`MCP Tool Result: ${JSON.stringify(result)}`);
    });
    
    context.subscriptions.push(disposable);
}

export function deactivate() {
    console.log('MCP Example Extension deactivated');
}

npm run compile
# 按 F5 在 VS Code 扩展开发主机中运行

GET /api/v1/tools/search
Params:
  - q: 搜索关键词
  - category: 工具分类
  - language: 支持的编程语言
  - page: 页码
  - per_page: 每页数量

Response:
{
  "tools": [
    {
      "id": "tool-123",
      "name": "Test Generator",
      "description": "Generate test code for your functions",
      "category": "testing",
      "languages": ["python", "javascript"],
      "download_url": "https://example.com/tools/test-generator.zip",
      "version": "1.0.0",
      "author": "MCP Team",
      "stars": 123,
      "downloads": 456
    }
  ],
  "total": 100,
  "page": 1,
  "per_page": 10
}

GET /api/v1/tools/{tool_id}

Response:
{
  "tool": {
    "id": "tool-123",
    "name": "Test Generator",
    "description": "Generate test code for your functions",
    "category": "testing",
    "languages": ["python", "javascript"],
    "download_url": "https://example.com/tools/test-generator.zip",
    "version": "1.0.0",
    "author": "MCP Team",
    "stars": 123,
    "downloads": 456,
    "mcp_schema": {
      "name": "test_generator",
      "description": "Generate test code",
      "parameters": {
        "type": "object",
        "properties": {
          "code": {
            "type": "string",
            "description": "Source code to generate tests for"
          },
          "language": {
            "type": "string",
            "description": "Programming language"
          }
        },
        "required": ["code"]
      }
    }
  }
}