FastMCP: Die pythonische Methode zum Erstellen von MCP-Servern und -Purchasers

python -c "from fastmcp import FastMCP; print('FastMCP put in efficiently')"

import logging
import sys
from typing import Dict
from fastmcp import FastMCP

# Configure logging to stderr (vital for MCP protocol integrity)
logging.basicConfig(
    stage=logging.DEBUG,
    format="%(asctime)s - %(identify)s - %(levelname)s - %(message)s",
    stream=sys.stderr
)
logger = logging.getLogger(__name__)

# Create the FastMCP server occasion
mcp = FastMCP(identify="CalculatorServer")

@mcp.device
def add(a: float, b: float) -> float:
    """
    Add two numbers collectively.
   
    Args:
        a: First quantity
        b: Second quantity
       
    Returns:
        Sum of a and b
    """
    attempt:
        end result = a + b
        logger.data(f"Addition carried out: {a} + {b} = {end result}")
        return end result
    besides TypeError as e:
        logger.error(f"Sort error in add: {e}")
        elevate ValueError(f"Invalid enter sorts: {e}")

@mcp.device
def subtract(a: float, b: float) -> float:
    """
    Subtract b from a.
   
    Args:
        a: First quantity (minuend)
        b: Second quantity (subtrahend)
       
    Returns:
        Distinction of a and b
    """
    attempt:
        end result = a - b
        logger.data(f"Subtraction carried out: {a} - {b} = {end result}")
        return end result
    besides TypeError as e:
        logger.error(f"Sort error in subtract: {e}")
        elevate ValueError(f"Invalid enter sorts: {e}")

@mcp.device
def multiply(a: float, b: float) -> float:
    """
    Multiply two numbers.
   
    Args:
        a: First quantity
        b: Second quantity
       
    Returns:
        Product of a and b
    """
    attempt:
        end result = a * b
        logger.data(f"Multiplication carried out: {a} * {b} = {end result}")
        return end result
    besides TypeError as e:
        logger.error(f"Sort error in multiply: {e}")
        elevate ValueError(f"Invalid enter sorts: {e}")

@mcp.device
def divide(a: float, b: float) -> float:
    """
    Divide a by b.
   
    Args:
        a: Dividend (numerator)
        b: Divisor (denominator)
       
    Returns:
        Quotient of a divided by b
       
    Raises:
        ValueError: If making an attempt to divide by zero
    """
    attempt:
        if b == 0:
            logger.warning(f"Division by zero tried: {a} / {b}")
            elevate ValueError("Can't divide by zero")
       
        end result = a / b
        logger.data(f"Division carried out: {a} / {b} = {end result}")
        return end result
    besides (TypeError, ZeroDivisionError) as e:
        logger.error(f"Error in divide: {e}")
        elevate ValueError(f"Division error: {e}")

@mcp.useful resource("config://calculator/settings")
def get_settings() -> Dict:
    """
    Offers calculator configuration and obtainable operations.
   
    Returns:
        Dictionary containing calculator settings and metadata
    """
    logger.debug("Fetching calculator settings")
   
    return {
        "model": "1.0.0",
        "operations": ("add", "subtract", "multiply", "divide"),
        "precision": "IEEE 754 double precision",
        "max_value": 1.7976931348623157e+308,
        "min_value": -1.7976931348623157e+308,
        "supports_negative": True,
        "supports_decimals": True
    }

@mcp.useful resource("docs://calculator/information")
def get_guide() -> str:
    """
    Offers a person information for the calculator server.
   
    Returns:
        String containing utilization information and examples
    """
    logger.debug("Retrieving calculator information")

    information = """
       
    1. **add(a, b)**: Returns a + b
       Instance: add(5, 3) = 8
   
    2. **subtract(a, b)**: Returns a - b
       Instance: subtract(10, 4) = 6
   
    3. **multiply(a, b)**: Returns a * b
       Instance: multiply(7, 6) = 42
   
    4. **divide(a, b)**: Returns a / b
       Instance: divide(20, 4) = 5.0
   
    ## Error Dealing with
   
    - Division by zero will elevate a ValueError
    - Non-numeric inputs will elevate a ValueError
    - All inputs ought to be legitimate numbers (int or float)
   
    ## Precision
   
    The calculator makes use of IEEE 754 double precision floating-point arithmetic.
    Outcomes might include minor rounding errors for some operations.
    """
   
    return information

