server.py

"""
MCP Server demonstrating HTTP tool patterns.

This template shows:
- Async HTTP boundaries with proper error handling
- Retry logic and partial failure
- Response parsing and validation

Run with: uv run server.py
"""

import logging
from typing import Any

import httpx
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

mcp = FastMCP("tools")

HTTP_TIMEOUT = 10.0
MAX_RETRIES = 2


async def fetch_json(url: str, headers: dict[str, str] | None = None) -> dict[str, Any] | None:
    """Make an HTTP GET request with retry logic."""
    async with httpx.AsyncClient() as client:
        for attempt in range(MAX_RETRIES):
            try:
                response = await client.get(url, headers=headers, timeout=HTTP_TIMEOUT)
                response.raise_for_status()
                return response.json()
            except httpx.TimeoutException:
                logger.warning(f"Timeout on attempt {attempt + 1} for {url}")
            except httpx.HTTPStatusError as e:
                logger.error(f"HTTP {e.response.status_code} for {url}")
                return None
            except httpx.RequestError as e:
                logger.error(f"Request failed: {e}")
                return None
    return None


@mcp.tool()
async def lookup_ip(ip_address: str) -> str:
    """Look up geolocation and network info for an IP address.

    Args:
        ip_address: IPv4 or IPv6 address to look up
    """
    data = await fetch_json(f"http://ip-api.com/json/{ip_address}")

    if not data:
        return f"Failed to look up IP: {ip_address}"

    if data.get("status") == "fail":
        return f"Lookup failed: {data.get('message', 'unknown error')}"

    return (
        f"IP: {data['query']}\n"
        f"Location: {data['city']}, {data['regionName']}, {data['country']}\n"
        f"ISP: {data['isp']}\n"
        f"Organization: {data['org']}\n"
        f"Timezone: {data['timezone']}"
    )


@mcp.tool()
async def get_random_user() -> str:
    """Generate a random user profile for testing or mock data."""
    data = await fetch_json("https://randomuser.me/api/")

    if not data or "results" not in data:
        return "Failed to generate random user."

    user = data["results"][0]
    name = user["name"]
    location = user["location"]

    return (
        f"Name: {name['first']} {name['last']}\n"
        f"Email: {user['email']}\n"
        f"Location: {location['city']}, {location['country']}\n"
        f"Phone: {user['phone']}"
    )


@mcp.tool()
async def fetch_post(post_id: int) -> str:
    """Fetch a post by ID from JSONPlaceholder API.

    Args:
        post_id: The post ID (1-100)
    """
    if not 1 <= post_id <= 100:
        return "Post ID must be between 1 and 100."

    data = await fetch_json(f"https://jsonplaceholder.typicode.com/posts/{post_id}")

    if not data:
        return f"Failed to fetch post {post_id}."

    return (
        f"Post #{data['id']}\n"
        f"Title: {data['title']}\n\n"
        f"{data['body']}"
    )


def main():
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()