Context and Dependencies

class Context:
    async def info(self, message: str) -> None:
        """
        Log an info message to the client.
        
        Parameters:
        - message: Info message to log
        """
    
    async def error(self, message: str) -> None:
        """
        Log an error message to the client.
        
        Parameters:
        - message: Error message to log
        """
    
    async def debug(self, message: str) -> None:
        """
        Log a debug message to the client.
        
        Parameters:
        - message: Debug message to log
        """
    
    async def warning(self, message: str) -> None:
        """
        Log a warning message to the client.
        
        Parameters:
        - message: Warning message to log
        """

async def sample(
    self,
    messages: list[dict],
    params: dict | None = None
) -> SamplingResult:
    """
    Request LLM completion from the client.
    
    Parameters:
    - messages: List of message objects for the LLM
    - params: Optional sampling parameters (temperature, max_tokens, etc.)
    
    Returns:
    Sampling result with generated text and metadata
    """

async def read_resource(self, uri: str) -> ResourceResult:
    """
    Read a resource from the server.
    
    Parameters:
    - uri: Resource URI to read
    
    Returns:
    Resource content and metadata
    """

async def http_request(
    self,
    method: str,
    url: str,
    headers: dict | None = None,
    data: Any | None = None,
    json: dict | None = None,
    params: dict | None = None,
    timeout: float = 30.0
) -> HttpResponse:
    """
    Make HTTP request to external service.
    
    Parameters:
    - method: HTTP method (GET, POST, PUT, DELETE, etc.)
    - url: Target URL
    - headers: Optional request headers
    - data: Request body as bytes/string
    - json: Request body as JSON object
    - params: URL query parameters
    - timeout: Request timeout in seconds
    
    Returns:
    HTTP response with status, headers, and content
    """

async def report_progress(
    self,
    progress: int,
    total: int | None = None
) -> None:
    """
    Report progress to the client.
    
    Parameters:
    - progress: Current progress value
    - total: Total progress value (optional)
    """

def get_context() -> Context:
    """
    Get the current execution context.
    
    Returns:
    Current Context instance
    """

def get_http_request() -> HttpRequest:
    """
    Get the current HTTP request object (for HTTP transport).
    
    Returns:
    Current HTTP request
    """

def get_http_headers() -> dict[str, str]:
    """
    Get the current HTTP request headers (for HTTP transport).
    
    Returns:
    Dictionary of HTTP headers
    """

def get_access_token() -> AccessToken | None:
    """
    Get the current access token (if authenticated).
    
    Returns:
    Access token or None if not authenticated
    """

class AccessToken:
    token: str
    token_type: str = "bearer"
    expires_at: datetime | None = None
    scope: list[str] | None = None

from fastmcp import FastMCP, Context

mcp = FastMCP("Context Demo Server")

@mcp.tool
async def analyze_data(data: str, ctx: Context) -> str:
    """Analyze data with context logging and progress."""
    await ctx.info("Starting data analysis")
    
    # Report initial progress
    await ctx.report_progress(0, 100)
    
    # Simulate processing steps
    steps = ["validation", "parsing", "analysis", "formatting"]
    
    for i, step in enumerate(steps):
        await ctx.info(f"Performing {step}")
        
        # Simulate work
        import asyncio
        await asyncio.sleep(0.1)
        
        # Report progress
        progress = int((i + 1) / len(steps) * 100)
        await ctx.report_progress(progress, 100)
    
    await ctx.info("Data analysis completed")
    return f"Analysis complete: {len(data)} characters processed"

@mcp.tool
async def safe_operation(input_data: str, ctx: Context) -> str:
    """Perform operation with error handling and logging."""
    try:
        await ctx.info("Starting safe operation")
        
        if not input_data:
            await ctx.warning("Empty input data provided")
            return "No data to process"
        
        # Process data
        result = input_data.upper().strip()
        await ctx.info(f"Processed {len(input_data)} characters")
        
        return result
        
    except Exception as e:
        await ctx.error(f"Operation failed: {str(e)}")
        raise

from fastmcp import FastMCP, Context

mcp = FastMCP("LLM Integration Server")

@mcp.tool
async def intelligent_summary(
    text: str,
    max_words: int = 100,
    ctx: Context = None
) -> str:
    """Generate intelligent summary using client's LLM."""
    await ctx.info(f"Generating summary of {len(text)} character text")
    
    # Prepare messages for LLM
    messages = [
        {
            "role": "system",
            "content": f"You are a professional summarizer. Create concise summaries in exactly {max_words} words or less."
        },
        {
            "role": "user", 
            "content": f"Please summarize the following text:\n\n{text}"
        }
    ]
    
    # Request completion from client's LLM
    result = await ctx.sample(
        messages=messages,
        params={
            "temperature": 0.3,
            "max_tokens": max_words * 2  # Buffer for token estimation
        }
    )
    
    await ctx.info("Summary generated successfully")
    return result.text

