// Before MCP:
you = "Claude, analyzuj ten screenshot z testu"
claude = "Nahraj mi screenshot..."
you = *downloads from MinIO*
you = *uploads to Claude*
claude = "OK, analyzujem..."
// 5 minút manuálnej práce 😰

// After MCP:
you = "Claude, analyzuj screenshot test-123/login.png"
claude = *fetches from MinIO directly*
claude = "Screenshot ukazuje error message: 'Invalid credentials'..."
// 5 sekúnd! ✅

// What happened?
// MCP = Model Context Protocol
// Claude má DIRECT access k MinIO! 🚀

Problém:
- Test automation generuje tisícky screenshotov v MinIO
- Chceš aby Claude analyzoval failure screenshoty
- Ale musíš každý screenshot manuálne stiahnuť a nahrať

S MCP:
- Claude má direct access k MinIO
- "Show me all screenshots from failed tests today"
- Claude listuje MinIO, analyzuje obrázky, vyhodnotí problémy
- Automaticky! 🎉

Time saved:
- Before: 30 minút na analýzu (download, upload, analyze)
- After: 30 sekúnd (Claude robí všetko)
- That's 60× faster! 🚀

MCP (Model Context Protocol):
  what: "Protocol pre AI agentov k externým dátam"
  created_by: "Anthropic"
  purpose: "Daj AI prístup k tvojim súborom, DB, API..."
  
How it works:
  ┌─────────────┐
  │   Claude    │
  │   (AI)      │
  └──────┬──────┘
         │ MCP Protocol
         │
  ┌──────▼──────────┐
  │  MCP Server     │
  │  (Your code)    │
  └──────┬──────────┘
         │
  ┌──────▼──────────┐
  │     MinIO       │
  │  (Your files)   │
  └─────────────────┘

Claude: "Show me test-123 screenshots"
  → MCP Server: fetches from MinIO
  → Returns images to Claude
  → Claude analyzes

✅ Direct access (no manual upload)
✅ Real-time data (always fresh)
✅ Secure (your infrastructure)
✅ Flexible (any data source)
✅ Powerful (AI + your data)

use_cases = {
    "Test Analysis": {
        "Ask": "Analyze failed test screenshots",
        "MCP": "Fetches screenshots, Claude analyzes",
        "Result": "Root cause identified in 30s"
    },
    
    "Document Processing": {
        "Ask": "Summarize all PDFs in bucket",
        "MCP": "Lists PDFs, Claude reads & summarizes",
        "Result": "All documents processed automatically"
    },
    
    "Log Analysis": {
        "Ask": "Find errors in today's logs",
        "MCP": "Fetches logs, Claude searches patterns",
        "Result": "All errors found & categorized"
    },
    
    "Video Analysis": {
        "Ask": "What happens in test video at 00:45?",
        "MCP": "Fetches video frame, Claude describes",
        "Result": "Detailed description of UI state"
    }
}

verdict = "MCP + MinIO = AI Superpowers! 🚀"

# 1. MinIO running (from previous article!)
docker ps | grep minio
# Should see: minio container running on 9000/9001

# 2. Python 3.10+ (for MCP server)
python --version
# Output: Python 3.10+

# 3. Node.js 18+ (for testing)
node --version
# Output: v18+

# 4. Claude Desktop App (for usage)
# Download: https://claude.ai/download

minio-mcp-server/
├── src/
│   ├── __init__.py
│   ├── server.py          # MCP server implementation
│   └── minio_client.py    # MinIO operations
├── pyproject.toml         # Dependencies
├── README.md
└── .env                   # Configuration

# Create project
mkdir minio-mcp-server
cd minio-mcp-server

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Create pyproject.toml
cat > pyproject.toml << 'EOF'
[project]
name = "minio-mcp-server"
version = "0.1.0"
description = "MCP server for MinIO object storage"
requires-python = ">=3.10"
dependencies = [
    "mcp>=0.9.0",
    "minio>=7.2.0",
    "python-dotenv>=1.0.0"
]

[project.scripts]
minio-mcp = "minio_mcp_server:main"
EOF

