Monitoring Model Context Protocol (MCP) servers

Weather MCP server

An MCP server with observability enabled is shown in the following example:

from __future__ import annotations

import os
import sys
from typing import Any, Dict, Optional

# Configure OTel exporters
os.environ.setdefault("OTEL_TRACES_EXPORTER", "none")
os.environ.setdefault("OTEL_METRICS_EXPORTER", "none")
os.environ.setdefault("OTEL_LOGS_EXPORTER", "none")

try:
    from traceloop.sdk import Traceloop
    from traceloop.sdk.decorators import workflow
except ImportError:
    Traceloop = None

import httpx
from mcp.server.fastmcp import FastMCP

# Initialize OpenLLMetry
if Traceloop is not None:
    try:
        Traceloop.init(app_name="Weather-MCP-Server", disable_batch=True)
    except Exception as e:
        print(f"OpenLLMetry initialization warning: {e}", file=sys.stderr)

mcp = FastMCP(name="Weather")
NWS_API_BASE = "https://api.weather.gov"
NWS_USER_AGENT = "Weather-MCP-Server (dev@example.com)"

async def make_nws_request(url: str) -> Optional[Dict[str, Any]]:
    """Make a GET request to NWS API."""
    headers = {
        "User-Agent": NWS_USER_AGENT,
        "Accept": "application/geo+json",
    }
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            resp = await client.get(url, headers=headers)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"NWS request failed: {e}", file=sys.stderr)
            return None

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude coordinate
        longitude: Longitude coordinate

    Returns:
        Weather forecast as formatted string
    """
    try:
        lat, lon = float(latitude), float(longitude)
    except ValueError:
        return "Latitude and longitude must be numeric."

    # Get forecast URL from points API
    points_url = f"{NWS_API_BASE}/points/{lat},{lon}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    forecast_url = (points_data.get("properties") or {}).get("forecast")
    if not forecast_url:
        return "Forecast URL not available for this location."

    # Get forecast data
    forecast_data = await make_nws_request(forecast_url)
    if not forecast_data:
        return "Unable to fetch detailed forecast."

    periods = (forecast_data.get("properties") or {}).get("periods") or []
    if not periods:
        return "No forecast periods available."

    # Format first 5 periods
    forecasts = []
    for period in periods[:5]:
        name = period.get("name", "N/A")
        temp = period.get("temperature", "N/A")
        unit = period.get("temperatureUnit", "")
        detail = period.get("detailedForecast", "N/A")
        forecasts.append(f"{name}: {temp}°{unit}\n{detail}")

    return "\n---\n".join(forecasts)

@workflow(name="weather_mcp_workflow")
def start_server():
    """Start MCP server."""
    mcp.run(transport="streamable-http")

if __name__ == "__main__":
    try:
        start_server()
    except KeyboardInterrupt:
        sys.exit(0)