┌─────────────────────────────────────────────────────────────┐
│                    AI Agent / Client                         │
│              (Claude, Local Agent, etc.)                     │
└────────────────────────┬────────────────────────────────────┘
                         │
                    MCP Protocol
                  (Streamable HTTP)
                         │
┌────────────────────────▼────────────────────────────────────┐
│                 File Manager MCP Server                      │
│              (FastMCP + Uvicorn on Port 8000)               │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌──────────────────┐   │
│  │File Service │  │Search Service│ │Batch Service    │   │
│  │(CRUD, I/O)  │  │(Find, Tree)  │  │(Bulk Ops)       │   │
│  └─────────────┘  └─────────────┘  └──────────────────┘   │
│  ┌─────────────┐  ┌─────────────┐  ┌──────────────────┐   │
│  │ ZIP Service │  │Utility Service│ │Security Service │   │
│  │(Archive)    │  │(Hash, Clean)  │  │(Path Validation)│   │
│  └─────────────┘  └─────────────┘  └──────────────────┘   │
├─────────────────────────────────────────────────────────────┤
│                  Filesystem (Local/Remote)                   │
└─────────────────────────────────────────────────────────────┘

file-manager-mcp/
├── main.py                      # Entry point, MCP server initialization
├── server.py                    # FastMCP server configuration
├── auth.py                      # Authentication & authorization
├── config.py                    # Configuration management
├── requirements.txt             # Python dependencies
├── README.md                    # This file
│
├── services/                    # Core business logic
│   ├── file_service.py         # File CRUD operations
│   ├── search_service.py       # Search and discovery tools
│   ├── batch_service.py        # Bulk operations
│   ├── zip_service.py          # Archive operations
│   ├── utility_service.py      # Hash, cleanup, comparison tools
│   └── security_service.py     # Security validations
│
└── server.log                   # Application logs

# Clone the repository
git clone https://github.com/parthRana26/file-manager-mcp.git
cd file-manager-mcp

# Create virtual environment
uv venv

# Activate environment
# On Linux/macOS:
source .venv/bin/activate
# On Windows:
.venv\Scripts\activate

# Install dependencies
uv pip install -r requirements.txt

# Start the MCP server
uv run python main.py

INFO: Uvicorn running on http://0.0.0.0:8000
INFO: File Manager MCP Server initialized
INFO: 34 tools registered
INFO: Ready to accept connections

# Server configuration
MCP_HOST=0.0.0.0
MCP_PORT=8000
LOG_LEVEL=DEBUG

# Optional: Set allowed base directories
ALLOWED_BASE_DIRS=/path/to/workspace

# Optional: Enable security features
ENABLE_PATH_VALIDATION=true
ENABLE_REQUEST_LOGGING=true

# Run with debug logging
LOG_LEVEL=DEBUG uv run python main.py

# Terminal 1: Start the File Manager MCP server
uv run python main.py

# Terminal 2: Open MCP Inspector
npx @modelcontextprotocol/inspector

1. Call: health_check_tool
   → Verify server is running
   
2. Call: available_tools_tool
   → List all registered tools
   
3. Call: list_files_tool with path: "."
   → List current directory contents
   
4. Call: tree_view_tool with path: "." and max_depth: 3
   → Generate directory tree
   
5. Call: search_files_tool with pattern: "*.py" and path: "."
   → Find all Python files

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t file-manager-mcp .
docker run -p 8000:8000 file-manager-mcp

# Using Python client
from mcp import Client

client = Client("http://localhost:8000/mcp")

# List files in directory
result = await client.call_tool("list_files_tool", {
    "path": "/home/user/documents"
})

# Create a new file
result = await client.call_tool("create_file_tool", {
    "path": "/home/user/documents/notes.txt",
    "content": "Hello, World!"
})

# Read file content
result = await client.call_tool("read_file_tool", {
    "path": "/home/user/documents/notes.txt"
})