# Install dependencies
pip install -e .

# Create .env file
cat > .env << 'EOF'
# MinIO Configuration
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=admin
MINIO_SECRET_KEY=password123
MINIO_SECURE=false

# MCP Configuration
MCP_SERVER_NAME=minio-server
EOF

# src/minio_client.py

from minio import Minio
from minio.error import S3Error
from typing import List, Dict, Optional
import io
import os
from datetime import timedelta

class MinioClient:
    """MinIO client for MCP server"""
    
    def __init__(self):
        self.client = Minio(
            endpoint=os.getenv('MINIO_ENDPOINT', 'localhost:9000'),
            access_key=os.getenv('MINIO_ACCESS_KEY', 'admin'),
            secret_key=os.getenv('MINIO_SECRET_KEY', 'password123'),
            secure=os.getenv('MINIO_SECURE', 'false').lower() == 'true'
        )
    
    def list_buckets(self) -> List[Dict[str, str]]:
        """List all buckets"""
        try:
            buckets = self.client.list_buckets()
            return [
                {
                    'name': bucket.name,
                    'creation_date': bucket.creation_date.isoformat()
                }
                for bucket in buckets
            ]
        except S3Error as e:
            raise Exception(f"Failed to list buckets: {e}")
    
    def list_objects(self, bucket: str, prefix: str = '', recursive: bool = True) -> List[Dict[str, any]]:
        """List objects in bucket"""
        try:
            objects = self.client.list_objects(
                bucket_name=bucket,
                prefix=prefix,
                recursive=recursive
            )
            
            return [
                {
                    'name': obj.object_name,
                    'size': obj.size,
                    'last_modified': obj.last_modified.isoformat(),
                    'etag': obj.etag,
                    'content_type': obj.content_type
                }
                for obj in objects
            ]
        except S3Error as e:
            raise Exception(f"Failed to list objects: {e}")
    
    def get_object(self, bucket: str, object_name: str) -> bytes:
        """Get object content"""
        try:
            response = self.client.get_object(bucket, object_name)
            data = response.read()
            response.close()
            response.release_conn()
            return data
        except S3Error as e:
            raise Exception(f"Failed to get object: {e}")
    
    def get_object_metadata(self, bucket: str, object_name: str) -> Dict[str, any]:
        """Get object metadata"""
        try:
            stat = self.client.stat_object(bucket, object_name)
            return {
                'name': object_name,
                'size': stat.size,
                'last_modified': stat.last_modified.isoformat(),
                'etag': stat.etag,
                'content_type': stat.content_type,
                'metadata': stat.metadata
            }
        except S3Error as e:
            raise Exception(f"Failed to get metadata: {e}")
    
    def put_object(self, bucket: str, object_name: str, data: bytes, content_type: str = 'application/octet-stream') -> Dict[str, str]:
        """Upload object"""
        try:
            result = self.client.put_object(
                bucket_name=bucket,
                object_name=object_name,
                data=io.BytesIO(data),
                length=len(data),
                content_type=content_type
            )
            return {
                'bucket': bucket,
                'object_name': object_name,
                'etag': result.etag,
                'version_id': result.version_id
            }
        except S3Error as e:
            raise Exception(f"Failed to upload object: {e}")
    
    def delete_object(self, bucket: str, object_name: str) -> Dict[str, str]:
        """Delete object"""
        try:
            self.client.remove_object(bucket, object_name)
            return {
                'bucket': bucket,
                'object_name': object_name,
                'status': 'deleted'
            }
        except S3Error as e:
            raise Exception(f"Failed to delete object: {e}")
    
    def get_presigned_url(self, bucket: str, object_name: str, expires: int = 3600) -> str:
        """Get presigned URL"""
        try:
            url = self.client.presigned_get_object(
                bucket_name=bucket,
                object_name=object_name,
                expires=timedelta(seconds=expires)
            )
            return url
        except S3Error as e:
            raise Exception(f"Failed to generate URL: {e}")
    
    def search_objects(self, bucket: str, pattern: str) -> List[Dict[str, any]]:
        """Search objects by pattern"""
        try:
            all_objects = self.list_objects(bucket, recursive=True)
            
            # Simple pattern matching (contains)
            matching = [
                obj for obj in all_objects
                if pattern.lower() in obj['name'].lower()
            ]
            
            return matching
        except Exception as e:
            raise Exception(f"Failed to search objects: {e}")

