Tools System

class Tool:
    def __init__(
        self,
        name: str,
        description: str,
        func: Callable,
        schema: dict | None = None
    ):
        """
        Base tool class.
        
        Parameters:
        - name: Tool name for LLM reference
        - description: Tool description for LLM understanding
        - func: Python function to execute
        - schema: Optional custom JSON schema (auto-generated if None)
        """

class FunctionTool(Tool):
    """Tool implementation for function-based tools with automatic schema generation."""

class ToolManager:
    def add_tool(self, tool: Tool) -> None:
        """
        Add a tool to the manager.
        
        Parameters:
        - tool: Tool instance to add
        """
    
    def remove_tool(self, name: str) -> None:
        """
        Remove a tool by name.
        
        Parameters:
        - name: Name of tool to remove
        """
    
    def get_tool(self, name: str) -> Tool | None:
        """
        Get a tool by name.
        
        Parameters:
        - name: Name of tool to retrieve
        
        Returns:
        Tool instance or None if not found
        """
    
    def list_tools(self) -> list[Tool]:
        """
        List all registered tools.
        
        Returns:
        List of all tool instances
        """

def forward(
    tool_name: str,
    target_tool_name: str | None = None,
    transform_args: Callable | None = None,
    transform_result: Callable | None = None
) -> Callable:
    """
    Forward tool calls to another tool with optional transformations.
    
    Parameters:
    - tool_name: Name of tool to forward
    - target_tool_name: Target tool name (defaults to same name)
    - transform_args: Function to transform arguments
    - transform_result: Function to transform result
    
    Returns:
    Tool transformation function
    """

def forward_raw(
    tool_name: str,
    target_function: Callable,
    transform_args: Callable | None = None,
    transform_result: Callable | None = None
) -> Callable:
    """
    Forward tool calls to a raw function with optional transformations.
    
    Parameters:
    - tool_name: Name of tool to forward
    - target_function: Target function to call
    - transform_args: Function to transform arguments
    - transform_result: Function to transform result
    
    Returns:
    Tool transformation function
    """

from fastmcp import FastMCP

mcp = FastMCP("Math Server")

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@mcp.tool  
def divide(a: float, b: float) -> float:
    """Divide two numbers with error handling."""
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

@mcp.tool
def factorial(n: int) -> int:
    """Calculate factorial of a number."""
    if n < 0:
        raise ValueError("Factorial not defined for negative numbers")
    if n == 0 or n == 1:
        return 1
    result = 1
    for i in range(2, n + 1):
        result *= i
    return result

from fastmcp import FastMCP
from fastmcp.utilities.types import Image, File
from typing import List, Dict, Optional
from dataclasses import dataclass

mcp = FastMCP("Advanced Server")

@dataclass
class ProcessingResult:
    success: bool
    message: str
    data: Dict[str, any]

@mcp.tool
def process_data(
    items: List[str],
    options: Optional[Dict[str, str]] = None,
    batch_size: int = 10
) -> ProcessingResult:
    """
    Process a list of data items with configurable options.
    
    Parameters:
    - items: List of items to process
    - options: Optional processing configuration
    - batch_size: Number of items to process at once
    
    Returns:
    ProcessingResult with success status and processed data
    """
    if not items:
        return ProcessingResult(
            success=False,
            message="No items provided",
            data={}
        )
    
    processed = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        processed.extend([item.upper() for item in batch])
    
    return ProcessingResult(
        success=True,
        message=f"Processed {len(items)} items",
        data={"processed_items": processed}
    )

@mcp.tool
def create_chart(
    data: Dict[str, List[float]], 
    title: str = "Chart"
) -> Image:
    """
    Create a chart from data and return as image.
    
    Parameters:
    - data: Dictionary with series names as keys and data points as values
    - title: Chart title
    
    Returns:
    Chart image as PNG
    """
    # This would use matplotlib or similar to create a chart
    import matplotlib.pyplot as plt
    import io
    
    plt.figure(figsize=(10, 6))
    for series_name, values in data.items():
        plt.plot(values, label=series_name)
    
    plt.title(title)
    plt.legend()
    plt.grid(True)
    
    # Save to bytes
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)
    
    return Image(buffer.read(), mime_type="image/png")

from fastmcp import FastMCP, Context
import httpx

mcp = FastMCP("API Server")

@mcp.tool
async def fetch_and_analyze(
    url: str,
    analysis_prompt: str,
    ctx: Context
) -> str:
    """
    Fetch data from URL and analyze it using LLM.
    
    Parameters:
    - url: URL to fetch data from
    - analysis_prompt: Prompt for LLM analysis
    - ctx: Execution context for capabilities
    
    Returns:
    Analysis result from LLM
    """
    # Log the operation
    await ctx.info(f"Fetching data from {url}")
    
    # Fetch data using HTTP request
    response = await ctx.http_request("GET", url)
    
    if response.status_code != 200:
        await ctx.error(f"Failed to fetch data: {response.status_code}")
        return f"Error: Unable to fetch data from {url}"
    
    data = response.text
    await ctx.info(f"Fetched {len(data)} characters of data")
    
    # Use LLM sampling for analysis
    messages = [
        {
            "role": "user", 
            "content": f"{analysis_prompt}\n\nData to analyze:\n{data[:1000]}..."
        }
    ]
    
    # Report progress
    await ctx.report_progress(50, 100)
    
    analysis = await ctx.sample(messages)
    
    await ctx.report_progress(100, 100)
    
    return analysis.text

