s3-tools

import asyncio import boto3 from botocore.exceptions import ClientError from mcp.server.models import InitializationOptions import mcp.types as types from mcp.server import NotificationOptions, Server import mcp.server.stdio server = Server("s3-tools") @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools. Each tool specifies its arguments using JSON Schema validation. """ return [ types.Tool( name="list-s3-buckets", description="List all S3 buckets in your AWS account", inputSchema={ "type": "object", "properties": { "region": { "type": "string", "description": "AWS region (optional, defaults to configured region)", } } }, ) ] @server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ Handle tool execution requests. """ if name == "list-s3-buckets": try: region = arguments.get("region") if arguments else None s3_client = boto3.client('s3', region_name=region) if region else boto3.client('s3') response = s3_client.list_buckets() buckets = [bucket['Name'] for bucket in response['Buckets']] return [ types.TextContent( type="text", text=f"Found {len(buckets)} S3 buckets:\n" + "\n".join(f"- {bucket}" for bucket in buckets) ) ] except ClientError as e: return [ types.TextContent( type="text", text=f"Error listing S3 buckets: {str(e)}" ) ] raise ValueError(f"Unknown tool: {name}") async def main(): # Run the server using stdin/stdout streams async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="s3-tools", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), )