MCP Tutorial: Build an AI Tool Server in Python

┌─────────────────────────────────────────┐
│  MCP Host (Claude Desktop, VS Code)     │
│                                         │
│  ┌──────────┐  ┌──────────┐             │
│  │ Client 1 │  │ Client 2 │   ...       │
│  └────┬─────┘  └────┬─────┘             │
└───────┼──────────────┼──────────────────┘
        │              │
   ┌────▼────┐    ┌────▼────┐
   │ Server A│    │ Server B│
   │(Database)│   │(GitHub) │
   └─────────┘    └─────────┘

# Install with uv (recommended)
uv init mcp-demo && cd mcp-demo
uv add "mcp[cli]"

# Or with pip
pip install "mcp[cli]"

from mcp.server.fastmcp import FastMCP

# Initialize server
mcp = FastMCP("demo-server")


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


@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


@mcp.resource("config://app")
def get_config() -> str:
    """Return application configuration."""
    return "debug=true\nlog_level=INFO\nmax_retries=3"


@mcp.prompt()
def explain_code(code: str, language: str = "python") -> str:
    """Generate a prompt to explain code."""
    return f"Explain this {language} code step by step:\n\n```{language}\n{code}\n```"


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

# Launch the MCP Inspector for interactive testing
npx -y @modelcontextprotocol/inspector uv run server.py

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("sqlite-explorer")

DB_PATH = "example.db"


@mcp.resource("db://schema")
def get_schema() -> str:
    """Return the complete database schema."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "SELECT sql FROM sqlite_master WHERE type='table'"
    )
    schemas = [row[0] for row in cursor.fetchall() if row[0]]
    conn.close()
    return "\n\n".join(schemas)


@mcp.resource("db://tables")
def list_tables() -> str:
    """List all tables in the database."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
    )
    tables = [row[0] for row in cursor.fetchall()]
    conn.close()
    return "\n".join(tables)


@mcp.tool()
def query(sql: str) -> str:
    """Execute a read-only SQL query.

    Args:
        sql: A SELECT query to run against the database.
             Only SELECT statements are allowed for safety.
    """
    sql_stripped = sql.strip().upper()
    if not sql_stripped.startswith("SELECT"):
        return "Error: Only SELECT queries allowed."

    dangerous = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE"]
    if any(kw in sql_stripped for kw in dangerous):
        return "Error: Write operations not permitted."

    conn = sqlite3.connect(DB_PATH)
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        if not rows:
            return "No results found."
        result = [dict(zip(columns, row)) for row in rows]
        return str(result[:50])  # Limit output size
    except Exception as e:
        return f"SQL Error: {e}"
    finally:
        conn.close()


@mcp.tool()
def table_stats(table_name: str) -> str:
    """Get row count and column info for a table.

    Args:
        table_name: Name of the table to analyze.
    """
    # Prevent SQL injection
    if not table_name.isalnum():
        return "Error: Invalid table name."

    conn = sqlite3.connect(DB_PATH)
    try:
        count = conn.execute(
            f"SELECT COUNT(*) FROM {table_name}"
        ).fetchone()[0]
        columns = conn.execute(
            f"PRAGMA table_info({table_name})"
        ).fetchall()
        col_info = [
            f"  {c[1]} ({c[2]})" + (" PRIMARY KEY" if c[5] else "")
            for c in columns
        ]
        return f"Table: {table_name}\nRows: {count}\nColumns:\n" + "\n".join(col_info)
    except Exception as e:
        return f"Error: {e}"
    finally:
        conn.close()


@mcp.prompt()
def analyze_table(table_name: str) -> str:
    """Create an analysis prompt for a database table."""
    return f"Analyze the '{table_name}' table:\n1. Check the schema using db://schema\n2. Run a query to see sample data\n3. Get table_stats for row counts\n4. Report any data quality issues"

{
  "mcpServers": {
    "sqlite-explorer": {
      "command": "uv",
      "args": ["--directory", "/path/to/your/project", "run", "server.py"]
    }
  }
}

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
import aiohttp
from mcp.server.fastmcp import FastMCP, Context


@dataclass
class AppContext:
    session: aiohttp.ClientSession


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage HTTP session lifecycle."""
    session = aiohttp.ClientSession()
    try:
        yield AppContext(session=session)
    finally:
        await session.close()


mcp = FastMCP("api-server", lifespan=app_lifespan)


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

    Args:
        url: The URL to fetch (must be HTTPS).
    """
    if not url.startswith("https://"):
        return "Error: Only HTTPS URLs allowed."
    session = ctx.request_context.lifespan_context.session
    async with session.get(url) as resp:
        text = await resp.text()
        return text[:5000]  # Truncate large responses

@mcp.tool()
def read_file(path: str) -> str:
    """Read a file from the allowed directory only."""
    import os
    # Prevent path traversal
    safe_path = os.path.realpath(path)
    if not safe_path.startswith("/allowed/directory/"):
        return "Error: Access denied."
    with open(safe_path) as f:
        return f.read()

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("remote-server")

# ... define tools, resources, prompts ...

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

claude mcp add --transport http my-server http://your-server:8000/mcp

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("name")

# Tool (AI-controlled)
@mcp.tool()
def my_tool(arg: str) -> str:
    """Docstring becomes the tool description."""
    return "result"

# Async tool with progress
@mcp.tool()
async def slow_tool(data: str, ctx: Context) -> str:
    """Long-running operation with progress."""
    for i in range(10):
        await ctx.report_progress(i, 10)
    return "done"

# Resource (app-controlled context)
@mcp.resource("scheme://path")
def static_resource() -> str:
    return "data"

@mcp.resource("scheme://path/{param}")
def dynamic_resource(param: str) -> str:
    return f"data for {param}"

# Prompt (user-controlled template)
@mcp.prompt()
def my_prompt(topic: str) -> str:
    return f"Analyze {topic} in detail."

# Run
mcp.run(transport="stdio")           # Local
mcp.run(transport="streamable-http") # Remote