Megh Iq

MeghIQ MCP服务器——Azure成本优化工具


A. 模型上下文协议（MCP） 暴露的服务器 22工具 用于Azure成本管理、预算控制、异常检测、利用率分析、Azure更新智能和PDF报告。

建于 FastMCP 具有流式HTTP传输，完全由Azure REST API支持。

______________________________________________________________________

目录

• 建筑
• 项目结构
• 工具参考

- 成本查询工具 - 预测工具 - 预算工具 - 警报工具 - 推荐工具 - 异常检测工具 - 使用工具 - Azure更新智能工具

• 先决条件
• 设置
• 运行服务器
• 环境变量
• API终点
• 响应格式
• 使用的Azure API
• 依赖项
• 测试
• 模块参考

______________________________________________________________________

建筑

┌─────────────────────────────────────────────────────────────────┐
│                        MCP Client                               │
│             (Claude, Copilot, custom agent, etc.)               │
└───────────────────────────┬─────────────────────────────────────┘
                            │  Streamable HTTP / stdio
┌───────────────────────────▼─────────────────────────────────────┐
│                    server.py (FastMCP)                           │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  22 @mcp.tool() wrappers + Health checks (/ , /health)   │   │
│  └──────────────────────────────────────────────────────────┘   │
│          │              │               │              │         │
│  ┌───────▼──────┐ ┌────▼─────┐ ┌───────▼──────┐ ┌────▼─────┐  │
│  │ auth.py      │ │response.py│ │ tools/*.py   │ │ starlette│  │
│  │ Credentials  │ │ JSON fmt  │ │ 11 modules   │ │ routes   │  │
│  └──────┬───────┘ └──────────┘ └───────┬───────┘ └──────────┘  │
└─────────┼──────────────────────────────┼────────────────────────┘
          │                              │
          ▼                              ▼
┌─────────────────┐   ┌──────────────────────────────────────────┐
│ DefaultAzure    │   │          Azure REST APIs                  │
│ Credential      │   │  • Cost Management (2023-11-01)           │
│ (az login /     │   │  • Advisor (2023-01-01)                   │
│  env vars /     │   │  • Resource Graph (2022-10-01)            │
│  managed id)    │   │  • Azure Monitor (2024-02-01)             │
└─────────────────┘   │  • Azure Updates RSS Feed                 │
                      └──────────────────────────────────────────┘

所有工具通过以下方式返回标准化的JSON response.py 助手。Azure API调用通过 httpx 使用承载令牌身份验证（使用 azure-mgmt-monitor SDK）。

______________________________________________________________________

项目结构

mcp-server/
├── server.py                  # Entry point — registers 22 MCP tools, health checks, CLI
├── auth.py                    # Azure credential management & token acquisition
├── response.py                # Standardised JSON response helpers (success/error)
├── requirements.txt           # Python dependencies
├── startup.sh                 # Azure App Service startup script
├── .env.example               # Environment variables template (copy to .env)
├── LICENSE                    # MIT License
├── CONTRIBUTING.md            # Contribution guidelines
└── tools/
    ├── __init__.py
    ├── validators.py          # Input validation & sanitization (SSRF, injection prevention)
    ├── cost_services.py       # Cost queries (subscription, resource group, management group, compare)
    ├── forecast.py            # Cost forecasting
    ├── budgets.py             # Budget CRUD operations
    ├── alerts.py              # Cost alert management
    ├── anomalies.py           # Anomaly detection (native API + heuristic fallback)
    ├── recommendations.py     # Azure Advisor cost recommendations
    ├── utilization.py         # Dynamic service utilization metrics
    ├── resource_discovery.py  # Azure Resource Graph discovery & token extraction
    ├── azure_updates.py            # Azure Updates RSS feed parser (defusedxml)
    ├── azure_updates_history.py    # Historical Azure Updates via OData API (up to 3 years)
    ├── update_intelligence.py      # Two-phase personalised update digest & drill-down engine
    └── pdf_report.py               # Professional colour-coded PDF report generation
├── tests/
    ├── test_mcp.py            # MCP protocol smoke tests
    └── test_cost.py           # Cost Management API integration test

______________________________________________________________________

工具参考

成本查询工具（4个工具）

常用参数：

• subscription_id --Azure订阅ID（默认为 AZURE_SUBSCRIPTION_ID 任何人）
• query_type — ActualCost 或 AmortizedCost
• timeframe — MonthToDate, BillingMonthToDate, TheLastMonth, TheLastBillingMonth, WeekToDate，或 Custom
• granularity — Daily, Monthly，或 None （仅总计）
• group_by --逗号分隔的尺寸： ServiceName, ResourceGroup, ResourceType, MeterCategory, ResourceLocation, ChargeType, PublisherType, TagKey:YourTag
• custom_from / custom_to --日期范围(YYYY-MM-DD)当时间范围为 Custom

预测工具（1个工具）

参数：

• subscription_id, resource_group --范围
• granularity — Daily 或 Monthly
• forecast_days --预测天数（默认30天，最多365天）
• group_by --可选逗号分隔的分组维度
• include_actual --包括实际成本和预测（默认 true)

预算工具（5个工具）

关键参数 create_budget_tool:

• budget_name --预算名称
• amount --以认购货币表示的预算金额
• time_grain — Monthly, Quarterly, Annually，或 BillingMonth
• notification_thresholds --逗号分隔的阈值（例如。 80,100,120).值>100会自动使用预测警报类型。
• contact_emails --通知的逗号分隔电子邮件地址
• start_date / end_date --预算期(YYYY-MM-DDT00:00:00Z)

警报工具（2个工具）

推荐工具（2个工具）

异常检测工具（1个工具）

参数：

• subscription_id --Azure订阅ID
• days_back --要分析的天数（默认值为30）

使用工具（1个工具）

参数：

• service_name --要查询的Azure服务（例如。 "Storage", "Virtual Machines", "SQL", "Cosmos DB", "AKS")
• subscription_id --Azure订阅ID

Azure更新智能工具（6个工具）

这些工具实现了 两阶段工作流 对于Azure更新：

Phase 1: get_personalized_azure_updates_tool
    → Compact service-grouped executive summary
    → IDF-weighted relevance scoring against YOUR deployed resources
    → 7 priority sections (Retirements → Security → Cost → Features → Preview → Regional → Other)

Phase 2: drill_down_azure_updates_tool
    → Expand any section, service, or status with AND filters
    → Full details for the slice you care about

评分管道（第一阶段）：

1. 通过Azure资源图发现已部署的资源
2. 从Azure更新RSS源获取并解析所有更新
3. 计算所有更新令牌的IDF权重
4. 根据每种部署的资源类型对每次更新进行评分（IDF加权集重叠，0.0-1.0）
5. 计算优先级=相关性×（1-紧急情况权重）+紧急程度×紧急情况权重
6. 分为7个部分，按服务分组，生成执行摘要

参数 search_azure_updates_history_tool:

优先部分：

______________________________________________________________________

先决条件

• Python 3.10+
• Azure命令行界面 --已登录 az login （或支持的其他凭据类型 DefaultAzureCredential)
• RBAC角色 在目标订阅上：

- Cost Management Reader --用于成本查询、预测、预算、警报、异常 - Reader --有关资源图、Azure Monitor和Advisor的建议

______________________________________________________________________

设置

# Navigate to the MCP server directory
cd mcp-server

# Create and activate a virtual environment
python -m venv .venv
.venv\Scripts\activate   # Windows
# source .venv/bin/activate  # macOS/Linux

# Install dependencies
pip install -r requirements.txt

创建一个 .env 文件在 mcp-server/ 目录：

AZURE_SUBSCRIPTION_ID=your-subscription-id-here

______________________________________________________________________

运行服务器

# Ensure Azure CLI is logged in
az login

# Run the MCP server (Streamable HTTP on port 8000, default)
python server.py

# Or specify transport mode explicitly
python server.py --transport streamable-http   # HTTP mode (default)
python server.py --transport stdio             # stdio mode (for local MCP clients)

服务器启动于 http://localhost:8000 默认情况下使用Streamable HTTP传输。

MCP端点： http://localhost:8000/mcp

______________________________________________________________________

环境变量

用于服务主体身份验证（而不是 az login)，设置：

• AZURE_CLIENT_ID
• AZURE_TENANT_ID
• AZURE_CLIENT_SECRET

______________________________________________________________________

API终点

______________________________________________________________________

响应格式

所有工具都返回标准化的JSON：

成功：

{
  "status": "success",
  "data": [ ... ],
  "metadata": {
    "timestamp": "2025-01-15T10:30:00+00:00",
    "scope": "subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    "timeframe": "MonthToDate",
    "currency": "USD",
    "rowCount": 15
  }
}

错误：

{
  "status": "error",
  "error": {
    "message": "Azure API error 401: ...",
    "code": "401"
  },
  "metadata": {
    "timestamp": "2025-01-15T10:30:00+00:00"
  }
}

______________________________________________________________________

使用的Azure API

______________________________________________________________________

依赖项

______________________________________________________________________

测试

烟雾测试脚本位于 tests/ 目录：

# With the server running on port 8000
python tests/test_mcp.py

测试验证：

1. 健康检查端点(GET /health)
2. MCP初始化（服务器名称和版本）
3. 工具列表（已注册的所有22个工具）
4. list_all_azure_updates_tool （RSS源解析）
5. query_subscription_costs_tool （Azure成本管理）

______________________________________________________________________

模块参考

server.py

主要入口点。将所有22个MCP工具注册为 @mcp.tool() 包装器（将空字符串转换为 None)，配置Starlette健康检查路由，设置日志记录，并提供CLI --transport 旗帜。

auth.py

具有令牌缓存的Azure凭据管理。提供：

• get_credential() --已缓存 DefaultAzureCredential 单例
• get_subscription_id() --阅读 AZURE_SUBSCRIPTION_ID 从环境
• get_token(scope) --通过自动缓存获取承载令牌，直到接近到期
• get_cost_management_client() --Azure成本管理SDK客户端
• get_resource_graph_client() --Azure资源图SDK客户端
• get_monitor_client(sub_id) --Azure监控SDK客户端
• get_resource_client(sub_id) --Azure资源管理SDK客户端

支持三种凭据来源： az login，环境变量(AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET)以及管理身份。

response.py

所有工具使用的标准化JSON响应助手：

• success_response(data, scope=, timeframe=, currency=, extra_meta=) --用元数据包装数据（时间戳、列表的rowCount）
• error_response(message, code=) --用时间戳包裹错误消息

tools/cost_services.py

成本查询引擎。构建成本管理查询API有效负载，针对订阅/资源组/管理组作用域执行，解析表响应，处理429速率限制。所有范围标识符均通过以下方式验证 validators.pyThe compare_costs 函数独立查询两个时段，并将结果与每个维度的成本变化和百分比增量合并。

tools/validators.py

所有工具模块使用的输入验证和净化助手：

• validate_subscription_id() --验证UUID格式
• validate_resource_group() / validate_budget_name() / validate_management_group_id() --验证Azure资源名称
• validate_azure_resource_id() --防止资源ID URL中的SSRF和路径遍历
• sanitize_kql_input() --KQL查询插值的安全字符白名单
• validate_output_path() / validate_output_directory() --防止文件输出的路径遍历
• sanitize_error_message() --截断错误消息以防止内部细节泄漏

tools/forecast.py

通过成本管理预测API进行成本预测。支持可配置的预测范围（最多365天）、粒度和维度分组。可以包括实际历史成本和预测数据。

tools/budgets.py

Azure成本预算的完整CRUD。使用以下参数自动计算12个月的默认结束日期 python-dateutil.通知阈值>100时自动使用“预测”警报类型，≤100时使用“实际”警报类型。根据当前支出与金额分析预算利用率。

tools/alerts.py

成本预警管理。列出警报的类型、类别、状态、阈值、当前支出和联系人电子邮件。支持通过PATCH解除警报 status=Dismissed.

tools/anomalies.py

双策略异常检测：

1. 本地API -首先尝试Azure成本异常检测API
2. 启发式回退 -如果本地API不可用（404/400/501），则查询每日成本并标记支出超过7天滚动平均值2个标准偏差的天数。返回严重性（高>100%，中>50%，低）和偏差百分比。

tools/recommendations.py

Azure Advisor成本优化建议。获取按以下条件筛选的推荐 Category eq 'Cost'，通过处理分页 nextLink，提取估计的节省，并按节省降序对结果进行排序。包含 totalEstimatedSavings 在元数据中。

tools/utilization.py

动态服务利用率指标。没有硬编码的资源/度量映射：

1. 通过Azure资源图发现所请求服务类型的资源
2. 查询每个资源的Azure Monitor的度量定义API以查找可用的度量
3. 获取过去3天的度量值（平均聚合）
4. 按资源利用率细分的回报

tools/resource_discovery.py

基于Azure资源图的资源发现，具有智能令牌提取功能：

• 通过Resource Graph API查询订阅中的所有资源
• 按类型、计数和位置汇总
• 从ARM资源类型字符串中提取语义标记（camelCase拆分、复合对生成、噪声词删除）
• 令牌由以下人员使用 update_intelligence.py 用于相关性评分

tools/azure_updates.py

Azure更新RSS提要解析器（使用 defusedxml 对于XXE安全的XML解析，否 feedparser 依赖性）。仅限于约200个最近的项目（约4个月）：

• 胎儿来自 https://www.microsoft.com/releasecommunications/api/v2/azure/rss
• 解析XML项，提取id、标题、状态（来自括号前缀）、类别（服务与元）、描述、链接
• 支持按类别、状态和关键字搜索进行筛选
• 生成内部令牌(_category_tokens, _meta_tokens, _title_tokens)用于匹配

tools/azure_updates_history.py

通过Microsoft Release Communications OData REST API搜索Azure历史更新：

• API完整终结点： https://www.microsoft.com/releasecommunications/api/v2/azure
• 返回带有OData的结构化JSON（不是RSS/XML） $top, $skip, $count, $filter 支持
• 服务器端过滤打开 products, productCategories，以及 status/tags 领域
• 标签感知状态过滤：退役项目存储在 tags （不是 status)；该工具透明地处理两者
• 跨标题和描述的客户端关键字搜索
• 可配置的日期范围过滤 from_date/to_date （最多3年回顾）
• 分页提取（100个项目/页，最多50页），结果按日期排序（最新优先）
• 每次查询最多返回500个结果
• 产品使用API分类法中的确切名称（例如。 Virtual Machines, Azure Kubernetes Service (AKS), Azure SQL Database)

tools/update_intelligence.py

实施两阶段个性化更新工作流程的核心智能引擎：

• 第一阶段(get_personalized_updates): 发现已部署的资源→ 获取RSS源→ 计算IDF令牌权重→ 通过IDF加权集重叠对每次更新进行评分→ 分为7个优先部分→ 按服务分组→ 生成紧凑的执行摘要，并可选择CSV导出（16列）
• 第二阶段(drill_down_updates): 接受部分/服务/状态过滤器（不区分大小写的部分匹配，AND逻辑）→ 返回高于最小相关性阈值的匹配更新的完整详细信息
• 评分： priority = relevance × (1 − urgency_weight) + urgency × urgency_weight
• 影响评估确定每次更新会影响哪些部署的资源类型

tools/pdf_report.py

专业彩色编码PDF报告生成器，使用 fpdf2:

• 封面 --订阅摘要、顶级服务、生成时间戳
• 执行摘要 --关键指标，带颜色编码的部分概述
• 最重要的更新 --所有部分的关键亮点
• 每节详细页面 --关键要点+服务组摘要（仅限完整报告）
• 调色板：红色（渐变）、橙色（安全）、绿色（成本）、蓝色（功能）、紫色（预览）、青色（区域）、灰色（其他）
• 处理Unicode→拉丁文-1文本净化
• 输出默认为带时间戳文件名的临时目录

Azure部署

MCP服务器可以部署到 Azure应用服务 （Linux、Python 3.11+）。

先决条件

• Azure订阅 Cost Management Reader 和 Reader RBAC角色
• Azure CLI已安装并登录(az login)

推荐的应用程序设置

安全最佳实践

• 使用 系统分配的管理身份 用于Azure API身份验证（未存储机密）
• 分配 RBAC角色 在目标订阅上：

- Cost Management Reader - Reader

部署

# From the mcp-server/ directory
cd mcp-server

# Create deployment zip (exclude venv, pycache, env files)
$exclude = @('.venv', '__pycache__', '.env', '*.pyc', 'test_*.py')
$items = Get-ChildItem -Path . -Exclude $exclude
Compress-Archive -Path $items -DestinationPath $env:TEMP\meghiq-mcp-deploy.zip -Force

# Deploy (replace with your resource names)
az webapp deploy --name  --resource-group  `
  --src-path "$env:TEMP\meghiq-mcp-deploy.zip" --type zip

______________________________________________________________________

MCP检验员测试

# Use the MCP CLI inspector to test tools interactively
mcp dev server.py

从代理框架连接

from agent_framework import MCPStreamableHTTPTool

# Local development
mcp_tool = MCPStreamableHTTPTool(
    name="MeghIQ Cost Optimizer",
    url="http://localhost:8000/mcp",
)

# Azure deployment (replace with your app URL)
mcp_tool = MCPStreamableHTTPTool(
    name="MeghIQ Cost Optimizer",
    url="https://.azurewebsites.net/mcp",
)

______________________________________________________________________

例题

一旦连接到MCP客户端（Copilot、Claude、自定义代理等），请尝试以下提示：

成本分析

• “本月我最昂贵的5项Azure服务是什么？”
• “按资源组显示过去7天的每日成本明细。”
• “比较我上个月和本月的Azure支出——有什么变化？”
• “到目前为止，我在存储和计算服务上的支出是多少？”
• “我的成本是多少 production-rg 资源组此计费周期？"

预测

• “预测我未来30天的Azure成本。”
• “按服务细分，我下一季度的支出会是什么样子？”

预算

• “列出我的所有Azure预算以及每个预算的使用量。”
• “创建每月5000美元的预算，并在80%和100%时发出警报。”
• “将我的‘生产预算’更新为8000美元，并添加120%的预测警报。”
• “删除‘测试预算’预算。”

警报和异常

• “是否有任何我应该知道的主动成本警报？”
• “在过去的30天里，是否出现了任何不寻常的支出激增？”
• “显示我过去两周订阅的成本异常情况。”

建议

• “Azure Advisor为我提供了哪些成本优化建议？”
• “请详细说明最重要的建议——我能节省多少钱？”

利用率

• “我的虚拟机的当前利用率是多少？”
• “显示我的SQL数据库的利用率指标。”
• “我的存储帐户的使用情况如何？”

Azure更新

• “哪些Azure更新与我的环境相关？”
• “深入了解我的服务的退役和弃用更新。”
• “是否有任何安全更新会影响我正在使用的服务？”
• “显示与Azure Kubernetes服务相关的所有Azure更新。”
• “生成针对我的订阅个性化的Azure更新的PDF报告。”

Azure历史更新（最长3年）

• “查找过去2年的所有虚拟机更新。”
• “显示过去3年中所有Azure SQL数据库的退役情况。”
• “搜索自2024年以来所有Azure服务中提到‘机密’的更新。”
• “2025年1月至2026年3月期间，AKS推出了哪些更新？”
• “列出当前处于预览状态的所有计算类别更新。”
• “查找去年Azure Functions的任何退役通知。”

组合成本优化（代理协调多种工具）

• “根据我当前的利用率和过去2年的Azure更新，为前5大昂贵资源提供成本建议。”
• “我的哪些资源闲置或未充分利用，删除它们可以节省多少钱？”
• “我的服务是否已经退役？有什么更便宜的替代品吗？”
• “将我当前的VM大小与最新的Azure VM系列进行比较——我可以调整大小或升级以节省资金吗？”
• “显示我所有资源组中的孤立资源（磁盘、IP、扩展）以及每月的成本影响。”
• “根据我过去3个月的支出，存在哪些预留实例或储蓄计划机会？”
• “本月是否存在任何成本异常，是什么资源导致了价格飙升？”
• “哪些资源组的成本环比增长最高，这些资源是否被实际使用？”
• “给我一个完整的成本优化摘要：顶级支出服务、顾问建议、未充分利用的资源和即将退休的人员。”
• “对于我的永远在线的虚拟机和应用程序服务，有哪些无服务器或基于消费的替代方案？”