Utilities and Types

class Image:
    def __init__(
        self,
        data: bytes | str,
        mime_type: str = "image/png"
    ):
        """
        Image helper for returning image data from tools.
        
        Parameters:
        - data: Image data as bytes or base64 string
        - mime_type: MIME type of the image
        """

class Audio:
    def __init__(
        self,
        data: bytes | str,
        mime_type: str = "audio/wav"
    ):
        """
        Audio helper for returning audio data from tools.
        
        Parameters:
        - data: Audio data as bytes or base64 string
        - mime_type: MIME type of the audio
        """

class File:
    def __init__(
        self,
        data: bytes | str,
        name: str,
        mime_type: str | None = None
    ):
        """
        File helper for returning file data from tools.
        
        Parameters:
        - data: File content as bytes or string
        - name: Filename
        - mime_type: MIME type of the file
        """

class FastMCPBaseModel:
    """Base Pydantic model for FastMCP with common configurations."""
    
    class Config:
        arbitrary_types_allowed = True
        extra = "forbid"
        validate_assignment = True

def get_cached_typeadapter(type_hint: type) -> TypeAdapter:
    """
    Get cached TypeAdapter for type validation.
    
    Parameters:
    - type_hint: Type to create adapter for
    
    Returns:
    Cached TypeAdapter instance
    """

class FastMCPComponent:
    """Base class for FastMCP components with lifecycle management."""
    
    def __init__(self, name: str | None = None):
        """
        Initialize component.
        
        Parameters:
        - name: Component name for identification
        """
    
    async def initialize(self) -> None:
        """Initialize component resources."""
    
    async def cleanup(self) -> None:
        """Clean up component resources."""
    
    def get_name(self) -> str:
        """Get component name."""

class Settings:
    """Main settings class for FastMCP configuration."""
    
    # Logging settings
    log_enabled: bool = True
    log_level: str = "INFO"
    enable_rich_tracebacks: bool = True
    
    # Development settings
    test_mode: bool = False
    debug_mode: bool = False
    
    # Feature flags
    deprecation_warnings: bool = True
    experimental_features: bool = False
    
    # Performance settings
    max_connections: int = 100
    request_timeout: float = 30.0
    
    # Security settings
    allow_dangerous_tools: bool = False
    require_auth: bool = False

class ExperimentalSettings:
    """Settings for experimental features."""
    
    enable_async_tools: bool = False
    enable_streaming: bool = False
    enable_caching: bool = False

class OpenAPIParser:
    """Parser for OpenAPI specifications."""
    
    def __init__(self, spec: dict | str):
        """
        Initialize OpenAPI parser.
        
        Parameters:
        - spec: OpenAPI specification as dict or JSON string
        """
    
    def parse_paths(self) -> list[HTTPRoute]:
        """Parse API paths into HTTPRoute objects."""
    
    def get_schemas(self) -> dict[str, dict]:
        """Get schema definitions from specification."""

class HTTPRoute:
    """HTTP route representation."""
    
    def __init__(
        self,
        path: str,
        method: str,
        operation_id: str | None = None,
        summary: str | None = None,
        description: str | None = None,
        parameters: list[dict] | None = None,
        request_body: dict | None = None,
        responses: dict | None = None
    ):
        """
        HTTP route definition.
        
        Parameters:
        - path: URL path with parameters
        - method: HTTP method (GET, POST, etc.)
        - operation_id: Unique operation identifier
        - summary: Brief operation summary
        - description: Detailed operation description
        - parameters: Request parameters
        - request_body: Request body schema
        - responses: Response schemas
        """

class MCPConfig:
    """MCP configuration format for server definitions."""
    
    def __init__(self, config: dict):
        """
        Initialize MCP configuration.
        
        Parameters:
        - config: Configuration dictionary
        """
    
    def get_servers(self) -> dict[str, dict]:
        """Get server configurations."""
    
    def validate(self) -> bool:
        """Validate configuration format."""

def create_test_server() -> FastMCP:
    """Create a FastMCP server configured for testing."""

def create_test_client(server: FastMCP) -> Client:
    """Create a client connected to test server via in-memory transport."""

async def assert_tool_exists(client: Client, tool_name: str) -> None:
    """Assert that a tool exists on the server."""

