s3-tools
by sofianhamiti
- src
- 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={},
),
),
)