"""MinIO tools for MCP server."""
import io
import logging
import os
from typing import Optional
from minio import Minio
from minio.error import S3Error
logger = logging.getLogger(__name__)
def get_minio_client() -> Minio:
"""Create and return a MinIO client."""
try:
endpoint = os.getenv("MINIO_ENDPOINT", "localhost:9000")
access_key = os.getenv("MINIO_ACCESS_KEY")
secret_key = os.getenv("MINIO_SECRET_KEY")
secure = os.getenv("MINIO_SECURE", "false").lower() == "true"
if not access_key or not secret_key:
raise ValueError("MinIO access key and secret key must be provided")
client = Minio(
endpoint=endpoint,
access_key=access_key,
secret_key=secret_key,
secure=secure
)
return client
except Exception as e:
logger.error(f"Failed to create MinIO client: {e}")
raise
async def list_buckets() -> str:
"""List all available MinIO buckets."""
try:
client = get_minio_client()
buckets = client.list_buckets()
if not buckets:
return "No buckets found"
output = ["Available buckets:\n"]
for bucket in buckets:
creation_date = bucket.creation_date.strftime("%Y-%m-%d %H:%M:%S") if bucket.creation_date else "Unknown"
output.append(f"• {bucket.name} (created: {creation_date})")
return "\n".join(output)
except Exception as e:
raise Exception(f"Failed to list buckets: {str(e)}")
async def list_objects(bucket_name: str, prefix: Optional[str] = None, max_keys: int = 1000) -> str:
"""List objects in a MinIO bucket."""
try:
client = get_minio_client()
if not client.bucket_exists(bucket_name):
return f"Bucket '{bucket_name}' does not exist"
objects = client.list_objects(bucket_name=bucket_name, prefix=prefix, recursive=True)
object_list = []
count = 0
for obj in objects:
if count >= max_keys:
break
size_mb = obj.size / (1024 * 1024) if obj.size else 0
last_modified = obj.last_modified.strftime("%Y-%m-%d %H:%M:%S") if obj.last_modified else "Unknown"
object_list.append({
'name': obj.object_name,
'size_mb': round(size_mb, 2),
'last_modified': last_modified
})
count += 1
if not object_list:
prefix_info = f" with prefix '{prefix}'" if prefix else ""
return f"No objects found in bucket '{bucket_name}'{prefix_info}"
output = [f"Objects in bucket '{bucket_name}':\n"]
output.append(f"{'Name':<50} {'Size (MB)':<12} {'Last Modified':<20}")
output.append("-" * 82)
for obj in object_list:
output.append(f"{obj['name']:<50} {obj['size_mb']:<12} {obj['last_modified']:<20}")
return "\n".join(output)
except Exception as e:
raise Exception(f"Failed to list objects: {str(e)}")
async def upload_object(bucket_name: str, object_name: str, data: str, content_type: str = "text/plain") -> str:
"""Upload data to MinIO object store."""
try:
client = get_minio_client()
# Create bucket if it doesn't exist
if not client.bucket_exists(bucket_name):
client.make_bucket(bucket_name)
# Convert string data to bytes
data_bytes = data.encode('utf-8')
data_stream = io.BytesIO(data_bytes)
# Upload object
result = client.put_object(
bucket_name=bucket_name,
object_name=object_name,
data=data_stream,
length=len(data_bytes),
content_type=content_type
)
return f"Successfully uploaded '{object_name}' to bucket '{bucket_name}'"
except Exception as e:
raise Exception(f"Failed to upload object: {str(e)}")
async def download_object(bucket_name: str, object_name: str) -> str:
"""Download object from MinIO."""
try:
client = get_minio_client()
if not client.bucket_exists(bucket_name):
return f"Bucket '{bucket_name}' does not exist"
response = client.get_object(bucket_name, object_name)
data = response.read().decode('utf-8')
response.close()
response.release_conn()
return data
except S3Error as e:
if e.code == 'NoSuchKey':
return f"Object '{object_name}' not found in bucket '{bucket_name}'"
else:
raise Exception(f"Failed to download object: {str(e)}")
except Exception as e:
raise Exception(f"Failed to download object: {str(e)}")
async def create_bucket(bucket_name: str, region: Optional[str] = None) -> str:
"""Create a new MinIO bucket."""
try:
client = get_minio_client()
if client.bucket_exists(bucket_name):
return f"Bucket '{bucket_name}' already exists"
client.make_bucket(bucket_name, location=region)
return f"Successfully created bucket '{bucket_name}'"
except Exception as e:
raise Exception(f"Failed to create bucket: {str(e)}")
async def delete_object(bucket_name: str, object_name: str) -> str:
"""Delete object from MinIO."""
try:
client = get_minio_client()
if not client.bucket_exists(bucket_name):
return f"Bucket '{bucket_name}' does not exist"
client.remove_object(bucket_name, object_name)
return f"Successfully deleted '{object_name}' from bucket '{bucket_name}'"
except S3Error as e:
if e.code == 'NoSuchKey':
return f"Object '{object_name}' not found in bucket '{bucket_name}'"
else:
raise Exception(f"Failed to delete object: {str(e)}")
except Exception as e:
raise Exception(f"Failed to delete object: {str(e)}")
