mkdir my-python-mcp-server
cd my-python-mcp-server
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

pip install mcp

uv init my-python-mcp-server
cd my-python-mcp-server
uv add mcp

from mcp.server.fastmcp import FastMCP

# Create a FastMCP server instance
mcp = FastMCP("my-python-server")

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("calculator-server")

@mcp.tool()
def add(a: float, b: float) -> str:
    """Add two numbers together.

    Args:
        a: The first number
        b: The second number
    """
    return str(a + b)

@mcp.tool()
def multiply(a: float, b: float) -> str:
    """Multiply two numbers together.

    Args:
        a: The first number
        b: The second number
    """
    return str(a * b)

@mcp.resource("config://app")
def get_config() -> str:
    """Return the application configuration."""
    return json.dumps({
        "version": "1.0.0",
        "environment": "production",
        "features": ["auth", "logging", "cache"]
    })

@mcp.resource("users://{user_id}")
def get_user(user_id: str) -> str:
    """Retrieve user information by ID."""
    users = {
        "1": {"name": "Alice", "role": "admin"},
        "2": {"name": "Bob", "role": "user"},
    }
    user = users.get(user_id, {"error": "User not found"})
    return json.dumps(user)

@mcp.prompt()
def analyze_data(dataset: str) -> str:
    """Create a prompt to analyze a dataset."""
    return f"""Please analyze the following dataset and provide insights:

Dataset: {dataset}

Please include:
1. Summary statistics
2. Key trends
3. Anomalies or outliers
4. Recommendations"""

import os
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("file-manager")

WORKSPACE = Path.home() / "mcp-workspace"
WORKSPACE.mkdir(exist_ok=True)

@mcp.tool()
def list_files(directory: str = "") -> str:
    """List files in the workspace directory.

    Args:
        directory: Subdirectory to list (relative to workspace root)
    """
    target = WORKSPACE / directory
    if not target.exists():
        return json.dumps({"error": f"Directory not found: {directory}"})

    files = []
    for item in target.iterdir():
        files.append({
            "name": item.name,
            "type": "directory" if item.is_dir() else "file",
            "size": item.stat().st_size if item.is_file() else None,
        })
    return json.dumps({"files": files, "count": len(files)})

@mcp.tool()
def read_file(filename: str) -> str:
    """Read the contents of a file in the workspace.

    Args:
        filename: Path to file relative to workspace root
    """
    filepath = WORKSPACE / filename
    if not filepath.exists():
        return json.dumps({"error": f"File not found: {filename}"})
    if not str(filepath.resolve()).startswith(str(WORKSPACE)):
        return json.dumps({"error": "Access denied: path traversal detected"})
    return filepath.read_text()

@mcp.tool()
def write_file(filename: str, content: str) -> str:
    """Write content to a file in the workspace.

    Args:
        filename: Path to file relative to workspace root
        content: Content to write to the file
    """
    filepath = WORKSPACE / filename
    if not str(filepath.resolve()).startswith(str(WORKSPACE)):
        return json.dumps({"error": "Access denied: path traversal detected"})
    filepath.parent.mkdir(parents=True, exist_ok=True)
    filepath.write_text(content)
    return json.dumps({"status": "success", "path": str(filepath)})

@mcp.tool()
def search_files(pattern: str, directory: str = "") -> str:
    """Search for files matching a glob pattern.

    Args:
        pattern: Glob pattern to match (e.g., '*.py', '**/*.json')
        directory: Subdirectory to search in
    """
    target = WORKSPACE / directory
    matches = [str(p.relative_to(WORKSPACE)) for p in target.glob(pattern)]
    return json.dumps({"matches": matches, "count": len(matches)})

@mcp.resource("workspace://structure")
def workspace_structure() -> str:
    """Return the workspace directory structure."""
    structure = []
    for root, dirs, files in os.walk(WORKSPACE):
        level = len(Path(root).relative_to(WORKSPACE).parts)
        indent = "  " * level
        structure.append(f"{indent}{Path(root).name}/")
        for f in files:
            structure.append(f"{indent}  {f}")
    return "\n".join(structure)

if __name__ == "__main__":
    mcp.run()

import asyncio
import httpx
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("async-server")

@mcp.tool()
async def fetch_url(url: str, ctx: Context) -> str:
    """Fetch content from a URL.

    Args:
        url: The URL to fetch
    """
    ctx.info(f"Fetching {url}...")
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        ctx.info(f"Received {len(response.text)} bytes")
        return response.text[:5000]

@mcp.tool()
async def parallel_fetch(urls: list[str], ctx: Context) -> str:
    """Fetch multiple URLs in parallel.

    Args:
        urls: List of URLs to fetch
    """
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

    results = {}
    for url, resp in zip(urls, responses):
        if isinstance(resp, Exception):
            results[url] = f"Error: {str(resp)}"
        else:
            results[url] = resp.text[:1000]

    return json.dumps(results)

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("robust-server")

@mcp.tool()
def divide(a: float, b: float) -> str:
    """Divide two numbers safely.

    Args:
        a: The dividend
        b: The divisor (must not be zero)
    """
    if b == 0:
        return json.dumps({"error": "Division by zero is not allowed"})
    result = a / b
    return json.dumps({"result": result})

@mcp.tool()
def process_json(data: str) -> str:
    """Parse and validate JSON data.

    Args:
        data: JSON string to process
    """
    try:
        parsed = json.loads(data)
        return json.dumps({
            "valid": True,
            "type": type(parsed).__name__,
            "keys": list(parsed.keys()) if isinstance(parsed, dict) else None,
            "length": len(parsed) if isinstance(parsed, (list, dict)) else None,
        })
    except json.JSONDecodeError as e:
        return json.dumps({"valid": False, "error": str(e)})

if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "file-manager": {
      "command": "python",
      "args": ["/path/to/your/server.py"]
    }
  }
}

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

npx @modelcontextprotocol/inspector python server.py

import pytest
from server import list_files, read_file, write_file

def test_write_and_read():
    result = write_file("test.txt", "Hello MCP")
    data = json.loads(result)
    assert data["status"] == "success"

    content = read_file("test.txt")
    assert content == "Hello MCP"

def test_list_files():
    result = list_files("")
    data = json.loads(result)
    assert "files" in data
    assert isinstance(data["count"], int)

my-mcp-server/
  pyproject.toml
  src/
    server/
      __init__.py
      main.py          # FastMCP instance and entry point
      tools/
        __init__.py
        file_tools.py  # File management tools
        data_tools.py  # Data processing tools
      resources/
        __init__.py
        config.py      # Configuration resources
      utils/
        __init__.py
        validation.py  # Input validation helpers

# src/server/main.py
from mcp.server.fastmcp import FastMCP
from .tools.file_tools import register_file_tools
from .tools.data_tools import register_data_tools

mcp = FastMCP("organized-server")

register_file_tools(mcp)
register_data_tools(mcp)

if __name__ == "__main__":
    mcp.run()

# src/server/tools/file_tools.py
def register_file_tools(mcp):
    @mcp.tool()
    def list_files(directory: str = "") -> str:
        """List files in a directory."""
        # implementation...
        pass