# Directory Structure
```
├── .gitignore
├── BUILD.md
├── Dockerfile
├── glama.json
├── LAMBDA.md
├── LICENSE
├── logo.png
├── MCP.md
├── PRICING.md
├── README.md
├── requirements.txt
├── smithery.yaml
└── src
├── lambda
│ ├── __init__.py
│ ├── lambda_handler.py
│ ├── requirements.txt
│ ├── test_event.json
│ └── test_lambda.py
└── server.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# Virtual Environment
venv/
env/
ENV/
.env
# IDE files
.idea/
.vscode/
*.swp
*.swo
.DS_Store
# Docker
.dockerignore
docker-compose.override.yml
# Logs
*.log
logs/
# Cache
.cache/
.pytest_cache/
.coverage
htmlcov/
# Data
*.csv
ec2_pricing.json
# Jupyter Notebooks
.ipynb_checkpoints
.aws-sam
samconfig.toml
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# AWS Pricing MCP
A Model Context Protocol (MCP) server that provides AWS EC2 instance pricing data. This project includes both a traditional server implementation and a serverless Lambda function.
## Quick Start
### Lambda Deployment (Recommended)
The Lambda function provides the same functionality as the server but with serverless benefits:
```bash
# Build and deploy
sam build
sam deploy --guided
# Get the Function URL
aws cloudformation describe-stacks \
--stack-name aws-pricing-mcp \
--query 'Stacks[0].Outputs[?OutputKey==`FunctionUrl`].OutputValue' \
--output text
```
For detailed Lambda documentation, see [LAMBDA.md](LAMBDA.md).
### Server Deployment
```bash
# Install dependencies
pip install -r requirements.txt
# Run the server
python src/server.py
```
## Features
- **EC2 Pricing Data**: Find the cheapest EC2 instances based on specifications
- **Multiple Pricing Models**: On Demand, Reserved Instances, CloudFix RightSpend
- **Flexible Filtering**: Region, platform, tenancy, vCPU, RAM, GPU, etc.
- **JSON-RPC 2.0**: Full MCP protocol compliance
- **Serverless Option**: Lambda function with Function URL
- **Dynamic Data**: Always up-to-date pricing from S3
## Documentation
- [LAMBDA.md](LAMBDA.md) - Comprehensive Lambda documentation
- [MCP.md](MCP.md) - MCP protocol examples
- [PRICING.md](PRICING.md) - Pricing data format and sources
- [BUILD.md](BUILD.md) - Build instructions
## License
[LICENSE](LICENSE)
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
fastmcp
```
--------------------------------------------------------------------------------
/src/lambda/__init__.py:
--------------------------------------------------------------------------------
```python
# AWS Pricing MCP Lambda Handler Package
```
--------------------------------------------------------------------------------
/glama.json:
--------------------------------------------------------------------------------
```json
{
"$schema": "https://glama.ai/mcp/schemas/server.json",
"maintainers": ["ai-1st"]
}
```
--------------------------------------------------------------------------------
/src/lambda/requirements.txt:
--------------------------------------------------------------------------------
```
# AWS Pricing MCP Lambda Handler Requirements
# No external dependencies required - uses only Python standard library
# If you need to add any external dependencies in the future, add them here
# Example:
# requests==2.31.0
# boto3==1.34.0
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery.ai configuration
startCommand:
type: stdio
configSchema:
{}
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({
"command": "python",
"args": ["server.py"],
"env": { }
})
```
--------------------------------------------------------------------------------
/src/lambda/test_event.json:
--------------------------------------------------------------------------------
```json
{
"requestContext": {
"http": {
"method": "POST"
}
},
"body": "{\"jsonrpc\": \"2.0\", \"id\": 1, \"method\": \"initialize\", \"params\": {\"protocolVersion\": \"2025-03-26\", \"capabilities\": {}, \"clientInfo\": {\"name\": \"TestClient\", \"version\": \"1.0.0\"}}}"
}
```
--------------------------------------------------------------------------------
/BUILD.md:
--------------------------------------------------------------------------------
```markdown
# Building the AWS EC2 Pricing MCP Server
To build the Docker image for multiple platforms (e.g., x86_64 and ARM64):
1. Enable Docker BuildKit:
```bash
export DOCKER_BUILDKIT=1
```
2. Create a builder that supports multi-platform builds:
```bash
docker buildx create --name multiplatform-builder --driver docker-container --use
```
3. Log in to Docker Hub:
```bash
docker login
```
4. Push the multi-platform image to Docker Hub:
```bash
docker buildx build --platform linux/amd64,linux/arm64 -t ai1st/aws-pricing-mcp:latest --push . --build-arg BUILD_DATE=$(date +%Y-%m-%d)
```
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Use Python 3.9 as the base image
FROM python:3.13-slim
# Set working directory in the container
WORKDIR /app
# Install curl for downloading files
RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
# Copy requirements file and install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy the MCP server code
COPY src/server.py .
# Invalidate pricing cache with this arg
ARG BUILD_DATE=2025-05-08
# Fetch pricing data from public S3 - bust cache daily
RUN echo $BUILD_DATE && curl https://cloudfix-public-aws-pricing.s3.us-east-1.amazonaws.com/pricing/ec2_pricing.json.gz | gunzip > ec2_pricing.json
# Run the server when the container starts
CMD ["python", "server.py"]
```
--------------------------------------------------------------------------------
/PRICING.md:
--------------------------------------------------------------------------------
```markdown
## Pricing Data JSON Format
The pricing data JSON file has the following structure:
```json
{
"instance_family_name": {
"Current Generation": "Yes/No",
"Instance Family": "General Purpose/Compute Optimized/etc.",
"Physical Processor": "Intel Xeon/AMD EPYC/etc.",
"Clock Speed": 2.5,
"Processor Features": "AVX, AVX2, etc.",
"Enhanced Networking Supported": "Yes/No",
"sizes": {
"instance_size": {
"vCPU": 2,
"Memory": 8.0,
"Ephemeral Storage": 0,
"Network Performance": 5000,
"Dedicated EBS Throughput": 650,
"GPU": 0,
"GPU Memory": 0,
"operations": {
"operation_code": {
"region": "price1,price2,price3,..."
}
}
}
}
}
}
```
### Field Descriptions
#### Instance Family Level Fields
- **Instance Family Name**: The name of the instance family (e.g., "t3a")
- **Current Generation**: Indicates if the instance family is current generation ("Yes") or previous generation ("No")
- **Instance Family**: The AWS classification of the instance family (e.g., "General Purpose", "Compute Optimized")
- **Physical Processor**: The CPU manufacturer and model (e.g., "Intel Xeon Platinum 8175", "AMD EPYC 7R13")
- **Clock Speed**: The processor clock speed in GHz (e.g., 2.5)
- **Processor Features**: Special CPU features or instruction sets (e.g., "AVX, AVX2, Intel AVX-512")
- **Enhanced Networking Supported**: Whether enhanced networking is supported ("Yes" or "No")
#### Instance Size Level Fields
- **Instance Size Name**: The name of the instance size (e.g., "2xlarge")
- **vCPU**: Number of virtual CPUs
- **Memory**: Amount of RAM in GiB
- **Ephemeral Storage**: Amount of instance store storage in GB (0 for EBS-only instances)
- **Network Performance**: Network performance in Mbps
- **Dedicated EBS Throughput**: EBS throughput in Mbps
- **GPU**: Number of GPUs (0 if none)
- **GPU Memory**: Amount of GPU memory in GB (0 if no GPU)
#### Operations and Pricing
The `operations` object contains mappings from operation codes to region-specific pricing. Each region has a comma-separated string of prices with the following positions:
- Position 0: OnDemand price for Shared tenancy
- Position 1: No Upfront 1yr Compute Savings Plan price for Shared tenancy
- Position 2: Partial Upfront 1yr Compute Savings Plan price for Shared tenancy
- Position 3: All Upfront 1yr Compute Savings Plan price for Shared tenancy
- Position 4: No Upfront 3yr Compute Savings Plan price for Shared tenancy
- Position 5: Partial Upfront 3yr Compute Savings Plan price for Shared tenancy
- Position 6: All Upfront 3yr Compute Savings Plan price for Shared tenancy
- Position 7: OnDemand price for Dedicated tenancy
- Position 8: No Upfront 1yr Compute Savings Plan price for Dedicated tenancy
- Position 9: Partial Upfront 1yr Compute Savings Plan price for Dedicated tenancy
- Position 10: All Upfront 1yr Compute Savings Plan price for Dedicated tenancy
- Position 11: No Upfront 3yr Compute Savings Plan price for Dedicated tenancy
- Position 12: Partial Upfront 3yr Compute Savings Plan price for Dedicated tenancy
- Position 13: All Upfront 3yr Compute Savings Plan price for Dedicated tenancy
Empty string values indicate that no pricing is available for that specific combination.
## Operation System to Operation Code Mapping
The following table shows the mapping between operating systems and their corresponding operation codes:
| Operating System | Operation Code |
|------------------|---------------|
| Linux/UNIX | "" (empty string) |
| Red Hat BYOL Linux | "00g0" |
| Red Hat Enterprise Linux | "0010" |
| Red Hat Enterprise Linux with HA | "1010" |
| Red Hat Enterprise Linux with SQL Server Standard and HA | "1014" |
| Red Hat Enterprise Linux with SQL Server Enterprise and HA | "1110" |
| Red Hat Enterprise Linux with SQL Server Standard | "0014" |
| Red Hat Enterprise Linux with SQL Server Web | "0210" |
| Red Hat Enterprise Linux with SQL Server Enterprise | "0110" |
| Linux with SQL Server Enterprise | "0100" |
| Linux with SQL Server Standard | "0004" |
| Linux with SQL Server Web | "0200" |
| SUSE Linux | "000g" |
| Windows | "0002" |
| Windows BYOL | "0800" |
| Windows with SQL Server Enterprise | "0102" |
| Windows with SQL Server Standard | "0006" |
| Windows with SQL Server Web | "0202" |
```
--------------------------------------------------------------------------------
/src/lambda/test_lambda.py:
--------------------------------------------------------------------------------
```python
"""
Test script for the AWS Pricing MCP Lambda Handler
This script demonstrates how to use the Lambda handler with sample MCP requests.
"""
import json
from lambda_handler import lambda_handler, process_mcp_request
def test_initialize():
"""Test the initialize request"""
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {
"roots": {
"listChanged": True
},
"sampling": {}
},
"clientInfo": {
"name": "TestClient",
"version": "1.0.0"
}
}
}
# Test direct MCP processing
response = process_mcp_request(request)
print("Initialize Response (Direct MCP):")
print(json.dumps(response, indent=2))
print("\n" + "="*50 + "\n")
# Test HTTP Lambda handler
http_event = {
"requestContext": {
"http": {
"method": "POST"
}
},
"body": json.dumps(request)
}
http_response = lambda_handler(http_event, None)
print("Initialize Response (HTTP Lambda):")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
def test_tools_list():
"""Test the tools/list request"""
request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
# Test direct MCP processing
response = process_mcp_request(request)
print("Tools List Response (Direct MCP):")
print(json.dumps(response, indent=2))
print("\n" + "="*50 + "\n")
# Test HTTP Lambda handler
http_event = {
"requestContext": {
"http": {
"method": "POST"
}
},
"body": json.dumps(request)
}
http_response = lambda_handler(http_event, None)
print("Tools List Response (HTTP Lambda):")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
def test_tools_call():
"""Test the tools/call request"""
request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "ec2_instances_pricing",
"arguments": {
"filter_region": "us-east-1",
"filter_platform": "Linux/UNIX",
"filter_tenancy": "Shared",
"filter_pricing_model": "On Demand",
"filter_min_vcpu": 2,
"filter_min_ram": 4.0,
"filter_max_price_per_hour": 0.1,
"sort_by": "Price",
"sort_order": "Ascending",
"page_num": 0
}
}
}
# Test direct MCP processing
response = process_mcp_request(request)
print("Tools Call Response (Direct MCP):")
print(json.dumps(response, indent=2))
print("\n" + "="*50 + "\n")
# Test HTTP Lambda handler
http_event = {
"requestContext": {
"http": {
"method": "POST"
}
},
"body": json.dumps(request)
}
http_response = lambda_handler(http_event, None)
print("Tools Call Response (HTTP Lambda):")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
def test_ping():
"""Test the ping request"""
request = {
"jsonrpc": "2.0",
"id": "ping-123",
"method": "ping"
}
# Test direct MCP processing
response = process_mcp_request(request)
print("Ping Response (Direct MCP):")
print(json.dumps(response, indent=2))
print("\n" + "="*50 + "\n")
# Test HTTP Lambda handler
http_event = {
"requestContext": {
"http": {
"method": "POST"
}
},
"body": json.dumps(request)
}
http_response = lambda_handler(http_event, None)
print("Ping Response (HTTP Lambda):")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
def test_error_handling():
"""Test error handling with invalid request"""
request = {
"jsonrpc": "2.0",
"id": 4,
"method": "nonexistent_method",
"params": {}
}
# Test direct MCP processing
response = process_mcp_request(request)
print("Error Handling Response (Direct MCP):")
print(json.dumps(response, indent=2))
print("\n" + "="*50 + "\n")
# Test HTTP Lambda handler
http_event = {
"requestContext": {
"http": {
"method": "POST"
}
},
"body": json.dumps(request)
}
http_response = lambda_handler(http_event, None)
print("Error Handling Response (HTTP Lambda):")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
def test_cors_preflight():
"""Test CORS preflight request"""
http_event = {
"requestContext": {
"http": {
"method": "OPTIONS"
}
}
}
http_response = lambda_handler(http_event, None)
print("CORS Preflight Response:")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
def test_invalid_json():
"""Test handling of invalid JSON"""
http_event = {
"requestContext": {
"http": {
"method": "POST"
}
},
"body": "invalid json"
}
http_response = lambda_handler(http_event, None)
print("Invalid JSON Response:")
print(json.dumps(http_response, indent=2))
print("\n" + "="*50 + "\n")
if __name__ == "__main__":
print("Testing AWS Pricing MCP Lambda Handler")
print("="*50)
try:
test_initialize()
test_tools_list()
test_tools_call()
test_ping()
test_error_handling()
test_cors_preflight()
test_invalid_json()
print("All tests completed successfully!")
except Exception as e:
print(f"Test failed with error: {str(e)}")
import traceback
traceback.print_exc()
```
--------------------------------------------------------------------------------
/src/server.py:
--------------------------------------------------------------------------------
```python
"""
AWS Pricing MCP Server
This server exposes AWS EC2 instance pricing data through the Model Context Protocol (MCP).
It provides tools to find the cheapest EC2 instances based on specified criteria.
"""
import json
import os
import sys
import traceback
from typing import List, Dict, Any
from fastmcp import FastMCP, Context
# Set up error logging
def log_error(message):
print(f"ERROR: {message}", file=sys.stderr)
# Load pricing data
try:
PRICING_FILE = os.path.join(os.path.dirname(__file__), "ec2_pricing.json")
if not os.path.exists(PRICING_FILE):
log_error(f"Pricing file not found at {PRICING_FILE}")
raise FileNotFoundError(f"Could not find ec2_pricing.json")
with open(PRICING_FILE, "r") as f:
PRICING_DATA = json.load(f)
except Exception as e:
log_error(f"Failed to load pricing data: {str(e)}")
log_error(traceback.format_exc())
raise
PLATFORM_TO_OP_CODE = {
"Linux/UNIX": "",
"Red Hat BYOL Linux": "00g0",
"Red Hat Enterprise Linux": "0010",
"Red Hat Enterprise Linux with HA": "1010",
"Red Hat Enterprise Linux with SQL Server Standard and HA": "1014",
"Red Hat Enterprise Linux with SQL Server Enterprise and HA": "1110",
"Red Hat Enterprise Linux with SQL Server Standard": "0014",
"Red Hat Enterprise Linux with SQL Server Web": "0210",
"Red Hat Enterprise Linux with SQL Server Enterprise": "0110",
"Linux with SQL Server Enterprise": "0100",
"Linux with SQL Server Standard": "0004",
"Linux with SQL Server Web": "0200",
"SUSE Linux": "000g",
"Windows": "0002",
"Windows BYOL": "0800",
"Windows with SQL Server Enterprise": "0102",
"Windows with SQL Server Standard": "0006",
"Windows with SQL Server Web": "0202",
}
PRICING_MODELS = ["On Demand", "1-yr No Upfront", "1-yr Partial Upfront", "1-yr All Upfront", "3-yr No Upfront", "3-yr Partial Upfront", "3-yr All Upfront"]
TENANCIES = ["Shared", "Dedicated"]
# Create an MCP server
mcp = FastMCP("AWS EC2 Pricing MCP", log_level="ERROR")
# Define Tools
@mcp.tool()
def ec2_instances_pricing(
filter_region: str = "us-east-1",
filter_platform: str = "Linux/UNIX",
filter_tenancy: str = "Shared",
filter_pricing_model: str = "On Demand",
filter_min_vcpu: int = 0,
filter_min_ram: float = 0,
filter_min_gpu: int = 0,
filter_min_gpu_memory: int = 0,
filter_min_cpu_ghz: float = 0,
filter_min_network_performance: int = 0,
filter_min_ebs_throughput: int = 0,
filter_min_ephemeral_storage: int = 0,
filter_max_price_per_hour: float = float('inf'),
filter_family: str = "",
filter_size: str = "",
filter_processor: str = "",
sort_by: str = "Price",
sort_order: str = "Descending",
page_num: int = 0
) -> List[Dict[str, Any]]:
"""
Find AWS EC2 instances based on specified criteria.
Filter Parameters:
- region: AWS region (default: us-east-1)
- platform: OS platform (one of: Linux/UNIX, Red Hat Enterprise Linux, Red Hat Enterprise Linux with HA, Red Hat Enterprise Linux with SQL Server Standard and HA, Red Hat Enterprise Linux with SQL Server Enterprise and HA, Red Hat Enterprise Linux with SQL Server Standard, Red Hat Enterprise Linux with SQL Server Web, Linux with SQL Server Enterprise, Linux with SQL Server Standard, Linux with SQL Server Web, SUSE Linux, Windows, Windows BYOL, Windows with SQL Server Enterprise, Windows with SQL Server Standard, Windows with SQL Server Web; default: Linux/UNIX)
- tenancy: Instance tenancy (one of: Shared, Dedicated; default: Shared)
- pricing_model: Pricing model (one of: On Demand, 1-yr No Upfront, 1-yr Partial Upfront, 1-yr All Upfront, 3-yr No Upfront, 3-yr Partial Upfront, 3-yr All Upfront; default: On Demand)
- min_vcpu: Minimum number of vCPUs (default: 0)
- min_ram: Minimum amount of RAM in GB (default: 0)
- min_gpu: Minimum number of GPUs (default: 0)
- min_gpu_memory: Minimum GPU memory in GB (default: 0)
- min_cpu_ghz: Minimum CPU clock speed in GHz (default: 0)
- min_network_performance: Minimum network performance in Mbps (default: 0)
- min_ebs_throughput: Minimum dedicated EBS throughput in Mbps (default: 0)
- min_ephemeral_storage: Minimum ephemeral storage in GB (default: 0)
- max_price_per_hour: Maximum price per hour in USD (default: no limit)
- sort_by: Field to sort by (one of: Price, Clock Speed GHz, vCPU cores, Memory GB, Ephemeral Storage GB, Network Performance Mbps, Dedicated EBS Throughput Mbps, GPU cores, GPU Memory GB; default: Price)
- sort_order: Sort order (one of: Ascending, Descending; default: Descending)
- family: Filter by instance family (e.g., "m5", "c6g"; default: "" for all families)
- size: Filter by instance size (e.g., "large", "2xlarge"; default: "" for all sizes)
- processor: Filter by physical processor (e.g., "Graviton", "Xeon", "AMD"; default: "" for all processors)
- page_num: Page number for pagination (default: 0)
Returns:
- List of instances matching the criteria (5 per page). CloudFix RightSpend pricing is provided when using the flexible cRIs provided by RightSpend (a third-party solution). The benefit of RightSpend is that it 1) eliminates the need for complex forecasting or frequent consultations with engineering about usage fluctuations 2) removes the risk of unused reservations 3) provides 3-yr All Upfront discounts without the need for prepayment.
"""
# Get the operation code for the platform
if filter_platform not in PLATFORM_TO_OP_CODE:
raise ValueError(f"Invalid platform: {filter_platform}; valid platforms: {list(PLATFORM_TO_OP_CODE.keys())}")
filter_op_code = PLATFORM_TO_OP_CODE.get(filter_platform, "")
if filter_tenancy not in TENANCIES:
raise ValueError(f"Invalid tenancy: {filter_tenancy}; valid tenancies: {list(TENANCIES)}")
if filter_pricing_model not in PRICING_MODELS:
raise ValueError(f"Invalid pricing model: {filter_pricing_model}; valid pricing models: {list(PRICING_MODELS)}")
# Find matching instances
on_demand_price_offset = 7 * TENANCIES.index(filter_tenancy)
price_offset = on_demand_price_offset + PRICING_MODELS.index(filter_pricing_model)
matching_instances = []
for family_name, family in PRICING_DATA.items():
# Filter by family if specified
if filter_family and family_name != filter_family:
continue
# Filter by processor if specified
if filter_processor and filter_processor.lower() not in family.get("Physical Processor", "").lower():
continue
if family["Clock Speed, GHz"] < filter_min_cpu_ghz:
continue
for size_name, size in family["sizes"].items():
# Filter by size if specified
if filter_size and size_name != filter_size:
continue
# Check if the instance meets the minimum requirements
if size.get("vCPU, cores", 0) < filter_min_vcpu:
continue
if size.get("Memory, GB", 0) < filter_min_ram:
continue
if size.get("GPU, cores", 0) < filter_min_gpu:
continue
if size.get("GPU Memory, GB", 0) < filter_min_gpu_memory:
continue
if size.get("Network Performance, Mbps", 0) < filter_min_network_performance:
continue
if size.get("Dedicated EBS Throughput, Mbps", 0) < filter_min_ebs_throughput:
continue
if size.get("Ephemeral Storage, GB", 0) < filter_min_ephemeral_storage:
continue
if "operations" not in size:
raise ValueError(f"Instance {family_name}.{size_name} does not have operations")
for op_code, regions in size["operations"].items():
if op_code != filter_op_code:
continue
for region, prices in regions.items():
if region != filter_region:
continue
price_arr = prices.split(",")
if len(price_arr) < price_offset:
continue
price = price_arr[price_offset]
if price == "":
continue
on_demand_price = price_arr[on_demand_price_offset]
instance = {
"Instance Type": f"{family_name}.{size_name}",
"Region": filter_region,
"Platform": filter_platform,
"Tenancy": filter_tenancy,
"Pricing Model": filter_pricing_model,
"Effective Price per hour, USD": float(price),
"Effective Price per month, USD": round(float(price) * 24 * 365 / 12, 2),
"Effective Price per year, USD": round(float(price) * 24 * 365, 2)
}
if filter_pricing_model == "1-yr Partial Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365 / 2, 2)
elif filter_pricing_model == "1-yr All Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365, 2)
elif filter_pricing_model == "3-yr Partial Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365 * 3 / 2, 2)
elif filter_pricing_model == "3-yr All Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365 * 3, 2)
if on_demand_price and on_demand_price != price:
instance["On-Demand Price per hour, USD"] = float(on_demand_price)
instance["Discount Percentage"] = (1 - (float(price) / float(on_demand_price))) * 100
if len(price_arr) > on_demand_price_offset+6:
cloudfix_rightspend_price = price_arr[on_demand_price_offset+6]
if cloudfix_rightspend_price != "":
instance["CloudFix RightSpend Price per hour, USD"] = float(cloudfix_rightspend_price)
instance["CloudFix RightSpend Price per month, USD"] = round(float(cloudfix_rightspend_price) * 24 * 365 / 12, 2)
instance["CloudFix RightSpend Price per year, USD"] = round(float(cloudfix_rightspend_price) * 24 * 365, 2)
instance["CloudFix RightSpend Upfront Payment, USD"] = round(float(cloudfix_rightspend_price) * 24 * 7, 2)
instance.update({key: value for key, value in family.items() if key != "sizes"})
instance.update({key: value for key, value in size.items() if key != "operations"})
matching_instances.append(instance)
# Filter by max price if specified
if filter_max_price_per_hour != float('inf'):
matching_instances = [i for i in matching_instances if i["Effective Price per hour, USD"] <= filter_max_price_per_hour]
# Define sort key mapping
sort_key_map = {
"Price": "Effective Price per hour, USD",
"Clock Speed GHz": "Clock Speed, GHz",
"vCPU cores": "vCPU, cores",
"Memory GB": "Memory, GB",
"Ephemeral Storage GB": "Ephemeral Storage, GB",
"Network Performance Mbps": "Network Performance, Mbps",
"Dedicated EBS Throughput Mbps": "Dedicated EBS Throughput, Mbps",
"GPU cores": "GPU, cores",
"GPU Memory GB": "GPU Memory, GB"
}
# Sort by selected field
if sort_by not in sort_key_map:
raise ValueError(f"Invalid sort by: {sort_by}; valid sort by: {list(sort_key_map.keys())}")
sort_key = sort_key_map.get(sort_by, "Effective Price per hour, USD")
# Two-pass sorting approach:
# 1. First sort by price (ascending) - this will be our secondary sort
matching_instances.sort(key=lambda x: x.get("Effective Price per hour, USD", 0))
# 2. Then sort by the primary field with the specified direction
# Since Python's sort is stable, when primary fields are equal, the price order is preserved
matching_instances.sort(
key=lambda x: x.get(sort_key, 0),
reverse=(sort_order == "Descending")
)
# Calculate pagination
items_per_page = 5
start_idx = page_num * items_per_page
end_idx = start_idx + items_per_page
# Return the requested page
return matching_instances[start_idx:end_idx]
# Run the server if executed directly
if __name__ == "__main__":
# res = find_instances()
# print(res)
mcp.run(transport="stdio")
```
--------------------------------------------------------------------------------
/MCP.md:
--------------------------------------------------------------------------------
```markdown
This document provides example JSON-RPC messages for the supported MCP commands.
# Example JSON-RPC Messages for Anthropic MCP (Stateless HTTP Mode)
Below are example JSON-RPC 2.0 request and response objects for each relevant Model Context Protocol (MCP) command in stateless HTTP mode. Each example shows a complete JSON structure with realistic field values, based on the MCP specification. (All streaming or SSE-based fields are omitted, as these examples assume a non-streaming HTTP interaction.)
## initialize
**Description:** The client begins a session by sending an `initialize` request with its supported protocol version, capabilities, and client info. The server replies with its own protocol version (which may be negotiated), supported server capabilities (e.g. logging, prompts, resources, tools), server info, and any optional instructions.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {
"roots": {
"listChanged": true
},
"sampling": {}
},
"clientInfo": {
"name": "ExampleClient",
"version": "1.0.0"
}
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"logging": {},
"prompts": {
"listChanged": true
},
"resources": {
"subscribe": true,
"listChanged": true
},
"tools": {
"listChanged": true
}
},
"serverInfo": {
"name": "ExampleServer",
"version": "1.0.0"
},
"instructions": "Optional instructions for the client"
}
}
```
## initialized (notification)
**Description:** After the server responds to `initialize`, the client sends an `initialized` notification to signal that it is ready for normal operations. This is a JSON-RPC notification (no `id` field and no response expected).
**Notification:**
```json
{
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
```
## ping
**Description:** Either party can send a `ping` request at any time to check connectivity. The `ping` request has no parameters, and the receiver must promptly return an empty result object if still alive.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": "123",
"method": "ping"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": "123",
"result": {}
}
```
## resources/list
**Description:** The client requests a list of available resources (files, data, etc.) from the server. The `resources/list` request may include an optional `cursor` for pagination. The response contains an array of resource descriptors (each with fields like `uri`, `name`, `description`, `mimeType`, etc.) and may include a `nextCursor` token if more results are available.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "resources/list",
"params": {
"cursor": "optional-cursor-value"
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"resources": [
{
"uri": "file:///project/src/main.rs",
"name": "main.rs",
"description": "Primary application entry point",
"mimeType": "text/x-rust"
}
],
"nextCursor": "next-page-cursor"
}
}
```
## resources/read
**Description:** The client retrieves the contents of a specific resource by sending `resources/read` with the resource's URI. The server's response includes a `contents` array with the resource data. If the resource is text-based, it appears under a `text` field (with an associated MIME type); for binary data, a `blob` (base64 string) would be used instead.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 2,
"method": "resources/read",
"params": {
"uri": "file:///project/src/main.rs"
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"contents": [
{
"uri": "file:///project/src/main.rs",
"mimeType": "text/x-rust",
"text": "fn main() {\n println!(\"Hello world!\");\n}"
}
]
}
}
```
## resources/templates/list
**Description:** The client can query available *resource templates* (parameterized resource URIs) by sending `resources/templates/list`. The response provides a list of resource template definitions, each with a `uriTemplate` (often containing placeholders), a human-readable `name` and `description`, and an optional `mimeType` indicating the type of resource produced.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 3,
"method": "resources/templates/list"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 3,
"result": {
"resourceTemplates": [
{
"uriTemplate": "file:///{path}",
"name": "Project Files",
"description": "Access files in the project directory",
"mimeType": "application/octet-stream"
}
]
}
}
```
## prompts/list
**Description:** The client requests a list of available prompt templates by sending `prompts/list`. This may also support pagination via a `cursor`. The server responds with an array of prompt definitions, where each prompt has a unique `name`, a `description` of what it does, and an optional list of expected `arguments` (each argument with a name, description, and whether it's required). A `nextCursor` may be provided if the list is paginated.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "prompts/list",
"params": {
"cursor": "optional-cursor-value"
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"prompts": [
{
"name": "code_review",
"description": "Asks the LLM to analyze code quality and suggest improvements",
"arguments": [
{
"name": "code",
"description": "The code to review",
"required": true
}
]
}
],
"nextCursor": "next-page-cursor"
}
}
```
## prompts/get
**Description:** To fetch the content of a specific prompt template (possibly filling in arguments), the client sends `prompts/get` with the prompt's `name` and an `arguments` object providing any required values. The server returns the resolved prompt: typically a `description` and a sequence of `messages` that make up the prompt. Each message has a `role` (e.g. "user" or "assistant") and `content` which could be text or other supported content types.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 2,
"method": "prompts/get",
"params": {
"name": "code_review",
"arguments": {
"code": "def hello():\n print('world')"
}
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"description": "Code review prompt",
"messages": [
{
"role": "user",
"content": {
"type": "text",
"text": "Please review this Python code:\n def hello():\n print('world')"
}
}
]
}
}
```
## tools/list
**Description:** The client sends `tools/list` to get the list of tools (functions/actions) the server provides. The response includes an array of tool definitions. Each tool has a `name`, a `description` of its functionality, and an `inputSchema` (a JSON Schema object) describing the expected parameters for that tool. The example below shows one tool with a required `location` parameter. A `nextCursor` may appear if the list is paginated.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {
"cursor": "optional-cursor-value"
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "get_weather",
"description": "Get current weather information for a location",
"inputSchema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name or zip code"
}
},
"required": ["location"]
}
}
],
"nextCursor": "next-page-cursor"
}
}
```
## tools/call
**Description:** To execute a specific tool, the client sends a `tools/call` request with the tool's `name` and an `arguments` object providing the needed inputs. The server will run the tool and return a result. The result includes a `content` array (which may contain text or other content types, depending on what the tool returns) and an `isError` boolean indicating whether the tool succeeded. In this example, the tool returns a text result (weather information) and `isError: false` to show success.
**Request:**
```json
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": {
"location": "New York"
}
}
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [
{
"type": "text",
"text": "Current weather in New York:\n Temperature: 72°F\n Conditions: Partly cloudy"
}
],
"isError": false
}
}
```
## notifications/cancelled (notification)
**Description:** If either side needs to cancel an in-progress request (e.g. due to a timeout or user action), it sends a `notifications/cancelled` notification. This one-way message includes the `requestId` of the original request to be aborted and an optional `reason` string. The receiver should stop work on that request but does not send any response to the notification.
**Notification:**
```json
{
"jsonrpc": "2.0",
"method": "notifications/cancelled",
"params": {
"requestId": "123",
"reason": "User requested cancellation"
}
}
```
Each JSON example above illustrates the structure and fields defined by the MCP specification for stateless HTTP usage, covering the full request/response cycle (or one-way notification) for that command. These messages can be sent over an HTTP-based JSON-RPC connection to manage the model's context and actions without using server-sent events or streaming protocols. All field names and nesting conform to the MCP spec, ensuring interoperability between MCP clients and servers.
## Introduction
This specification aims to describe a simple protocol for LLMs to discover and use remote APIs, while minimizing the context window used. While Anthropic's Model Context Protocol (MCP) is primarily used for the purpose, its design is non-optimal. MCP began as a simple STDIO‑based local solution but evolved into a complex, stateful HTTP/SSE system that burdens developers with session management and infrastructure headaches. Maintaining persistent connections conflicts with stateless microservice patterns, leading to scalability and load‑balancing challenges. This specification adopts a minimal, stateless design to avoid these pitfalls.
## Overview
Webtools expose a lightweight, HTTP‑based contract that allows consumers to
* **Discover** capabilities through self‑describing metadata
* **Validate** inputs and outputs via JSON Schema definitions
* one schema for the request object (`requestSchema`)
* one schema for the response object (`responseSchema`)
* **Execute** actions with optional per‑request configuration
* **Consume** predictable, strongly‑typed responses
* **Lock-in** specific API versions to improve security
## Use Case Scenario
Webtools are defined by URLs. The typical workflow follows these steps:
1. **Discovery**: A user finds a webtool URL from a tool provider, marketplace, or other source
2. **Metadata Retrieval**: The user's system issues a GET request to the URL to retrieve the webtool's metadata
3. **Configuration**: The user fills in configuration data according to the `configSchema` defined in the metadata
4. **Integration**: The system is now able to use the webtool with LLMs, passing the configuration and handling requests/responses
### Security Considerations
After reviewing the schemas and metadata, users may choose to lock-in a specific version by storing the validated metadata on their side and no longer fetching it from the remote server. This prevents potential security risks where malicious instructions could be injected into the LLM context through schema changes in newer versions of the webtool metadata.
## HTTP Methods
### GET {webtoolUrl}/ — Webtool Metadata (latest)
Returns metadata about the **latest** version of the webtool.
```json
{
"name": "webtool_name",
"description": "Human‑readable description of what this webtool does",
"version": "2.1.0",
"actions": [
{
"name": "action_name",
"description": "What this action does",
"requestSchema": { /* JSON Schema for request */ },
"responseSchema": { /* JSON Schema for response */ }
}
],
"configSchema": { /* JSON Schema for configuration */ },
"defaultConfig": { /* Default configuration values */ }
}
```
### GET {webtoolUrl}/{version} — Webtool Metadata (specific version)
Returns metadata **for the specified semantic version**. Use this to fetch historical versions or pin a client to a stable release. The `version` parameter is optional; if omitted, the server SHOULD default to the latest version.
```http
GET /1.0.0
```
```json
{
"name": "weather",
"description": "Provides weather information",
"version": "1.0.0",
"actions": [ /* …as above… */ ],
"configSchema": { /* … */ },
"defaultConfig": { /* … */ }
}
```
> **Note**: If the version is not found, the endpoint should return `404 Not Found` with an error envelope identical to the standard error response.
### POST {webtoolUrl}/ — Webtool Execution
```json
{
"sessionId": "unique-session-identifier",
"version": "1.0.0",
"action": "action_name",
"config": { /* Optional configuration object matching configSchema */ },
"request": { /* Required data matching requestSchema */ }
}
```
The `sessionId` field is optional and used for maintaining state across multiple requests to the same webtool. The `version` field is optional; if omitted, the server SHOULD default to the latest version. The `config` property contains data that is not generated by the LLM but rather supplied by the environment, allowing minimization of context use, passing security credentials, parameter defaults, or user-configured values.
#### Response Examples
##### Successful Response
```json
{
"status": "ok",
"data": { /* Action‑specific result */ }
}
```
##### Error Response
```json
{
"status": "error",
"error": {
"code": "INVALID_INPUT",
"message": "Validation failed for field 'location'"
}
}
```
## Content Types
All requests and responses MUST use `application/json` content type. Servers MUST include `Content-Type: application/json` headers in their responses.
## JSON Schema Requirements
Webtools MAY use any JSON Schema features. Schemas SHOULD include descriptions and example values to help LLMs understand the expected data structure and format.
## Error Handling
The specification defines standard error codes for common validation errors:
- `WEBTOOL_NOT_FOUND` - Requested webtool or version does not exist
- `SCHEMA_ERROR` - Request data does not match the action's requestSchema
- `CONFIG_ERROR` - Configuration data does not match the webtool's configSchema
- `RATE_LIMIT` - Request rate limit exceeded
- `INTERNAL_ERROR` - Unrecoverable server error
Webtools MAY define their own custom error codes for domain-specific errors. Error messages SHOULD be human-readable.
## Integration Guides
### Using Webtools with the **Vercel AI SDK**
The Vercel AI SDK supports OpenAI‑style *tool calling* out‑of‑the‑box.
#### 1 – Fetch Metadata at Build Time
```ts
// lib/tools/weather.ts
import type { Tool } from "ai";
export async function getWeatherTool(): Promise<Tool> {
const res = await fetch("/api/webtools/weather");
const meta = await res.json();
return {
name: meta.name,
description: meta.description,
parameters: meta.actions[0].requestSchema, // <-- uses requestSchema
execute: async (args) => {
const exec = await fetch("/api/webtools/weather", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
sessionId: crypto.randomUUID(),
action: args.action,
request: args
})
});
return (await exec.json()).data; // unwrap the envelope
}
};
}
```
#### 2 – Create a Tool Caller
```ts
import { createToolCaller } from "ai/tool-caller";
import OpenAI from "@ai-sdk/openai";
import { getWeatherTool } from "@/lib/tools/weather";
const toolCaller = createToolCaller([await getWeatherTool()]);
export async function chat(messages) {
const llm = new OpenAI();
const modelResponse = await llm.chat({ messages, tools: toolCaller.tools });
const final = await toolCaller.call(modelResponse);
return final;
}
```
> **Tip:** The AI SDK automatically translates `requestSchema` into the function‑calling format the model expects.
### Using Webtools with **LangChain**
LangChain's `StructuredTool` helper lets you wrap a webtool with schema metadata so agents can invoke it.
```python
from langchain_core.tools import StructuredTool
import requests, uuid
WEATHER_ENDPOINT = "https://api.example.com/webtools/weather"
def run_get_current(location: str, units: str = "metric"):
body = {
"sessionId": str(uuid.uuid4()),
"action": "get_current",
"request": {"location": location},
"config": {"units": units}
}
return requests.post(WEATHER_ENDPOINT, json=body, timeout=10).json()
weather_tool = StructuredTool.from_function(
func=run_get_current,
name="get_current_weather",
description="Return the current weather for a given location via the Weather webtool",
schema={
"type": "object",
"properties": {
"location": {"type": "string"},
"units": {"type": "string", "enum": ["metric", "imperial"], "default": "metric"}
},
"required": ["location"]
}
)
```
Then add `weather_tool` to any LCEL runnable or agent:
```python
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
llm = ChatOpenAI(model_name="gpt-4o")
agent = create_openai_functions_agent(llm, tools=[weather_tool])
executor = AgentExecutor(agent=agent, tools=[weather_tool])
result = executor.invoke("Should I take an umbrella to Paris today?")
print(result)
```
## Error Handling & Best Practices
* Validate client input against each action's `requestSchema` before issuing a POST.
* For recoverable failures (4xx), return `status: "error"` with appropriate error codes.
* For unrecoverable server errors (5xx), set code `INTERNAL_ERROR` and avoid leaking internals.
* Use standard HTTP status codes alongside the JSON response for broad compatibility.
* Rate limiting, quotas, and other implementation details are left to implementers.
## Authentication & Authorization
This specification is agnostic regarding authorization schemes. Consumers MAY include an Authorization: Bearer <token> header on every request. Additional details—such as token scopes, expiration, or refresh flows—are considered implementation-specific and are not mandated by this spec.
---
**Version:** 2025‑06‑30
```
--------------------------------------------------------------------------------
/LAMBDA.md:
--------------------------------------------------------------------------------
```markdown
# AWS Pricing MCP Lambda Handler
This document provides comprehensive documentation for the AWS Pricing MCP Lambda handler implementation, including deployment, usage, and technical details.
## Table of Contents
1. [Overview](#overview)
2. [Quick Start](#quick-start)
3. [Architecture](#architecture)
4. [Usage](#usage)
5. [Deployment](#deployment)
6. [Configuration](#configuration)
7. [Testing](#testing)
8. [Monitoring](#monitoring)
9. [Troubleshooting](#troubleshooting)
10. [Implementation Details](#implementation-details)
11. [Performance](#performance)
12. [Security](#security)
13. [Future Enhancements](#future-enhancements)
## Overview
The Lambda handler implements the Model Context Protocol (MCP) specification as described in the `MCP.md` file, providing a serverless way to access AWS EC2 pricing data through JSON-RPC 2.0 requests. The function automatically downloads the latest pricing data from S3 and caches it for optimal performance. It's deployed using AWS SAM with a Function URL for direct HTTP access.
### Key Features
- **JSON-RPC 2.0**: Full compliance with JSON-RPC 2.0 specification
- **Dynamic Pricing Data**: Downloads latest pricing data from S3 at runtime
- **HTTP Function URL**: Direct HTTP access with no authentication required
- **CORS Support**: Full CORS headers for web browser access
- **Caching**: Intelligent caching for optimal performance
- **Error Handling**: Comprehensive error handling and validation
- **No Dependencies**: Uses only Python standard library modules
## Quick Start
### Prerequisites
1. **AWS CLI** installed and configured
2. **AWS SAM CLI** installed
3. **Python 3.9+** installed
4. **AWS Account** with appropriate permissions
### Install Prerequisites
```bash
# Install AWS CLI
curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install
# Install AWS SAM CLI
pip install aws-sam-cli
# Configure AWS CLI
aws configure
```
### Deploy the Function
```bash
# From the project root directory
sam build
sam deploy --guided
```
The guided deployment will prompt for configuration values. Use the defaults or customize as needed.
### Get the Function URL
After deployment, get the Function URL:
```bash
aws cloudformation describe-stacks \
--stack-name aws-pricing-mcp \
--query 'Stacks[0].Outputs[?OutputKey==`InvokeUrl`].OutputValue' \
--output text
```
### Test the Function
```bash
# Test with curl
curl -X POST YOUR_FUNCTION_URL \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {},
"clientInfo": {
"name": "TestClient",
"version": "1.0.0"
}
}
}'
```
## Architecture
### Components
1. **Lambda Function**: Processes MCP protocol requests
2. **Function URL**: Provides HTTP access without authentication
3. **IAM Role**: Provides execution permissions
4. **CloudWatch Logs**: Stores function logs
### Request/Response Flow
```
Client Request (JSON-RPC 2.0)
↓
Lambda Handler (lambda_handler function)
↓
Method Router (based on "method" field)
↓
Specific Handler Function
↓
JSON-RPC 2.0 Response
```
### Function URL Features
- **No Authentication**: Public access
- **CORS Enabled**: Web browser compatible
- **HTTP Methods**: GET, POST, OPTIONS
- **Direct Access**: No API Gateway required
## Usage
### Local Testing
```bash
cd src/lambda
python test_lambda.py
```
### HTTP API Usage
Once deployed, the function is accessible via HTTP POST requests:
```bash
# Get the Function URL from SAM outputs
FUNCTION_URL=$(aws cloudformation describe-stacks --stack-name aws-pricing-mcp --query 'Stacks[0].Outputs[?OutputKey==`InvokeUrl`].OutputValue' --output text)
# Test the function
curl -X POST $FUNCTION_URL \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {},
"clientInfo": {
"name": "TestClient",
"version": "1.0.0"
}
}
}'
```
### Supported MCP Methods
| Method | Handler Function | Status |
|--------|------------------|--------|
| `initialize` | `handle_initialize()` | ✅ Implemented |
| `ping` | `handle_ping()` | ✅ Implemented |
| `tools/list` | `handle_tools_list()` | ✅ Implemented |
| `tools/call` | `handle_tools_call()` | ✅ Implemented |
| `resources/list` | `handle_resources_list()` | ✅ Empty implementation |
| `resources/read` | `handle_resources_read()` | ✅ Empty implementation |
| `prompts/list` | `handle_prompts_list()` | ✅ Empty implementation |
| `prompts/get` | `handle_prompts_get()` | ✅ Empty implementation |
### Sample MCP Requests
#### Initialize
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {},
"clientInfo": {
"name": "TestClient",
"version": "1.0.0"
}
}
}
```
#### Get Tools List
```json
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
```
#### Call EC2 Pricing Tool
```json
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "ec2_instances_pricing",
"arguments": {
"filter_region": "us-east-1",
"filter_platform": "Linux/UNIX",
"filter_min_vcpu": 2,
"filter_min_ram": 4.0,
"filter_max_price_per_hour": 0.1,
"sort_by": "Price",
"sort_order": "Ascending"
}
}
}
```
### EC2 Pricing Tool
The main tool provided is `ec2_instances_pricing` which allows filtering and searching EC2 instances based on:
- **Region**: AWS region (default: us-east-1)
- **Platform**: OS platform (Linux/UNIX, Windows, Red Hat, etc.)
- **Tenancy**: Shared or Dedicated
- **Pricing Model**: On Demand, Reserved Instances, etc.
- **Specifications**: vCPU, RAM, GPU, network performance, etc.
- **Cost**: Maximum price per hour
- **Sorting**: By price, specifications, etc.
- **Pagination**: 5 results per page
## Deployment
### SAM Template Parameters
The `template.yaml` file supports the following parameters:
| Parameter | Default | Description |
|-----------|---------|-------------|
| `FunctionName` | `aws-pricing-mcp` | Name of the Lambda function |
| `Runtime` | `python3.12` | Python runtime version |
| `Architecture` | `x86_64` | Lambda function architecture (x86_64 or arm64) |
| `Timeout` | `30` | Function timeout in seconds |
| `MemorySize` | `512` | Function memory size in MB |
### Custom Deployment
```bash
# Deploy with custom parameters
sam deploy \
--stack-name my-pricing-mcp \
--parameter-overrides \
FunctionName=my-pricing-function \
Runtime=python3.12 \
Architecture=arm64 \
Timeout=60 \
MemorySize=1024
```
### SAM Template
The deployment uses `template.yaml` in the project root which defines:
- Lambda function with Function URL
- IAM roles and policies
- CORS configuration
- Output values for Function URL
## Configuration
### Lambda Function Settings
- **Runtime**: Python 3.12
- **Handler**: `lambda_handler.lambda_handler`
- **Timeout**: 30 seconds (configurable)
- **Memory**: 512 MB (increased for pricing data processing)
- **Architecture**: x86_64 or arm64 (configurable)
### Environment Variables
No environment variables are required.
### IAM Permissions
The Lambda function requires:
- **CloudWatch Logs** for logging
- **Internet access** to download pricing data from S3
- **Basic execution permissions**
### Pricing Data Configuration
- **Download URL**: https://cloudfix-public-aws-pricing.s3.us-east-1.amazonaws.com/pricing/ec2_pricing.json.gz
- **Cache Location**: In-memory (global variable)
- **Cache Duration**: 1 hour (3600 seconds)
- **Compression**: Gzip compressed JSON (decompressed in memory)
### Function URL Configuration
- **Auth Type**: NONE (no authentication required)
- **CORS**: Enabled for all origins
- **Methods**: GET, POST, OPTIONS
- **Headers**: All headers allowed
## Testing
The `test_lambda.py` script provides comprehensive testing of all MCP methods with sample requests and expected responses.
### Testing Results
The test script successfully validates:
1. **Initialize**: Returns proper MCP capabilities and server info
2. **Tools List**: Returns complete tool schema with all parameters
3. **Tools Call**: Successfully executes pricing queries and returns results
4. **Ping**: Returns empty result (health check)
5. **Error Handling**: Properly handles invalid methods
Sample successful query returned 5 EC2 instances matching criteria:
- t4g.medium ($0.0336/hour)
- t3a.medium ($0.0376/hour)
- t3.medium ($0.0416/hour)
- t2.medium ($0.0464/hour)
- a1.large ($0.051/hour)
## Monitoring
### CloudWatch Logs
```bash
# View function logs
aws logs tail /aws/lambda/aws-pricing-mcp --follow
```
### CloudWatch Metrics
Monitor key metrics:
- **Invocations**: Number of function calls
- **Duration**: Function execution time
- **Errors**: Number of errors
- **Throttles**: Number of throttled requests
### Custom Metrics
The function logs important events:
- Pricing data download attempts
- Cache hits/misses
- Error conditions
### Key Metrics to Monitor
- **Download Success Rate**: Percentage of successful pricing data downloads
- **Cache Hit Rate**: Percentage of requests using cached data
- **Response Time**: Time to process requests
- **Error Rate**: Percentage of failed requests
- **HTTP Status Codes**: Monitor 4xx and 5xx errors
## Troubleshooting
### Common Issues
1. **Deployment Fails**
```bash
# Check SAM build
sam build --debug
# Check CloudFormation events
aws cloudformation describe-stack-events --stack-name aws-pricing-mcp
```
2. **Function URL Not Working**
```bash
# Verify Function URL exists
aws lambda get-function-url-config --function-name aws-pricing-mcp
# Test with curl
curl -v YOUR_FUNCTION_URL
```
3. **Pricing Data Download Fails**
```bash
# Check function logs
aws logs tail /aws/lambda/aws-pricing-mcp --since 1h
# Verify internet access
# Ensure Lambda is not in a private VPC without NAT Gateway
```
4. **CORS Issues**
- Verify CORS headers in function response
- Check browser console for CORS errors
- Ensure preflight OPTIONS requests are handled
### Debugging Tips
- Check CloudWatch logs for detailed error messages
- Monitor pricing data download events
- Verify cache file existence and size
- Test with simple requests first
- Use curl or Postman to test HTTP endpoints
## Implementation Details
### Protocol Implementation
**Original Server (fastmcp-based):**
- Uses `fastmcp` library for MCP protocol handling
- HTTP/SSE-based communication
- Stateful server with session management
- Complex dependency management
**Lambda Handler (standard library):**
- Manual JSON-RPC 2.0 implementation
- Stateless request handling
- Direct method routing
- No external dependencies
### Lambda Handler Function
The `lambda_handler` function is the entry point that:
1. Handles HTTP requests from Function URL
2. Parses incoming JSON-RPC requests
3. Validates request format
4. Routes to appropriate handler based on method
5. Returns HTTP response with JSON-RPC response
### Handler Functions
Each MCP method has a dedicated handler function:
- `handle_initialize()` - Protocol initialization
- `handle_tools_list()` - Tool discovery
- `handle_tools_call()` - Tool execution
- `handle_ping()` - Health check
- etc.
### Pricing Data Management
The pricing data is managed through:
- `download_and_cache_pricing_data()` - Downloads and caches pricing data in memory
- **Source**: https://cloudfix-public-aws-pricing.s3.us-east-1.amazonaws.com/pricing/ec2_pricing.json.gz
- **Cache Location**: In-memory global variable
- **Cache Duration**: 1 hour (3600 seconds)
- **Fallback**: Uses cached data if download fails
### Pricing Logic
The `ec2_instances_pricing()` function contains the core pricing logic:
- Ensures pricing data is loaded (downloads if needed)
- Applies filters based on parameters
- Calculates pricing for different models
- Sorts and paginates results
### Error Handling
**JSON-RPC Error Codes:**
- `-32700`: Parse error (invalid JSON)
- `-32600`: Invalid request (wrong jsonrpc version)
- `-32601`: Method not found
- `-32603`: Internal error
**Validation:**
- Input parameter validation
- Platform/tenancy/pricing model validation
- Graceful handling of missing pricing data
### Dependencies
The Lambda handler uses only Python standard library modules:
- `json` - JSON parsing and serialization
- `os` - File system operations
- `sys` - System-specific parameters
- `traceback` - Exception handling
- `typing` - Type hints
- `datetime` - Date/time operations
- `gzip` - Gzip file decompression
- `urllib.request` - HTTP downloads
- `time` - Time-based operations
- `pathlib` - Path operations
No external dependencies are required, making deployment simple and lightweight.
## Performance
### Lambda Configuration
- **Cold Start**: ~100-500ms (first request, includes data download)
- **Warm Start**: ~10-50ms (subsequent requests use cached data)
- **Memory Usage**: ~100-200 MB for pricing data in memory
- **Concurrency**: Handles multiple concurrent requests
### Caching Strategy
- **Cache Duration**: 1 hour balances freshness with performance
- **Cache Location**: In-memory global variable
- **Cache Size**: ~100MB uncompressed in memory
- **Cache Hit Rate**: High for typical usage patterns
### Network Considerations
- **Download Time**: ~2-5 seconds for initial data download
- **Retry Logic**: Falls back to cached data if download fails
- **Bandwidth**: ~25MB download per cache refresh
### HTTP Performance
- **Function URL**: Direct HTTP access without API Gateway
- **CORS**: Pre-configured for web browser access
- **Response Time**: Fast JSON-RPC responses
### Performance Characteristics
#### Download Performance
- **Download Time**: ~2-5 seconds for initial download
- **Decompression**: ~1-2 seconds for gzip extraction
- **File Size**: ~25MB compressed, ~100MB uncompressed
- **Network**: Requires internet access from Lambda
#### Caching Performance
- **Cache Hit**: ~10-50ms response time
- **Cache Miss**: ~3-7 seconds (includes download)
- **Memory Usage**: ~100-200 MB for pricing data
- **Storage**: ~100MB in memory
## Security
### Function URL Security
- **No Authentication**: Function URL is publicly accessible
- **CORS**: Configured for all origins (customize as needed)
- **Rate Limiting**: Consider adding rate limiting for production use
### IAM Permissions
The function uses minimal IAM permissions:
- CloudWatch Logs access
- Basic Lambda execution permissions
- No additional AWS service access required
### Network Security
- **Internet Access**: Required for pricing data download
- **VPC**: If using VPC, ensure NAT Gateway for internet access
- **Security Groups**: Allow outbound HTTPS traffic
## Cost Optimization
### Lambda Costs
- **Memory**: 512 MB (adjust based on usage)
- **Timeout**: 30 seconds (sufficient for most requests)
- **Concurrency**: Auto-scaling based on demand
### Optimization Tips
1. **Memory Tuning**: Monitor memory usage and adjust
2. **Timeout Tuning**: Set appropriate timeout values
3. **Caching**: Function caches pricing data for 1 hour
4. **Concurrency**: Monitor and adjust if needed
## Updates and Maintenance
### Updating the Function
```bash
# Deploy updates
sam build
sam deploy
```
### Updating Configuration
```bash
# Update parameters
sam deploy --parameter-overrides Timeout=60 MemorySize=1024
```
### Monitoring Updates
- Monitor pricing data freshness
- Check for new AWS instance types
- Review CloudWatch metrics regularly
## Production Considerations
### High Availability
- **Multi-Region**: Deploy to multiple regions if needed
- **Backup**: Consider backup strategies for critical data
- **Monitoring**: Set up comprehensive monitoring and alerting
### Performance
- **Caching**: Function caches pricing data for optimal performance
- **Cold Starts**: Consider provisioned concurrency for consistent performance
- **Memory**: Monitor and optimize memory usage
### Security
- **Authentication**: Consider adding authentication for production use
- **Rate Limiting**: Implement rate limiting to prevent abuse
- **Monitoring**: Set up security monitoring and alerting
## Future Enhancements
### Potential Improvements
1. **Multi-Region Support**: Download pricing data for multiple regions
2. **Incremental Updates**: Only download changed pricing data
3. **Compression**: Use more efficient compression formats
4. **CDN Integration**: Use CloudFront for faster downloads
5. **Background Updates**: Update cache in background threads
6. **API Gateway**: Add API Gateway for additional features
7. **Custom Domain**: Use custom domain with Function URL
### Monitoring Enhancements
1. **Custom Metrics**: Track download performance and cache efficiency
2. **Alerts**: Set up CloudWatch alarms for download failures
3. **Dashboard**: Create comprehensive monitoring dashboard
4. **Health Checks**: Implement pricing data freshness checks
5. **Performance Monitoring**: Track HTTP response times
## Support
For issues and questions:
1. Check CloudWatch Logs for error details
2. Review this documentation
3. Check AWS Lambda documentation
4. Review SAM documentation
## Cleanup
To remove the deployment:
```bash
sam delete
```
This will remove:
- Lambda function
- Function URL
- IAM role
- CloudWatch log group
- All associated resources
## File Structure
```
src/lambda/
├── __init__.py # Package initialization
├── lambda_handler.py # Main Lambda handler
├── test_lambda.py # Test script
├── test_event.json # Test event for local testing
├── requirements.txt # Dependencies (empty)
└── README.md # Documentation (merged into this file)
```
## Migration Path
### From Original Server
1. **Replace fastmcp calls** with direct JSON-RPC handling
2. **Update deployment** to use Lambda instead of HTTP server
3. **Modify client code** to call Lambda function instead of HTTP endpoint
4. **Update monitoring** to use CloudWatch instead of server logs
### Client Integration
```python
# Example client code
import boto3
import json
lambda_client = boto3.client('lambda')
def call_mcp_server(method, params=None):
request = {
"jsonrpc": "2.0",
"id": 1,
"method": method,
"params": params or {}
}
response = lambda_client.invoke(
FunctionName='aws-pricing-mcp',
Payload=json.dumps(request)
)
return json.loads(response['Payload'].read())
```
## Conclusion
The Lambda handler implementation successfully provides the same functionality as the original fastmcp-based server while offering significant advantages in terms of deployment simplicity, cost-effectiveness, and scalability. The implementation is production-ready and can be deployed immediately to AWS Lambda.
The dynamic pricing data download implementation provides significant benefits:
- **Always up-to-date pricing data**
- **Reduced deployment complexity**
- **Improved reliability and error handling**
- **Better performance through intelligent caching**
- **Simplified maintenance and updates**
The implementation is production-ready and provides a robust foundation for serving current AWS EC2 pricing data through the MCP protocol.
```
--------------------------------------------------------------------------------
/src/lambda/lambda_handler.py:
--------------------------------------------------------------------------------
```python
"""
AWS Pricing MCP Lambda Handler
This Lambda function implements the Model Context Protocol (MCP) for AWS EC2 pricing data
without the fastmcp dependency. It handles JSON-RPC requests and provides tools for
finding EC2 instances based on specified criteria.
"""
import json
import os
import sys
import traceback
import gzip
import urllib.request
import time
from typing import List, Dict, Any, Optional, Union
from datetime import datetime
from pathlib import Path
# Set up error logging
def log_error(message):
print(f"ERROR: {message}", file=sys.stderr)
# Global variable to store pricing data
PRICING_DATA = None
PRICING_DATA_LAST_UPDATE = 0
CACHE_DURATION = 3600 # Cache for 1 hour (3600 seconds)
def download_and_cache_pricing_data():
"""
Download pricing data from S3 and store it in memory.
Returns the pricing data.
"""
global PRICING_DATA, PRICING_DATA_LAST_UPDATE
current_time = time.time()
# Check if we have cached data that's still valid
if PRICING_DATA is not None and (current_time - PRICING_DATA_LAST_UPDATE) < CACHE_DURATION:
return PRICING_DATA
# S3 URL for pricing data
pricing_url = "https://cloudfix-public-aws-pricing.s3.us-east-1.amazonaws.com/pricing/ec2_pricing.json.gz"
try:
# Download the gzipped file
log_error(f"Downloading pricing data from {pricing_url}")
# Download and decompress in memory
with urllib.request.urlopen(pricing_url) as response:
compressed_data = response.read()
# Decompress the gzipped data
decompressed_data = gzip.decompress(compressed_data)
# Parse the JSON data
PRICING_DATA = json.loads(decompressed_data.decode('utf-8'))
PRICING_DATA_LAST_UPDATE = current_time
log_error(f"Successfully downloaded and cached pricing data. Size: {len(decompressed_data)} bytes")
return PRICING_DATA
except Exception as e:
log_error(f"Failed to download pricing data: {str(e)}")
log_error(traceback.format_exc())
# If download fails and we have cached data, use it
if PRICING_DATA is not None:
log_error("Using cached pricing data as fallback")
return PRICING_DATA
# If all else fails, raise the original error
raise
# Initialize pricing data on module load
try:
download_and_cache_pricing_data()
except Exception as e:
log_error(f"Failed to initialize pricing data: {str(e)}")
# Don't raise here - let the handler try to download when needed
PLATFORM_TO_OP_CODE = {
"Linux/UNIX": "",
"Red Hat BYOL Linux": "00g0",
"Red Hat Enterprise Linux": "0010",
"Red Hat Enterprise Linux with HA": "1010",
"Red Hat Enterprise Linux with SQL Server Standard and HA": "1014",
"Red Hat Enterprise Linux with SQL Server Enterprise and HA": "1110",
"Red Hat Enterprise Linux with SQL Server Standard": "0014",
"Red Hat Enterprise Linux with SQL Server Web": "0210",
"Red Hat Enterprise Linux with SQL Server Enterprise": "0110",
"Linux with SQL Server Enterprise": "0100",
"Linux with SQL Server Standard": "0004",
"Linux with SQL Server Web": "0200",
"SUSE Linux": "000g",
"Windows": "0002",
"Windows BYOL": "0800",
"Windows with SQL Server Enterprise": "0102",
"Windows with SQL Server Standard": "0006",
"Windows with SQL Server Web": "0202",
}
PRICING_MODELS = ["On Demand", "1-yr No Upfront", "1-yr Partial Upfront", "1-yr All Upfront", "3-yr No Upfront", "3-yr Partial Upfront", "3-yr All Upfront"]
TENANCIES = ["Shared", "Dedicated"]
# MCP Server Info
SERVER_INFO = {
"name": "AWS EC2 Pricing MCP",
"version": "1.0.0"
}
# MCP Protocol Version
PROTOCOL_VERSION = "2025-03-26"
def create_error_response(request_id: str, code: int, message: str, data: Optional[Dict] = None) -> Dict:
"""Create a JSON-RPC error response"""
error = {
"code": code,
"message": message
}
if data:
error["data"] = data
return {
"jsonrpc": "2.0",
"id": request_id,
"error": error
}
def create_success_response(request_id: str, result: Any) -> Dict:
"""Create a JSON-RPC success response"""
return {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
def handle_initialize(params: Dict) -> Dict:
"""Handle MCP initialize request"""
return {
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {
"logging": {},
"prompts": {
"listChanged": True
},
"resources": {
"subscribe": True,
"listChanged": True
},
"tools": {
"listChanged": True
}
},
"serverInfo": SERVER_INFO,
"instructions": "AWS EC2 Pricing MCP Server - Provides tools to find EC2 instances based on pricing and specifications"
}
def handle_ping() -> Dict:
"""Handle MCP ping request"""
return {}
def handle_tools_list(params: Dict) -> Dict:
"""Handle MCP tools/list request"""
tools = [
{
"name": "ec2_instances_pricing",
"description": "Find AWS EC2 instances based on specified criteria including pricing, specifications, and filters",
"inputSchema": {
"type": "object",
"properties": {
"filter_region": {
"type": "string",
"description": "AWS region (default: us-east-1)",
"default": "us-east-1"
},
"filter_platform": {
"type": "string",
"description": "OS platform",
"enum": list(PLATFORM_TO_OP_CODE.keys()),
"default": "Linux/UNIX"
},
"filter_tenancy": {
"type": "string",
"description": "Instance tenancy",
"enum": TENANCIES,
"default": "Shared"
},
"filter_pricing_model": {
"type": "string",
"description": "Pricing model",
"enum": PRICING_MODELS,
"default": "On Demand"
},
"filter_min_vcpu": {
"type": "integer",
"description": "Minimum number of vCPUs",
"default": 0
},
"filter_min_ram": {
"type": "number",
"description": "Minimum amount of RAM in GB",
"default": 0
},
"filter_min_gpu": {
"type": "integer",
"description": "Minimum number of GPUs",
"default": 0
},
"filter_min_gpu_memory": {
"type": "integer",
"description": "Minimum GPU memory in GB",
"default": 0
},
"filter_min_cpu_ghz": {
"type": "number",
"description": "Minimum CPU clock speed in GHz",
"default": 0
},
"filter_min_network_performance": {
"type": "integer",
"description": "Minimum network performance in Mbps",
"default": 0
},
"filter_min_ebs_throughput": {
"type": "integer",
"description": "Minimum dedicated EBS throughput in Mbps",
"default": 0
},
"filter_min_ephemeral_storage": {
"type": "integer",
"description": "Minimum ephemeral storage in GB",
"default": 0
},
"filter_max_price_per_hour": {
"type": "number",
"description": "Maximum price per hour in USD",
"default": None
},
"filter_family": {
"type": "string",
"description": "Filter by instance family (e.g., 'm5', 'c6g')",
"default": ""
},
"filter_size": {
"type": "string",
"description": "Filter by instance size (e.g., 'large', '2xlarge')",
"default": ""
},
"filter_processor": {
"type": "string",
"description": "Filter by physical processor (e.g., 'Graviton', 'Xeon', 'AMD')",
"default": ""
},
"sort_by": {
"type": "string",
"description": "Field to sort by",
"enum": ["Price", "Clock Speed GHz", "vCPU cores", "Memory GB", "Ephemeral Storage GB", "Network Performance Mbps", "Dedicated EBS Throughput Mbps", "GPU cores", "GPU Memory GB"],
"default": "Price"
},
"sort_order": {
"type": "string",
"description": "Sort order",
"enum": ["Ascending", "Descending"],
"default": "Descending"
},
"page_num": {
"type": "integer",
"description": "Page number for pagination",
"default": 0
}
}
}
}
]
return {
"tools": tools
}
def ec2_instances_pricing(
filter_region: str = "us-east-1",
filter_platform: str = "Linux/UNIX",
filter_tenancy: str = "Shared",
filter_pricing_model: str = "On Demand",
filter_min_vcpu: int = 0,
filter_min_ram: float = 0,
filter_min_gpu: int = 0,
filter_min_gpu_memory: int = 0,
filter_min_cpu_ghz: float = 0,
filter_min_network_performance: int = 0,
filter_min_ebs_throughput: int = 0,
filter_min_ephemeral_storage: int = 0,
filter_max_price_per_hour: Optional[float] = None,
filter_family: str = "",
filter_size: str = "",
filter_processor: str = "",
sort_by: str = "Price",
sort_order: str = "Descending",
page_num: int = 0
) -> List[Dict[str, Any]]:
"""
Find AWS EC2 instances based on specified criteria.
Returns:
- List of instances matching the criteria (5 per page). CloudFix RightSpend pricing is provided when using the flexible cRIs provided by RightSpend (a third-party solution). The benefit of RightSpend is that it 1) eliminates the need for complex forecasting or frequent consultations with engineering about usage fluctuations 2) removes the risk of unused reservations 3) provides 3-yr All Upfront discounts without the need for prepayment.
"""
global PRICING_DATA
# Ensure we have pricing data
if PRICING_DATA is None:
try:
download_and_cache_pricing_data()
except Exception as e:
raise RuntimeError(f"Failed to load pricing data: {str(e)}")
# Get the operation code for the platform
if filter_platform not in PLATFORM_TO_OP_CODE:
raise ValueError(f"Invalid platform: {filter_platform}; valid platforms: {list(PLATFORM_TO_OP_CODE.keys())}")
filter_op_code = PLATFORM_TO_OP_CODE.get(filter_platform, "")
if filter_tenancy not in TENANCIES:
raise ValueError(f"Invalid tenancy: {filter_tenancy}; valid tenancies: {list(TENANCIES)}")
if filter_pricing_model not in PRICING_MODELS:
raise ValueError(f"Invalid pricing model: {filter_pricing_model}; valid pricing models: {list(PRICING_MODELS)}")
# Find matching instances
on_demand_price_offset = 7 * TENANCIES.index(filter_tenancy)
price_offset = on_demand_price_offset + PRICING_MODELS.index(filter_pricing_model)
matching_instances = []
for family_name, family in PRICING_DATA.items():
# Filter by family if specified
if filter_family and family_name != filter_family:
continue
# Filter by processor if specified
if filter_processor and filter_processor.lower() not in family.get("Physical Processor", "").lower():
continue
if family["Clock Speed, GHz"] < filter_min_cpu_ghz:
continue
for size_name, size in family["sizes"].items():
# Filter by size if specified
if filter_size and size_name != filter_size:
continue
# Check if the instance meets the minimum requirements
if size.get("vCPU, cores", 0) < filter_min_vcpu:
continue
if size.get("Memory, GB", 0) < filter_min_ram:
continue
if size.get("GPU, cores", 0) < filter_min_gpu:
continue
if size.get("GPU Memory, GB", 0) < filter_min_gpu_memory:
continue
if size.get("Network Performance, Mbps", 0) < filter_min_network_performance:
continue
if size.get("Dedicated EBS Throughput, Mbps", 0) < filter_min_ebs_throughput:
continue
if size.get("Ephemeral Storage, GB", 0) < filter_min_ephemeral_storage:
continue
if "operations" not in size:
raise ValueError(f"Instance {family_name}.{size_name} does not have operations")
for op_code, regions in size["operations"].items():
if op_code != filter_op_code:
continue
for region, prices in regions.items():
if region != filter_region:
continue
price_arr = prices.split(",")
if len(price_arr) < price_offset:
continue
price = price_arr[price_offset]
if price == "":
continue
on_demand_price = price_arr[on_demand_price_offset]
instance = {
"Instance Type": f"{family_name}.{size_name}",
"Region": filter_region,
"Platform": filter_platform,
"Tenancy": filter_tenancy,
"Pricing Model": filter_pricing_model,
"Effective Price per hour, USD": float(price),
"Effective Price per month, USD": round(float(price) * 24 * 365 / 12, 2),
"Effective Price per year, USD": round(float(price) * 24 * 365, 2)
}
if filter_pricing_model == "1-yr Partial Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365 / 2, 2)
elif filter_pricing_model == "1-yr All Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365, 2)
elif filter_pricing_model == "3-yr Partial Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365 * 3 / 2, 2)
elif filter_pricing_model == "3-yr All Upfront":
instance["Upfront Payment, USD"] = round(float(price) * 24 * 365 * 3, 2)
if on_demand_price and on_demand_price != price:
instance["On-Demand Price per hour, USD"] = float(on_demand_price)
instance["Discount Percentage"] = (1 - (float(price) / float(on_demand_price))) * 100
if len(price_arr) > on_demand_price_offset+6:
cloudfix_rightspend_price = price_arr[on_demand_price_offset+6]
if cloudfix_rightspend_price != "":
instance["CloudFix RightSpend Price per hour, USD"] = float(cloudfix_rightspend_price)
instance["CloudFix RightSpend Price per month, USD"] = round(float(cloudfix_rightspend_price) * 24 * 365 / 12, 2)
instance["CloudFix RightSpend Price per year, USD"] = round(float(cloudfix_rightspend_price) * 24 * 365, 2)
instance["CloudFix RightSpend Upfront Payment, USD"] = round(float(cloudfix_rightspend_price) * 24 * 7, 2)
instance.update({key: value for key, value in family.items() if key != "sizes"})
instance.update({key: value for key, value in size.items() if key != "operations"})
matching_instances.append(instance)
# Filter by max price if specified
if filter_max_price_per_hour is not None:
matching_instances = [i for i in matching_instances if i["Effective Price per hour, USD"] <= filter_max_price_per_hour]
# Define sort key mapping
sort_key_map = {
"Price": "Effective Price per hour, USD",
"Clock Speed GHz": "Clock Speed, GHz",
"vCPU cores": "vCPU, cores",
"Memory GB": "Memory, GB",
"Ephemeral Storage GB": "Ephemeral Storage, GB",
"Network Performance Mbps": "Network Performance, Mbps",
"Dedicated EBS Throughput Mbps": "Dedicated EBS Throughput, Mbps",
"GPU cores": "GPU, cores",
"GPU Memory GB": "GPU Memory, GB"
}
# Sort by selected field
if sort_by not in sort_key_map:
raise ValueError(f"Invalid sort by: {sort_by}; valid sort by: {list(sort_key_map.keys())}")
sort_key = sort_key_map.get(sort_by, "Effective Price per hour, USD")
# Two-pass sorting approach:
# 1. First sort by price (ascending) - this will be our secondary sort
matching_instances.sort(key=lambda x: x.get("Effective Price per hour, USD", 0))
# 2. Then sort by the primary field with the specified direction
# Since Python's sort is stable, when primary fields are equal, the price order is preserved
matching_instances.sort(
key=lambda x: x.get(sort_key, 0),
reverse=(sort_order == "Descending")
)
# Calculate pagination
items_per_page = 5
start_idx = page_num * items_per_page
end_idx = start_idx + items_per_page
# Return the requested page
return matching_instances[start_idx:end_idx]
def handle_tools_call(params: Dict) -> Dict:
"""Handle MCP tools/call request"""
try:
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "ec2_instances_pricing":
result = ec2_instances_pricing(**arguments)
return {
"content": [
{
"type": "text",
"text": json.dumps(result, indent=2)
}
],
"isError": False
}
else:
return {
"content": [
{
"type": "text",
"text": f"Unknown tool: {tool_name}"
}
],
"isError": True
}
except Exception as e:
return {
"content": [
{
"type": "text",
"text": f"Error executing tool: {str(e)}"
}
],
"isError": True
}
def handle_resources_list(params: Dict) -> Dict:
"""Handle MCP resources/list request"""
return {
"resources": []
}
def handle_resources_read(params: Dict) -> Dict:
"""Handle MCP resources/read request"""
return {
"contents": []
}
def handle_prompts_list(params: Dict) -> Dict:
"""Handle MCP prompts/list request"""
return {
"prompts": []
}
def handle_prompts_get(params: Dict) -> Dict:
"""Handle MCP prompts/get request"""
return {
"description": "",
"messages": []
}
def process_mcp_request(request: Dict) -> Dict:
"""
Process MCP protocol request and return response
Args:
request: JSON-RPC request object
Returns:
JSON-RPC response object
"""
try:
# Extract request details
jsonrpc = request.get("jsonrpc")
request_id = request.get("id")
method = request.get("method")
params = request.get("params", {})
# Validate JSON-RPC version
if jsonrpc != "2.0":
return create_error_response(
request_id,
-32600,
"Invalid Request: jsonrpc must be '2.0'"
)
# Handle different MCP methods
if method == "initialize":
result = handle_initialize(params)
return create_success_response(request_id, result)
elif method == "ping":
result = handle_ping()
return create_success_response(request_id, result)
elif method == "tools/list":
result = handle_tools_list(params)
return create_success_response(request_id, result)
elif method == "tools/call":
result = handle_tools_call(params)
return create_success_response(request_id, result)
elif method == "resources/list":
result = handle_resources_list(params)
return create_success_response(request_id, result)
elif method == "resources/read":
result = handle_resources_read(params)
return create_success_response(request_id, result)
elif method == "prompts/list":
result = handle_prompts_list(params)
return create_success_response(request_id, result)
elif method == "prompts/get":
result = handle_prompts_get(params)
return create_success_response(request_id, result)
elif method == "notifications/initialized":
# This is a notification, no response needed
return None
else:
return create_error_response(
request_id,
-32601,
f"Method not found: {method}"
)
except json.JSONDecodeError as e:
return create_error_response(
None,
-32700,
f"Parse error: {str(e)}"
)
except Exception as e:
log_error(f"Unexpected error: {str(e)}")
log_error(traceback.format_exc())
return create_error_response(
request_id if 'request_id' in locals() else None,
-32603,
f"Internal error: {str(e)}"
)
def lambda_handler(event: Dict, context: Any) -> Dict:
"""
AWS Lambda handler for HTTP requests via Function URL
Args:
event: Lambda event containing HTTP request details
context: Lambda context object
Returns:
HTTP response with JSON-RPC response
"""
try:
# Handle CORS preflight requests
if event.get("requestContext", {}).get("http", {}).get("method") == "OPTIONS":
return {
"statusCode": 200,
"headers": {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "*",
"Access-Control-Allow-Methods": "GET, POST, OPTIONS",
"Access-Control-Allow-Credentials": "true"
},
"body": ""
}
# Get the HTTP method and body
http_method = event.get("requestContext", {}).get("http", {}).get("method", "POST")
body = event.get("body", "{}")
# Parse the request body
if isinstance(body, str):
try:
request = json.loads(body)
except json.JSONDecodeError:
return {
"statusCode": 400,
"headers": {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*"
},
"body": json.dumps({
"error": "Invalid JSON in request body"
})
}
else:
request = body
# Process the MCP request
response = process_mcp_request(request)
# Return HTTP response
if response is None:
# For notifications, return empty success
return {
"statusCode": 200,
"headers": {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*"
},
"body": ""
}
else:
return {
"statusCode": 200,
"headers": {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*"
},
"body": json.dumps(response)
}
except Exception as e:
log_error(f"Unexpected error in lambda_handler: {str(e)}")
log_error(traceback.format_exc())
return {
"statusCode": 500,
"headers": {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*"
},
"body": json.dumps({
"error": "Internal server error",
"message": str(e)
})
}
```