async def assert_resource_exists(client: Client, resource_uri: str) -> None:
    """Assert that a resource exists on the server."""

def generate_schema_from_function(func: Callable) -> dict:
    """
    Generate JSON schema from function signature.
    
    Parameters:
    - func: Function to analyze
    
    Returns:
    JSON schema dictionary
    """

def validate_against_schema(data: Any, schema: dict) -> bool:
    """
    Validate data against JSON schema.
    
    Parameters:
    - data: Data to validate
    - schema: JSON schema
    
    Returns:
    True if valid, False otherwise
    """

async def make_http_request(
    method: str,
    url: str,
    headers: dict | None = None,
    data: Any | None = None,
    timeout: float = 30.0
) -> dict:
    """
    Make HTTP request with error handling.
    
    Parameters:
    - method: HTTP method
    - url: Target URL
    - headers: Request headers
    - data: Request data
    - timeout: Request timeout
    
    Returns:
    Response data
    """

def parse_http_headers(headers: dict) -> dict[str, str]:
    """Parse and normalize HTTP headers."""

from fastmcp import FastMCP
from fastmcp.utilities.types import Image, Audio, File
import base64

mcp = FastMCP("Media Server")

@mcp.tool
def create_chart() -> Image:
    """Create a chart and return as image."""
    import matplotlib.pyplot as plt
    import io
    
    # Create chart
    plt.figure(figsize=(10, 6))
    plt.plot([1, 2, 3, 4], [1, 4, 2, 3])
    plt.title("Sample Chart")
    
    # Save to bytes
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)
    
    return Image(buffer.read(), mime_type="image/png")

@mcp.tool
def generate_audio() -> Audio:
    """Generate audio data."""
    import numpy as np
    import wave
    import io
    
    # Generate sine wave
    sample_rate = 44100
    duration = 1.0
    frequency = 440.0
    
    t = np.linspace(0, duration, int(sample_rate * duration))
    waveform = np.sin(2 * np.pi * frequency * t)
    
    # Convert to audio bytes
    buffer = io.BytesIO()
    with wave.open(buffer, 'wb') as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(sample_rate)
        wav_file.writeframes((waveform * 32767).astype(np.int16).tobytes())
    
    buffer.seek(0)
    return Audio(buffer.read(), mime_type="audio/wav")

@mcp.tool
def create_document(content: str, filename: str = "document.txt") -> File:
    """Create document file."""
    return File(
        data=content.encode('utf-8'),
        name=filename,
        mime_type="text/plain"
    )

@mcp.tool
def create_json_file(data: dict, filename: str = "data.json") -> File:
    """Create JSON file."""
    import json
    
    json_content = json.dumps(data, indent=2)
    
    return File(
        data=json_content,
        name=filename,
        mime_type="application/json"
    )

from fastmcp import FastMCP, Settings
from fastmcp.utilities.logging import configure_logging

# Create custom settings
settings = Settings()
settings.log_level = "DEBUG"
settings.debug_mode = True
settings.max_connections = 50
settings.request_timeout = 60.0
settings.allow_dangerous_tools = False

# Configure logging
configure_logging(
    level=settings.log_level,
    enable_rich_tracebacks=settings.enable_rich_tracebacks
)

# Create server with custom settings
mcp = FastMCP(
    name="Configured Server",
    settings=settings
)

@mcp.tool
def debug_info() -> dict:
    """Get debug information."""
    return {
        "debug_mode": settings.debug_mode,
        "log_level": settings.log_level,
        "max_connections": settings.max_connections,
        "request_timeout": settings.request_timeout
    }

mcp.run()

from fastmcp import FastMCP
from fastmcp.utilities.components import FastMCPComponent
import asyncio

class DatabaseComponent(FastMCPComponent):
    """Database connection component."""
    
    def __init__(self):
        super().__init__("database")
        self.connection = None
    
    async def initialize(self):
        """Initialize database connection."""
        # Mock database connection
        await asyncio.sleep(0.1)
        self.connection = {"status": "connected", "pool_size": 10}
        print(f"Component {self.get_name()} initialized")
    
    async def cleanup(self):
        """Clean up database connection."""
        if self.connection:
            self.connection = None
            print(f"Component {self.get_name()} cleaned up")
    
    def query(self, sql: str) -> list[dict]:
        """Execute database query."""
        if not self.connection:
            raise RuntimeError("Database not connected")
        
        # Mock query result
        return [{"id": 1, "name": "example", "sql": sql}]

