Mcp Builder

"""
FastMCP API Client Pattern
===========================
Manual API integration with connection pooling, caching, and retry logic.
"""

from fastmcp import FastMCP
import httpx
import os
import time
import asyncio
from typing import Optional, Any, Dict
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

mcp = FastMCP("API Client Pattern")

# ============================================================================
# Configuration
# ============================================================================

class Config:
    """API configuration from environment."""
    API_BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com")
    API_KEY = os.getenv("API_KEY", "")
    API_TIMEOUT = int(os.getenv("API_TIMEOUT", "30"))
    CACHE_TTL = int(os.getenv("CACHE_TTL", "300"))  # 5 minutes
    MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3"))


# ============================================================================
# API Client with Connection Pooling
# ============================================================================

class APIClient:
    """Singleton API client with connection pooling."""
    _instance: Optional[httpx.AsyncClient] = None

    @classmethod
    async def get_client(cls) -> httpx.AsyncClient:
        """Get or create the shared HTTP client."""
        if cls._instance is None:
            cls._instance = httpx.AsyncClient(
                base_url=Config.API_BASE_URL,
                headers={
                    "Authorization": f"Bearer {Config.API_KEY}",
                    "Content-Type": "application/json",
                    "User-Agent": "FastMCP-Client/1.0"
                },
                timeout=httpx.Timeout(Config.API_TIMEOUT),
                limits=httpx.Limits(
                    max_keepalive_connections=5,
                    max_connections=10
                )
            )
        return cls._instance

    @classmethod
    async def cleanup(cls):
        """Cleanup the HTTP client."""
        if cls._instance:
            await cls._instance.aclose()
            cls._instance = None


# ============================================================================
# Cache Implementation
# ============================================================================

class SimpleCache:
    """Time-based cache for API responses."""

    def __init__(self, ttl: int = 300):
        self.ttl = ttl
        self.cache: Dict[str, Any] = {}
        self.timestamps: Dict[str, float] = {}

    def get(self, key: str) -> Optional[Any]:
        """Get cached value if not expired."""
        if key in self.cache:
            if time.time() - self.timestamps[key] < self.ttl:
                return self.cache[key]
            else:
                # Expired, remove it
                del self.cache[key]
                del self.timestamps[key]
        return None

    def set(self, key: str, value: Any):
        """Set cache value with timestamp."""
        self.cache[key] = value
        self.timestamps[key] = time.time()

    def invalidate(self, pattern: Optional[str] = None):
        """Invalidate cache entries."""
        if pattern:
            keys_to_delete = [k for k in self.cache if pattern in k]
            for key in keys_to_delete:
                del self.cache[key]
                del self.timestamps[key]
        else:
            self.cache.clear()
            self.timestamps.clear()


# Global cache instance
cache = SimpleCache(ttl=Config.CACHE_TTL)


# ============================================================================
# Retry Logic with Exponential Backoff
# ============================================================================

async def retry_with_backoff(
    func,
    max_retries: int = 3,
    initial_delay: float = 1.0,
    exponential_base: float = 2.0
):
    """Retry function with exponential backoff."""
    delay = initial_delay
    last_exception = None

    for attempt in range(max_retries):
        try:
            return await func()
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            last_exception = e
            if attempt < max_retries - 1:
                print(f"Attempt {attempt + 1} failed, retrying in {delay}s...")
                await asyncio.sleep(delay)
                delay *= exponential_base
        except httpx.HTTPStatusError as e:
            # Don't retry client errors (4xx)
            if 400 <= e.response.status_code < 500:
                raise
            last_exception = e
            if attempt < max_retries - 1:
                await asyncio.sleep(delay)
                delay *= exponential_base

    raise last_exception


# ============================================================================
# API Tools
# ============================================================================

@mcp.tool()
async def api_get(
    endpoint: str,
    use_cache: bool = True
) -> dict:
    """
    Make a GET request to the API.

    Args:
        endpoint: API endpoint path (e.g., "/users/123")
        use_cache: Whether to use cached response if available

    Returns:
        API response data or error
    """
    cache_key = f"GET:{endpoint}"

    # Check cache
    if use_cache:
        cached = cache.get(cache_key)
        if cached:
            return {
                "success": True,
                "data": cached,
                "from_cache": True,
                "timestamp": datetime.now().isoformat()
            }

    # Make request with retry
    async def make_request():
        client = await APIClient.get_client()
        response = await client.get(endpoint)
        response.raise_for_status()
        return response.json()

    try:
        data = await retry_with_backoff(make_request, max_retries=Config.MAX_RETRIES)

        # Cache successful response
        if use_cache:
            cache.set(cache_key, data)

        return {
            "success": True,
            "data": data,
            "from_cache": False,
            "timestamp": datetime.now().isoformat()
        }

    except httpx.HTTPStatusError as e:
        return {
            "success": False,
            "error": f"HTTP {e.response.status_code}",
            "message": e.response.text,
            "endpoint": endpoint
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "endpoint": endpoint
        }


@mcp.tool()
async def api_post(
    endpoint: str,
    data: dict,
    invalidate_cache: bool = True
) -> dict:
    """
    Make a POST request to the API.

    Args:
        endpoint: API endpoint path
        data: Request body data
        invalidate_cache: Whether to invalidate related cache entries

    Returns:
        API response or error
    """
    async def make_request():
        client = await APIClient.get_client()
        response = await client.post(endpoint, json=data)
        response.raise_for_status()
        return response.json()

    try:
        result = await retry_with_backoff(make_request, max_retries=Config.MAX_RETRIES)

        # Invalidate cache for related endpoints
        if invalidate_cache:
            cache.invalidate(endpoint.split('/')[1] if '/' in endpoint else endpoint)

        return {
            "success": True,
            "data": result,
            "timestamp": datetime.now().isoformat()
        }

    except httpx.HTTPStatusError as e:
        return {
            "success": False,
            "error": f"HTTP {e.response.status_code}",
            "message": e.response.text
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }


@mcp.tool()
async def api_put(
    endpoint: str,
    data: dict,
    invalidate_cache: bool = True
) -> dict:
    """Make a PUT request to the API."""
    async def make_request():
        client = await APIClient.get_client()
        response = await client.put(endpoint, json=data)
        response.raise_for_status()
        return response.json()

    try:
        result = await retry_with_backoff(make_request)

        if invalidate_cache:
            cache.invalidate(endpoint)

        return {"success": True, "data": result}

    except Exception as e:
        return {"success": False, "error": str(e)}