@mcp.tool
async def code_explanation(
    code: str,
    language: str,
    ctx: Context
) -> str:
    """Explain code using LLM sampling."""
    await ctx.info(f"Explaining {language} code")
    
    messages = [
        {
            "role": "system",
            "content": f"You are a programming expert. Explain {language} code in clear, simple terms."
        },
        {
            "role": "user",
            "content": f"Explain this {language} code:\n\n```{language}\n{code}\n```"
        }
    ]
    
    explanation = await ctx.sample(messages)
    await ctx.info("Code explanation generated")
    
    return explanation.text

@mcp.tool
async def creative_content(
    prompt: str,
    style: str = "professional",
    ctx: Context = None
) -> str:
    """Generate creative content with LLM."""
    await ctx.info(f"Generating {style} content")
    
    messages = [
        {
            "role": "system",
            "content": f"You are a {style} content creator. Write engaging, high-quality content."
        },
        {
            "role": "user",
            "content": prompt
        }
    ]
    
    # Use sampling parameters for creativity
    params = {
        "temperature": 0.8 if style == "creative" else 0.5,
        "max_tokens": 1000
    }
    
    content = await ctx.sample(messages, params)
    await ctx.info("Content generated successfully")
    
    return content.text

from fastmcp import FastMCP, Context
import json

mcp = FastMCP("API Integration Server")

@mcp.tool
async def fetch_weather(city: str, ctx: Context) -> dict:
    """Fetch weather data from external API."""
    await ctx.info(f"Fetching weather for {city}")
    
    # Make HTTP request to weather API
    response = await ctx.http_request(
        method="GET",
        url="https://api.weather.com/v1/current",
        params={"city": city, "units": "metric"},
        headers={
            "Authorization": "Bearer weather-api-key",
            "User-Agent": "FastMCP-Weather/1.0"
        },
        timeout=10.0
    )
    
    if response.status_code == 200:
        weather_data = response.json()
        await ctx.info("Weather data retrieved successfully")
        
        return {
            "city": city,
            "temperature": weather_data.get("temp"),
            "condition": weather_data.get("condition"),
            "humidity": weather_data.get("humidity"),
            "source": "weather.com"
        }
    else:
        await ctx.error(f"Weather API error: {response.status_code}")
        return {"error": "Failed to fetch weather data"}

@mcp.tool
async def post_data(
    endpoint: str,
    data: dict,
    ctx: Context
) -> dict:
    """Post data to external endpoint."""
    await ctx.info(f"Posting data to {endpoint}")
    
    response = await ctx.http_request(
        method="POST",
        url=endpoint,
        json=data,
        headers={
            "Content-Type": "application/json",
            "Authorization": "Bearer api-token"
        }
    )
    
    if response.status_code in [200, 201]:
        await ctx.info("Data posted successfully")
        return {
            "success": True,
            "response": response.json(),
            "status_code": response.status_code
        }
    else:
        await ctx.error(f"POST failed: {response.status_code}")
        return {
            "success": False,
            "error": response.text,
            "status_code": response.status_code
        }

@mcp.tool
async def aggregate_apis(
    endpoints: list[str],
    ctx: Context
) -> dict:
    """Aggregate data from multiple API endpoints."""
    await ctx.info(f"Aggregating data from {len(endpoints)} endpoints")
    
    results = {}
    total_endpoints = len(endpoints)
    
    for i, endpoint in enumerate(endpoints):
        await ctx.info(f"Fetching from endpoint {i+1}/{total_endpoints}")
        await ctx.report_progress(i, total_endpoints)
        
        try:
            response = await ctx.http_request("GET", endpoint, timeout=5.0)
            
            if response.status_code == 200:
                results[endpoint] = {
                    "success": True,
                    "data": response.json()
                }
            else:
                results[endpoint] = {
                    "success": False,
                    "error": f"HTTP {response.status_code}"
                }
                
        except Exception as e:
            await ctx.warning(f"Failed to fetch from {endpoint}: {str(e)}")
            results[endpoint] = {
                "success": False,
                "error": str(e)
            }
    
    await ctx.report_progress(total_endpoints, total_endpoints)
    await ctx.info("API aggregation completed")
    
    # Summary statistics
    successful = sum(1 for r in results.values() if r["success"])
    
    return {
        "summary": {
            "total_endpoints": total_endpoints,
            "successful": successful,
            "failed": total_endpoints - successful
        },
        "results": results
    }

