Developing Your MCP Server¶

from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("my_echo_server", port="8000")

@mcp.tool()
def echo(text: str) -> str:
    """Echo the provided text back to the caller"""
    return text

if __name__ == "__main__":
    mcp.run()  # STDIO mode by default

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

from mcp.server.fastmcp import FastMCP
import datetime

# Create an MCP server
mcp = FastMCP("my_enhanced_server", port="8000")

@mcp.tool()
def echo(text: str) -> str:
    """Echo the provided text back to the caller"""
    return text

@mcp.tool()
def get_timestamp() -> str:
    """Get the current timestamp"""
    return datetime.datetime.now().isoformat()

@mcp.tool()
def calculate(a: float, b: float, operation: str) -> float:
    """Perform basic math operations: add, subtract, multiply, divide"""
    operations = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b if b != 0 else float('inf')
    }

    if operation not in operations:
        raise ValueError(f"Unknown operation: {operation}")

    return operations[operation]

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
import datetime

mcp = FastMCP("structured_server", port="8000")

class EchoResponse(BaseModel):
    """Response structure for echo tool"""
    original_text: str = Field(description="The original input text")
    echo_text: str = Field(description="The echoed text")
    length: int = Field(description="Length of the text")
    timestamp: str = Field(description="When the echo was processed")

@mcp.tool()
def structured_echo(text: str) -> EchoResponse:
    """Echo with structured response data"""
    return EchoResponse(
        original_text=text,
        echo_text=text,
        length=len(text),
        timestamp=datetime.datetime.now().isoformat()
    )

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

from mcp.server.fastmcp import FastMCP
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

mcp = FastMCP("robust_server", port="8000")

@mcp.tool()
def safe_echo(text: str) -> str:
    """Echo with validation and error handling"""
    try:
        # Log the request
        logger.info(f"Processing echo request for text of length {len(text)}")

        # Validate input
        if not text.strip():
            raise ValueError("Text cannot be empty")

        if len(text) > 1000:
            raise ValueError("Text too long (max 1000 characters)")

        # Process and return
        return text

    except Exception as e:
        logger.error(f"Error in safe_echo: {e}")
        raise

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

import pytest
from my_echo_server import mcp

@pytest.mark.asyncio
async def test_echo_tool():
    """Test the echo tool directly"""
    # This would require setting up the MCP server context
    # For integration testing, use the MCP Inspector instead
    pass

def test_basic_functionality():
    """Test basic server setup"""
    assert mcp.name == "my_echo_server"
    # Add more server validation tests

import os
from mcp.server.fastmcp import FastMCP

# Configuration from environment
SERVER_NAME = os.getenv("MCP_SERVER_NAME", "my_echo_server")
PORT = os.getenv("MCP_SERVER_PORT", "8000")
DEBUG_MODE = os.getenv("MCP_DEBUG", "false").lower() == "true"

mcp = FastMCP(SERVER_NAME, port=PORT)

@mcp.tool()
def echo(text: str) -> str:
    """Echo the provided text"""
    if DEBUG_MODE:
        print(f"Debug: Processing text of length {len(text)}")
    return text

if __name__ == "__main__":
    transport = os.getenv("MCP_TRANSPORT", "streamable-http")
    print(f"Starting {SERVER_NAME} with {transport} transport")
    mcp.run(transport=transport)

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("resource_server", port="8000")

@mcp.resource("config://settings")
def get_settings() -> str:
    """Provide server configuration as a resource"""
    return """{
  "server_name": "my_echo_server",
  "version": "1.0.0",
  "features": ["echo", "timestamp"]
}"""

@mcp.resource("status://health")
def get_health() -> str:
    """Provide server health status"""
    return "Server is running normally"

@mcp.tool()
def echo(text: str) -> str:
    """Echo the provided text"""
    return text

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("context_server", port="8000")

@mcp.tool()
async def echo_with_logging(text: str, ctx: Context) -> str:
    """Echo with context logging"""
    await ctx.info(f"Processing echo request for: {text[:50]}...")
    await ctx.debug(f"Full text length: {len(text)}")

    result = text

    await ctx.info("Echo completed successfully")
    return result

if __name__ == "__main__":
    mcp.run(transport="streamable-http")