@mcp.tool()
async def api_delete(
    endpoint: str,
    invalidate_cache: bool = True
) -> dict:
    """Make a DELETE request to the API."""
    async def make_request():
        client = await APIClient.get_client()
        response = await client.delete(endpoint)
        response.raise_for_status()
        return response.status_code

    try:
        status = await retry_with_backoff(make_request)

        if invalidate_cache:
            cache.invalidate(endpoint)

        return {
            "success": True,
            "status_code": status,
            "deleted": True
        }

    except Exception as e:
        return {"success": False, "error": str(e)}


@mcp.tool()
async def batch_api_requests(
    endpoints: list[str],
    use_cache: bool = True
) -> dict:
    """
    Make multiple GET requests in parallel.

    Args:
        endpoints: List of endpoint paths
        use_cache: Whether to use cache

    Returns:
        Batch results with successes and failures
    """
    async def fetch_one(endpoint: str):
        return await api_get(endpoint, use_cache=use_cache)

    results = await asyncio.gather(*[fetch_one(ep) for ep in endpoints])

    successful = [r for r in results if r.get("success")]
    failed = [r for r in results if not r.get("success")]

    return {
        "total": len(endpoints),
        "successful": len(successful),
        "failed": len(failed),
        "results": results
    }


@mcp.tool()
def clear_cache(pattern: Optional[str] = None) -> dict:
    """
    Clear API response cache.

    Args:
        pattern: Optional pattern to match cache keys (clears all if not provided)

    Returns:
        Cache clear status
    """
    try:
        cache.invalidate(pattern)
        return {
            "success": True,
            "message": f"Cache cleared{f' for pattern: {pattern}' if pattern else ''}"
        }
    except Exception as e:
        return {"success": False, "error": str(e)}


# ============================================================================
# Resources
# ============================================================================

@mcp.resource("info://api-status")
async def api_status() -> dict:
    """Check API connectivity and status."""
    try:
        client = await APIClient.get_client()
        response = await client.get("/health", timeout=5)
        return {
            "api_reachable": True,
            "status_code": response.status_code,
            "healthy": response.status_code == 200,
            "timestamp": datetime.now().isoformat()
        }
    except Exception as e:
        return {
            "api_reachable": False,
            "error": str(e),
            "timestamp": datetime.now().isoformat()
        }


@mcp.resource("info://cache-stats")
def cache_statistics() -> dict:
    """Get cache statistics."""
    return {
        "total_entries": len(cache.cache),
        "ttl_seconds": cache.ttl,
        "entries": list(cache.cache.keys())
    }


# ============================================================================
# Main
# ============================================================================

if __name__ == "__main__":
    try:
        mcp.run()
    finally:
        # Cleanup on exit
        import asyncio
        asyncio.run(APIClient.cleanup())

"""
Basic FastMCP Server Template
==============================
A minimal working FastMCP server with essential patterns.
"""

from fastmcp import FastMCP
import os

# Load environment variables (optional)
from dotenv import load_dotenv
load_dotenv()

# ============================================================================
# CRITICAL: Server must be at module level for FastMCP Cloud
# ============================================================================

mcp = FastMCP(
    name="My Basic Server",
    instructions="""
    This is a basic MCP server demonstrating core patterns.

    Available tools:
    - greet: Say hello to someone
    - calculate: Perform basic math operations

    Available resources:
    - info://status: Server status information
    """
)

# ============================================================================
# Tools
# ============================================================================

@mcp.tool()
def greet(name: str, greeting: str = "Hello") -> str:
    """
    Greet someone by name.

    Args:
        name: The name of the person to greet
        greeting: The greeting to use (default: "Hello")

    Returns:
        A greeting message
    """
    return f"{greeting}, {name}!"


@mcp.tool()
async def calculate(operation: str, a: float, b: float) -> dict:
    """
    Perform a mathematical operation.

    Args:
        operation: The operation to perform (add, subtract, multiply, divide)
        a: First number
        b: Second number

    Returns:
        Dictionary with the result or error message
    """
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else None
    }

    if operation not in operations:
        return {
            "error": f"Unknown operation: {operation}",
            "valid_operations": list(operations.keys())
        }

    result = operations[operation](a, b)

    if result is None:
        return {"error": "Division by zero"}

    return {
        "operation": operation,
        "a": a,
        "b": b,
        "result": result
    }


# ============================================================================
# Resources
# ============================================================================

@mcp.resource("info://status")
def server_status() -> dict:
    """Get current server status."""
    from datetime import datetime

    return {
        "server": "My Basic Server",
        "status": "operational",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0"
    }


# ============================================================================
# Main Execution
# ============================================================================

if __name__ == "__main__":
    # Run with stdio transport (default)
    mcp.run()

    # Alternative: HTTP transport for testing
    # mcp.run(transport="http", port=8000)

"""
FastMCP Client Example
======================
Testing MCP servers with the FastMCP Client.
"""

import asyncio
from fastmcp import Client
from typing import Optional


async def test_basic_server():
    """Test a basic MCP server."""
    print("\n=== Testing Basic Server ===\n")

    async with Client("basic-server.py") as client:
        # List available tools
        tools = await client.list_tools()
        print(f"Available tools: {len(tools)}")
        for tool in tools:
            print(f"  - {tool.name}: {tool.description}")

        # Call a tool
        print("\nCalling 'greet' tool...")
        result = await client.call_tool("greet", {"name": "World"})
        print(f"Result: {result.data}")

        # Call another tool
        print("\nCalling 'calculate' tool...")
        result = await client.call_tool("calculate", {
            "operation": "add",
            "a": 10,
            "b": 5
        })
        print(f"Result: {result.data}")

        # List resources
        print("\nAvailable resources:")
        resources = await client.list_resources()
        for resource in resources:
            print(f"  - {resource.uri}: {resource.description}")

        # Read a resource
        print("\nReading 'info://status' resource...")
        status = await client.read_resource("info://status")
        print(f"Status: {status}")