from fastmcp import FastMCP, Context

mcp = FastMCP("Resource Integration Server")

@mcp.tool
async def process_config(setting_name: str, ctx: Context) -> dict:
    """Process configuration using resource access."""
    await ctx.info(f"Processing configuration: {setting_name}")
    
    # Read configuration resource
    config_resource = await ctx.read_resource("config://settings")
    config_data = json.loads(config_resource.content)
    
    if setting_name not in config_data:
        await ctx.warning(f"Setting '{setting_name}' not found in configuration")
        return {"error": f"Setting '{setting_name}' not found"}
    
    setting_value = config_data[setting_name]
    await ctx.info(f"Retrieved setting: {setting_name} = {setting_value}")
    
    return {
        "setting": setting_name,
        "value": setting_value,
        "type": type(setting_value).__name__
    }

@mcp.tool
async def compile_report(report_type: str, ctx: Context) -> str:
    """Compile report using multiple resources."""
    await ctx.info(f"Compiling {report_type} report")
    
    # Read multiple resources
    resources_to_read = [
        "data://users/summary",
        "data://orders/recent", 
        "config://report_settings"
    ]
    
    report_data = {}
    
    for i, resource_uri in enumerate(resources_to_read):
        await ctx.info(f"Reading resource: {resource_uri}")
        await ctx.report_progress(i, len(resources_to_read))
        
        try:
            resource = await ctx.read_resource(resource_uri)  
            report_data[resource_uri] = json.loads(resource.content)
        except Exception as e:
            await ctx.warning(f"Failed to read {resource_uri}: {str(e)}")
            report_data[resource_uri] = {"error": str(e)}
    
    await ctx.report_progress(len(resources_to_read), len(resources_to_read))
    
    # Compile report
    report = f"# {report_type.title()} Report\n\n"
    
    for uri, data in report_data.items():
        if "error" in data:
            report += f"## {uri}\nError: {data['error']}\n\n"
        else:
            report += f"## {uri}\n{json.dumps(data, indent=2)}\n\n"
    
    await ctx.info("Report compilation completed")
    return report

from fastmcp import FastMCP, Context
from fastmcp.server.dependencies import get_access_token, get_http_headers

mcp = FastMCP("Secure Server")

@mcp.tool
async def secure_operation(data: str, ctx: Context) -> dict:
    """Perform secure operation with authentication info."""
    await ctx.info("Starting secure operation")
    
    # Get access token
    token = get_access_token()
    if not token:
        await ctx.error("No access token available")
        return {"error": "Authentication required"}
    
    await ctx.info(f"Authenticated with token type: {token.token_type}")
    
    # Get HTTP headers if available
    try:
        headers = get_http_headers()
        user_agent = headers.get("user-agent", "unknown")
        await ctx.info(f"Request from: {user_agent}")
    except:
        await ctx.debug("No HTTP headers available (not HTTP transport)")
    
    # Process data securely
    result = {
        "processed_data": data.upper(),
        "user_info": {
            "token_type": token.token_type,
            "has_scope": bool(token.scope),
            "expires_at": token.expires_at.isoformat() if token.expires_at else None
        },
        "timestamp": "2024-01-01T00:00:00Z"
    }
    
    await ctx.info("Secure operation completed")
    return result

@mcp.tool
async def user_specific_action(action: str, ctx: Context) -> str:
    """Perform action based on user authentication."""
    token = get_access_token()
    
    if not token:
        await ctx.error("Authentication required")
        return "Error: Please authenticate to perform this action"
    
    # Check token scopes
    required_scope = f"{action}_access"
    if token.scope and required_scope not in token.scope:
        await ctx.warning(f"Insufficient permissions for action: {action}")
        return f"Error: Missing required scope '{required_scope}'"
    
    await ctx.info(f"Performing {action} for authenticated user")
    
    # Perform the action
    return f"Successfully performed {action} with proper authentication"

class SamplingResult:
    """Result from LLM sampling request."""
    text: str
    finish_reason: str | None
    usage: dict | None

class HttpResponse:
    """Result from HTTP request."""
    status_code: int
    headers: dict[str, str]
    text: str  
    content: bytes
    
    def json(self) -> dict:
        """Parse response as JSON."""