class CacheComponent(FastMCPComponent):
    """Cache component."""
    
    def __init__(self):
        super().__init__("cache")
        self.cache = {}
    
    async def initialize(self):
        """Initialize cache."""
        self.cache = {}
        print(f"Component {self.get_name()} initialized")
    
    async def cleanup(self):
        """Clean up cache."""
        self.cache.clear()
        print(f"Component {self.get_name()} cleaned up")
    
    def get(self, key: str) -> Any:
        """Get value from cache."""
        return self.cache.get(key)
    
    def set(self, key: str, value: Any) -> None:
        """Set value in cache."""
        self.cache[key] = value

# Set up components
db_component = DatabaseComponent()
cache_component = CacheComponent()

mcp = FastMCP("Component Server")

@mcp.tool
async def query_with_cache(sql: str) -> list[dict]:
    """Query database with caching."""
    # Check cache first
    cached_result = cache_component.get(sql)
    if cached_result:
        return {"cached": True, "data": cached_result}
    
    # Query database
    result = db_component.query(sql)
    
    # Cache result
    cache_component.set(sql, result)
    
    return {"cached": False, "data": result}

# Initialize components before running server
async def main():
    await db_component.initialize()
    await cache_component.initialize()
    
    try:
        mcp.run()
    finally:
        await db_component.cleanup()
        await cache_component.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

from fastmcp import FastMCP
from fastmcp.utilities.openapi import OpenAPIParser, HTTPRoute

# Load OpenAPI specification
openapi_spec = {
    "openapi": "3.0.0",
    "info": {"title": "Example API", "version": "1.0.0"},
    "paths": {
        "/users": {
            "get": {
                "operationId": "getUsers",
                "summary": "Get all users",
                "responses": {
                    "200": {
                        "description": "List of users",
                        "content": {
                            "application/json": {
                                "schema": {
                                    "type": "array",
                                    "items": {"$ref": "#/components/schemas/User"}
                                }
                            }
                        }
                    }
                }
            },
            "post": {
                "operationId": "createUser",
                "summary": "Create new user",
                "requestBody": {
                    "content": {
                        "application/json": {
                            "schema": {"$ref": "#/components/schemas/User"}
                        }
                    }
                },
                "responses": {
                    "201": {"description": "User created"}
                }
            }
        }
    },
    "components": {
        "schemas": {
            "User": {
                "type": "object",
                "properties": {
                    "id": {"type": "integer"},
                    "name": {"type": "string"},
                    "email": {"type": "string"}
                }
            }
        }
    }
}

# Parse OpenAPI spec
parser = OpenAPIParser(openapi_spec)
routes = parser.parse_paths()
schemas = parser.get_schemas()

mcp = FastMCP("OpenAPI Server")

# Create tools from OpenAPI routes
for route in routes:
    if route.method == "GET" and route.path == "/users":
        @mcp.tool
        def get_users() -> list[dict]:
            """Get all users."""
            return [
                {"id": 1, "name": "Alice", "email": "alice@example.com"},
                {"id": 2, "name": "Bob", "email": "bob@example.com"}
            ]
    
    elif route.method == "POST" and route.path == "/users":
        @mcp.tool
        def create_user(name: str, email: str) -> dict:
            """Create new user."""
            return {
                "id": 123,
                "name": name,
                "email": email,
                "created": "2024-01-01T00:00:00Z"
            }

mcp.run()

from fastmcp import FastMCP, Client
from fastmcp.utilities.tests import create_test_client, assert_tool_exists
import asyncio