@mcp.tool
async def multi_step_process(
    input_data: str,
    ctx: Context
) -> Dict[str, str]:
    """
    Perform multi-step processing with progress reporting.
    
    Parameters:
    - input_data: Data to process
    - ctx: Execution context
    
    Returns:
    Dictionary with processing results
    """
    results = {}
    total_steps = 4
    
    # Step 1: Validation
    await ctx.info("Step 1: Validating input")
    await ctx.report_progress(1, total_steps)
    if not input_data.strip():
        raise ValueError("Input data cannot be empty")
    results["validation"] = "passed"
    
    # Step 2: Processing
    await ctx.info("Step 2: Processing data")
    await ctx.report_progress(2, total_steps)
    processed = input_data.upper().strip()
    results["processed"] = processed
    
    # Step 3: Analysis
    await ctx.info("Step 3: Analyzing results")
    await ctx.report_progress(3, total_steps)
    analysis = f"Data contains {len(processed)} characters"
    results["analysis"] = analysis
    
    # Step 4: Completion
    await ctx.info("Step 4: Finalizing results")
    await ctx.report_progress(4, total_steps)
    results["status"] = "complete"
    
    await ctx.info("Multi-step process completed successfully")
    
    return results

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("Robust Server")

@mcp.tool  
def safe_divide(a: float, b: float) -> float:
    """
    Safely divide two numbers with proper error handling.
    
    Parameters:
    - a: Dividend
    - b: Divisor
    
    Returns:
    Result of division
    
    Raises:
    ToolError: If division by zero or invalid input
    """
    try:
        if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
            raise ToolError("Both arguments must be numbers")
        
        if b == 0:
            raise ToolError("Division by zero is not allowed")
        
        result = a / b
        
        # Check for infinite or NaN results
        if not isinstance(result, (int, float)) or result != result:  # NaN check
            raise ToolError("Division resulted in invalid number")
        
        return result
        
    except Exception as e:
        if isinstance(e, ToolError):
            raise
        raise ToolError(f"Unexpected error during division: {str(e)}")

@mcp.tool
def validate_and_process(data: Dict[str, any]) -> Dict[str, any]:
    """
    Validate and process input data with comprehensive error handling.
    
    Parameters:
    - data: Input data dictionary
    
    Returns:
    Processed data dictionary
    """
    if not isinstance(data, dict):
        raise ToolError("Input must be a dictionary")
    
    required_fields = ["name", "value"]
    missing_fields = [field for field in required_fields if field not in data]
    
    if missing_fields:
        raise ToolError(f"Missing required fields: {', '.join(missing_fields)}")
    
    # Validate field types 
    if not isinstance(data["name"], str):
        raise ToolError("Field 'name' must be a string")
    
    if not isinstance(data["value"], (int, float)):
        raise ToolError("Field 'value' must be a number")
    
    # Process the data
    return {
        "processed_name": data["name"].strip().title(),
        "processed_value": round(data["value"], 2),
        "timestamp": "2024-01-01T00:00:00Z",  # Would use real timestamp
        "status": "processed"
    }

from fastmcp import FastMCP
from fastmcp.tools import forward, forward_raw

mcp = FastMCP("Transform Server")

# Original tools
@mcp.tool
def original_add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

@mcp.tool 
def original_multiply(x: float, y: float) -> float:
    """Multiply two floats.""" 
    return x * y

# Transform arguments: convert strings to numbers
def string_to_number_transform(args):
    """Transform string arguments to numbers."""
    return {
        key: float(value) if isinstance(value, str) and value.replace('.', '').isdigit() else value
        for key, value in args.items()
    }

# Transform result: add metadata
def add_metadata_transform(result):
    """Add metadata to result."""
    return {
        "result": result,
        "type": type(result).__name__,
        "timestamp": "2024-01-01T00:00:00Z"
    }

# Apply transformations
mcp.add_tool_transformation(forward(
    "string_add",
    target_tool_name="original_add", 
    transform_args=string_to_number_transform,
    transform_result=add_metadata_transform
))

# Forward to external function
def external_power(base: float, exponent: float) -> float:
    """Calculate base raised to exponent."""
    return base ** exponent

mcp.add_tool_transformation(forward_raw(
    "power",
    target_function=external_power,
    transform_result=lambda result: {"power_result": result}
))

# Simple types
@mcp.tool
def get_string() -> str:
    return "Hello, world!"

@mcp.tool  
def get_number() -> float:
    return 42.5

@mcp.tool
def get_boolean() -> bool:
    return True

# Complex types
@mcp.tool
def get_dict() -> Dict[str, any]:
    return {"key": "value", "count": 10}

@mcp.tool
def get_list() -> List[str]:
    return ["item1", "item2", "item3"]

# Media types
@mcp.tool
def get_image() -> Image:
    # Return image data
    with open("chart.png", "rb") as f:
        return Image(f.read(), mime_type="image/png")

@mcp.tool  
def get_file() -> File:
    # Return file data
    return File(
        data="file content",
        name="output.txt",
        mime_type="text/plain"
    )