async def test_tools_examples():
    """Test the tools examples server."""
    print("\n=== Testing Tools Examples ===\n")

    async with Client("tools-examples.py") as client:
        # Test sync tool
        print("Testing sync tool...")
        result = await client.call_tool("simple_sync_tool", {"text": "hello"})
        print(f"Sync result: {result.data}")

        # Test async tool
        print("\nTesting async tool...")
        result = await client.call_tool("simple_async_tool", {"text": "WORLD"})
        print(f"Async result: {result.data}")

        # Test validated search
        print("\nTesting validated search...")
        result = await client.call_tool("validated_search", {
            "params": {
                "query": "python",
                "limit": 5
            }
        })
        print(f"Search result: {result.data}")

        # Test batch processing
        print("\nTesting batch process...")
        result = await client.call_tool("batch_process", {
            "items": ["item1", "item2", "item3"]
        })
        print(f"Batch result: {result.data}")


async def test_resources_examples():
    """Test the resources examples server."""
    print("\n=== Testing Resources Examples ===\n")

    async with Client("resources-examples.py") as client:
        # List all resources
        resources = await client.list_resources()
        print(f"Total resources: {len(resources)}")

        # Test static resource
        print("\nReading static resource...")
        config = await client.read_resource("data://config")
        print(f"Config: {config}")

        # Test dynamic resource
        print("\nReading dynamic resource...")
        status = await client.read_resource("info://status")
        print(f"Status: {status}")

        # Test resource template
        print("\nReading resource template...")
        profile = await client.read_resource("user://123/profile")
        print(f"User profile: {profile}")


async def test_with_error_handling():
    """Test server with comprehensive error handling."""
    print("\n=== Testing with Error Handling ===\n")

    async with Client("error-handling.py") as client:
        # Test successful operation
        print("Testing successful operation...")
        try:
            result = await client.call_tool("divide_numbers", {
                "a": 10,
                "b": 2
            })
            print(f"Success: {result.data}")
        except Exception as e:
            print(f"Error: {e}")

        # Test error case
        print("\nTesting error case (division by zero)...")
        try:
            result = await client.call_tool("divide_numbers", {
                "a": 10,
                "b": 0
            })
            print(f"Result: {result.data}")
        except Exception as e:
            print(f"Error: {e}")

        # Test validation error
        print("\nTesting validation error...")
        try:
            result = await client.call_tool("validated_operation", {
                "data": ""
            })
            print(f"Result: {result.data}")
        except Exception as e:
            print(f"Error: {e}")


async def test_http_server():
    """Test server running on HTTP transport."""
    print("\n=== Testing HTTP Server ===\n")

    # Note: Server must be running on http://localhost:8000
    # Start with: python server.py --transport http --port 8000

    try:
        async with Client("http://localhost:8000/mcp") as client:
            print("Connected to HTTP server")

            tools = await client.list_tools()
            print(f"Available tools: {len(tools)}")

            if tools:
                result = await client.call_tool(tools[0].name, {})
                print(f"Tool result: {result.data}")

    except Exception as e:
        print(f"Could not connect to HTTP server: {e}")
        print("Make sure server is running with: python server.py --transport http --port 8000")


async def test_with_handlers():
    """Test server with client handlers (elicitation, progress, sampling)."""
    print("\n=== Testing with Handlers ===\n")

    # Define handlers
    async def elicitation_handler(message: str, response_type: type, context: dict):
        """Handle elicitation requests."""
        print(f"\n[ELICIT] {message}")
        return input("Your response: ")

    async def progress_handler(progress: float, total: Optional[float], message: Optional[str]):
        """Handle progress updates."""
        if total:
            pct = (progress / total) * 100
            print(f"\r[PROGRESS] {pct:.1f}% - {message}", end="", flush=True)
        else:
            print(f"\n[PROGRESS] {message}")

    async def sampling_handler(messages, params, context):
        """Handle sampling requests (LLM completions)."""
        print(f"\n[SAMPLE] LLM request with {len(messages)} messages")
        # In production, call actual LLM
        return {
            "content": "Mock LLM response",
            "model": params.get("model", "mock"),
            "usage": {"tokens": 100}
        }

    # Create client with handlers
    async with Client(
        "server.py",
        elicitation_handler=elicitation_handler,
        progress_handler=progress_handler,
        sampling_handler=sampling_handler
    ) as client:
        print("Client created with handlers")

        # Test tools that use handlers
        # Note: Requires server to have tools using context.request_elicitation, etc.
        tools = await client.list_tools()
        print(f"Available tools: {len(tools)}")


async def comprehensive_test():
    """Run comprehensive test suite."""
    print("=" * 60)
    print("FastMCP Client Test Suite")
    print("=" * 60)

    tests = [
        ("Basic Server", test_basic_server),
        ("Tools Examples", test_tools_examples),
        ("Resources Examples", test_resources_examples),
        ("Error Handling", test_with_error_handling),
        # ("HTTP Server", test_http_server),  # Uncomment if HTTP server is running
        # ("With Handlers", test_with_handlers),  # Uncomment if server supports handlers
    ]

    for test_name, test_func in tests:
        try:
            await test_func()
        except Exception as e:
            print(f"\n❌ {test_name} failed: {e}")
        else:
            print(f"\n✅ {test_name} passed")

        print("\n" + "-" * 60)

    print("\nTest suite completed!")


async def interactive_client():
    """Interactive client for manual testing."""
    print("\n=== Interactive FastMCP Client ===\n")

    server_path = input("Enter server path or URL: ").strip()

    if not server_path:
        server_path = "basic-server.py"
        print(f"Using default: {server_path}")

    async with Client(server_path) as client:
        print(f"\n✅ Connected to: {server_path}\n")

        while True:
            print("\nOptions:")
            print("1. List tools")
            print("2. List resources")
            print("3. Call tool")
            print("4. Read resource")
            print("5. Exit")

            choice = input("\nChoice: ").strip()

            if choice == "1":
                tools = await client.list_tools()
                print(f"\n📋 Available tools ({len(tools)}):")
                for i, tool in enumerate(tools, 1):
                    print(f"  {i}. {tool.name}")
                    print(f"     {tool.description}")

            elif choice == "2":
                resources = await client.list_resources()
                print(f"\n📋 Available resources ({len(resources)}):")
                for i, resource in enumerate(resources, 1):
                    print(f"  {i}. {resource.uri}")
                    print(f"     {resource.description}")

            elif choice == "3":
                tool_name = input("Tool name: ").strip()
                print("Arguments (as JSON): ")
                import json
                try:
                    args = json.loads(input().strip())
                    result = await client.call_tool(tool_name, args)
                    print(f"\n✅ Result: {result.data}")
                except Exception as e:
                    print(f"\n❌ Error: {e}")

            elif choice == "4":
                uri = input("Resource URI: ").strip()
                try:
                    data = await client.read_resource(uri)
                    print(f"\n✅ Data: {data}")
                except Exception as e:
                    print(f"\n❌ Error: {e}")

            elif choice == "5":
                print("\nGoodbye!")
                break

            else:
                print("Invalid choice")


