# 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:
--------------------------------------------------------------------------------
```
1 | # Entornos virtuales
2 | .venv/
3 | env/
4 | venv/
5 | ENV/
6 |
7 | # Archivos de Python
8 | __pycache__/
9 | *.py[cod]
10 | *$py.class
11 | *.so
12 | .Python
13 | build/
14 | develop-eggs/
15 | dist/
16 | downloads/
17 | eggs/
18 | .eggs/
19 | lib/
20 | lib64/
21 | parts/
22 | sdist/
23 | var/
24 | wheels/
25 | *.egg-info/
26 | .installed.cfg
27 | *.egg
28 |
29 | # Archivos de entorno
30 | .env
31 | .env.local
32 | .env.*.local
33 | !.env.example
34 |
35 | # Logs
36 | *.log
37 | logs/
38 |
39 | # Archivos del sistema operativo
40 | .DS_Store
41 | Thumbs.db
42 |
43 | # IDEs y editores
44 | .idea/
45 | .vscode/
46 | *.swp
47 | *.swo
48 | *~
49 |
50 | # Archivos de caché
51 | .cache/
52 | .pytest_cache/
53 | .mypy_cache/
54 |
55 | # Archivos de cobertura
56 | htmlcov/
57 | .coverage
58 | .coverage.*
59 | coverage.xml
60 | *.cover
61 | .hypothesis/
62 | .pytest_cache/
63 |
64 | # Archivos de documentación generados
65 | docs/_build/
66 |
67 | # Archivos de Jupyter Notebook
68 | .ipynb_checkpoints
69 |
70 | # Archivos temporales
71 | tmp/
72 | temp/
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
1 | # ============================================
2 | # CONFIGURACIÓN OBLIGATORIA
3 | # ============================================
4 |
5 | # Configuración del servidor MCP (obligatorio)
6 | MCP_HOST=0.0.0.0 # Dirección IP para escuchar (ej: 0.0.0.0 para todas las interfaces)
7 | MCP_PORT=8080 # Puerto para el servidor
8 |
9 | # Configuración de Azure Retail Prices API (obligatorio)
10 | MCP_AZURE_RETAIL_PRICES_URL=https://prices.azure.com/api/retail/prices
11 | MCP_AZURE_API_VERSION=2023-01-01-preview
12 |
13 | # Configuración de cálculos (obligatorio)
14 | MCP_HOURS_IN_MONTH=730 # Horas en un mes (aprox. 24/7)
15 |
16 | # Configuración de alternativas (obligatorio)
17 | MCP_MAX_ALTERNATIVES_TO_SHOW=3 # Número máximo de alternativas a mostrar
18 |
19 | # Configuración de precios (obligatorio)
20 | MCP_PRICE_TYPE=Consumption # Tipo de precio (ej: 'Consumption')
21 |
22 | # ============================================
23 | # CONFIGURACIÓN OPCIONAL
24 | # ============================================
25 |
26 | # Configuración de CORS (opcional, separar orígenes por comas)
27 | MCP_CORS_ORIGINS=*
28 |
29 | # Nivel de logging (opcional, valores: DEBUG, INFO, WARNING, ERROR, CRITICAL)
30 | MCP_LOG_LEVEL=INFO
31 |
32 | # Modo desarrollo (opcional)
33 | MCP_DEBUG=false # Habilita el modo debug
34 | MCP_RELOAD=false # Recarga automática en desarrollo
35 |
36 | # ============================================
37 | # NOTAS
38 | # ============================================
39 | # 1. Todas las variables marcadas como (obligatorio) deben tener un valor
40 | # 2. Las cadenas no necesitan comillas
41 | # 3. Los valores booleanos deben ser 'true' o 'false' (minúsculas)
42 | # 4. Para múltiples orígenes CORS, separar por comas: http://localhost:3000,http://otro-dominio.com
43 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Azure Pricing MCP Server
2 |
3 | [](https://www.python.org/downloads/)
4 | [](https://opensource.org/licenses/MIT)
5 |
6 | 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.
7 |
8 | ## Features
9 |
10 | - Query Azure pricing data through a simple, structured workflow
11 | - Get real-time pricing information from the Azure Retail Prices API
12 | - Navigate through Azure service families, service names, and products
13 | - Calculate monthly costs for Azure resources
14 |
15 | ## System Requirements
16 |
17 | - Python 3.8 or higher
18 | - Internet connection to access the Azure Retail Prices API
19 | - Permission to install Python packages
20 | - No Azure account or credentials required (uses public pricing API)
21 |
22 | ## Installation
23 |
24 | 1. **Clone the repository:**
25 | ```bash
26 | git clone https://github.com/sboludaf/mcp-azure-pricing.git
27 | cd mcp-azure-pricing
28 | ```
29 |
30 | 2. **Create and activate a virtual environment:**
31 | ```bash
32 | python -m venv .venv
33 | ```
34 | * Windows:
35 | ```bash
36 | .venv\Scripts\activate
37 | ```
38 | * macOS/Linux:
39 | ```bash
40 | source .venv/bin/activate
41 | ```
42 |
43 | 3. **Install the dependencies:**
44 | ```bash
45 | pip install -r requirements.txt
46 | ```
47 |
48 | ## Usage
49 |
50 | The MCP server provides a structured four-step workflow for accessing Azure pricing information:
51 |
52 | 1. **Get service families** - Retrieve the list of available Azure service families
53 | 2. **Get service names** - Get service names within a specific family
54 | 3. **Get products** - Get products associated with a service
55 | 4. **Calculate monthly costs** - Calculate the monthly cost for a specific product
56 |
57 | ## Starting the MCP Server
58 |
59 | ```bash
60 | source .venv/bin/activate # Activate the virtual environment
61 | python azure_pricing_mcp_server.py
62 | ```
63 |
64 | The server will start at `http://0.0.0.0:8080` by default.
65 |
66 | ### Available Endpoints
67 |
68 | - `GET /sse`: Server-Sent Events endpoint for MCP communication
69 | - `GET /tools`: Lists the available tools in the MCP server
70 |
71 | ### MCP Client Configuration
72 |
73 | To configure an MCP client to connect to this server, add the following to your `mcp_config.json` file:
74 |
75 | ```json
76 | "azure-pricing": {
77 | "serverUrl": "http://localhost:8080/sse"
78 | }
79 | ```
80 |
81 | 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.
82 |
83 | ## MCP Tools
84 |
85 | The server provides four main tools that form a logical workflow for querying Azure pricing:
86 |
87 | ### 1. list_service_families
88 |
89 | **Description**: Lists all available service families in Azure according to Microsoft's official documentation.
90 |
91 | ### 2. get_service_names
92 |
93 | **Description**: Gets all unique service names within a specified service family.
94 |
95 | **Parameters**:
96 | - `service_family`: The service family to query (e.g., 'Compute', 'Storage')
97 | - `region`: Azure region (default: 'westeurope')
98 | - `max_results`: Maximum number of results to process
99 |
100 | ### 3. get_products
101 |
102 | **Description**: Gets product names from a specific service family.
103 |
104 | **Parameters**:
105 | - `service_family`: The service family to query
106 | - `region`: Azure region (default: 'westeurope')
107 | - `type`: Price type (optional, e.g., 'Consumption', 'Reservation')
108 | - `service_name`: Service name to filter by (optional)
109 | - `product_name_contains`: Filter products whose name contains this text (optional)
110 | - `limit`: Maximum number of products to return (optional)
111 |
112 | ### 4. get_monthly_cost
113 |
114 | **Description**: Calculates the monthly cost of a specific Azure product.
115 |
116 | **Parameters**:
117 | - `product_name`: Exact name of the product (e.g., 'Azure App Service Premium v3 Plan')
118 | - `region`: Azure region (default: 'westeurope')
119 | - `monthly_hours`: Number of hours per month (default: 730)
120 | - `type`: Price type (optional, e.g., 'Consumption')
121 |
122 | ## Error Handling
123 |
124 | The MCP server includes a robust error handling system that:
125 |
126 | - Provides descriptive error messages when resources cannot be found
127 | - Properly handles Azure API errors
128 | - Logs detailed information for debugging purposes
129 |
130 | ### Common Error Scenarios
131 |
132 | - **Product not found**: When a product name doesn't exist in the specified region
133 | - **Service family not found**: When an invalid service family is specified
134 | - **API rate limits**: When the Azure Retail Prices API rate limits are exceeded
135 | - **Network errors**: When the server cannot connect to the Azure API
136 |
137 | ## Limitations
138 |
139 | * Prices are estimates based on public information from the Azure Retail Prices API
140 | * Does not include all possible discounts, account-specific offers, or additional costs like taxes or support
141 | * The Azure Retail Prices API has rate limits that can affect performance with a high volume of requests
142 | * Prices may vary depending on the region and currency selected
143 | * Not all Azure resources are available in all regions
144 |
145 | ## Contributing
146 |
147 | Contributions are welcome! Here's how you can contribute to this project:
148 |
149 | 1. Fork the repository
150 | 2. Create a feature branch: `git checkout -b feature/amazing-feature`
151 | 3. Commit your changes: `git commit -m 'Add some amazing feature'`
152 | 4. Push to the branch: `git push origin feature/amazing-feature`
153 | 5. Open a Pull Request
154 |
155 | ## License
156 |
157 | This project is licensed under the MIT License - see the LICENSE file for details.
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | # Core dependencies
2 | requests>=2.28.0
3 | fastapi>=0.89.0
4 | uvicorn>=0.20.0
5 | starlette>=0.25.0
6 | pydantic>=1.10.0
7 |
8 | # MCP server
9 | mcp-server>=0.1.0
10 |
11 | # Utilities
12 | tabulate>=0.9.0
13 | python-dotenv>=0.21.0
14 |
15 | # Azure-specific (if needed)
16 | # azure-core>=1.26.0
17 | # azure-identity>=1.12.0
18 |
```
--------------------------------------------------------------------------------
/tests/test_api.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | import requests
3 | import json
4 | import sys
5 |
6 | def test_api_call(service_family, region="westeurope", type="", service_name=""):
7 | """
8 | Prueba directa de la API de Azure Retail Prices con los parámetros especificados.
9 | """
10 | print(f"Consultando API para familia: {service_family}, región: {region}, tipo: {type}, servicio: {service_name}")
11 |
12 | # Configuración de la API
13 | AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
14 | API_VERSION = "2023-01-01-preview"
15 |
16 | # Construir los parámetros de la consulta
17 | filter_params = f"serviceFamily eq '{service_family}'"
18 | if region:
19 | filter_params += f" and armRegionName eq '{region}'"
20 | if type:
21 | filter_params += f" and type eq '{type}'"
22 | if service_name:
23 | filter_params += f" and serviceName eq '{service_name}'"
24 |
25 | params = {
26 | 'api-version': API_VERSION,
27 | '$filter': filter_params,
28 | '$top': 10 # Limitamos a 10 resultados para la prueba
29 | }
30 |
31 | print(f"Filtro: {filter_params}")
32 |
33 | try:
34 | # Realizar la petición a la API
35 | response = requests.get(AZURE_PRICE_API, params=params)
36 | response.raise_for_status()
37 | result = response.json()
38 |
39 | # Mostrar información sobre los resultados
40 | items = result.get("Items", [])
41 | print(f"Se encontraron {len(items)} productos")
42 |
43 | # Mostrar los primeros 3 resultados
44 | for i, item in enumerate(items[:3]):
45 | print(f"\nProducto {i+1}:")
46 | print(f" Nombre: {item.get('productName', 'N/A')}")
47 | print(f" Servicio: {item.get('serviceName', 'N/A')}")
48 | print(f" SKU: {item.get('skuName', 'N/A')}")
49 | print(f" Región: {item.get('armRegionName', 'N/A')}")
50 | print(f" Precio: {item.get('retailPrice', 'N/A')} {item.get('currencyCode', '')}")
51 |
52 | return result
53 |
54 | except requests.exceptions.RequestException as e:
55 | print(f"Error al conectar con la API: {str(e)}")
56 | return None
57 |
58 | if __name__ == "__main__":
59 | # Procesar argumentos de línea de comandos
60 | service_family = sys.argv[1] if len(sys.argv) > 1 else "Networking"
61 | region = sys.argv[2] if len(sys.argv) > 2 else "westeurope"
62 | service_name = sys.argv[3] if len(sys.argv) > 3 else "Virtual Network"
63 | type_param = sys.argv[4] if len(sys.argv) > 4 else ""
64 |
65 | # Llamar a la función de prueba
66 | test_api_call(service_family, region, type_param, service_name)
67 |
```
--------------------------------------------------------------------------------
/tests/test_monthly_cost.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | import requests
3 | import json
4 |
5 | def calculate_monthly_cost(product_name, region="westeurope", monthly_hours=730, type="Consumption"):
6 | """
7 | Calcula el coste mensual de un producto específico de Azure.
8 |
9 | Args:
10 | product_name: Nombre exacto del producto
11 | region: Región de Azure
12 | monthly_hours: Número de horas al mes (por defecto 730)
13 | type: Tipo de precio
14 | """
15 | # Configuración de la API
16 | AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
17 | API_VERSION = "2023-01-01-preview"
18 |
19 | # Construir la consulta
20 | filter_params = f"productName eq '{product_name}' and armRegionName eq '{region}'"
21 | if type:
22 | filter_params += f" and type eq '{type}'"
23 |
24 | params = {
25 | 'api-version': API_VERSION,
26 | '$filter': filter_params
27 | }
28 |
29 | print(f"Consultando API para: {product_name} en {region}")
30 | print(f"Filtro: {filter_params}")
31 |
32 | try:
33 | # Realizar la petición a la API
34 | response = requests.get(AZURE_PRICE_API, params=params)
35 | response.raise_for_status()
36 | result = response.json()
37 |
38 | # Extraer los items relevantes
39 | items = result.get("Items", [])
40 |
41 | if len(items) == 0:
42 | print(f"No se encontraron productos para: {product_name}")
43 | return
44 |
45 | print(f"Se encontraron {len(items)} variantes del producto.")
46 | total_monthly_cost = 0
47 |
48 | # Mostrar información para cada variante
49 | for i, item in enumerate(items):
50 | sku_name = item.get("skuName", "")
51 | meter_name = item.get("meterName", "")
52 | retail_price = item.get("retailPrice", 0)
53 | currency = item.get("currencyCode", "USD")
54 | unit_of_measure = item.get("unitOfMeasure", "")
55 |
56 | # Calcular coste mensual
57 | monthly_cost = retail_price * monthly_hours if "Hour" in unit_of_measure else retail_price
58 | total_monthly_cost += monthly_cost
59 |
60 | print(f"\nVariante {i+1}:")
61 | print(f" SKU: {sku_name}")
62 | print(f" Meter: {meter_name}")
63 | print(f" Precio por unidad: {retail_price} {currency} por {unit_of_measure}")
64 | print(f" Coste mensual estimado: {monthly_cost:.2f} {currency}")
65 |
66 | print(f"\nCoste mensual total estimado: {total_monthly_cost:.2f} {currency}")
67 |
68 | except requests.exceptions.RequestException as e:
69 | print(f"Error al conectar con la API: {str(e)}")
70 |
71 |
72 | if __name__ == "__main__":
73 | import sys
74 |
75 | # Valores por defecto
76 | product_name = "Azure App Service Premium v3 Plan"
77 | region = "westeurope"
78 | monthly_hours = 730
79 | type = "Consumption"
80 |
81 | # Procesar argumentos de línea de comandos si se proporcionan
82 | if len(sys.argv) > 1:
83 | product_name = sys.argv[1]
84 | if len(sys.argv) > 2:
85 | region = sys.argv[2]
86 | if len(sys.argv) > 3:
87 | monthly_hours = int(sys.argv[3])
88 | if len(sys.argv) > 4:
89 | type = sys.argv[4]
90 |
91 | # Calcular el coste mensual
92 | calculate_monthly_cost(product_name, region, monthly_hours, type)
93 |
```
--------------------------------------------------------------------------------
/config.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Configuración del servidor MCP de Azure Pricing.
3 |
4 | Este archivo puede ser sobreescrito mediante variables de entorno con prefijo MCP_
5 | """
6 | from typing import List, Optional, Union
7 | from pydantic import Field, field_validator, HttpUrl, validator
8 | from pydantic_settings import BaseSettings, SettingsConfigDict
9 | import os
10 |
11 | # Valores por defecto
12 | DEFAULT_HOST = '0.0.0.0'
13 | DEFAULT_PORT = 8080
14 | DEFAULT_LOG_LEVEL = 'INFO'
15 |
16 | class Settings(BaseSettings):
17 | # Configuración del servidor
18 | MCP_HOST: str = Field(
19 | default=DEFAULT_HOST,
20 | description='Dirección IP para escuchar (ej: 0.0.0.0 para todas las interfaces)'
21 | )
22 |
23 | MCP_PORT: int = Field(
24 | default=DEFAULT_PORT,
25 | description='Puerto para el servidor',
26 | gt=0,
27 | lt=65536
28 | )
29 |
30 | MCP_DEBUG: bool = Field(
31 | default=True,
32 | description='Habilita el modo debug (no usar en producción)'
33 | )
34 |
35 | MCP_RELOAD: bool = Field(
36 | default=False,
37 | description='Habilita la recarga automática en desarrollo'
38 | )
39 |
40 | # Configuración de CORS
41 | CORS_ORIGINS: Union[str, List[str]] = Field(
42 | default='*',
43 | description='Orígenes permitidos para CORS (separados por comas o lista)'
44 | )
45 |
46 | # Configuración de logging
47 | LOG_LEVEL: str = Field(
48 | default=DEFAULT_LOG_LEVEL,
49 | description='Nivel de logging (DEBUG, INFO, WARNING, ERROR, CRITICAL)'
50 | )
51 |
52 | # Configuración de Azure Retail Prices API
53 | AZURE_RETAIL_PRICES_URL: str = Field(
54 | default='https://prices.azure.com/api/retail/prices',
55 | description='URL base de la API de precios de Azure'
56 | )
57 |
58 | AZURE_API_VERSION: str = Field(
59 | default='2023-01-01-preview',
60 | description='Versión de la API de precios de Azure'
61 | )
62 |
63 | # Configuración de cálculos
64 | HOURS_IN_MONTH: int = Field(
65 | default=730, # 24 horas * 365 días / 12 meses ≈ 730
66 | description='Horas en un mes (aprox. 24/7)',
67 | gt=0
68 | )
69 |
70 | # Configuración de alternativas
71 | MAX_ALTERNATIVES_TO_SHOW: int = Field(
72 | default=3,
73 | description='Número máximo de alternativas a mostrar',
74 | gt=0
75 | )
76 |
77 | # Configuración de precios
78 | PRICE_TYPE: str = Field(
79 | default='Consumption',
80 | description='Tipo de precio (ej: Consumption)'
81 | )
82 |
83 | # Configuración del modelo
84 | model_config = SettingsConfigDict(
85 | env_file='.env',
86 | env_file_encoding='utf-8',
87 | env_prefix='MCP_',
88 | case_sensitive=False,
89 | extra='ignore',
90 | validate_default=True,
91 | env_nested_delimiter='__'
92 | )
93 |
94 | # Validadores
95 | @field_validator('CORS_ORIGINS', mode='before')
96 | def parse_cors_origins(cls, v):
97 | if not v:
98 | return ['*']
99 | if isinstance(v, str):
100 | return [origin.strip() for origin in v.split(',')]
101 | if isinstance(v, (list, set)):
102 | return list(v)
103 | return str(v).split(',')
104 |
105 | @field_validator('LOG_LEVEL')
106 | def validate_log_level(cls, v):
107 | if not v:
108 | return DEFAULT_LOG_LEVEL
109 | v = v.upper()
110 | valid_levels = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'}
111 | if v not in valid_levels:
112 | raise ValueError(f'Nivel de log inválido: {v}. Debe ser uno de: {", ".join(valid_levels)}')
113 | return v
114 |
115 | @field_validator('AZURE_RETAIL_PRICES_URL')
116 | def validate_azure_url(cls, v):
117 | if not v:
118 | v = 'https://prices.azure.com/api/retail/prices'
119 | v = str(v).strip()
120 | if not v.startswith(('http://', 'https://')):
121 | v = f'https://{v}'
122 | return v.rstrip('/')
123 |
124 | @field_validator('AZURE_API_VERSION')
125 | def validate_api_version(cls, v):
126 | if not v:
127 | return '2023-01-01-preview'
128 | v = str(v).strip()
129 | # Validar formato de versión (ej: 2023-01-01-preview)
130 | try:
131 | parts = v.split('-')
132 | if not all(part.isdigit() for part in parts[0].split('.')):
133 | raise ValueError()
134 | except (ValueError, AttributeError):
135 | # Si el formato no es válido, usar el valor por defecto
136 | return '2023-01-01-preview'
137 | return v
138 |
139 | @field_validator('PRICE_TYPE')
140 | def validate_price_type(cls, v):
141 | if not v:
142 | return 'Consumption'
143 | return str(v).strip()
144 |
145 | @field_validator('MCP_DEBUG', 'MCP_RELOAD', mode='before')
146 | def validate_bool(cls, v):
147 | if isinstance(v, str):
148 | v = v.lower()
149 | if v in ('true', '1', 'yes'):
150 | return True
151 | if v in ('false', '0', 'no', ''):
152 | return False
153 | return bool(v)
154 |
155 | # Cargar configuración
156 | try:
157 | settings = Settings()
158 | except Exception as e:
159 | print(f"Error al cargar la configuración: {e}")
160 | raise
161 |
162 | # Configuración de logging
163 | import logging
164 | from logging.config import dictConfig
165 |
166 | logging_config = {
167 | 'version': 1,
168 | 'disable_existing_loggers': False,
169 | 'formatters': {
170 | 'default': {
171 | 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
172 | 'datefmt': '%Y-%m-%d %H:%M:%S'
173 | },
174 | 'simple': {
175 | 'format': '%(levelname)s: %(message)s'
176 | },
177 | },
178 | 'handlers': {
179 | 'console': {
180 | 'class': 'logging.StreamHandler',
181 | 'formatter': 'default' if settings.MCP_DEBUG else 'simple',
182 | 'stream': 'ext://sys.stdout',
183 | },
184 | },
185 | 'loggers': {
186 | '': { # root logger
187 | 'handlers': ['console'],
188 | 'level': settings.LOG_LEVEL,
189 | 'propagate': True
190 | },
191 | 'azure': {
192 | 'level': 'WARNING', # Reduce el ruido de las librerías de Azure
193 | 'propagate': False
194 | },
195 | 'urllib3': {
196 | 'level': 'WARNING', # Reduce el ruido de las peticiones HTTP
197 | 'propagate': False
198 | },
199 | }
200 | }
201 |
202 | # Aplicar configuración de logging
203 | dictConfig(logging_config)
204 |
205 | # Configurar el logger para este módulo
206 | logger = logging.getLogger(__name__)
207 | logger.debug("Configuración cargada correctamente")
208 |
```
--------------------------------------------------------------------------------
/azure_pricing_mcp_server.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | import logging.config
3 | import uvicorn
4 | import requests
5 | from typing import List, Dict, Any
6 |
7 | # MCP imports
8 | from mcp.server.fastmcp import FastMCP
9 | from starlette.applications import Starlette
10 | from starlette.routing import Mount, Route
11 | from starlette.responses import JSONResponse
12 | from starlette.endpoints import HTTPEndpoint
13 |
14 | # Import configuration
15 | from config import settings, logging_config
16 |
17 | # Configure logging
18 | logging.config.dictConfig(logging_config)
19 | logger = logging.getLogger(__name__)
20 |
21 | # Configure debug mode based on settings
22 | if settings.MCP_DEBUG:
23 | logger.setLevel(logging.DEBUG)
24 | logger.debug("DEBUG mode activated")
25 |
26 | def log(message: str, level: str = "info"):
27 | """Helper function for consistent logging."""
28 | log_func = getattr(logger, level.lower(), logger.info)
29 | log_func(message)
30 |
31 | # Create MCP server
32 | mcp = FastMCP("Azure Pricing MCP")
33 |
34 | @mcp.tool(description="""
35 | [STEP 1] List all available service families in Azure.
36 |
37 | This endpoint returns the official list of service families available in Azure
38 | according to Microsoft's official documentation.
39 |
40 | This should be the FIRST STEP in the pricing query workflow:
41 | 1. First, get the list of service families with list_service_families
42 | 2. Then, get service names with get_service_names
43 | 3. Next, get products with get_products
44 | 4. Finally, calculate monthly costs with get_monthly_cost
45 |
46 | Official reference: https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices#supported-servicefamily-values
47 |
48 | Returns:
49 | dict: Dictionary with the list of service families and the total number of items
50 | """)
51 | def list_service_families():
52 | # Official list of service families according to Microsoft documentation
53 | official_service_families = [
54 | "Analytics",
55 | "Azure Arc",
56 | "Azure Communication Services",
57 | "Azure Security",
58 | "Azure Stack",
59 | "Compute",
60 | "Containers",
61 | "Data",
62 | "Databases",
63 | "Developer Tools",
64 | "Dynamics",
65 | "Gaming",
66 | "Integration",
67 | "Internet of Things",
68 | "Management and Governance",
69 | "Microsoft Syntex",
70 | "Mixed Reality",
71 | "Networking",
72 | "Other",
73 | "Power Platform",
74 | "Quantum Computing",
75 | "Security",
76 | "Storage",
77 | "Telecommunications",
78 | "Web",
79 | "Windows Virtual Desktop"
80 | ]
81 |
82 | log("Returning official list of Azure service families")
83 | return {
84 | "service_families": official_service_families,
85 | "count": len(official_service_families),
86 | "source": "official_documentation",
87 | "reference": "https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices#supported-servicefamily-values"
88 | }
89 |
90 | @mcp.tool(description="""
91 | [STEP 3] Get product names from a specific service family.
92 |
93 | This endpoint returns a list of product names (productName) that belong to
94 | the specified service family, without including all the complete details of each product.
95 |
96 | This is the THIRD STEP in the pricing query workflow:
97 | 1. First, get the list of service families with list_service_families
98 | 2. Then, get service names with get_service_names
99 | 3. Now, use this tool to get specific products
100 | 4. Finally, calculate monthly costs with get_monthly_cost
101 |
102 | Args:
103 | service_family (str): The service family to query (e.g. 'Compute', 'Storage', 'Networking')
104 | region (str): Azure region (default: 'westeurope')
105 | type (str, optional): Price type (e.g. 'Consumption', 'Reservation')
106 | service_name (str, optional): Service name to filter by (e.g. 'Virtual Machines', 'Storage Accounts')
107 | product_name_contains (str, optional): Filter products whose name contains this text (e.g. 'Redis', 'SQL')
108 | limit (int, optional): Maximum number of products to return (default: 0, which means no limit)
109 | """)
110 | def get_products(service_family, region="westeurope", type="", service_name="", product_name_contains="", limit=0):
111 | # Ensure parameters are of the correct type
112 | service_family = str(service_family)
113 | region = str(region)
114 | type = str(type)
115 | service_name = str(service_name)
116 | product_name_contains = str(product_name_contains)
117 | limit = int(limit) if not isinstance(limit, int) else limit
118 |
119 | log_message = f"Getting products for family {service_family} in region {region}"
120 | if service_name:
121 | log_message += f", filtered by service '{service_name}'"
122 | if product_name_contains:
123 | log_message += f", filtered by products containing '{product_name_contains}'"
124 | if limit > 0:
125 | log_message += f", limited to {limit} results"
126 | log(log_message)
127 |
128 | # API Configuration
129 | AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
130 | API_VERSION = "2023-01-01-preview"
131 |
132 | # Build query parameters
133 | filter_params = f"serviceFamily eq '{service_family}'"
134 | if region:
135 | filter_params += f" and armRegionName eq '{region}'"
136 | if type:
137 | filter_params += f" and type eq '{type}'"
138 | if service_name:
139 | filter_params += f" and serviceName eq '{service_name}'"
140 |
141 | params = {
142 | 'api-version': API_VERSION,
143 | '$filter': filter_params
144 | }
145 |
146 | # We don't set a limit to get all available results
147 | # The API can automatically paginate if there are many results
148 |
149 | try:
150 | # Make the request to the Azure API
151 | log(f"Querying API with filter: {filter_params}")
152 | response = requests.get(AZURE_PRICE_API, params=params)
153 | response.raise_for_status()
154 | result = response.json()
155 |
156 | # Check if the response is empty
157 | items = result.get("Items", [])
158 | if len(items) == 0:
159 | log(f"No products found for the specified criteria: {filter_params}", "warning")
160 | return {
161 | "product_names": [],
162 | "count": 0,
163 | "total_products": 0,
164 | "total_products_processed": 0,
165 | "was_limited": False,
166 | "limit_applied": limit if limit > 0 else None,
167 | "status": "success",
168 | "message": f"No products found for family '{service_family}' with the applied filters.",
169 | "filter_applied": filter_params,
170 | "product_name_filter": product_name_contains if product_name_contains else None
171 | }
172 |
173 | # Check if there are more pages
174 | next_page_link = result.get("NextPageLink")
175 | all_items = items
176 | page_count = 1
177 | max_pages = 3 # Reasonable page limit to avoid too many calls
178 |
179 | # Continue getting more pages if there is a NextPageLink
180 | while next_page_link and page_count < max_pages:
181 | log(f"Getting additional page {page_count + 1}: {next_page_link}")
182 | try:
183 | next_response = requests.get(next_page_link)
184 | next_response.raise_for_status()
185 | next_result = next_response.json()
186 |
187 | # Add items from the next page
188 | next_items = next_result.get("Items", [])
189 | all_items.extend(next_items)
190 |
191 | # Update for the next iteration
192 | next_page_link = next_result.get("NextPageLink")
193 | page_count += 1
194 | except Exception as e:
195 | log(f"Error getting additional page: {str(e)}", "error")
196 | break
197 |
198 | # Filter by product name if specified
199 | if product_name_contains:
200 | filtered_items = [item for item in all_items if product_name_contains.lower() in item.get("productName", "").lower()]
201 | log(f"Filtering: from {len(all_items)} products, {len(filtered_items)} contain '{product_name_contains}'")
202 | all_items = filtered_items
203 |
204 | # Extract product names
205 | product_names = list(set(item.get("productName", "") for item in all_items if item.get("productName")))
206 | product_names.sort() # Sort alphabetically
207 |
208 | # Limit results if specified
209 | if limit > 0 and len(product_names) > limit:
210 | limited_product_names = product_names[:limit]
211 | log(f"Limiting results: showing {len(limited_product_names)} of {len(product_names)} total products")
212 | was_limited = True
213 | else:
214 | limited_product_names = product_names
215 | was_limited = False
216 |
217 | # Return product names
218 | log(f"Retrieved {len(limited_product_names)} unique product names from {len(all_items)} total products across {page_count} pages")
219 | return {
220 | "product_names": limited_product_names,
221 | "count": len(limited_product_names),
222 | "total_products": len(product_names),
223 | "total_products_processed": len(all_items),
224 | "was_limited": was_limited,
225 | "limit_applied": limit if limit > 0 else None,
226 | "status": "success",
227 | "filter_applied": filter_params,
228 | "product_name_filter": product_name_contains if product_name_contains else None
229 | }
230 |
231 | except requests.exceptions.RequestException as e:
232 | log(f"Error connecting to Azure API: {str(e)}", "error")
233 | return {
234 | "error": f"Error connecting to API: {str(e)}",
235 | "status": "error"
236 | }
237 |
238 | @mcp.tool(description="""
239 | [STEP 2] Get all unique service names within a service family.
240 |
241 | This endpoint returns a list of all unique service names that belong
242 | to the specified service family. For the 'Compute' family, which has many results,
243 | an optimized strategy is used to limit overhead.
244 |
245 | This is the SECOND STEP in the pricing query workflow:
246 | 1. First, get the list of service families with list_service_families
247 | 2. Then, use this tool to get the service names within that family
248 | 3. Next, get specific products with get_products
249 | 4. Finally, calculate monthly costs with get_monthly_cost
250 |
251 | Args:
252 | service_family (str): The service family to query (e.g. 'Compute', 'Storage', 'Networking')
253 | region (str): Azure region (default: 'westeurope')
254 | max_results (int): Maximum number of results to process (for 'Compute')
255 | """)
256 | def get_service_names(service_family, region="westeurope", max_results=500):
257 | # Ensure parameters are of the correct type
258 | service_family = str(service_family)
259 | region = str(region)
260 | max_results = int(max_results)
261 |
262 | log(f"Getting unique service names for family {service_family} in region {region}")
263 |
264 | # Special handling for 'Compute' which has many results
265 | if service_family.lower() == "compute":
266 | log(f"Using optimized approach for Compute family, limiting to {max_results} results")
267 |
268 | # API Configuration
269 | AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
270 | API_VERSION = "2023-01-01-preview"
271 |
272 | # Build query parameters with a specific top to optimize
273 | filter_params = f"serviceFamily eq '{service_family}'"
274 | if region:
275 | filter_params += f" and armRegionName eq '{region}'"
276 |
277 | params = {
278 | 'api-version': API_VERSION,
279 | '$filter': filter_params,
280 | '$top': min(int(max_results), 1000) # Maximum 1000 allowed by the API
281 | }
282 |
283 | try:
284 | # Make the request to the Azure API
285 | log(f"Querying API with optimized filter: {filter_params}")
286 | response = requests.get(AZURE_PRICE_API, params=params)
287 | response.raise_for_status()
288 | result = response.json()
289 |
290 | # Check if the response is empty
291 | items = result.get("Items", [])
292 | if len(items) == 0:
293 | log(f"No products found for the specified criteria: {filter_params}", "warning")
294 | return {
295 | "service_family": service_family,
296 | "service_names": [],
297 | "count": 0,
298 | "is_complete": True,
299 | "status": "success",
300 | "message": f"No products found for family '{service_family}' in region '{region}'.",
301 | "filter_applied": filter_params
302 | }
303 |
304 | # Extract unique service names
305 | service_names = list(set(item.get("serviceName", "") for item in items if item.get("serviceName")))
306 | service_names.sort() # Sort alphabetically
307 |
308 | return {
309 | "service_family": service_family,
310 | "service_names": service_names,
311 | "count": len(service_names),
312 | "is_complete": False, # Indicate that it might not be the complete list
313 | "region": region,
314 | "note": f"Optimized results for Compute family, limited to {len(items)} products"
315 | }
316 |
317 | except requests.exceptions.RequestException as e:
318 | log(f"Error connecting to Azure API: {str(e)}", "error")
319 | return {
320 | "error": f"Error connecting to API: {str(e)}",
321 | "status": "error"
322 | }
323 |
324 | # For other families, we get all products and extract service names
325 | try:
326 | # Use existing get_products function
327 | products_result = get_products(service_family, region)
328 |
329 | # Check if response is empty or has an error
330 | if products_result.get("status") == "error" or products_result.get("count", 0) == 0:
331 | log(f"No products found for family {service_family}", "warning")
332 | return {
333 | "service_family": service_family,
334 | "service_names": [],
335 | "count": 0,
336 | "is_complete": True,
337 | "status": "success",
338 | "message": products_result.get("message", f"No products found for family '{service_family}' in region '{region}'."),
339 | "filter_applied": products_result.get("filter_applied", "")
340 | }
341 |
342 | # Since get_products now returns only product names, we need to make another call
343 | # to the API to get the service names.
344 | log(f"Querying API to get service names for {service_family}")
345 |
346 | # API Configuration
347 | AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
348 | API_VERSION = "2023-01-01-preview"
349 |
350 | # API Configuration
351 | filter_params = f"serviceFamily eq '{service_family}'"
352 | if region:
353 | filter_params += f" and armRegionName eq '{region}'"
354 |
355 | params = {
356 | 'api-version': API_VERSION,
357 | '$filter': filter_params,
358 | '$top': 100 # Limit to 100 results to get a reasonable sample
359 | }
360 |
361 | try:
362 | # Make the request to the Azure API
363 | log(f"Querying API with filter: {filter_params}")
364 | response = requests.get(AZURE_PRICE_API, params=params)
365 | response.raise_for_status()
366 | result = response.json()
367 |
368 | # Extract unique service names
369 | items = result.get("Items", [])
370 | service_names = list(set(item.get("serviceName", "") for item in items if item.get("serviceName")))
371 | service_names.sort() # Sort alphabetically
372 | except Exception as e:
373 | log(f"Error getting service names: {str(e)}", "error")
374 | return {
375 | "service_family": service_family,
376 | "service_names": [],
377 | "count": 0,
378 | "is_complete": True,
379 | "status": "error",
380 | "message": f"Error getting service names: {str(e)}"
381 | }
382 |
383 | return {
384 | "service_family": service_family,
385 | "service_names": service_names,
386 | "count": len(service_names),
387 | "is_complete": products_result.get("NextPageLink") is None, # Indicate if it's the complete list
388 | "region": region,
389 | "processed_items": len(items)
390 | }
391 |
392 | except Exception as e:
393 | log(f"Error getting service names: {str(e)}", "error")
394 | return {
395 | "error": f"Error getting service names: {str(e)}",
396 | "status": "error"
397 | }
398 |
399 | @mcp.tool(description="""
400 | [STEP 4] Calculate the monthly cost of a specific Azure product.
401 |
402 | This endpoint queries the Azure Retail Prices API to get the hourly price
403 | of a specific product and calculates its monthly cost based on the number of hours.
404 |
405 | This is the FINAL STEP in the pricing query workflow:
406 | 1. First, get the list of service families with list_service_families
407 | 2. Then, get service names with get_service_names
408 | 3. Next, get specific products with get_products
409 | 4. Finally, use this tool to calculate the monthly cost of the selected product
410 |
411 | IMPORTANT: You must use the EXACT product name as it appears in the results of get_products.
412 |
413 | Args:
414 | product_name (str): Exact name of the product (e.g. 'Azure App Service Premium v3 Plan')
415 | region (str): Azure region (default: 'westeurope')
416 | monthly_hours (int): Number of hours per month (default: 730, which is approximately one month)
417 | type (str, optional): Price type (e.g. 'Consumption', 'Reservation')
418 | """)
419 | def get_monthly_cost(product_name, region="westeurope", monthly_hours=730, type="Consumption"):
420 | # Ensure parameters are of the correct type
421 | product_name = str(product_name)
422 | region = str(region)
423 | type = str(type)
424 | monthly_hours = int(monthly_hours) if not isinstance(monthly_hours, int) else monthly_hours
425 |
426 | # API Configuration
427 | AZURE_PRICE_API = "https://prices.azure.com/api/retail/prices"
428 | API_VERSION = "2023-01-01-preview"
429 |
430 | filter_params = f"productName eq '{product_name}' and armRegionName eq '{region}'"
431 | if type:
432 | filter_params += f" and type eq '{type}'"
433 |
434 | params = {
435 | 'api-version': API_VERSION,
436 | '$filter': filter_params,
437 | }
438 |
439 | try:
440 | # Make the request to the Azure API
441 | log(f"Querying API to get the price of {product_name} in {region}")
442 | response = requests.get(AZURE_PRICE_API, params=params)
443 | response.raise_for_status()
444 | result = response.json()
445 |
446 | # Extract relevant items
447 | items = result.get("Items", [])
448 |
449 | if len(items) == 0:
450 | log(f"No products found for: {product_name}", "warning")
451 | return {
452 | "product_name": product_name,
453 | "region": region,
454 | "status": "error",
455 | "message": f"Product '{product_name}' not found in region '{region}'.",
456 | "filter_applied": filter_params
457 | }
458 |
459 | # Prepare the response with costs
460 | products_costs = []
461 | total_monthly_cost = 0
462 |
463 | for item in items:
464 | # Extract relevant information
465 | sku_name = item.get("skuName", "")
466 | meter_name = item.get("meterName", "")
467 | retail_price = item.get("retailPrice", 0)
468 | currency = item.get("currencyCode", "USD")
469 | unit_of_measure = item.get("unitOfMeasure", "")
470 |
471 | # Calculate monthly cost
472 | monthly_cost = retail_price * monthly_hours if "Hour" in unit_of_measure else retail_price
473 | total_monthly_cost += monthly_cost
474 |
475 | # Add to the costs list
476 | products_costs.append({
477 | "sku_name": sku_name,
478 | "meter_name": meter_name,
479 | "retail_price": retail_price,
480 | "unit_of_measure": unit_of_measure,
481 | "monthly_cost": monthly_cost,
482 | "currency": currency
483 | })
484 |
485 | # Sort by monthly cost in descending order
486 | products_costs.sort(key=lambda x: x["monthly_cost"], reverse=True)
487 |
488 | return {
489 | "product_name": product_name,
490 | "region": region,
491 | "monthly_hours": monthly_hours,
492 | "products": products_costs,
493 | "total_monthly_cost": total_monthly_cost,
494 | "currency": items[0].get("currencyCode", "USD") if items else "USD",
495 | "count": len(products_costs),
496 | "status": "success"
497 | }
498 |
499 | except requests.exceptions.RequestException as e:
500 | log(f"Error connecting to Azure API: {str(e)}", "error")
501 | return {
502 | "product_name": product_name,
503 | "region": region,
504 | "status": "error",
505 | "error": f"Error connecting to API: {str(e)}"
506 | }
507 |
508 |
509 | # Endpoint to list available tools in the MCP
510 | class ToolsEndpoint(HTTPEndpoint):
511 | """
512 | Endpoint to list available tools in the MCP.
513 |
514 | This endpoint returns information about all registered tools
515 | in the MCP, including their names, descriptions, and expected parameters.
516 | """
517 | async def get(self, request):
518 | tools = mcp.list_tools()
519 | return JSONResponse(tools)
520 |
521 | # Create the Starlette application with routes
522 | app = Starlette(routes=[
523 | Mount("/", app=mcp.sse_app()),
524 | Route("/tools", ToolsEndpoint)
525 | ])
526 |
527 | # Create the FastAPI application with Model Context Protocol
528 | def get_application():
529 | """Create and return the Starlette application."""
530 | return app
531 |
532 | if __name__ == "__main__":
533 | log(f"Starting MCP server at http://{settings.MCP_HOST}:{settings.MCP_PORT}")
534 | log(f"SSE Endpoint: http://{settings.MCP_HOST}:{settings.MCP_PORT}/sse")
535 | log(f"Tools Endpoint: http://{settings.MCP_HOST}:{settings.MCP_PORT}/tools")
536 | log(f"Debug mode: {'ON' if settings.MCP_DEBUG else 'OFF'}")
537 | log(f"Auto-reload: {'ENABLED' if settings.MCP_RELOAD else 'DISABLED'}")
538 |
539 | # Configure uvicorn
540 | uvicorn_config = {
541 | "app": "azure_pricing_mcp_server:get_application",
542 | "host": settings.MCP_HOST,
543 | "port": settings.MCP_PORT,
544 | "reload": settings.MCP_RELOAD,
545 | "log_level": "debug" if settings.MCP_DEBUG else settings.LOG_LEVEL.lower()
546 | }
547 |
548 | logger.debug(f"Uvicorn configuration: {uvicorn_config}")
549 |
550 | if settings.MCP_DEBUG:
551 | logger.debug("DEBUG mode enabled for uvicorn")
552 |
553 | uvicorn.run(**uvicorn_config)
554 |
```