# Directory Structure
```
├── .env.example
├── .gitignore
├── azure_pricing_mcp_server.py
├── config.py
├── env
├── README.md
├── requirements.txt
└── tests
├── test_api.py
└── test_monthly_cost.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Entornos virtuales
.venv/
env/
venv/
ENV/
# Archivos de 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
# Archivos de entorno
.env
.env.local
.env.*.local
!.env.example
# Logs
*.log
logs/
# Archivos del sistema operativo
.DS_Store
Thumbs.db
# IDEs y editores
.idea/
.vscode/
*.swp
*.swo
*~
# Archivos de caché
.cache/
.pytest_cache/
.mypy_cache/
# Archivos de cobertura
htmlcov/
.coverage
.coverage.*
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Archivos de documentación generados
docs/_build/
# Archivos de Jupyter Notebook
.ipynb_checkpoints
# Archivos temporales
tmp/
temp/
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# ============================================
# CONFIGURACIÓN OBLIGATORIA
# ============================================
# Configuración del servidor MCP (obligatorio)
MCP_HOST=0.0.0.0 # Dirección IP para escuchar (ej: 0.0.0.0 para todas las interfaces)
MCP_PORT=8080 # Puerto para el servidor
# Configuración de Azure Retail Prices API (obligatorio)
MCP_AZURE_RETAIL_PRICES_URL=https://prices.azure.com/api/retail/prices
MCP_AZURE_API_VERSION=2023-01-01-preview
# Configuración de cálculos (obligatorio)
MCP_HOURS_IN_MONTH=730 # Horas en un mes (aprox. 24/7)
# Configuración de alternativas (obligatorio)
MCP_MAX_ALTERNATIVES_TO_SHOW=3 # Número máximo de alternativas a mostrar
# Configuración de precios (obligatorio)
MCP_PRICE_TYPE=Consumption # Tipo de precio (ej: 'Consumption')
# ============================================
# CONFIGURACIÓN OPCIONAL
# ============================================
# Configuración de CORS (opcional, separar orígenes por comas)
MCP_CORS_ORIGINS=*
# Nivel de logging (opcional, valores: DEBUG, INFO, WARNING, ERROR, CRITICAL)
MCP_LOG_LEVEL=INFO
# Modo desarrollo (opcional)
MCP_DEBUG=false # Habilita el modo debug
MCP_RELOAD=false # Recarga automática en desarrollo
# ============================================
# NOTAS
# ============================================
# 1. Todas las variables marcadas como (obligatorio) deben tener un valor
# 2. Las cadenas no necesitan comillas
# 3. Los valores booleanos deben ser 'true' o 'false' (minúsculas)
# 4. Para múltiples orígenes CORS, separar por comas: http://localhost:3000,http://otro-dominio.com
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Azure Pricing MCP Server
[](https://www.python.org/downloads/)
[](https://opensource.org/licenses/MIT)
This project provides a Model Context Protocol (MCP) server that allows you to programmatically query Azure resource pricing. The server provides a structured workflow to retrieve pricing information from the Azure Retail Prices API.
## Features
- Query Azure pricing data through a simple, structured workflow
- Get real-time pricing information from the Azure Retail Prices API
- Navigate through Azure service families, service names, and products
- Calculate monthly costs for Azure resources
## System Requirements
- Python 3.8 or higher
- Internet connection to access the Azure Retail Prices API
- Permission to install Python packages
- No Azure account or credentials required (uses public pricing API)
## Installation
1. **Clone the repository:**
```bash
git clone https://github.com/sboludaf/mcp-azure-pricing.git
cd mcp-azure-pricing
```
2. **Create and activate a virtual environment:**
```bash
python -m venv .venv
```
* Windows:
```bash
.venv\Scripts\activate
```
* macOS/Linux:
```bash
source .venv/bin/activate
```
3. **Install the dependencies:**
```bash
pip install -r requirements.txt
```
## Usage
The MCP server provides a structured four-step workflow for accessing Azure pricing information:
1. **Get service families** - Retrieve the list of available Azure service families
2. **Get service names** - Get service names within a specific family
3. **Get products** - Get products associated with a service
4. **Calculate monthly costs** - Calculate the monthly cost for a specific product
## Starting the MCP Server
```bash
source .venv/bin/activate # Activate the virtual environment
python azure_pricing_mcp_server.py
```
The server will start at `http://0.0.0.0:8080` by default.
### Available Endpoints
- `GET /sse`: Server-Sent Events endpoint for MCP communication
- `GET /tools`: Lists the available tools in the MCP server
### MCP Client Configuration
To configure an MCP client to connect to this server, add the following to your `mcp_config.json` file:
```json
"azure-pricing": {
"serverUrl": "http://localhost:8080/sse"
}
```
This configuration tells the MCP client to connect to the local server on port 8080 using the SSE endpoint. Make sure the URL matches the address and port your server is running on.
## MCP Tools
The server provides four main tools that form a logical workflow for querying Azure pricing:
### 1. list_service_families
**Description**: Lists all available service families in Azure according to Microsoft's official documentation.
### 2. get_service_names
**Description**: Gets all unique service names within a specified service family.
**Parameters**:
- `service_family`: The service family to query (e.g., 'Compute', 'Storage')
- `region`: Azure region (default: 'westeurope')
- `max_results`: Maximum number of results to process
### 3. get_products
**Description**: Gets product names from a specific service family.
**Parameters**:
- `service_family`: The service family to query
- `region`: Azure region (default: 'westeurope')
- `type`: Price type (optional, e.g., 'Consumption', 'Reservation')
- `service_name`: Service name to filter by (optional)
- `product_name_contains`: Filter products whose name contains this text (optional)
- `limit`: Maximum number of products to return (optional)
### 4. get_monthly_cost
**Description**: Calculates the monthly cost of a specific Azure product.
**Parameters**:
- `product_name`: Exact name of the product (e.g., 'Azure App Service Premium v3 Plan')
- `region`: Azure region (default: 'westeurope')
- `monthly_hours`: Number of hours per month (default: 730)
- `type`: Price type (optional, e.g., 'Consumption')
## Error Handling
The MCP server includes a robust error handling system that:
- Provides descriptive error messages when resources cannot be found
- Properly handles Azure API errors
- Logs detailed information for debugging purposes
### Common Error Scenarios
- **Product not found**: When a product name doesn't exist in the specified region
- **Service family not found**: When an invalid service family is specified
- **API rate limits**: When the Azure Retail Prices API rate limits are exceeded
- **Network errors**: When the server cannot connect to the Azure API
## Limitations
* Prices are estimates based on public information from the Azure Retail Prices API
* Does not include all possible discounts, account-specific offers, or additional costs like taxes or support
* The Azure Retail Prices API has rate limits that can affect performance with a high volume of requests
* Prices may vary depending on the region and currency selected
* Not all Azure resources are available in all regions
## Contributing
Contributions are welcome! Here's how you can contribute to this project:
1. Fork the repository
2. Create a feature branch: `git checkout -b feature/amazing-feature`
3. Commit your changes: `git commit -m 'Add some amazing feature'`
4. Push to the branch: `git push origin feature/amazing-feature`
5. Open a Pull Request
## License
This project is licensed under the MIT License - see the LICENSE file for details.
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
# Core dependencies
requests>=2.28.0
fastapi>=0.89.0
uvicorn>=0.20.0
starlette>=0.25.0
pydantic>=1.10.0
# MCP server
mcp-server>=0.1.0
# Utilities
tabulate>=0.9.0
python-dotenv>=0.21.0
# Azure-specific (if needed)
# azure-core>=1.26.0
# azure-identity>=1.12.0
```
--------------------------------------------------------------------------------
/tests/test_api.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import requests
import json
import sys
def test_api_call(service_family, region="westeurope", type="", service_name=""):
"""
Prueba directa de la API de Azure Retail Prices con los parámetros especificados.
"""
print(f"Consultando API para familia: {service_family}, región: {region}, tipo: {type}, servicio: {service_name}")
# Configuración de la API
AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
API_VERSION = "2023-01-01-preview"
# Construir los parámetros de la consulta
filter_params = f"serviceFamily eq '{service_family}'"
if region:
filter_params += f" and armRegionName eq '{region}'"
if type:
filter_params += f" and type eq '{type}'"
if service_name:
filter_params += f" and serviceName eq '{service_name}'"
params = {
'api-version': API_VERSION,
'$filter': filter_params,
'$top': 10 # Limitamos a 10 resultados para la prueba
}
print(f"Filtro: {filter_params}")
try:
# Realizar la petición a la API
response = requests.get(AZURE_PRICE_API, params=params)
response.raise_for_status()
result = response.json()
# Mostrar información sobre los resultados
items = result.get("Items", [])
print(f"Se encontraron {len(items)} productos")
# Mostrar los primeros 3 resultados
for i, item in enumerate(items[:3]):
print(f"\nProducto {i+1}:")
print(f" Nombre: {item.get('productName', 'N/A')}")
print(f" Servicio: {item.get('serviceName', 'N/A')}")
print(f" SKU: {item.get('skuName', 'N/A')}")
print(f" Región: {item.get('armRegionName', 'N/A')}")
print(f" Precio: {item.get('retailPrice', 'N/A')} {item.get('currencyCode', '')}")
return result
except requests.exceptions.RequestException as e:
print(f"Error al conectar con la API: {str(e)}")
return None
if __name__ == "__main__":
# Procesar argumentos de línea de comandos
service_family = sys.argv[1] if len(sys.argv) > 1 else "Networking"
region = sys.argv[2] if len(sys.argv) > 2 else "westeurope"
service_name = sys.argv[3] if len(sys.argv) > 3 else "Virtual Network"
type_param = sys.argv[4] if len(sys.argv) > 4 else ""
# Llamar a la función de prueba
test_api_call(service_family, region, type_param, service_name)
```
--------------------------------------------------------------------------------
/tests/test_monthly_cost.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import requests
import json
def calculate_monthly_cost(product_name, region="westeurope", monthly_hours=730, type="Consumption"):
"""
Calcula el coste mensual de un producto específico de Azure.
Args:
product_name: Nombre exacto del producto
region: Región de Azure
monthly_hours: Número de horas al mes (por defecto 730)
type: Tipo de precio
"""
# Configuración de la API
AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
API_VERSION = "2023-01-01-preview"
# Construir la consulta
filter_params = f"productName eq '{product_name}' and armRegionName eq '{region}'"
if type:
filter_params += f" and type eq '{type}'"
params = {
'api-version': API_VERSION,
'$filter': filter_params
}
print(f"Consultando API para: {product_name} en {region}")
print(f"Filtro: {filter_params}")
try:
# Realizar la petición a la API
response = requests.get(AZURE_PRICE_API, params=params)
response.raise_for_status()
result = response.json()
# Extraer los items relevantes
items = result.get("Items", [])
if len(items) == 0:
print(f"No se encontraron productos para: {product_name}")
return
print(f"Se encontraron {len(items)} variantes del producto.")
total_monthly_cost = 0
# Mostrar información para cada variante
for i, item in enumerate(items):
sku_name = item.get("skuName", "")
meter_name = item.get("meterName", "")
retail_price = item.get("retailPrice", 0)
currency = item.get("currencyCode", "USD")
unit_of_measure = item.get("unitOfMeasure", "")
# Calcular coste mensual
monthly_cost = retail_price * monthly_hours if "Hour" in unit_of_measure else retail_price
total_monthly_cost += monthly_cost
print(f"\nVariante {i+1}:")
print(f" SKU: {sku_name}")
print(f" Meter: {meter_name}")
print(f" Precio por unidad: {retail_price} {currency} por {unit_of_measure}")
print(f" Coste mensual estimado: {monthly_cost:.2f} {currency}")
print(f"\nCoste mensual total estimado: {total_monthly_cost:.2f} {currency}")
except requests.exceptions.RequestException as e:
print(f"Error al conectar con la API: {str(e)}")
if __name__ == "__main__":
import sys
# Valores por defecto
product_name = "Azure App Service Premium v3 Plan"
region = "westeurope"
monthly_hours = 730
type = "Consumption"
# Procesar argumentos de línea de comandos si se proporcionan
if len(sys.argv) > 1:
product_name = sys.argv[1]
if len(sys.argv) > 2:
region = sys.argv[2]
if len(sys.argv) > 3:
monthly_hours = int(sys.argv[3])
if len(sys.argv) > 4:
type = sys.argv[4]
# Calcular el coste mensual
calculate_monthly_cost(product_name, region, monthly_hours, type)
```
--------------------------------------------------------------------------------
/config.py:
--------------------------------------------------------------------------------
```python
"""
Configuración del servidor MCP de Azure Pricing.
Este archivo puede ser sobreescrito mediante variables de entorno con prefijo MCP_
"""
from typing import List, Optional, Union
from pydantic import Field, field_validator, HttpUrl, validator
from pydantic_settings import BaseSettings, SettingsConfigDict
import os
# Valores por defecto
DEFAULT_HOST = '0.0.0.0'
DEFAULT_PORT = 8080
DEFAULT_LOG_LEVEL = 'INFO'
class Settings(BaseSettings):
# Configuración del servidor
MCP_HOST: str = Field(
default=DEFAULT_HOST,
description='Dirección IP para escuchar (ej: 0.0.0.0 para todas las interfaces)'
)
MCP_PORT: int = Field(
default=DEFAULT_PORT,
description='Puerto para el servidor',
gt=0,
lt=65536
)
MCP_DEBUG: bool = Field(
default=True,
description='Habilita el modo debug (no usar en producción)'
)
MCP_RELOAD: bool = Field(
default=False,
description='Habilita la recarga automática en desarrollo'
)
# Configuración de CORS
CORS_ORIGINS: Union[str, List[str]] = Field(
default='*',
description='Orígenes permitidos para CORS (separados por comas o lista)'
)
# Configuración de logging
LOG_LEVEL: str = Field(
default=DEFAULT_LOG_LEVEL,
description='Nivel de logging (DEBUG, INFO, WARNING, ERROR, CRITICAL)'
)
# Configuración de Azure Retail Prices API
AZURE_RETAIL_PRICES_URL: str = Field(
default='https://prices.azure.com/api/retail/prices',
description='URL base de la API de precios de Azure'
)
AZURE_API_VERSION: str = Field(
default='2023-01-01-preview',
description='Versión de la API de precios de Azure'
)
# Configuración de cálculos
HOURS_IN_MONTH: int = Field(
default=730, # 24 horas * 365 días / 12 meses ≈ 730
description='Horas en un mes (aprox. 24/7)',
gt=0
)
# Configuración de alternativas
MAX_ALTERNATIVES_TO_SHOW: int = Field(
default=3,
description='Número máximo de alternativas a mostrar',
gt=0
)
# Configuración de precios
PRICE_TYPE: str = Field(
default='Consumption',
description='Tipo de precio (ej: Consumption)'
)
# Configuración del modelo
model_config = SettingsConfigDict(
env_file='.env',
env_file_encoding='utf-8',
env_prefix='MCP_',
case_sensitive=False,
extra='ignore',
validate_default=True,
env_nested_delimiter='__'
)
# Validadores
@field_validator('CORS_ORIGINS', mode='before')
def parse_cors_origins(cls, v):
if not v:
return ['*']
if isinstance(v, str):
return [origin.strip() for origin in v.split(',')]
if isinstance(v, (list, set)):
return list(v)
return str(v).split(',')
@field_validator('LOG_LEVEL')
def validate_log_level(cls, v):
if not v:
return DEFAULT_LOG_LEVEL
v = v.upper()
valid_levels = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'}
if v not in valid_levels:
raise ValueError(f'Nivel de log inválido: {v}. Debe ser uno de: {", ".join(valid_levels)}')
return v
@field_validator('AZURE_RETAIL_PRICES_URL')
def validate_azure_url(cls, v):
if not v:
v = 'https://prices.azure.com/api/retail/prices'
v = str(v).strip()
if not v.startswith(('http://', 'https://')):
v = f'https://{v}'
return v.rstrip('/')
@field_validator('AZURE_API_VERSION')
def validate_api_version(cls, v):
if not v:
return '2023-01-01-preview'
v = str(v).strip()
# Validar formato de versión (ej: 2023-01-01-preview)
try:
parts = v.split('-')
if not all(part.isdigit() for part in parts[0].split('.')):
raise ValueError()
except (ValueError, AttributeError):
# Si el formato no es válido, usar el valor por defecto
return '2023-01-01-preview'
return v
@field_validator('PRICE_TYPE')
def validate_price_type(cls, v):
if not v:
return 'Consumption'
return str(v).strip()
@field_validator('MCP_DEBUG', 'MCP_RELOAD', mode='before')
def validate_bool(cls, v):
if isinstance(v, str):
v = v.lower()
if v in ('true', '1', 'yes'):
return True
if v in ('false', '0', 'no', ''):
return False
return bool(v)
# Cargar configuración
try:
settings = Settings()
except Exception as e:
print(f"Error al cargar la configuración: {e}")
raise
# Configuración de logging
import logging
from logging.config import dictConfig
logging_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'default': {
'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
'datefmt': '%Y-%m-%d %H:%M:%S'
},
'simple': {
'format': '%(levelname)s: %(message)s'
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'default' if settings.MCP_DEBUG else 'simple',
'stream': 'ext://sys.stdout',
},
},
'loggers': {
'': { # root logger
'handlers': ['console'],
'level': settings.LOG_LEVEL,
'propagate': True
},
'azure': {
'level': 'WARNING', # Reduce el ruido de las librerías de Azure
'propagate': False
},
'urllib3': {
'level': 'WARNING', # Reduce el ruido de las peticiones HTTP
'propagate': False
},
}
}
# Aplicar configuración de logging
dictConfig(logging_config)
# Configurar el logger para este módulo
logger = logging.getLogger(__name__)
logger.debug("Configuración cargada correctamente")
```
--------------------------------------------------------------------------------
/azure_pricing_mcp_server.py:
--------------------------------------------------------------------------------
```python
import logging
import logging.config
import uvicorn
import requests
from typing import List, Dict, Any
# MCP imports
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.responses import JSONResponse
from starlette.endpoints import HTTPEndpoint
# Import configuration
from config import settings, logging_config
# Configure logging
logging.config.dictConfig(logging_config)
logger = logging.getLogger(__name__)
# Configure debug mode based on settings
if settings.MCP_DEBUG:
logger.setLevel(logging.DEBUG)
logger.debug("DEBUG mode activated")
def log(message: str, level: str = "info"):
"""Helper function for consistent logging."""
log_func = getattr(logger, level.lower(), logger.info)
log_func(message)
# Create MCP server
mcp = FastMCP("Azure Pricing MCP")
@mcp.tool(description="""
[STEP 1] List all available service families in Azure.
This endpoint returns the official list of service families available in Azure
according to Microsoft's official documentation.
This should be the FIRST STEP in the pricing query workflow:
1. First, get the list of service families with list_service_families
2. Then, get service names with get_service_names
3. Next, get products with get_products
4. Finally, calculate monthly costs with get_monthly_cost
Official reference: https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices#supported-servicefamily-values
Returns:
dict: Dictionary with the list of service families and the total number of items
""")
def list_service_families():
# Official list of service families according to Microsoft documentation
official_service_families = [
"Analytics",
"Azure Arc",
"Azure Communication Services",
"Azure Security",
"Azure Stack",
"Compute",
"Containers",
"Data",
"Databases",
"Developer Tools",
"Dynamics",
"Gaming",
"Integration",
"Internet of Things",
"Management and Governance",
"Microsoft Syntex",
"Mixed Reality",
"Networking",
"Other",
"Power Platform",
"Quantum Computing",
"Security",
"Storage",
"Telecommunications",
"Web",
"Windows Virtual Desktop"
]
log("Returning official list of Azure service families")
return {
"service_families": official_service_families,
"count": len(official_service_families),
"source": "official_documentation",
"reference": "https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices#supported-servicefamily-values"
}
@mcp.tool(description="""
[STEP 3] Get product names from a specific service family.
This endpoint returns a list of product names (productName) that belong to
the specified service family, without including all the complete details of each product.
This is the THIRD STEP in the pricing query workflow:
1. First, get the list of service families with list_service_families
2. Then, get service names with get_service_names
3. Now, use this tool to get specific products
4. Finally, calculate monthly costs with get_monthly_cost
Args:
service_family (str): The service family to query (e.g. 'Compute', 'Storage', 'Networking')
region (str): Azure region (default: 'westeurope')
type (str, optional): Price type (e.g. 'Consumption', 'Reservation')
service_name (str, optional): Service name to filter by (e.g. 'Virtual Machines', 'Storage Accounts')
product_name_contains (str, optional): Filter products whose name contains this text (e.g. 'Redis', 'SQL')
limit (int, optional): Maximum number of products to return (default: 0, which means no limit)
""")
def get_products(service_family, region="westeurope", type="", service_name="", product_name_contains="", limit=0):
# Ensure parameters are of the correct type
service_family = str(service_family)
region = str(region)
type = str(type)
service_name = str(service_name)
product_name_contains = str(product_name_contains)
limit = int(limit) if not isinstance(limit, int) else limit
log_message = f"Getting products for family {service_family} in region {region}"
if service_name:
log_message += f", filtered by service '{service_name}'"
if product_name_contains:
log_message += f", filtered by products containing '{product_name_contains}'"
if limit > 0:
log_message += f", limited to {limit} results"
log(log_message)
# API Configuration
AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
API_VERSION = "2023-01-01-preview"
# Build query parameters
filter_params = f"serviceFamily eq '{service_family}'"
if region:
filter_params += f" and armRegionName eq '{region}'"
if type:
filter_params += f" and type eq '{type}'"
if service_name:
filter_params += f" and serviceName eq '{service_name}'"
params = {
'api-version': API_VERSION,
'$filter': filter_params
}
# We don't set a limit to get all available results
# The API can automatically paginate if there are many results
try:
# Make the request to the Azure API
log(f"Querying API with filter: {filter_params}")
response = requests.get(AZURE_PRICE_API, params=params)
response.raise_for_status()
result = response.json()
# Check if the response is empty
items = result.get("Items", [])
if len(items) == 0:
log(f"No products found for the specified criteria: {filter_params}", "warning")
return {
"product_names": [],
"count": 0,
"total_products": 0,
"total_products_processed": 0,
"was_limited": False,
"limit_applied": limit if limit > 0 else None,
"status": "success",
"message": f"No products found for family '{service_family}' with the applied filters.",
"filter_applied": filter_params,
"product_name_filter": product_name_contains if product_name_contains else None
}
# Check if there are more pages
next_page_link = result.get("NextPageLink")
all_items = items
page_count = 1
max_pages = 3 # Reasonable page limit to avoid too many calls
# Continue getting more pages if there is a NextPageLink
while next_page_link and page_count < max_pages:
log(f"Getting additional page {page_count + 1}: {next_page_link}")
try:
next_response = requests.get(next_page_link)
next_response.raise_for_status()
next_result = next_response.json()
# Add items from the next page
next_items = next_result.get("Items", [])
all_items.extend(next_items)
# Update for the next iteration
next_page_link = next_result.get("NextPageLink")
page_count += 1
except Exception as e:
log(f"Error getting additional page: {str(e)}", "error")
break
# Filter by product name if specified
if product_name_contains:
filtered_items = [item for item in all_items if product_name_contains.lower() in item.get("productName", "").lower()]
log(f"Filtering: from {len(all_items)} products, {len(filtered_items)} contain '{product_name_contains}'")
all_items = filtered_items
# Extract product names
product_names = list(set(item.get("productName", "") for item in all_items if item.get("productName")))
product_names.sort() # Sort alphabetically
# Limit results if specified
if limit > 0 and len(product_names) > limit:
limited_product_names = product_names[:limit]
log(f"Limiting results: showing {len(limited_product_names)} of {len(product_names)} total products")
was_limited = True
else:
limited_product_names = product_names
was_limited = False
# Return product names
log(f"Retrieved {len(limited_product_names)} unique product names from {len(all_items)} total products across {page_count} pages")
return {
"product_names": limited_product_names,
"count": len(limited_product_names),
"total_products": len(product_names),
"total_products_processed": len(all_items),
"was_limited": was_limited,
"limit_applied": limit if limit > 0 else None,
"status": "success",
"filter_applied": filter_params,
"product_name_filter": product_name_contains if product_name_contains else None
}
except requests.exceptions.RequestException as e:
log(f"Error connecting to Azure API: {str(e)}", "error")
return {
"error": f"Error connecting to API: {str(e)}",
"status": "error"
}
@mcp.tool(description="""
[STEP 2] Get all unique service names within a service family.
This endpoint returns a list of all unique service names that belong
to the specified service family. For the 'Compute' family, which has many results,
an optimized strategy is used to limit overhead.
This is the SECOND STEP in the pricing query workflow:
1. First, get the list of service families with list_service_families
2. Then, use this tool to get the service names within that family
3. Next, get specific products with get_products
4. Finally, calculate monthly costs with get_monthly_cost
Args:
service_family (str): The service family to query (e.g. 'Compute', 'Storage', 'Networking')
region (str): Azure region (default: 'westeurope')
max_results (int): Maximum number of results to process (for 'Compute')
""")
def get_service_names(service_family, region="westeurope", max_results=500):
# Ensure parameters are of the correct type
service_family = str(service_family)
region = str(region)
max_results = int(max_results)
log(f"Getting unique service names for family {service_family} in region {region}")
# Special handling for 'Compute' which has many results
if service_family.lower() == "compute":
log(f"Using optimized approach for Compute family, limiting to {max_results} results")
# API Configuration
AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
API_VERSION = "2023-01-01-preview"
# Build query parameters with a specific top to optimize
filter_params = f"serviceFamily eq '{service_family}'"
if region:
filter_params += f" and armRegionName eq '{region}'"
params = {
'api-version': API_VERSION,
'$filter': filter_params,
'$top': min(int(max_results), 1000) # Maximum 1000 allowed by the API
}
try:
# Make the request to the Azure API
log(f"Querying API with optimized filter: {filter_params}")
response = requests.get(AZURE_PRICE_API, params=params)
response.raise_for_status()
result = response.json()
# Check if the response is empty
items = result.get("Items", [])
if len(items) == 0:
log(f"No products found for the specified criteria: {filter_params}", "warning")
return {
"service_family": service_family,
"service_names": [],
"count": 0,
"is_complete": True,
"status": "success",
"message": f"No products found for family '{service_family}' in region '{region}'.",
"filter_applied": filter_params
}
# Extract unique service names
service_names = list(set(item.get("serviceName", "") for item in items if item.get("serviceName")))
service_names.sort() # Sort alphabetically
return {
"service_family": service_family,
"service_names": service_names,
"count": len(service_names),
"is_complete": False, # Indicate that it might not be the complete list
"region": region,
"note": f"Optimized results for Compute family, limited to {len(items)} products"
}
except requests.exceptions.RequestException as e:
log(f"Error connecting to Azure API: {str(e)}", "error")
return {
"error": f"Error connecting to API: {str(e)}",
"status": "error"
}
# For other families, we get all products and extract service names
try:
# Use existing get_products function
products_result = get_products(service_family, region)
# Check if response is empty or has an error
if products_result.get("status") == "error" or products_result.get("count", 0) == 0:
log(f"No products found for family {service_family}", "warning")
return {
"service_family": service_family,
"service_names": [],
"count": 0,
"is_complete": True,
"status": "success",
"message": products_result.get("message", f"No products found for family '{service_family}' in region '{region}'."),
"filter_applied": products_result.get("filter_applied", "")
}
# Since get_products now returns only product names, we need to make another call
# to the API to get the service names.
log(f"Querying API to get service names for {service_family}")
# API Configuration
AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
API_VERSION = "2023-01-01-preview"
# API Configuration
filter_params = f"serviceFamily eq '{service_family}'"
if region:
filter_params += f" and armRegionName eq '{region}'"
params = {
'api-version': API_VERSION,
'$filter': filter_params,
'$top': 100 # Limit to 100 results to get a reasonable sample
}
try:
# Make the request to the Azure API
log(f"Querying API with filter: {filter_params}")
response = requests.get(AZURE_PRICE_API, params=params)
response.raise_for_status()
result = response.json()
# Extract unique service names
items = result.get("Items", [])
service_names = list(set(item.get("serviceName", "") for item in items if item.get("serviceName")))
service_names.sort() # Sort alphabetically
except Exception as e:
log(f"Error getting service names: {str(e)}", "error")
return {
"service_family": service_family,
"service_names": [],
"count": 0,
"is_complete": True,
"status": "error",
"message": f"Error getting service names: {str(e)}"
}
return {
"service_family": service_family,
"service_names": service_names,
"count": len(service_names),
"is_complete": products_result.get("NextPageLink") is None, # Indicate if it's the complete list
"region": region,
"processed_items": len(items)
}
except Exception as e:
log(f"Error getting service names: {str(e)}", "error")
return {
"error": f"Error getting service names: {str(e)}",
"status": "error"
}
@mcp.tool(description="""
[STEP 4] Calculate the monthly cost of a specific Azure product.
This endpoint queries the Azure Retail Prices API to get the hourly price
of a specific product and calculates its monthly cost based on the number of hours.
This is the FINAL STEP in the pricing query workflow:
1. First, get the list of service families with list_service_families
2. Then, get service names with get_service_names
3. Next, get specific products with get_products
4. Finally, use this tool to calculate the monthly cost of the selected product
IMPORTANT: You must use the EXACT product name as it appears in the results of get_products.
Args:
product_name (str): Exact name of the product (e.g. 'Azure App Service Premium v3 Plan')
region (str): Azure region (default: 'westeurope')
monthly_hours (int): Number of hours per month (default: 730, which is approximately one month)
type (str, optional): Price type (e.g. 'Consumption', 'Reservation')
""")
def get_monthly_cost(product_name, region="westeurope", monthly_hours=730, type="Consumption"):
# Ensure parameters are of the correct type
product_name = str(product_name)
region = str(region)
type = str(type)
monthly_hours = int(monthly_hours) if not isinstance(monthly_hours, int) else monthly_hours
# API Configuration
AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
API_VERSION = "2023-01-01-preview"
filter_params = f"productName eq '{product_name}' and armRegionName eq '{region}'"
if type:
filter_params += f" and type eq '{type}'"
params = {
'api-version': API_VERSION,
'$filter': filter_params,
}
try:
# Make the request to the Azure API
log(f"Querying API to get the price of {product_name} in {region}")
response = requests.get(AZURE_PRICE_API, params=params)
response.raise_for_status()
result = response.json()
# Extract relevant items
items = result.get("Items", [])
if len(items) == 0:
log(f"No products found for: {product_name}", "warning")
return {
"product_name": product_name,
"region": region,
"status": "error",
"message": f"Product '{product_name}' not found in region '{region}'.",
"filter_applied": filter_params
}
# Prepare the response with costs
products_costs = []
total_monthly_cost = 0
for item in items:
# Extract relevant information
sku_name = item.get("skuName", "")
meter_name = item.get("meterName", "")
retail_price = item.get("retailPrice", 0)
currency = item.get("currencyCode", "USD")
unit_of_measure = item.get("unitOfMeasure", "")
# Calculate monthly cost
monthly_cost = retail_price * monthly_hours if "Hour" in unit_of_measure else retail_price
total_monthly_cost += monthly_cost
# Add to the costs list
products_costs.append({
"sku_name": sku_name,
"meter_name": meter_name,
"retail_price": retail_price,
"unit_of_measure": unit_of_measure,
"monthly_cost": monthly_cost,
"currency": currency
})
# Sort by monthly cost in descending order
products_costs.sort(key=lambda x: x["monthly_cost"], reverse=True)
return {
"product_name": product_name,
"region": region,
"monthly_hours": monthly_hours,
"products": products_costs,
"total_monthly_cost": total_monthly_cost,
"currency": items[0].get("currencyCode", "USD") if items else "USD",
"count": len(products_costs),
"status": "success"
}
except requests.exceptions.RequestException as e:
log(f"Error connecting to Azure API: {str(e)}", "error")
return {
"product_name": product_name,
"region": region,
"status": "error",
"error": f"Error connecting to API: {str(e)}"
}
# Endpoint to list available tools in the MCP
class ToolsEndpoint(HTTPEndpoint):
"""
Endpoint to list available tools in the MCP.
This endpoint returns information about all registered tools
in the MCP, including their names, descriptions, and expected parameters.
"""
async def get(self, request):
tools = mcp.list_tools()
return JSONResponse(tools)
# Create the Starlette application with routes
app = Starlette(routes=[
Mount("/", app=mcp.sse_app()),
Route("/tools", ToolsEndpoint)
])
# Create the FastAPI application with Model Context Protocol
def get_application():
"""Create and return the Starlette application."""
return app
if __name__ == "__main__":
log(f"Starting MCP server at http://{settings.MCP_HOST}:{settings.MCP_PORT}")
log(f"SSE Endpoint: http://{settings.MCP_HOST}:{settings.MCP_PORT}/sse")
log(f"Tools Endpoint: http://{settings.MCP_HOST}:{settings.MCP_PORT}/tools")
log(f"Debug mode: {'ON' if settings.MCP_DEBUG else 'OFF'}")
log(f"Auto-reload: {'ENABLED' if settings.MCP_RELOAD else 'DISABLED'}")
# Configure uvicorn
uvicorn_config = {
"app": "azure_pricing_mcp_server:get_application",
"host": settings.MCP_HOST,
"port": settings.MCP_PORT,
"reload": settings.MCP_RELOAD,
"log_level": "debug" if settings.MCP_DEBUG else settings.LOG_LEVEL.lower()
}
logger.debug(f"Uvicorn configuration: {uvicorn_config}")
if settings.MCP_DEBUG:
logger.debug("DEBUG mode enabled for uvicorn")
uvicorn.run(**uvicorn_config)
```