if __name__ == "__main__":
    import sys

    if len(sys.argv) > 1 and sys.argv[1] == "--interactive":
        asyncio.run(interactive_client())
    else:
        asyncio.run(comprehensive_test())

"""
FastMCP Error Handling Template
================================
Comprehensive error handling patterns with structured responses and retry logic.
"""

from fastmcp import FastMCP
import asyncio
import httpx
from enum import Enum
from typing import Dict, Any, Optional
from datetime import datetime

mcp = FastMCP("Error Handling Examples")

# ============================================================================
# Error Code Enum
# ============================================================================

class ErrorCode(Enum):
    """Standard error codes."""
    VALIDATION_ERROR = "VALIDATION_ERROR"
    NOT_FOUND = "NOT_FOUND"
    UNAUTHORIZED = "UNAUTHORIZED"
    RATE_LIMITED = "RATE_LIMITED"
    API_ERROR = "API_ERROR"
    TIMEOUT = "TIMEOUT"
    NETWORK_ERROR = "NETWORK_ERROR"
    UNKNOWN_ERROR = "UNKNOWN_ERROR"


# ============================================================================
# Response Formatters
# ============================================================================

def create_success(data: Any, message: str = "Success") -> Dict[str, Any]:
    """Create structured success response."""
    return {
        "success": True,
        "message": message,
        "data": data,
        "timestamp": datetime.now().isoformat()
    }


