Como MCP funciona: o protocolo que conecta agents ao mundo

{
  "name": "restart_service",
  "description": "Reinicia um serviço systemd no servidor especificado",
  "inputSchema": {
    "type": "object",
    "properties": {
      "hostname": {
        "type": "string",
        "description": "Hostname do servidor"
      },
      "service_name": {
        "type": "string",
        "description": "Nome do serviço systemd"
      }
    },
    "required": ["hostname", "service_name"]
  }
}

{
  "uri": "server://web-prod-01/metrics",
  "name": "Métricas de web-prod-01",
  "description": "CPU, memória, disco e rede do servidor web-prod-01",
  "mimeType": "application/json"
}

{
  "name": "diagnose_alert",
  "description": "Template pra diagnosticar um alerta de monitoramento",
  "arguments": [
    {"name": "alert_name", "description": "Nome do alerta", "required": true},
    {"name": "resource", "description": "Recurso afetado", "required": true}
  ]
}

import os
import subprocess
from typing import Annotated

from mcp.server import MCPServer
from pydantic import Field

SUBSCRIPTION_ID = os.environ["AZURE_SUBSCRIPTION_ID"]

mcp = MCPServer("azure-monitor")


@mcp.tool()
def get_vm_metrics(
    resource_group: str,
    vm_name: str,
    offset: Annotated[str, Field(description="Ex: 1h ou 24h")] = "1h",
) -> str:
    """Retorna métricas de CPU de uma VM Azure."""
    resource_id = (
        f"/subscriptions/{SUBSCRIPTION_ID}"
        f"/resourceGroups/{resource_group}"
        f"/providers/Microsoft.Compute/virtualMachines/{vm_name}"
    )
    result = subprocess.run(
        [
            "az", "monitor", "metrics", "list",
            "--resource", resource_id,
            "--metric", "Percentage CPU",
            "--interval", "PT5M",
            "--offset", offset,
            "--output", "json",
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    return result.stdout


@mcp.tool()
def list_metric_alert_rules(resource_group: str) -> str:
    """Lista regras de alerta baseadas em métricas no resource group."""
    result = subprocess.run(
        [
            "az", "monitor", "metrics", "alert", "list",
            "--resource-group", resource_group,
            "--output", "json",
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    return result.stdout


@mcp.resource("azure://monitor/metric-alert-rules")
def metric_alert_rules() -> str:
    """Regras de alertas de métricas disponíveis na subscription."""
    result = subprocess.run(
        [
            "az", "monitor", "metrics", "alert", "list",
            "--output", "json",
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    return result.stdout


@mcp.prompt()
def diagnose_alert(alert_name: str, resource: str) -> str:
    """Monta um prompt inicial pra diagnosticar um alerta."""
    return (
        f"Diagnostique o alerta {alert_name} no recurso {resource}. "
        "Liste hipóteses, sinais pra confirmar cada uma e a ordem de checagem."
    )


if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "azure-monitor": {
      "command": "python",
      "args": ["./mcp-servers/azure-monitor/server.py"],
      "env": {
        "AZURE_SUBSCRIPTION_ID": "seu-sub-id"
      }
    }
  }
}

from server import mcp


if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=8000,
        streamable_http_path="/mcp",
    )

# Adicionar auth no server HTTP
from starlette.authentication import requires
from starlette.responses import Response


@requires("authenticated")
async def handle_mcp_http_request(request):
    payload = await request.json()

    if payload.get("method") == "tools/call":
        token = request.auth.credentials
        allowed_tools = get_allowed_tools(token)
        tool_name = payload["params"]["name"]

        if tool_name not in allowed_tools:
            return Response(status_code=403)

    # Encaminhar a requisição pro handler MCP real
    ...

# Não exponha tools destrutivas por padrão
SAFE_TOOLS = ["get_metrics", "list_alerts", "get_logs"]
DANGEROUS_TOOLS = ["restart_service", "scale_resource", "delete_resource"]


def exposed_tools_for(client_level: str) -> list[str]:
    tool_names = SAFE_TOOLS.copy()
    if client_level == "admin":
        tool_names += DANGEROUS_TOOLS

    return tool_names

def validate_get_vm_metrics(arguments: dict) -> dict:
    if not validate_resource_group(arguments["resource_group"]):
        raise ValueError("resource group inválido ou sem permissão")

    return {
        "resource_group": arguments["resource_group"],
        "vm_name": sanitize_resource_name(arguments["vm_name"]),
        "offset": arguments.get("offset", "1h"),
    }