aws_advisor_server.py•9.06 kB
#!/usr/bin/env python3
"""
AWS Service Advisor MCP Server
A simple MCP server that provides AWS service recommendations based on use cases.
"""
import asyncio
import json
from typing import Any
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio
# AWS service database with use cases
AWS_SERVICES = {
"compute": {
"EC2": "Virtual servers for flexible compute capacity, long-running applications",
"Lambda": "Serverless functions for event-driven, short-duration workloads",
"ECS": "Container orchestration with Docker, managed container service",
"EKS": "Managed Kubernetes for container orchestration at scale",
"Fargate": "Serverless container compute, no server management",
"Lightsail": "Simple virtual private servers for small projects, pre-configured apps"
},
"storage": {
"S3": "Object storage for files, backups, static websites, data lakes",
"EBS": "Block storage volumes for EC2 instances, database storage",
"EFS": "Managed file storage for Linux workloads, shared file systems",
"Glacier": "Long-term archival storage, compliance, cold data",
"FSx": "High-performance file systems (Windows, Lustre, NetApp, OpenZFS)"
},
"database": {
"RDS": "Managed relational databases (MySQL, PostgreSQL, Oracle, SQL Server)",
"DynamoDB": "NoSQL key-value database, high performance, serverless",
"Aurora": "MySQL/PostgreSQL compatible, high-performance relational database",
"DocumentDB": "MongoDB-compatible document database",
"ElastiCache": "In-memory caching (Redis, Memcached), session storage",
"Neptune": "Graph database for connected data, social networks, recommendations",
"Redshift": "Data warehouse for analytics, OLAP, business intelligence"
},
"networking": {
"VPC": "Virtual private cloud, network isolation, custom networking",
"CloudFront": "CDN for content delivery, low latency, global distribution",
"Route53": "DNS service, domain registration, traffic routing",
"API Gateway": "Create, publish, maintain APIs, REST/WebSocket APIs",
"Direct Connect": "Dedicated network connection from on-premises to AWS",
"ELB": "Load balancing (Application, Network, Gateway Load Balancers)"
},
"analytics": {
"Athena": "Query S3 data with SQL, serverless analytics",
"EMR": "Big data processing (Hadoop, Spark), data transformation",
"Kinesis": "Real-time data streaming, log analytics, IoT data",
"Glue": "ETL service, data cataloging, data preparation",
"QuickSight": "Business intelligence, dashboards, visualizations"
},
"ml_ai": {
"SageMaker": "Build, train, deploy ML models, managed Jupyter notebooks",
"Rekognition": "Image and video analysis, face detection, content moderation",
"Comprehend": "Natural language processing, sentiment analysis, entity extraction",
"Polly": "Text-to-speech, voice synthesis",
"Transcribe": "Speech-to-text, audio transcription",
"Translate": "Language translation service",
"Lex": "Conversational interfaces, chatbots, voice assistants"
},
"security": {
"IAM": "Identity and access management, user permissions, roles",
"Cognito": "User authentication, identity pools, user management",
"KMS": "Key management, encryption keys, cryptographic operations",
"Secrets Manager": "Store and rotate secrets, credentials, API keys",
"WAF": "Web application firewall, DDoS protection, traffic filtering",
"GuardDuty": "Threat detection, security monitoring, anomaly detection"
},
"messaging": {
"SQS": "Message queuing, decouple services, asynchronous processing",
"SNS": "Pub/sub messaging, notifications, mobile push, email/SMS",
"EventBridge": "Event bus, application integration, serverless workflows",
"SES": "Email sending service, transactional emails, marketing campaigns"
},
"monitoring": {
"CloudWatch": "Monitoring, logs, metrics, alarms, dashboards",
"X-Ray": "Distributed tracing, application debugging, performance analysis",
"CloudTrail": "API logging, governance, compliance, audit trails"
}
}
def search_aws_services(use_case: str) -> dict[str, Any]:
"""
Search for AWS services matching a use case description.
Args:
use_case: Description of the use case or requirements
Returns:
Dictionary with recommended services and explanations
"""
use_case_lower = use_case.lower()
recommendations = {}
# Search through all services
for category, services in AWS_SERVICES.items():
matching_services = {}
for service_name, description in services.items():
# Check if use case keywords match service description
if any(word in description.lower() for word in use_case_lower.split()):
matching_services[service_name] = description
if matching_services:
recommendations[category] = matching_services
return recommendations
async def main():
"""Main entry point for the AWS Advisor MCP server."""
server = Server("aws-advisor")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="suggest_aws_service",
description="Get AWS service recommendations based on your use case. Provide a description of what you want to build or accomplish, and get relevant AWS service suggestions.",
inputSchema={
"type": "object",
"properties": {
"use_case": {
"type": "string",
"description": "Description of your use case, requirements, or what you want to build (e.g., 'serverless API', 'store images', 'real-time data processing')"
}
},
"required": ["use_case"]
}
),
Tool(
name="list_aws_categories",
description="List all available AWS service categories (compute, storage, database, etc.) to help you explore services by type.",
inputSchema={
"type": "object",
"properties": {}
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool calls."""
if name == "suggest_aws_service":
use_case = arguments.get("use_case", "")
if not use_case:
return [TextContent(
type="text",
text="Error: Please provide a use case description."
)]
recommendations = search_aws_services(use_case)
if not recommendations:
return [TextContent(
type="text",
text=f"No specific AWS services found matching '{use_case}'. Try describing your use case differently (e.g., 'serverless compute', 'object storage', 'relational database')."
)]
# Format the response
response_lines = [f"## AWS Service Recommendations for: '{use_case}'\n"]
for category, services in recommendations.items():
response_lines.append(f"\n### {category.upper().replace('_', ' ')}")
for service_name, description in services.items():
response_lines.append(f"\n**{service_name}**")
response_lines.append(f"- {description}")
return [TextContent(
type="text",
text="\n".join(response_lines)
)]
elif name == "list_aws_categories":
response_lines = ["## AWS Service Categories\n"]
for category, services in AWS_SERVICES.items():
service_count = len(services)
response_lines.append(f"\n### {category.upper().replace('_', ' ')} ({service_count} services)")
response_lines.append(f"Services: {', '.join(services.keys())}")
response_lines.append("\n\n**Tip**: Use the 'suggest_aws_service' tool with a use case description to get specific recommendations!")
return [TextContent(
type="text",
text="\n".join(response_lines)
)]
else:
return [TextContent(
type="text",
text=f"Error: Unknown tool '{name}'"
)]
# Run the server
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())