def create_error(
    code: ErrorCode,
    message: str,
    details: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """Create structured error response."""
    return {
        "success": False,
        "error": {
            "code": code.value,
            "message": message,
            "details": details or {},
            "timestamp": datetime.now().isoformat()
        }
    }


# ============================================================================
# Retry Logic
# ============================================================================

async def retry_with_backoff(
    func,
    max_retries: int = 3,
    initial_delay: float = 1.0,
    exponential_base: float = 2.0,
    catch_exceptions: tuple = (Exception,)
):
    """
    Retry function with exponential backoff.

    Args:
        func: Async function to retry
        max_retries: Maximum number of retry attempts
        initial_delay: Initial delay in seconds
        exponential_base: Base for exponential backoff
        catch_exceptions: Tuple of exceptions to catch and retry

    Returns:
        Function result if successful

    Raises:
        Last exception if all retries fail
    """
    delay = initial_delay
    last_exception = None

    for attempt in range(max_retries):
        try:
            return await func()
        except catch_exceptions as e:
            last_exception = e
            if attempt < max_retries - 1:
                await asyncio.sleep(delay)
                delay *= exponential_base

    raise last_exception


# ============================================================================
# Tools with Error Handling
# ============================================================================

@mcp.tool()
def divide_numbers(a: float, b: float) -> dict:
    """
    Divide two numbers with error handling.

    Args:
        a: Numerator
        b: Denominator

    Returns:
        Result or error
    """
    try:
        if b == 0:
            return create_error(
                ErrorCode.VALIDATION_ERROR,
                "Division by zero is not allowed",
                {"a": a, "b": b}
            )

        result = a / b
        return create_success(
            {"result": result, "a": a, "b": b},
            "Division successful"
        )

    except Exception as e:
        return create_error(
            ErrorCode.UNKNOWN_ERROR,
            f"Unexpected error: {str(e)}"
        )


@mcp.tool()
async def validated_operation(data: str, min_length: int = 1) -> dict:
    """
    Operation with input validation.

    Args:
        data: Input data to validate
        min_length: Minimum required length

    Returns:
        Processed result or validation error
    """
    # Validate input
    if not data:
        return create_error(
            ErrorCode.VALIDATION_ERROR,
            "Data is required",
            {"field": "data", "constraint": "not_empty"}
        )

    if len(data) < min_length:
        return create_error(
            ErrorCode.VALIDATION_ERROR,
            f"Data must be at least {min_length} characters",
            {"field": "data", "min_length": min_length, "actual_length": len(data)}
        )

    # Process data
    try:
        processed = data.upper()
        return create_success(
            {"original": data, "processed": processed},
            "Data processed successfully"
        )
    except Exception as e:
        return create_error(
            ErrorCode.UNKNOWN_ERROR,
            str(e)
        )


@mcp.tool()
async def resilient_api_call(url: str) -> dict:
    """
    API call with retry logic and comprehensive error handling.

    Args:
        url: URL to fetch

    Returns:
        API response or detailed error
    """
    async def make_request():
        async with httpx.AsyncClient(timeout=10) as client:
            response = await client.get(url)
            response.raise_for_status()
            return response.json()

    try:
        data = await retry_with_backoff(
            make_request,
            max_retries=3,
            catch_exceptions=(httpx.TimeoutException, httpx.NetworkError)
        )

        return create_success(
            data,
            "API call successful"
        )

    except httpx.TimeoutException:
        return create_error(
            ErrorCode.TIMEOUT,
            "Request timed out",
            {"url": url, "timeout_seconds": 10}
        )

    except httpx.NetworkError as e:
        return create_error(
            ErrorCode.NETWORK_ERROR,
            "Network error occurred",
            {"url": url, "error": str(e)}
        )

    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return create_error(
                ErrorCode.NOT_FOUND,
                "Resource not found",
                {"url": url, "status_code": 404}
            )
        elif e.response.status_code == 401:
            return create_error(
                ErrorCode.UNAUTHORIZED,
                "Unauthorized access",
                {"url": url, "status_code": 401}
            )
        elif e.response.status_code == 429:
            return create_error(
                ErrorCode.RATE_LIMITED,
                "Rate limit exceeded",
                {"url": url, "status_code": 429}
            )
        else:
            return create_error(
                ErrorCode.API_ERROR,
                f"HTTP {e.response.status_code}",
                {"url": url, "status_code": e.response.status_code}
            )

    except Exception as e:
        return create_error(
            ErrorCode.UNKNOWN_ERROR,
            f"Unexpected error: {str(e)}",
            {"url": url}
        )


@mcp.tool()
async def batch_with_error_recovery(items: list[str]) -> dict:
    """
    Batch process items with individual error recovery.

    Args:
        items: List of items to process

    Returns:
        Results with successes and failures tracked separately
    """
    results = []
    errors = []

    for i, item in enumerate(items):
        try:
            # Simulate processing
            if not item:
                raise ValueError("Empty item")

            await asyncio.sleep(0.1)

            results.append({
                "index": i,
                "item": item,
                "processed": item.upper(),
                "success": True
            })

        except ValueError as e:
            errors.append({
                "index": i,
                "item": item,
                "error": str(e),
                "code": ErrorCode.VALIDATION_ERROR.value
            })

        except Exception as e:
            errors.append({
                "index": i,
                "item": item,
                "error": str(e),
                "code": ErrorCode.UNKNOWN_ERROR.value
            })

    return {
        "success": len(errors) == 0,
        "total": len(items),
        "successful": len(results),
        "failed": len(errors),
        "results": results,
        "errors": errors,
        "timestamp": datetime.now().isoformat()
    }


@mcp.tool()
async def safe_database_operation(query: str) -> dict:
    """
    Simulated database operation with error handling.

    Args:
        query: SQL query (simulated)

    Returns:
        Query result or error
    """
    try:
        # Validate query
        if "DROP" in query.upper():
            return create_error(
                ErrorCode.UNAUTHORIZED,
                "DROP operations not allowed",
                {"query": query}
            )

        if not query.strip():
            return create_error(
                ErrorCode.VALIDATION_ERROR,
                "Query cannot be empty"
            )

        # Simulate query execution
        await asyncio.sleep(0.1)

        # Simulate success
        mock_data = [
            {"id": 1, "name": "Alice"},
            {"id": 2, "name": "Bob"}
        ]

        return create_success(
            {"rows": mock_data, "count": len(mock_data)},
            "Query executed successfully"
        )

    except Exception as e:
        return create_error(
            ErrorCode.API_ERROR,
            f"Database error: {str(e)}",
            {"query": query}
        )


# ============================================================================
# Resources with Error Handling
# ============================================================================

@mcp.resource("health://detailed")
async def detailed_health_check() -> dict:
    """Comprehensive health check with error tracking."""
    checks = {}
    errors = []

    # Check API connectivity
    try:
        async with httpx.AsyncClient(timeout=5) as client:
            response = await client.get("https://api.example.com/health")
            checks["api"] = {
                "status": "healthy",
                "status_code": response.status_code
            }
    except Exception as e:
        checks["api"] = {
            "status": "unhealthy",
            "error": str(e)
        }
        errors.append(f"API check failed: {e}")

    # Check system resources
    try:
        import psutil
        checks["system"] = {
            "status": "healthy",
            "cpu_percent": psutil.cpu_percent(),
            "memory_percent": psutil.virtual_memory().percent
        }
    except Exception as e:
        checks["system"] = {
            "status": "error",
            "error": str(e)
        }

    # Overall status
    all_healthy = all(
        check.get("status") == "healthy"
        for check in checks.values()
    )

    return {
        "status": "healthy" if all_healthy else "degraded",
        "checks": checks,
        "errors": errors if errors else None,
        "timestamp": datetime.now().isoformat()
    }


# ============================================================================
# Main
# ============================================================================

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

"""
FastMCP OpenAPI Integration Template
=====================================
Auto-generate MCP server from OpenAPI/Swagger specification.
"""

from fastmcp import FastMCP
from fastmcp.server.openapi import RouteMap, MCPType
import httpx
import os
from dotenv import load_dotenv

load_dotenv()

# ============================================================================
# Configuration
# ============================================================================

API_BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com")
API_KEY = os.getenv("API_KEY", "")
OPENAPI_SPEC_URL = os.getenv("OPENAPI_SPEC_URL", f"{API_BASE_URL}/openapi.json")

# ============================================================================
# Load OpenAPI Specification
# ============================================================================

def load_openapi_spec():
    """Load OpenAPI specification from URL or file."""
    try:
        # Try loading from URL
        response = httpx.get(OPENAPI_SPEC_URL, timeout=10)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"Failed to load OpenAPI spec from URL: {e}")

        # Fallback: try loading from local file
        try:
            import json
            with open("openapi.json") as f:
                return json.load(f)
        except FileNotFoundError:
            print("Error: OpenAPI spec not found. Please provide OPENAPI_SPEC_URL or openapi.json file")
            return None


spec = load_openapi_spec()

# ============================================================================
# Create Authenticated HTTP Client
# ============================================================================

client = httpx.AsyncClient(
    base_url=API_BASE_URL,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    timeout=httpx.Timeout(30.0),
    limits=httpx.Limits(
        max_keepalive_connections=5,
        max_connections=10
    )
)

# ============================================================================
# Define Route Mapping Strategy
# ============================================================================

route_maps = [
    # GET endpoints with path parameters → Resource Templates
    # Example: /users/{user_id} → resource template
    RouteMap(
        methods=["GET"],
        pattern=r".*\{.*\}.*",  # Has path parameters
        mcp_type=MCPType.RESOURCE_TEMPLATE
    ),

    # GET endpoints without parameters → Static Resources
    # Example: /users → static resource
    RouteMap(
        methods=["GET"],
        pattern=r"^(?!.*\{.*\}).*$",  # No path parameters
        mcp_type=MCPType.RESOURCE
    ),

    # POST/PUT/PATCH → Tools (create/update operations)
    RouteMap(
        methods=["POST", "PUT", "PATCH"],
        mcp_type=MCPType.TOOL
    ),

    # DELETE → Tools (delete operations)
    RouteMap(
        methods=["DELETE"],
        mcp_type=MCPType.TOOL
    ),

    # Exclude internal endpoints
    RouteMap(
        pattern=r"/internal/.*",
        mcp_type=MCPType.EXCLUDE
    ),

    # Exclude health checks
    RouteMap(
        pattern=r"/(health|healthz|readiness|liveness)",
        mcp_type=MCPType.EXCLUDE
    )
]