# Update file
result = await client.call_tool("update_file_tool", {
    "path": "/home/user/documents/notes.txt",
    "content": "Updated content"
})

# Delete file
result = await client.call_tool("delete_file_tool", {
    "path": "/home/user/documents/notes.txt"
})

# Find all Python files
result = await client.call_tool("search_files_tool", {
    "path": "/home/user/projects",
    "pattern": "*.py"
})

# Get file information
result = await client.call_tool("get_file_info_tool", {
    "path": "/home/user/projects/main.py"
})

# Generate directory tree
result = await client.call_tool("tree_view_tool", {
    "path": "/home/user/projects",
    "max_depth": 3
})

# Batch create multiple files
result = await client.call_tool("batch_create_files_tool", {
    "files": [
        {"path": "/tmp/file1.txt", "content": "File 1"},
        {"path": "/tmp/file2.txt", "content": "File 2"},
        {"path": "/tmp/file3.txt", "content": "File 3"}
    ]
})

# Batch delete files
result = await client.call_tool("batch_delete_files_tool", {
    "paths": [
        "/tmp/file1.txt",
        "/tmp/file2.txt",
        "/tmp/file3.txt"
    ]
})

# Batch move files
result = await client.call_tool("batch_move_files_tool", {
    "operations": [
        {"source": "/tmp/src1.txt", "destination": "/home/user/dst1.txt"},
        {"source": "/tmp/src2.txt", "destination": "/home/user/dst2.txt"}
    ]
})

# Create ZIP archive
result = await client.call_tool("create_zip_tool", {
    "source_path": "/home/user/documents",
    "zip_path": "/home/user/backup.zip"
})

# Extract ZIP archive
result = await client.call_tool("extract_zip_tool", {
    "zip_path": "/home/user/backup.zip",
    "extract_path": "/home/user/restored"
})

# List ZIP contents
result = await client.call_tool("list_zip_contents_tool", {
    "zip_path": "/home/user/backup.zip"
})

# Find large files
result = await client.call_tool("find_large_files_tool", {
    "path": "/home/user",
    "size_mb": 100
})

# Generate file hash
result = await client.call_tool("file_hash_tool", {
    "path": "/home/user/file.bin",
    "algorithm": "sha256"
})

# Find duplicate files
result = await client.call_tool("duplicate_file_finder_tool", {
    "path": "/home/user/documents"
})

# Compare files
result = await client.call_tool("compare_files_tool", {
    "path1": "/home/user/file1.txt",
    "path2": "/home/user/file2.txt"
})

# Compare directories
result = await client.call_tool("compare_directories_tool", {
    "path1": "/home/user/backup1",
    "path2": "/home/user/backup2"
})

# Clean temporary files
result = await client.call_tool("clean_temp_files_tool", {
    "path": "/home/user"
})

# Clean empty directories
result = await client.call_tool("clean_empty_directories_tool", {
    "path": "/home/user"
})

import asyncio
from mcp import Client