@mcp.immediate
def calculate_expression(expression: str) -> str:
    """
    Offers directions for evaluating a mathematical expression.
    Args:
        expression: A mathematical expression to guage        
    Returns:
        Formatted immediate instructing the LLM methods to consider the expression
    """
    logger.debug(f"Producing calculation immediate for: {expression}")

    immediate = f"""
    Please consider the next mathematical expression step-by-step:
   
    Expression: {expression}
   
    Directions:
    1. Break down the expression into particular person operations
    2. Use the suitable calculator device for every operation
    3. Comply with order of operations (parentheses, multiplication/division, addition/subtraction)
    4. Present all intermediate steps
    5. Present the ultimate end result
   
    Obtainable instruments: add, subtract, multiply, divide
    """
   
    return immediate.strip()

if __name__ == "__main__":
    logger.data("Beginning Calculator MCP Server...")
   
    attempt:
        # Run the server with stdio transport (default for Claude Desktop)
        mcp.run(transport="stdio")
    besides KeyboardInterrupt:
        logger.data("Server interrupted by person")
        sys.exit(0)
    besides Exception as e:
        logger.error(f"Deadly error: {e}", exc_info=True)
        sys.exit(1)

import asyncio
import logging
import sys
from typing import Any
from fastmcp import Consumer, FastMCP

logging.basicConfig(
    stage=logging.INFO,
    format="%(asctime)s - %(identify)s - %(levelname)s - %(message)s",
    stream=sys.stderr
)
logger = logging.getLogger(__name__)

async def primary():
    """
    Essential shopper operate demonstrating server interplay.
    """
   
    from calculator_server import mcp as server
   
    logger.data("Initializing Calculator Consumer...")

    attempt:
        async with Consumer(server) as shopper:
            logger.data("✓ Linked to Calculator Server")
           
            # DISCOVER CAPABILITIEs            
            print("n" + "="*60)
            print("1. DISCOVERING SERVER CAPABILITIES")
            print("="*60)
           
            # Listing obtainable instruments
            instruments = await shopper.list_tools()
            print(f"nAvailable Instruments ({len(instruments)}):")
            for device in instruments:
                print(f"  • {device.identify}: {device.description}")
           
            # Listing obtainable assets
            assets = await shopper.list_resources()
            print(f"nAvailable Sources ({len(assets)}):")
            for useful resource in assets:
                print(f"  • {useful resource.uri}: {useful resource.identify or useful resource.uri}")
           
            # Listing obtainable prompts
            prompts = await shopper.list_prompts()
            print(f"nAvailable Prompts ({len(prompts)}):")
            for immediate in prompts:
                print(f"  • {immediate.identify}: {immediate.description}")
           
            # CALL TOOLS
           
            print("n" + "="*60)
            print("2. CALLING TOOLS")
            print("="*60)
           
            # Easy addition
            print("nTest 1: Including 15 + 27")
            end result = await shopper.call_tool("add", {"a": 15, "b": 27})
            result_value = extract_tool_result(end result)
            print(f"  Consequence: 15 + 27 = {result_value}")
           
            # Division with error dealing with
            print("nTest 2: Dividing 100 / 5")
            end result = await shopper.call_tool("divide", {"a": 100, "b": 5})
            result_value = extract_tool_result(end result)
            print(f"  Consequence: 100 / 5 = {result_value}")
           
            # Error case: division by zero
            print("nTest 3: Division by Zero (Error Dealing with)")
            attempt:
                end result = await shopper.call_tool("divide", {"a": 10, "b": 0})
                print(f"  Sudden success: {end result}")
            besides Exception as e:
                print(f"  ✓ Error caught appropriately: {str(e)}")
           
            # READ RESOURCES
            print("n" + "="*60)
            print("3. READING RESOURCES")
            print("="*60)
           
            # Learn settings useful resource
            print("nFetching Calculator Settings...")
            settings_resource = await shopper.read_resource("config://calculator/settings")
            print(f"  Model: {settings_resource(0).textual content}")
           
            # Learn information useful resource
            print("nFetching Calculator Information...")
            guide_resource = await shopper.read_resource("docs://calculator/information")
            # Print first 200 characters of information
            guide_text = guide_resource(0).textual content(:200) + "..."
            print(f"  {guide_text}")
           
            # CHAINING OPERATIONS
           
            print("n" + "="*60)
            print("4. CHAINING MULTIPLE OPERATIONS")
            print("="*60)
           
            # Calculate: (10 + 5) * 3 - 7
            print("nCalculating: (10 + 5) * 3 - 7")
           
            # Step 1: Add
            print("  Step 1: Add 10 + 5")
            add_result = await shopper.call_tool("add", {"a": 10, "b": 5})
            step1 = extract_tool_result(add_result)
            print(f"    Consequence: {step1}")
           
            # Step 2: Multiply
            print("  Step 2: Multiply 15 * 3")
            mult_result = await shopper.call_tool("multiply", {"a": step1, "b": 3})
            step2 = extract_tool_result(mult_result)
            print(f"    Consequence: {step2}")
           
            # Step 3: Subtract
            print("  Step 3: Subtract 45 - 7")
            final_result = await shopper.call_tool("subtract", {"a": step2, "b": 7})
            ultimate = extract_tool_result(final_result)
            print(f"    Last Consequence: {ultimate}")
           
            # GET PROMPT TEMPLATE
           
            print("n" + "="*60)
            print("5. USING PROMPT TEMPLATES")
            print("="*60)
           
            expression = "25 * 4 + 10 / 2"
            print(f"nPrompt Template for: {expression}")
            prompt_response = await shopper.get_prompt(
                "calculate_expression",
                {"expression": expression}
            )
            print(f"  Template:n{prompt_response.messages(0).content material.textual content}")
           
            logger.data("✓ Consumer operations accomplished efficiently")
   
    besides Exception as e:
        logger.error(f"Consumer error: {e}", exc_info=True)
        sys.exit(1)