# ============================================================================
# Generate MCP Server from OpenAPI
# ============================================================================

if spec:
    mcp = FastMCP.from_openapi(
        openapi_spec=spec,
        client=client,
        name="API Integration Server",
        route_maps=route_maps
    )

    print(f"✅ Generated MCP server from OpenAPI spec")
    print(f"   Base URL: {API_BASE_URL}")

else:
    # Fallback: create empty server if spec not available
    mcp = FastMCP("API Integration Server")
    print("⚠️  Running without OpenAPI spec - please configure OPENAPI_SPEC_URL")


# ============================================================================
# Add Custom Tools (on top of auto-generated ones)
# ============================================================================

@mcp.tool()
async def process_api_response(data: dict, operation: str = "format") -> dict:
    """
    Process API response data.

    Custom tool to transform or analyze data from API endpoints.
    """
    if operation == "format":
        # Format the data nicely
        return {
            "formatted": True,
            "data": data,
            "count": len(data) if isinstance(data, (list, dict)) else 1
        }

    elif operation == "summarize":
        # Summarize the data
        if isinstance(data, list):
            return {
                "type": "list",
                "count": len(data),
                "sample": data[:3] if len(data) > 3 else data
            }
        elif isinstance(data, dict):
            return {
                "type": "dict",
                "keys": list(data.keys()),
                "size": len(data)
            }
        else:
            return {
                "type": type(data).__name__,
                "value": str(data)
            }

    return {"error": f"Unknown operation: {operation}"}


@mcp.tool()
async def batch_api_request(endpoints: list[str]) -> dict:
    """
    Make multiple API requests in parallel.

    Useful for gathering data from multiple endpoints efficiently.
    """
    import asyncio

    async def fetch_endpoint(endpoint: str):
        try:
            response = await client.get(endpoint)
            response.raise_for_status()
            return {
                "endpoint": endpoint,
                "success": True,
                "data": response.json()
            }
        except Exception as e:
            return {
                "endpoint": endpoint,
                "success": False,
                "error": str(e)
            }

    # Execute all requests in parallel
    results = await asyncio.gather(*[fetch_endpoint(ep) for ep in endpoints])

    successful = [r for r in results if r["success"]]
    failed = [r for r in results if not r["success"]]

    return {
        "total": len(endpoints),
        "successful": len(successful),
        "failed": len(failed),
        "results": results
    }


# ============================================================================
# Add Custom Resources
# ============================================================================

@mcp.resource("info://api-config")
def api_configuration() -> dict:
    """Get API configuration details."""
    return {
        "base_url": API_BASE_URL,
        "spec_url": OPENAPI_SPEC_URL,
        "authenticated": bool(API_KEY),
        "spec_loaded": spec is not None
    }


@mcp.resource("info://available-endpoints")
def list_available_endpoints() -> dict:
    """List all available API endpoints."""
    if not spec:
        return {"error": "OpenAPI spec not loaded"}

    endpoints = []

    for path, path_item in spec.get("paths", {}).items():
        for method in path_item.keys():
            if method.upper() in ["GET", "POST", "PUT", "PATCH", "DELETE"]:
                operation = path_item[method]
                endpoints.append({
                    "path": path,
                    "method": method.upper(),
                    "summary": operation.get("summary", ""),
                    "description": operation.get("description", "")
                })

    return {
        "total": len(endpoints),
        "endpoints": endpoints
    }


# ============================================================================
# Main Execution
# ============================================================================

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

"""
FastMCP Prompts Examples
=========================
Examples of pre-configured prompts for LLMs.
"""

from fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("Prompts Examples")

# ============================================================================
# Basic Prompts
# ============================================================================

@mcp.prompt("help")
def help_prompt() -> str:
    """Generate help text for the server."""
    return """
Welcome to the FastMCP Prompts Examples Server!

This server demonstrates various prompt patterns for LLM interactions.

Available Tools:
- search: Search for items in the database
- analyze: Analyze data and generate insights
- summarize: Create summaries of text content

Available Resources:
- info://status: Current server status
- data://config: Server configuration
- data://users: List of all users

How to Use:
1. Use the search tool to find items
2. Use the analyze tool to generate insights from data
3. Use the summarize tool to create concise summaries

For specific tasks, use the pre-configured prompts:
- /analyze: Analyze a topic in depth
- /report: Generate a comprehensive report
- /review: Review and provide feedback
"""


@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
    """Generate a prompt for analyzing a topic."""
    return f"""
Please analyze the following topic: {topic}

Consider the following aspects:
1. Current State: What is the current situation?
2. Challenges: What are the main challenges or issues?
3. Opportunities: What opportunities exist for improvement?
4. Data Points: What data supports your analysis?
5. Recommendations: What specific actions do you recommend?

Use the available tools to:
- Search for relevant data using the search tool
- Gather statistics and metrics
- Review related information

Provide a structured analysis with:
- Executive Summary
- Detailed Findings
- Data-Driven Insights
- Actionable Recommendations
"""


# ============================================================================
# Prompts with Parameters
# ============================================================================

@mcp.prompt("report")
def report_prompt(
    subject: str,
    timeframe: str = "last month",
    detail_level: str = "summary"
) -> str:
    """Generate a report prompt with parameters."""
    return f"""
Generate a comprehensive report on: {subject}

Timeframe: {timeframe}
Detail Level: {detail_level}

Report Structure:
1. Executive Summary
   - Key findings
   - Critical metrics
   - Main recommendations

2. Data Analysis
   - Quantitative metrics
   - Trend analysis
   - Comparative analysis

3. Insights
   - Patterns discovered
   - Anomalies identified
   - Correlations found

4. Recommendations
   - Short-term actions
   - Long-term strategies
   - Resource requirements

Please use the available tools to gather:
- Statistical data
- User information
- System metrics
- Historical trends

Format: {detail_level.upper()}
- "summary": High-level overview with key points
- "detailed": In-depth analysis with supporting data
- "comprehensive": Full analysis with all available data points
"""