async def test_server():
    """Test FastMCP server functionality."""
    # Create test server
    mcp = FastMCP("Test Server")
    
    @mcp.tool
    def add(a: int, b: int) -> int:
        """Add two numbers."""
        return a + b
    
    @mcp.resource("config://test")
    def get_test_config():
        """Get test configuration."""
        return {"test": True, "env": "testing"}
    
    # Create test client
    client = create_test_client(mcp)
    
    async with client:
        # Test tool existence
        await assert_tool_exists(client, "add")
        
        # Test tool functionality
        result = await client.call_tool("add", {"a": 5, "b": 3})
        assert result.text == "8", f"Expected 8, got {result.text}"
        
        # Test resource
        await assert_resource_exists(client, "config://test")
        resource = await client.read_resource("config://test")
        assert "test" in resource.content
        
        print("All tests passed!")

if __name__ == "__main__":
    asyncio.run(test_server())

from fastmcp import FastMCP, Context
from fastmcp.utilities.types import Image
import json
import hashlib
from datetime import datetime
from typing import Any

def serialize_for_logging(obj: Any) -> str:
    """Serialize object for logging with proper handling of complex types."""
    if isinstance(obj, (str, int, float, bool, type(None))):
        return str(obj)
    elif isinstance(obj, (list, tuple)):
        return f"[{len(obj)} items]"
    elif isinstance(obj, dict):
        return f"{{keys: {list(obj.keys())[:5]}...}}"
    else:
        return f"<{type(obj).__name__}>"

def calculate_hash(data: str | bytes) -> str:
    """Calculate SHA-256 hash of data."""
    if isinstance(data, str):
        data = data.encode('utf-8')
    return hashlib.sha256(data).hexdigest()

def format_timestamp() -> str:
    """Get current timestamp in ISO format."""
    return datetime.utcnow().isoformat() + "Z"

mcp = FastMCP("Utility Demo Server")

@mcp.tool
async def process_data(
    data: dict,
    operation: str = "analyze",
    ctx: Context = None
) -> dict:
    """Process data with utility functions."""
    await ctx.info(f"Processing data: {serialize_for_logging(data)}")
    
    # Calculate data hash for integrity
    data_str = json.dumps(data, sort_keys=True)
    data_hash = calculate_hash(data_str)
    
    await ctx.info(f"Data hash: {data_hash}")
    
    # Process based on operation
    if operation == "analyze":
        result = {
            "analysis": f"Analyzed {len(data)} fields",
            "keys": list(data.keys())[:10],  # Limit for safety
            "field_types": {k: type(v).__name__ for k, v in data.items()}
        }
    elif operation == "transform":
        result = {
            "transformed": {k: str(v).upper() if isinstance(v, str) else v 
                          for k, v in data.items()},
            "transformation": "uppercase_strings"
        }
    else:
        result = {"error": f"Unknown operation: {operation}"}
    
    # Add metadata
    result.update({
        "data_hash": data_hash,
        "processed_at": format_timestamp(),
        "operation": operation,
        "status": "completed"
    })
    
    await ctx.info("Data processing completed")
    return result

@mcp.tool
def create_visualization(data: dict, chart_type: str = "bar") -> Image:
    """Create visualization from data."""
    import matplotlib.pyplot as plt
    import io
    
    # Extract data for plotting
    if isinstance(data, dict) and all(isinstance(v, (int, float)) for v in data.values()):
        keys = list(data.keys())
        values = list(data.values())
    else:
        # Fallback data
        keys = ["A", "B", "C", "D"]
        values = [1, 3, 2, 4]
    
    # Create chart
    plt.figure(figsize=(10, 6))
    
    if chart_type == "bar":
        plt.bar(keys, values)
    elif chart_type == "line":
        plt.plot(keys, values, marker='o')
    elif chart_type == "pie":
        plt.pie(values, labels=keys, autopct='%1.1f%%')
    else:
        plt.bar(keys, values)  # Default to bar
    
    plt.title(f"{chart_type.title()} Chart")
    plt.tight_layout()
    
    # Save to bytes
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png', dpi=150)
    buffer.seek(0)
    
    return Image(buffer.read(), mime_type="image/png")

mcp.run()

# Common type aliases
ToolFunction = Callable[..., Any]
ResourceFunction = Callable[..., str | bytes | dict]
PromptFunction = Callable[..., str | list[dict]]

# Transport types
TransportType = Literal["stdio", "sse", "http", "ws"]

# Log levels
LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

# Duplicate behavior options
DuplicateBehavior = Literal["error", "ignore", "replace"]

# Component types
ComponentFn = Callable[..., Any]