Resource templates

from typing import Any
import httpx
from mcp.server import MCPServer

# Initialize the MCP server
mcp = MCPServer("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"


async def make_nws_request(url: str) -> dict[str, Any] | None:
    """Make a request to the NWS API with proper error handling."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None
        
def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    props = feature["properties"]
    return f"""
        Event: {props.get('event', 'Unknown')}
        Area: {props.get('areaDesc', 'Unknown')}
        Severity: {props.get('severity', 'Unknown')}
        Description: {props.get('description', 'No description available')}
        Instructions: {props.get('instruction', 'No specific instructions provided')}
        """

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)


@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
    """Echo a message as a resource"""
    return f"Resource echo: {message}"

from typing import Annotated, Literal

from pydantic import BaseModel, Field

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ResourceNotFoundError, ToolError
from mcp.types import ToolAnnotations

mcp = MCPServer("Shop support")

ORDERS = {
    "A17": {"item": "blue mug", "status": "shipped", "total": 12.50},
    "B42": {"item": "desk lamp", "status": "processing", "total": 48.00},
}

ARTICLES = {
    "Where is my order?": "orders",
    "Changing an order": "orders",
    "How refunds work": "refunds",
    "Refunds for damaged items": "refunds",
    "Resetting your password": "account",
}


class Order(BaseModel):
    id: str
    item: str
    status: Literal["processing", "shipped", "delivered"]
    total: float


@mcp.tool(annotations=ToolAnnotations(read_only_hint=True))
def lookup_order(order_id: str) -> Order:
    """Look up an order by its id."""
    if order_id not in ORDERS:
        raise ToolError(f"No order with id {order_id!r}. Order ids look like A17.")
    return Order(id=order_id, **ORDERS[order_id])


@mcp.tool()
def search_help(
    query: Annotated[str, Field(description="Words to look for in the help articles.")],
    topic: Literal["orders", "refunds", "account"] | None = None,
    limit: Annotated[int, Field(ge=1, le=5)] = 3,
) -> str:
    """Search the help centre and return matching article titles."""
    found = [title for title, t in ARTICLES.items() if query.lower() in title.lower() and topic in (None, t)]
    return "; ".join(found[:limit]) or "No articles found."


@mcp.tool(
    title="Refund an order",
    annotations=ToolAnnotations(read_only_hint=False, destructive_hint=False, idempotent_hint=False),
)
def refund_order(order_id: str, reason: str) -> str:
    """Refund the full total of an order to the customer's card."""
    if order_id not in ORDERS:
        raise ToolError(f"No order with id {order_id!r}.")
    return f"Refunded {ORDERS[order_id]['total']:.2f} for order {order_id}: {reason}."


@mcp.resource("policy://refunds", mime_type="text/markdown")
def refund_policy() -> str:
    """The shop's refund policy."""
    return "# Refunds\n\nFull refund within 30 days of delivery. Damaged items: refund or replacement."


@mcp.resource("orders://{order_id}", mime_type="application/json")
def order_record(order_id: str) -> Order:
    """The full record for one order."""
    if order_id not in ORDERS:
        raise ResourceNotFoundError(f"No order with id {order_id!r}.")
    return Order(id=order_id, **ORDERS[order_id])