@mcp.prompt("review")
def review_prompt(
    item_type: str,
    item_id: str,
    focus_areas: str = "all"
) -> str:
    """Generate a review prompt."""
    return f"""
Review the {item_type} (ID: {item_id})

Focus Areas: {focus_areas}

Review Criteria:
1. Quality Assessment
   - Overall quality rating
   - Strengths identified
   - Areas for improvement

2. Completeness
   - Required elements present
   - Missing components
   - Suggestions for additions

3. Consistency
   - Internal consistency
   - Alignment with standards
   - Conformance to guidelines

4. Performance
   - Efficiency metrics
   - Resource utilization
   - Optimization opportunities

5. Recommendations
   - Priority improvements
   - Nice-to-have enhancements
   - Long-term considerations

Please gather relevant data using available tools and provide a structured review.
"""


# ============================================================================
# Task-Specific Prompts
# ============================================================================

@mcp.prompt("summarize")
def summarize_prompt(content_type: str = "text") -> str:
    """Generate a summarization prompt."""
    return f"""
Create a comprehensive summary of the {content_type}.

Summary Guidelines:
1. Key Points
   - Extract the most important information
   - Identify main themes or topics
   - Highlight critical details

2. Structure
   - Opening: Context and overview
   - Body: Main points organized logically
   - Closing: Conclusions and implications

3. Audience Consideration
   - Write for clarity and understanding
   - Define technical terms if needed
   - Provide context where necessary

4. Length
   - Brief: 2-3 sentences
   - Standard: 1 paragraph
   - Detailed: 2-3 paragraphs

Output Format:
- Start with a one-sentence overview
- Follow with detailed points
- End with key takeaways

Use available tools to gather additional context if needed.
"""


@mcp.prompt("compare")
def compare_prompt(item1: str, item2: str, criteria: str = "general") -> str:
    """Generate a comparison prompt."""
    return f"""
Compare and contrast: {item1} vs {item2}

Comparison Criteria: {criteria}

Analysis Framework:
1. Similarities
   - Common features
   - Shared characteristics
   - Aligned goals or purposes

2. Differences
   - Unique features
   - Distinct characteristics
   - Divergent approaches

3. Strengths and Weaknesses
   - {item1} strengths
   - {item1} weaknesses
   - {item2} strengths
   - {item2} weaknesses

4. Use Cases
   - When to choose {item1}
   - When to choose {item2}
   - Situational recommendations

5. Conclusion
   - Overall assessment
   - Best fit scenarios
   - Decision factors

Please gather data using available tools and provide a balanced comparison.
"""


# ============================================================================
# Workflow Prompts
# ============================================================================

@mcp.prompt("troubleshoot")
def troubleshoot_prompt(problem_description: str) -> str:
    """Generate a troubleshooting prompt."""
    return f"""
Troubleshoot the following issue:

Problem: {problem_description}

Troubleshooting Process:
1. Problem Definition
   - Describe the issue clearly
   - Identify symptoms
   - Note when it started

2. Information Gathering
   - Use available tools to gather:
     * System status
     * Error logs
     * Configuration details
     * Recent changes

3. Analysis
   - Identify potential causes
   - Determine root cause
   - Assess impact

4. Solution Development
   - Propose solutions (short-term and long-term)
   - Evaluate each solution
   - Recommend best approach

5. Implementation Plan
   - Step-by-step resolution
   - Required resources
   - Expected timeline
   - Verification steps

6. Prevention
   - Preventive measures
   - Monitoring recommendations
   - Documentation needs

Please be systematic and thorough in your analysis.
"""


@mcp.prompt("plan")
def plan_prompt(objective: str, constraints: str = "none") -> str:
    """Generate a planning prompt."""
    return f"""
Create a detailed plan for: {objective}

Constraints: {constraints}

Planning Framework:
1. Objective Analysis
   - Clear definition of success
   - Key success criteria
   - Expected outcomes

2. Current State Assessment
   - Available resources
   - Existing capabilities
   - Known limitations

3. Strategy Development
   - Approach options
   - Recommended strategy
   - Rationale

4. Action Plan
   - Phase 1: Foundation
   - Phase 2: Implementation
   - Phase 3: Optimization

5. Resource Requirements
   - Personnel
   - Technology
   - Budget
   - Time

6. Risk Management
   - Identified risks
   - Mitigation strategies
   - Contingency plans

7. Success Metrics
   - KPIs to track
   - Measurement methods
   - Review milestones

Use available tools to gather supporting data and insights.
"""


# ============================================================================
# Main
# ============================================================================

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

[project]
name = "my-fastmcp-server"
version = "1.0.0"
description = "My FastMCP Server"
readme = "README.md"
requires-python = ">=3.10"
license = {text = "MIT"}
authors = [
    {name = "Your Name", email = "your.email@example.com"}
]
dependencies = [
    "fastmcp>=2.12.0",
    "httpx>=0.27.0",
    "python-dotenv>=1.0.0",
    "pydantic>=2.0.0",
    "psutil>=5.9.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.23.0",
    "ruff>=0.3.0",
    "mypy>=1.8.0",
]

[project.urls]
Homepage = "https://github.com/yourusername/my-fastmcp-server"
Repository = "https://github.com/yourusername/my-fastmcp-server"
Issues = "https://github.com/yourusername/my-fastmcp-server/issues"

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[tool.setuptools]
packages = ["src"]

[tool.setuptools.package-data]
src = ["py.typed"]

# Ruff Configuration
[tool.ruff]
line-length = 100
target-version = "py310"

[tool.ruff.lint]
select = [
    "E",  # pycodestyle errors
    "W",  # pycodestyle warnings
    "F",  # pyflakes
    "I",  # isort
    "B",  # flake8-bugbear
    "C4", # flake8-comprehensions
]
ignore = [
    "E501",  # line too long (handled by formatter)
    "B008",  # do not perform function calls in argument defaults
]

# Mypy Configuration
[tool.mypy]
python_version = "3.10"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = false
disallow_any_generics = false
check_untyped_defs = true

# Pytest Configuration
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
asyncio_mode = "auto"
addopts = "-v --tb=short"

# Coverage Configuration
[tool.coverage.run]
source = ["src"]
omit = ["*/tests/*", "*/test_*"]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise AssertionError",
    "raise NotImplementedError",
    "if __name__ == .__main__.:",
    "if TYPE_CHECKING:",
]