def extract_tool_result(response: Any) -> Any:
    """
    Extract the precise end result worth from a device response.
   
    MCP wraps leads to content material objects, this helper unwraps them.
    """
    attempt:
        if hasattr(response, 'content material') and response.content material:
            content material = response.content material(0)
            # Want express textual content content material when obtainable (TextContent)
            if hasattr(content material, 'textual content') and content material.textual content just isn't None:
                # If the textual content is JSON, attempt to parse and extract a `end result` area
                import json as _json
                text_val = content material.textual content
                attempt:
                    parsed_text = _json.hundreds(text_val)
                    # If JSON comprises a end result area, return it
                    if isinstance(parsed_text, dict) and 'end result' in parsed_text:
                        return parsed_text.get('end result')
                    return parsed_text
                besides _json.JSONDecodeError:
                    # Attempt to convert plain textual content to quantity
                    attempt:
                        if '.' in text_val:
                            return float(text_val)
                        return int(text_val)
                    besides Exception:
                        return text_val

            # Attempt to extract JSON end result by way of mannequin `.json()` or dict-like `.json`
            if hasattr(content material, 'json'):
                attempt:
                    if callable(content material.json):
                        json_str = content material.json()
                        import json as _json
                        attempt:
                            parsed = _json.hundreds(json_str)
                        besides _json.JSONDecodeError:
                            return json_str
                    else:
                        parsed = content material.json

                    # If parsed is a dict, attempt widespread shapes
                    if isinstance(parsed, dict):
                        # If nested end result exists
                        if 'end result' in parsed:
                            res = parsed.get('end result')
                        elif 'textual content' in parsed:
                            res = parsed.get('textual content')
                        else:
                            res = parsed

                        # If res is str that appears like a quantity, convert
                        if isinstance(res, str):
                            attempt:
                                if '.' in res:
                                    return float(res)
                                return int(res)
                            besides Exception:
                                return res
                        return res

                    return parsed
                besides Exception:
                    cross
        return response
    besides Exception as e:
        logger.warning(f"Couldn't extract end result: {e}")
        return response


if __name__ == "__main__":
    logger.data("Calculator Consumer Beginning...")
    asyncio.run(primary())

python calculator_server.py

python calculator_client.py