# src/server.py

import asyncio
import os
from typing import Any
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
from dotenv import load_dotenv

from .minio_client import MinioClient

# Load environment variables
load_dotenv()

# Create MCP server
app = Server("minio-server")

# Create MinIO client
minio_client = MinioClient()

@app.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """List available tools"""
    return [
        types.Tool(
            name="list_buckets",
            description="List all MinIO buckets",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        types.Tool(
            name="list_objects",
            description="List objects in a bucket",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "prefix": {
                        "type": "string",
                        "description": "Object prefix filter (optional)",
                        "default": ""
                    }
                },
                "required": ["bucket"]
            }
        ),
        types.Tool(
            name="get_object",
            description="Get object content from MinIO",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "object_name": {
                        "type": "string",
                        "description": "Object name/path"
                    }
                },
                "required": ["bucket", "object_name"]
            }
        ),
        types.Tool(
            name="get_object_metadata",
            description="Get object metadata",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "object_name": {
                        "type": "string",
                        "description": "Object name/path"
                    }
                },
                "required": ["bucket", "object_name"]
            }
        ),
        types.Tool(
            name="search_objects",
            description="Search objects by name pattern",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "pattern": {
                        "type": "string",
                        "description": "Search pattern (substring match)"
                    }
                },
                "required": ["bucket", "pattern"]
            }
        ),
        types.Tool(
            name="upload_object",
            description="Upload object to MinIO",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "object_name": {
                        "type": "string",
                        "description": "Object name/path"
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to upload (base64 for binary)"
                    },
                    "content_type": {
                        "type": "string",
                        "description": "Content type",
                        "default": "application/octet-stream"
                    }
                },
                "required": ["bucket", "object_name", "content"]
            }
        ),
        types.Tool(
            name="delete_object",
            description="Delete object from MinIO",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "object_name": {
                        "type": "string",
                        "description": "Object name/path"
                    }
                },
                "required": ["bucket", "object_name"]
            }
        ),
        types.Tool(
            name="get_presigned_url",
            description="Generate presigned URL for object",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket": {
                        "type": "string",
                        "description": "Bucket name"
                    },
                    "object_name": {
                        "type": "string",
                        "description": "Object name/path"
                    },
                    "expires": {
                        "type": "integer",
                        "description": "URL expiry in seconds",
                        "default": 3600
                    }
                },
                "required": ["bucket", "object_name"]
            }
        )
    ]