# FastMCP Core
fastmcp>=2.14.2

# Storage Backends (for production persistence)
py-key-value-aio>=0.1.0  # Memory, Disk, Redis, DynamoDB storage

# Encryption (for secure token storage)
cryptography>=42.0.0

# HTTP Client (for API integrations)
httpx>=0.27.0

# Environment Variables
python-dotenv>=1.0.0

# Validation (optional, for complex validation)
pydantic>=2.0.0

# System Monitoring (optional, for health checks)
psutil>=5.9.0

# Async Support
asyncio-extras>=1.3.2

# Optional Storage Backend Dependencies
# Uncomment as needed for specific backends:
# redis>=5.0.0  # For RedisStore
# boto3>=1.34.0  # For DynamoDB
# pymongo>=4.6.0  # For MongoDB
# elasticsearch>=8.12.0  # For Elasticsearch

# Development Dependencies (optional)
# Uncomment for development
# pytest>=8.0.0
# pytest-asyncio>=0.23.0
# ruff>=0.3.0
# mypy>=1.8.0

"""
FastMCP Resources Examples
===========================
Examples of static resources, dynamic resources, and resource templates.
"""

from fastmcp import FastMCP
from datetime import datetime
from typing import Dict, List, Any
import os

mcp = FastMCP("Resources Examples")

# ============================================================================
# Static Resources
# ============================================================================

@mcp.resource("data://config")
def get_config() -> dict:
    """Static configuration resource."""
    return {
        "version": "1.0.0",
        "environment": os.getenv("ENVIRONMENT", "development"),
        "features": ["search", "analytics", "notifications"],
        "limits": {
            "max_requests": 1000,
            "max_results": 100
        }
    }


@mcp.resource("info://server")
def server_info() -> dict:
    """Server metadata."""
    return {
        "name": "Resources Examples Server",
        "description": "Demonstrates various resource patterns",
        "version": "1.0.0",
        "capabilities": [
            "static_resources",
            "dynamic_resources",
            "resource_templates"
        ]
    }


# ============================================================================
# Dynamic Resources
# ============================================================================

@mcp.resource("info://status")
async def server_status() -> dict:
    """Dynamic status resource (updated on each read)."""
    import psutil

    return {
        "status": "operational",
        "timestamp": datetime.now().isoformat(),
        "uptime_seconds": 0,  # Would track actual uptime
        "system": {
            "cpu_percent": psutil.cpu_percent(interval=1),
            "memory_percent": psutil.virtual_memory().percent,
            "disk_percent": psutil.disk_usage('/').percent
        }
    }


@mcp.resource("data://statistics")
async def get_statistics() -> dict:
    """Dynamic statistics resource."""
    return {
        "timestamp": datetime.now().isoformat(),
        "total_requests": 1234,  # Would track actual requests
        "active_connections": 5,
        "cache_hit_rate": 0.87,
        "average_response_time_ms": 45.3
    }


# ============================================================================
# Resource Templates (with parameters)
# ============================================================================

@mcp.resource("user://{user_id}/profile")
async def get_user_profile(user_id: str) -> dict:
    """Get user profile by ID."""
    # In production, fetch from database
    return {
        "id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com",
        "created_at": "2024-01-01T00:00:00Z",
        "role": "user"
    }


@mcp.resource("user://{user_id}/posts")
async def get_user_posts(user_id: str) -> List[dict]:
    """Get posts for a specific user."""
    # In production, fetch from database
    return [
        {
            "id": 1,
            "user_id": user_id,
            "title": "First Post",
            "content": "Hello, world!",
            "created_at": "2024-01-01T00:00:00Z"
        },
        {
            "id": 2,
            "user_id": user_id,
            "title": "Second Post",
            "content": "Another post",
            "created_at": "2024-01-02T00:00:00Z"
        }
    ]


@mcp.resource("org://{org_id}/team/{team_id}/members")
async def get_team_members(org_id: str, team_id: str) -> List[dict]:
    """Get team members with org and team context."""
    # In production, fetch from database with filters
    return [
        {
            "id": 1,
            "name": "Alice",
            "role": "engineer",
            "org_id": org_id,
            "team_id": team_id
        },
        {
            "id": 2,
            "name": "Bob",
            "role": "designer",
            "org_id": org_id,
            "team_id": team_id
        }
    ]


@mcp.resource("api://{version}/config")
async def get_versioned_config(version: str) -> dict:
    """Get configuration for specific API version."""
    configs = {
        "v1": {
            "api_version": "v1",
            "endpoints": ["/users", "/posts"],
            "deprecated": True
        },
        "v2": {
            "api_version": "v2",
            "endpoints": ["/users", "/posts", "/comments", "/likes"],
            "deprecated": False
        }
    }

    return configs.get(version, {"error": f"Unknown version: {version}"})


# ============================================================================
# File-based Resources
# ============================================================================

@mcp.resource("file://docs/{filename}")
async def get_documentation(filename: str) -> dict:
    """Get documentation file content."""
    # In production, read actual files
    docs = {
        "getting-started.md": {
            "filename": "getting-started.md",
            "content": "# Getting Started\n\nWelcome to the docs!",
            "last_modified": "2024-01-01T00:00:00Z"
        },
        "api-reference.md": {
            "filename": "api-reference.md",
            "content": "# API Reference\n\nAPI documentation here.",
            "last_modified": "2024-01-02T00:00:00Z"
        }
    }

    return docs.get(filename, {"error": f"File not found: {filename}"})


# ============================================================================
# List-style Resources
# ============================================================================

@mcp.resource("data://users")
async def list_users() -> List[dict]:
    """List all users."""
    # In production, fetch from database
    return [
        {"id": "1", "name": "Alice", "email": "alice@example.com"},
        {"id": "2", "name": "Bob", "email": "bob@example.com"},
        {"id": "3", "name": "Charlie", "email": "charlie@example.com"}
    ]


@mcp.resource("data://categories")
def list_categories() -> List[str]:
    """List available categories."""
    return [
        "Technology",
        "Science",
        "Business",
        "Entertainment",
        "Sports"
    ]


# ============================================================================
# Main
# ============================================================================

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