File Manager MCP Server

#!/usr/bin/env python3 """文件管理器MCP服务器主程序 这个模块实现了一个MCP服务器，提供基本的文件系统操作工具。 """ import asyncio import logging from typing import Any, Sequence from mcp.server import NotificationOptions, Server from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import mcp.types as types from .tools import ( list_directory, read_file, create_file, delete_file, search_files, copy_files, ) from .validators import validate_tool_arguments, ValidationError # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 创建服务器实例 server = Server("file-manager-mcp") # 定义可用的工具 TOOLS = [ Tool( name="list_directory", description="列出指定目录的内容，包括文件和子目录", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "要列出内容的目录路径" }, "show_hidden": { "type": "boolean", "description": "是否显示隐藏文件（以.开头的文件）", "default": False } }, "required": ["path"] } ), Tool( name="read_file", description="读取指定文件的内容", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "要读取的文件路径" }, "encoding": { "type": "string", "description": "文件编码格式", "default": "utf-8" } }, "required": ["path"] } ), Tool( name="create_file", description="创建新文件并写入内容", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "要创建的文件路径" }, "content": { "type": "string", "description": "文件内容", "default": "" }, "encoding": { "type": "string", "description": "文件编码格式", "default": "utf-8" } }, "required": ["path"] } ), Tool( name="delete_file", description="删除指定的文件或目录", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "要删除的文件或目录路径" }, "recursive": { "type": "boolean", "description": "是否递归删除目录及其内容", "default": False } }, "required": ["path"] } ), Tool( name="search_files", description="在指定目录中搜索文件", inputSchema={ "type": "object", "properties": { "directory": { "type": "string", "description": "搜索的根目录路径" }, "pattern": { "type": "string", "description": "搜索模式（支持通配符）" }, "recursive": { "type": "boolean", "description": "是否递归搜索子目录", "default": True } }, "required": ["directory", "pattern"] } ), Tool( name="copy_files", description="将一个或多个文件复制到指定的目标文件夹中", inputSchema={ "type": "object", "properties": { "source_paths": { "type": "array", "items": { "type": "string" }, "description": "要复制的源文件路径列表" }, "destination_folder": { "type": "string", "description": "目标文件夹路径" }, "overwrite": { "type": "boolean", "description": "是否覆盖已存在的文件", "default": False } }, "required": ["source_paths", "destination_folder"] } ) ] @server.list_tools() async def handle_list_tools() -> list[Tool]: """返回可用工具列表""" return TOOLS @server.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[TextContent]: """处理工具调用请求""" if arguments is None: arguments = {} try: # 验证参数 validated_args = validate_tool_arguments(name, arguments) # 调用相应的工具 if name == "list_directory": result = await list_directory( path=validated_args["path"], show_hidden=validated_args["show_hidden"] ) elif name == "read_file": result = await read_file( path=validated_args["path"], encoding=validated_args["encoding"] ) elif name == "create_file": result = await create_file( path=validated_args["path"], content=validated_args["content"], encoding=validated_args["encoding"] ) elif name == "delete_file": result = await delete_file( path=validated_args["path"], recursive=validated_args["recursive"] ) elif name == "search_files": result = await search_files( directory=validated_args["directory"], pattern=validated_args["pattern"], recursive=validated_args["recursive"] ) elif name == "copy_files": result = await copy_files( source_paths=validated_args["source_paths"], destination_folder=validated_args["destination_folder"], overwrite=validated_args["overwrite"] ) else: raise ValidationError(f"未知的工具: {name}") return [TextContent(type="text", text=result)] except ValidationError as e: logger.warning(f"参数验证失败 {name}: {e}") return [TextContent(type="text", text=f"参数错误: {str(e)}")] except Exception as e: logger.error(f"工具调用失败 {name}: {e}") return [TextContent(type="text", text=f"错误: {str(e)}")] async def main(): """主函数 - 启动MCP服务器""" logger.info("启动文件管理器MCP服务器...") async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="file-manager-mcp", server_version="1.0.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={} ) ) ) if __name__ == "__main__": asyncio.run(main())