@app.call_tool()
async def handle_call_tool(
    name: str,
    arguments: dict[str, Any]
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """Handle tool calls"""
    
    try:
        if name == "list_buckets":
            buckets = minio_client.list_buckets()
            return [
                types.TextContent(
                    type="text",
                    text=f"Found {len(buckets)} buckets:\n" + 
                         "\n".join([f"- {b['name']}" for b in buckets])
                )
            ]
        
        elif name == "list_objects":
            bucket = arguments["bucket"]
            prefix = arguments.get("prefix", "")
            
            objects = minio_client.list_objects(bucket, prefix)
            
            if not objects:
                return [types.TextContent(type="text", text=f"No objects found in bucket '{bucket}'")]
            
            text = f"Found {len(objects)} objects in '{bucket}':\n"
            for obj in objects[:50]:  # Limit to 50 for readability
                text += f"\n- {obj['name']} ({obj['size']} bytes)"
            
            if len(objects) > 50:
                text += f"\n\n... and {len(objects) - 50} more objects"
            
            return [types.TextContent(type="text", text=text)]
        
        elif name == "get_object":
            bucket = arguments["bucket"]
            object_name = arguments["object_name"]
            
            # Get metadata first
            metadata = minio_client.get_object_metadata(bucket, object_name)
            content_type = metadata.get('content_type', 'application/octet-stream')
            
            # Get content
            data = minio_client.get_object(bucket, object_name)
            
            # Handle images
            if content_type.startswith('image/'):
                import base64
                return [
                    types.ImageContent(
                        type="image",
                        data=base64.b64encode(data).decode('utf-8'),
                        mimeType=content_type
                    )
                ]
            
            # Handle text
            elif content_type.startswith('text/') or 'json' in content_type:
                try:
                    text = data.decode('utf-8')
                    return [types.TextContent(type="text", text=text)]
                except:
                    return [types.TextContent(type="text", text="[Binary content - cannot display as text]")]
            
            # Handle other files
            else:
                return [
                    types.TextContent(
                        type="text",
                        text=f"Object: {object_name}\nType: {content_type}\nSize: {len(data)} bytes\n\n[Binary content]"
                    )
                ]
        
        elif name == "get_object_metadata":
            bucket = arguments["bucket"]
            object_name = arguments["object_name"]
            
            metadata = minio_client.get_object_metadata(bucket, object_name)
            
            text = f"Metadata for '{object_name}':\n"
            text += f"- Size: {metadata['size']} bytes\n"
            text += f"- Last Modified: {metadata['last_modified']}\n"
            text += f"- Content Type: {metadata['content_type']}\n"
            text += f"- ETag: {metadata['etag']}\n"
            
            return [types.TextContent(type="text", text=text)]
        
        elif name == "search_objects":
            bucket = arguments["bucket"]
            pattern = arguments["pattern"]
            
            objects = minio_client.search_objects(bucket, pattern)
            
            if not objects:
                return [types.TextContent(type="text", text=f"No objects matching '{pattern}' found")]
            
            text = f"Found {len(objects)} objects matching '{pattern}':\n"
            for obj in objects[:20]:  # Limit to 20
                text += f"\n- {obj['name']} ({obj['size']} bytes)"
            
            if len(objects) > 20:
                text += f"\n\n... and {len(objects) - 20} more matches"
            
            return [types.TextContent(type="text", text=text)]
        
        elif name == "upload_object":
            bucket = arguments["bucket"]
            object_name = arguments["object_name"]
            content = arguments["content"]
            content_type = arguments.get("content_type", "application/octet-stream")
            
            # Convert content to bytes
            import base64
            try:
                data = base64.b64decode(content)
            except:
                data = content.encode('utf-8')
            
            result = minio_client.put_object(bucket, object_name, data, content_type)
            
            return [
                types.TextContent(
                    type="text",
                    text=f"Uploaded '{object_name}' to bucket '{bucket}' successfully"
                )
            ]
        
        elif name == "delete_object":
            bucket = arguments["bucket"]
            object_name = arguments["object_name"]
            
            result = minio_client.delete_object(bucket, object_name)
            
            return [
                types.TextContent(
                    type="text",
                    text=f"Deleted '{object_name}' from bucket '{bucket}' successfully"
                )
            ]
        
        elif name == "get_presigned_url":
            bucket = arguments["bucket"]
            object_name = arguments["object_name"]
            expires = arguments.get("expires", 3600)
            
            url = minio_client.get_presigned_url(bucket, object_name, expires)
            
            return [
                types.TextContent(
                    type="text",
                    text=f"Presigned URL (expires in {expires}s):\n{url}"
                )
            ]
        
        else:
            raise ValueError(f"Unknown tool: {name}")
    
    except Exception as e:
        return [
            types.TextContent(
                type="text",
                text=f"Error: {str(e)}"
            )
        ]

async def main():
    """Run MCP server"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="minio-server",
                server_version="0.1.0",
                capabilities=app.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

# src/__init__.py

from .server import main

__all__ = ['main']

# Run MCP server (stdio mode)
python -m minio_mcp_server

# Server is running, waiting for input...

# Test with MCP Inspector (optional)
npx @modelcontextprotocol/inspector python -m minio_mcp_server

# Open browser: http://localhost:5173
# You can test all tools interactively!

# Install mcp CLI
npm install -g @modelcontextprotocol/cli

# Test list_buckets
echo '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"list_buckets","arguments":{}}}' | \
  python -m minio_mcp_server

# Test list_objects
echo '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"list_objects","arguments":{"bucket":"test-screenshots"}}}' | \
  python -m minio_mcp_server

// macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
// Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "minio": {
      "command": "python",
      "args": [
        "-m",
        "minio_mcp_server"
      ],
      "env": {
        "MINIO_ENDPOINT": "localhost:9000",
        "MINIO_ACCESS_KEY": "admin",
        "MINIO_SECRET_KEY": "password123",
        "MINIO_SECURE": "false"
      }
    }
  }
}

{
  "mcpServers": {
    "minio": {
      "command": "/path/to/venv/bin/python",
      "args": [
        "-m",
        "minio_mcp_server"
      ],
      "env": {
        "MINIO_ENDPOINT": "localhost:9000",
        "MINIO_ACCESS_KEY": "admin",
        "MINIO_SECRET_KEY": "password123"
      }
    }
  }
}

# macOS
killall Claude
open -a Claude

# Windows
# Close Claude from taskbar
# Open Claude again

Open Claude Desktop
→ Look for 🔌 icon (MCP connected)
→ Click 🔌 to see available servers
→ Should see "minio" server listed ✅

You: "What buckets do I have in MinIO?"

Claude: [calls list_buckets tool]

Claude: "You have 3 buckets in MinIO:
- test-screenshots
- test-videos
- documents"

You: "Show me the latest screenshot from test-screenshots bucket"

Claude: [calls list_objects with bucket="test-screenshots"]
Claude: [calls get_object with latest screenshot]
Claude: [analyzes image]

Claude: "The latest screenshot shows a login page with 
an error message: 'Invalid username or password'. 
The error is displayed in red text below the login form."

You: "Find all files with 'error' in the name from test-screenshots"

Claude: [calls search_objects with pattern="error"]

Claude: "Found 15 files with 'error' in the name:
- test-123/error-login.png
- test-124/error-checkout.png
- test-125/error-payment.png
..."

You: "Compare screenshots test-123/before.png and test-123/after.png"

Claude: [calls get_object for both images]
Claude: [analyzes both images]

Claude: "Comparing the two screenshots:

Before:
- Button is blue
- Text says 'Submit'
- No loading indicator

After:
- Button is gray
- Text says 'Processing...'
- Loading spinner visible

The after screenshot shows the button in a disabled/loading state."

You: "Analyze all error logs from today"

Claude: [calls list_objects with prefix="logs/2024-12-23"]
Claude: [calls get_object for each log file]
Claude: [analyzes content]

Claude: "Found 47 error logs from today. Summary:

Top errors:
1. Connection timeout (15 occurrences)
2. Invalid token (12 occurrences)
3. Database deadlock (8 occurrences)

Most problematic service: payment-service (23 errors)

Recommendation: Check database connection pool and 
investigate token validation logic."

You: "Summarize all PDFs in the documents bucket"

Claude: [calls list_objects with bucket="documents"]
Claude: [filters .pdf files]
Claude: [calls get_object for each PDF]
Claude: [extracts and summarizes text]

Claude: "Found 12 PDF documents:

1. Q4-2024-Report.pdf (15 pages)
   Summary: Financial results showing 23% growth...

2. Product-Roadmap-2025.pdf (8 pages)
   Summary: Plans for 5 major features...

3. Security-Audit.pdf (42 pages)
   Summary: Comprehensive audit identifying 3 high-priority issues...

..."

# src/server.py (add this)

@app.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    """List available resources"""
    buckets = minio_client.list_buckets()
    
    resources = []
    for bucket in buckets:
        # Add bucket as resource
        resources.append(
            types.Resource(
                uri=f"minio://{bucket['name']}/",
                name=f"Bucket: {bucket['name']}",
                description=f"MinIO bucket '{bucket['name']}'",
                mimeType="application/x-directory"
            )
        )
    
    return resources

@app.read_resource()
async def handle_read_resource(uri: str) -> str:
    """Read resource content"""
    # Parse URI: minio://bucket/path
    if not uri.startswith("minio://"):
        raise ValueError("Invalid URI scheme")
    
    path = uri.replace("minio://", "")
    parts = path.split("/", 1)
    
    bucket = parts[0]
    object_name = parts[1] if len(parts) > 1 else ""
    
    if not object_name:
        # List bucket contents
        objects = minio_client.list_objects(bucket)
        return "\n".join([obj['name'] for obj in objects])
    else:
        # Get object content
        data = minio_client.get_object(bucket, object_name)
        try:
            return data.decode('utf-8')
        except:
            return "[Binary content]"

# src/server.py (add this)

@app.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
    """List available prompts"""
    return [
        types.Prompt(
            name="analyze_test_failures",
            description="Analyze failed test screenshots",
            arguments=[
                types.PromptArgument(
                    name="bucket",
                    description="Bucket name",
                    required=True
                ),
                types.PromptArgument(
                    name="test_id",
                    description="Test ID",
                    required=True
                )
            ]
        ),
        types.Prompt(
            name="summarize_logs",
            description="Summarize log files",
            arguments=[
                types.PromptArgument(
                    name="bucket",
                    description="Bucket name",
                    required=True
                ),
                types.PromptArgument(
                    name="date",
                    description="Date (YYYY-MM-DD)",
                    required=True
                )
            ]
        )
    ]

@app.get_prompt()
async def handle_get_prompt(
    name: str,
    arguments: dict[str, str]
) -> types.GetPromptResult:
    """Get prompt content"""
    
    if name == "analyze_test_failures":
        bucket = arguments["bucket"]
        test_id = arguments["test_id"]
        
        return types.GetPromptResult(
            description=f"Analyze test failures for {test_id}",
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"""Please analyze the test failure for test ID: {test_id}

1. List all screenshots from bucket '{bucket}' with prefix '{test_id}/'
2. Examine each screenshot
3. Identify the failure point
4. Suggest root cause
5. Recommend fixes

Be thorough and specific."""
                    )
                )
            ]
        )
    
    elif name == "summarize_logs":
        bucket = arguments["bucket"]
        date = arguments["date"]
        
        return types.GetPromptResult(
            description=f"Summarize logs for {date}",
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"""Please analyze logs from {date}

1. List all log files from bucket '{bucket}' with prefix 'logs/{date}/'
2. Read and analyze each log file
3. Categorize errors by type
4. Count occurrences
5. Identify patterns
6. Recommend actions

Provide a clear summary with priorities."""
                    )
                )
            ]
        )
    
    raise ValueError(f"Unknown prompt: {name}")

# src/server.py (add validation)

def validate_bucket_access(bucket: str) -> bool:
    """Validate bucket access"""
    allowed_buckets = os.getenv('ALLOWED_BUCKETS', '').split(',')
    
    if allowed_buckets and bucket not in allowed_buckets:
        raise PermissionError(f"Access to bucket '{bucket}' denied")
    
    return True

# Use in tools:
@app.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if 'bucket' in arguments:
        validate_bucket_access(arguments['bucket'])
    
    # ... rest of tool logic

# src/server.py (add logging)

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('minio-mcp.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

@app.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    logger.info(f"Tool called: {name} with args: {arguments}")
    
    try:
        result = # ... tool logic
        logger.info(f"Tool {name} succeeded")
        return result
    except Exception as e:
        logger.error(f"Tool {name} failed: {e}")
        raise

# src/minio_client.py (improve error handling)

class MinioError(Exception):
    """Base MinIO error"""
    pass

class BucketNotFoundError(MinioError):
    """Bucket not found"""
    pass

class ObjectNotFoundError(MinioError):
    """Object not found"""
    pass

def get_object(self, bucket: str, object_name: str) -> bytes:
    """Get object with better error handling"""
    try:
        response = self.client.get_object(bucket, object_name)
        data = response.read()
        response.close()
        response.release_conn()
        return data
    except S3Error as e:
        if e.code == 'NoSuchBucket':
            raise BucketNotFoundError(f"Bucket '{bucket}' not found")
        elif e.code == 'NoSuchKey':
            raise ObjectNotFoundError(f"Object '{object_name}' not found")
        else:
            raise MinioError(f"MinIO error: {e}")

# src/minio_client.py (add caching)

from functools import lru_cache
from datetime import datetime, timedelta

class MinioClient:
    def __init__(self):
        self.cache = {}
        self.cache_ttl = 60  # seconds
    
    def list_objects_cached(self, bucket: str, prefix: str = '') -> List[Dict]:
        """List objects with caching"""
        cache_key = f"{bucket}:{prefix}"
        
        # Check cache
        if cache_key in self.cache:
            cached_data, cached_time = self.cache[cache_key]
            if datetime.now() - cached_time < timedelta(seconds=self.cache_ttl):
                return cached_data
        
        # Fetch fresh data
        data = self.list_objects(bucket, prefix)
        self.cache[cache_key] = (data, datetime.now())
        
        return data

# Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY pyproject.toml .
RUN pip install -e .

# Copy source
COPY src/ src/

# Set environment
ENV PYTHONUNBUFFERED=1

# Run server
CMD ["python", "-m", "minio_mcp_server"]

# docker-compose.yml

version: '3.8'

services:
  minio-mcp:
    build: .
    container_name: minio-mcp-server
    environment:
      MINIO_ENDPOINT: minio:9000
      MINIO_ACCESS_KEY: admin
      MINIO_SECRET_KEY: password123
      MINIO_SECURE: "false"
    depends_on:
      - minio
    networks:
      - minio-network

  minio:
    image: minio/minio:latest
    container_name: minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: admin
      MINIO_ROOT_PASSWORD: password123
    command: server /data --console-address ":9001"
    networks:
      - minio-network

networks:
  minio-network:
    driver: bridge

# /etc/systemd/system/minio-mcp.service

[Unit]
Description=MinIO MCP Server
After=network.target

[Service]
Type=simple
User=mcp
WorkingDirectory=/opt/minio-mcp-server
Environment="MINIO_ENDPOINT=localhost:9000"
Environment="MINIO_ACCESS_KEY=admin"
Environment="MINIO_SECRET_KEY=password123"
ExecStart=/opt/minio-mcp-server/venv/bin/python -m minio_mcp_server
Restart=always

[Install]
WantedBy=multi-user.target

# Enable and start service
sudo systemctl enable minio-mcp
sudo systemctl start minio-mcp

# Check status
sudo systemctl status minio-mcp

# View logs
sudo journalctl -u minio-mcp -f

MCP Server for MinIO:
  tools: 8 (list, get, search, upload, delete, etc.)
  resources: MinIO buckets and objects
  prompts: Analysis templates
  
Features:
  ✅ Direct Claude ↔ MinIO integration
  ✅ Image analysis (screenshots)
  ✅ Text processing (logs, documents)
  ✅ Search and discovery
  ✅ Automated workflows
  
Benefits:
  ✅ No manual file transfers
  ✅ Real-time data access
  ✅ Powerful AI analysis
  ✅ Automated troubleshooting
  ✅ Secure (your infrastructure)

Test Automation:
- "Analyze all failed test screenshots"
- "What caused test-123 to fail?"
- "Compare before/after screenshots"

Log Analysis:
- "Find errors in today's logs"
- "Which service has most issues?"
- "Summarize error patterns"

Document Processing:
- "Summarize all PDFs"
- "Extract data from invoices"
- "Find contracts mentioning X"

Video Analysis:
- "What happens at 00:45 in test video?"
- "Describe UI changes in recording"
- "Find frame where error appears"

Before MCP:
- Download file from MinIO: 30s
- Upload to Claude: 30s
- Wait for analysis: 60s
- Repeat for each file: ×10 = 20 minutes

After MCP:
- "Claude, analyze all files": 30s
- Done! 🎉

That's 40× faster! 🚀

# 1. Add more tools
- Batch operations
- File conversion
- Metadata tagging

# 2. Integrate with other services
- Database queries
- API calls
- Slack notifications

# 3. Create workflows
- Automated test analysis
- Daily log summaries
- Report generation