async def backup_and_cleanup():
    """
    Automated workflow:
    1. Find large files that aren't backed up
    2. Create ZIP archive of important directories
    3. Identify and remove duplicate files
    4. Clean up temporary files
    5. Verify backup integrity with hashing
    """
    
    client = Client("http://localhost:8000/mcp")
    
    # Step 1: Find large files
    print("📊 Analyzing large files...")
    large_files = await client.call_tool("find_large_files_tool", {
        "path": "/home/user/documents",
        "size_mb": 50
    })
    print(f"Found {len(large_files['files'])} files > 50MB")
    
    # Step 2: Create backup
    print("📦 Creating backup archive...")
    backup = await client.call_tool("create_zip_tool", {
        "source_path": "/home/user/documents",
        "zip_path": "/backups/documents-latest.zip"
    })
    print(f"Backup created: {backup['zip_path']}")
    
    # Step 3: Find duplicates
    print("🔍 Detecting duplicate files...")
    duplicates = await client.call_tool("duplicate_file_finder_tool", {
        "path": "/home/user/documents"
    })
    print(f"Found {len(duplicates['duplicates'])} duplicate sets")
    
    # Step 4: Remove duplicates (keep one of each)
    for dup_group in duplicates['duplicates'][1:]:  # Keep first, delete rest
        for dup_file in dup_group[1:]:
            await client.call_tool("delete_file_tool", {
                "path": dup_file
            })
    print("✅ Removed duplicate files")
    
    # Step 5: Clean temporary files
    print("🧹 Cleaning temporary files...")
    cleaned = await client.call_tool("clean_temp_files_tool", {
        "path": "/home/user"
    })
    print(f"Removed {cleaned['count']} temporary files")
    
    # Step 6: Verify backup integrity
    print("🔐 Verifying backup integrity...")
    hash_result = await client.call_tool("file_hash_tool", {
        "path": "/backups/documents-latest.zip",
        "algorithm": "sha256"
    })
    print(f"Backup hash: {hash_result['hash']}")
    
    print("\n✨ Backup and cleanup workflow completed!")

# Run the workflow
asyncio.run(backup_and_cleanup())

# ✅ DO: Validate user input
result = await client.call_tool("create_file_tool", {
    "path": secure_path_join(base_dir, user_input),
    "content": sanitized_content
})

# ❌ DON'T: Trust user paths directly
result = await client.call_tool("create_file_tool", {
    "path": user_provided_path,  # Potential traversal attack!
    "content": content
})

# ❌ Inefficient: Multiple individual calls
for file in files:
    await client.call_tool("create_file_tool", {"path": file, "content": content})

# ✅ Efficient: Single batch call
await client.call_tool("batch_create_files_tool", {"files": files})

# ✅ Better: Narrow search scope
await client.call_tool("search_files_tool", {
    "path": "/home/user/documents",  # Specific directory
    "pattern": "*.pdf"
})

# ❌ Slower: Full filesystem search
await client.call_tool("search_files_tool", {
    "path": "/",  # Entire filesystem
    "pattern": "*.pdf"
})

# When listing directories with thousands of files
# Use pagination and limit results
result = await client.call_tool("list_files_tool", {
    "path": "/large/directory",
    "max_results": 1000
})

# ❌ Slower: Full SHA-256 hash
result = await client.call_tool("file_hash_tool", {
    "path": "/large/file.iso",
    "algorithm": "sha256"
})

# ✅ Faster: Quick MD5 for deduplication
result = await client.call_tool("file_hash_tool", {
    "path": "/large/file.iso",
    "algorithm": "md5"
})

# Server configuration for optimal performance
MAX_CONCURRENT_OPERATIONS=10
REQUEST_TIMEOUT_SECONDS=300
BATCH_OPERATION_SIZE=1000
LOG_LEVEL=INFO  # Use INFO in production, not DEBUG

# Future: Watch files for changes
await client.call_tool("watch_directory_tool", {
    "path": "/home/user/documents",
    "callback_url": "https://your-api.com/file-changes",
    "filter": "*.pdf"
})

# Future: Encrypt/decrypt files
await client.call_tool("encrypt_file_tool", {
    "path": "/home/user/sensitive.txt",
    "algorithm": "aes-256",
    "password": "secure-password"
})

# Future: Seamless cloud integration
await client.call_tool("upload_to_s3_tool", {
    "local_path": "/home/user/backup.zip",
    "s3_path": "s3://my-bucket/backups/",
    "region": "us-west-2"
})

# Future: Image/video operations
await client.call_tool("create_thumbnail_tool", {
    "image_path": "/home/user/photo.jpg",
    "output_path": "/home/user/thumb.jpg",
    "size": "256x256"
})

# Future: Two-way sync between directories
await client.call_tool("sync_directories_tool", {
    "source": "/home/user/local",
    "destination": "/mnt/nas/backup",
    "bidirectional": True,
    "ignore_patterns": ["*.tmp", "node_modules"]
})