Cracking ML Interviews

Open in the interactive study hub

Model Context Protocol (MCP)

Model Context Protocol (MCP) is an open standard created by Anthropic that defines how LLMs connect to external tools, data sources, and services. Launched in November 2024, MCP has become the USB-C of AI integrations: one protocol to connect any LLM to any tool.


Table of Contents

  1. What is MCP?
  2. Why MCP Matters
  3. Architecture
  4. Core Primitives
  5. MCP in Claude Code
  6. Popular MCP Servers
  7. Building an MCP Server (Python)
  8. Building an MCP Server (TypeScript)
  9. MCP Client (Using a Server)
  10. MCP in 2026: Ecosystem

What is MCP?

Before MCP, connecting an LLM to external tools was bespoke:

MCP standardizes this with a client-server protocol:

┌─────────────────────────────────────────────────────┐
│                   MCP Client (LLM Host)              │
│  Claude Desktop / Claude Code / Cursor / Zed / etc. │
└─────────────────────────┬───────────────────────────┘
                          │ MCP Protocol (JSON-RPC 2.0)
          ┌───────────────┼──────────────────┐
          ▼               ▼                  ▼
   ┌─────────────┐ ┌─────────────┐ ┌──────────────┐
   │ GitHub MCP  │ │Postgres MCP │ │ Slack MCP    │
   │   Server   │ │   Server   │ │   Server    │
   └─────────────┘ └─────────────┘ └──────────────┘

One MCP server can be used by any MCP-compatible client (Claude, GPT, Gemini, local LLMs, etc.).


Why MCP Matters

Before MCP

# Every integration was custom per LLM
# OpenAI format:
{"type": "function", "function": {"name": "get_data", ...}}

# Anthropic format:
{"name": "get_data", "input_schema": {...}}

# Every developer had to rewrite integrations

After MCP

One server definition → Works with any MCP client

Benefits:


Architecture

Transport Layers

MCP defines two standard transports:

Transport Use Case
stdio Local servers. Client spawns server as subprocess. Most common.
Streamable HTTP Remote servers. Client connects over network (single HTTP endpoint, optional SSE streaming for responses). Replaced the older HTTP + SSE transport, which is deprecated but still seen in older servers.

Protocol Flow

Client                          Server
  │                               │
  │──── initialize ──────────────>│  (announce capabilities)
  │<─── initialized ──────────────│
  │                               │
  │──── tools/list ──────────────>│  (discover available tools)
  │<─── tools list ───────────────│
  │                               │
  │──── tools/call ──────────────>│  (invoke tool)
  │<─── tool result ──────────────│
  │                               │
  │──── resources/list ──────────>│  (discover resources)
  │<─── resources list ───────────│
  │                               │
  │──── resources/read ──────────>│  (read resource)
  │<─── resource content ─────────│

Core Primitives

MCP servers expose three types of capabilities:

1. Tools (LLM calls these)

Functions the LLM can invoke to take actions or retrieve data.

{
  "name": "execute_sql",
  "description": "Run a SQL query against the data warehouse",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {"type": "string", "description": "SQL query"},
      "database": {"type": "string", "enum": ["prod", "staging"]}
    },
    "required": ["query"]
  }
}

2. Resources (LLM reads these)

Read-only data that can be embedded in the LLM's context. Like files or database views.

{
  "uri": "postgres://localhost/mydb/users",
  "name": "Users table schema",
  "mimeType": "application/json"
}

3. Prompts (pre-built prompt templates)

Reusable prompt templates with arguments.

{
  "name": "analyze_dataset",
  "description": "Analyze a dataset for quality issues",
  "arguments": [
    {"name": "table_name", "required": true},
    {"name": "focus_area", "required": false}
  ]
}

MCP in Claude Code

Claude Code has first-class MCP support.

# Add a local (stdio) MCP server: everything after -- is the command to run
claude mcp add server-name -- command [args]

# Add a remote MCP server over HTTP
claude mcp add --transport http server-name https://example.com/mcp

# Examples:
claude mcp add filesystem -- npx -y @modelcontextprotocol/server-filesystem /path/to/dir
claude mcp add memory -- npx -y @modelcontextprotocol/server-memory

# List configured servers
claude mcp list

# Remove a server
claude mcp remove server-name

# Test server connectivity
claude mcp get server-name

Claude Desktop Configuration

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."
      }
    },
    "postgres": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-postgres",
        "postgresql://localhost/mydb"
      ]
    },
    "my-custom-server": {
      "command": "python",
      "args": ["/path/to/my_mcp_server.py"]
    }
  }
}

Reference Servers (MCP project)

The modelcontextprotocol/servers repo keeps a small set of reference servers meant as examples:

# Filesystem: read/write local files
npx -y @modelcontextprotocol/server-filesystem /allowed/path

# Memory: knowledge-graph style persistent memory
npx -y @modelcontextprotocol/server-memory

# Sequential Thinking: structured reasoning tool
npx -y @modelcontextprotocol/server-sequential-thinking

Several early reference servers (GitHub, PostgreSQL, Slack, Brave Search, Google Drive, Puppeteer) were moved to an archive and are no longer maintained there. For those services, prefer the server published by the vendor itself (for example GitHub's own GitHub MCP server) or a maintained community server, and check it before trusting it with credentials.

Community Servers (2026 Ecosystem)

Server Description
mcp-server-snowflake Query Snowflake DWH
mcp-server-dbt Run dbt models, tests, docs
mcp-server-kubernetes Manage K8s clusters
mcp-server-terraform Plan/apply Terraform
mcp-server-jira Create/update Jira tickets
mcp-server-linear Linear issue tracking
mcp-server-notion Read/write Notion pages
mcp-server-playwright Browser automation
mcp-server-puppeteer Browser automation (Node)
mcp-server-sentry Query Sentry errors
mcp-server-datadog Query Datadog metrics
mcp-server-aws AWS operations
mcp-server-gcp GCP operations

Building an MCP Server (Python)

pip install mcp

Simple Data Engineering MCP Server

# my_data_mcp_server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

# Initialize the MCP server
app = Server("data-engineering-tools")

# ─── Tools ────────────────────────────────────────────

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="run_sql",
            description="Execute a SQL query and return results as JSON",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL query to execute"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max rows to return (default 100)",
                        "default": 100
                    }
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="list_tables",
            description="List all available tables and their schemas",
            inputSchema={
                "type": "object",
                "properties": {
                    "schema": {
                        "type": "string",
                        "description": "Database schema to inspect (default: public)"
                    }
                }
            }
        ),
        types.Tool(
            name="get_table_stats",
            description="Get row count, size, and last updated time for a table",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {"type": "string"}
                },
                "required": ["table_name"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "run_sql":
        result = await execute_sql(
            arguments["query"],
            limit=arguments.get("limit", 100)
        )
        return [types.TextContent(
            type="text",
            text=json.dumps(result, indent=2, default=str)
        )]

    elif name == "list_tables":
        schema = arguments.get("schema", "public")
        tables = await get_tables(schema)
        return [types.TextContent(type="text", text=json.dumps(tables, indent=2))]

    elif name == "get_table_stats":
        stats = await get_stats(arguments["table_name"])
        return [types.TextContent(type="text", text=json.dumps(stats, indent=2))]

    raise ValueError(f"Unknown tool: {name}")

# ─── Resources ────────────────────────────────────────

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="data://schemas/all",
            name="All Database Schemas",
            description="Complete schema documentation for all tables",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri) -> str:  # uri arrives as a pydantic AnyUrl, not a str
    if str(uri) == "data://schemas/all":
        schemas = await get_all_schemas()
        return json.dumps(schemas, indent=2)
    raise ValueError(f"Unknown resource: {uri}")

# ─── Actual Implementation ────────────────────────────

import asyncpg

DB_URL = "postgresql://localhost/mydb"

async def execute_sql(query: str, limit: int = 100) -> list[dict]:
    conn = await asyncpg.connect(DB_URL)
    try:
        # Add LIMIT to SELECT queries for safety
        if query.strip().upper().startswith("SELECT") and "LIMIT" not in query.upper():
            query = f"{query} LIMIT {limit}"
        rows = await conn.fetch(query)
        return [dict(row) for row in rows]
    finally:
        await conn.close()

async def get_tables(schema: str = "public") -> list[dict]:
    conn = await asyncpg.connect(DB_URL)
    try:
        rows = await conn.fetch("""
            SELECT table_name, column_name, data_type
            FROM information_schema.columns
            WHERE table_schema = $1
            ORDER BY table_name, ordinal_position
        """, schema)
        # Group by table
        tables = {}
        for row in rows:
            t = row["table_name"]
            if t not in tables:
                tables[t] = []
            tables[t].append({"column": row["column_name"], "type": row["data_type"]})
        return [{"table": k, "columns": v} for k, v in tables.items()]
    finally:
        await conn.close()

async def get_stats(table_name: str) -> dict:
    conn = await asyncpg.connect(DB_URL)
    try:
        # table_name comes from the model, so never interpolate it directly:
        # confirm it is a real table, then quote it as an identifier.
        exists = await conn.fetchval(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = 'public' AND table_name = $1",
            table_name,
        )
        if not exists:
            raise ValueError(f"Unknown table: {table_name!r}")
        quoted = '"' + table_name.replace('"', '""') + '"'
        count = await conn.fetchval(f"SELECT COUNT(*) FROM {quoted}")
        return {"table": table_name, "row_count": count}
    finally:
        await conn.close()

async def get_all_schemas():
    return await get_tables()

# ─── Entry Point ──────────────────────────────────────

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Register in Claude Desktop

{
  "mcpServers": {
    "my-data-server": {
      "command": "python",
      "args": ["/path/to/my_data_mcp_server.py"],
      "env": {
        "DATABASE_URL": "postgresql://localhost/mydb"
      }
    }
  }
}

Building an MCP Server (TypeScript)

npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript ts-node @types/node
// src/server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

const server = new Server(
  { name: "my-tools-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// Define tool schemas with Zod
const SearchSchema = z.object({
  query: z.string().describe("Search query"),
  limit: z.number().optional().default(10),
});

// List available tools
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "search",
      description: "Search for information",
      inputSchema: {
        type: "object",
        properties: {
          query: { type: "string", description: "Search query" },
          limit: { type: "number", description: "Max results", default: 10 },
        },
        required: ["query"],
      },
    },
  ],
}));

// Handle tool calls
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "search") {
    const { query, limit } = SearchSchema.parse(request.params.arguments);

    // Perform actual search...
    const results = await performSearch(query, limit);

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(results, null, 2),
        },
      ],
    };
  }

  throw new Error(`Unknown tool: ${request.params.name}`);
});

async function performSearch(query: string, limit: number) {
  // Your implementation here
  return [{ title: "Result 1", url: "https://example.com" }];
}

// Start server
const transport = new StdioServerTransport();
await server.connect(transport);

MCP Client (Using a Server)

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def use_mcp_server():
    server_params = StdioServerParameters(
        command="python",
        args=["my_mcp_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print("Available tools:", [t.name for t in tools.tools])

            # Call a tool
            result = await session.call_tool(
                "run_sql",
                {"query": "SELECT COUNT(*) FROM orders"}
            )
            print("Result:", result.content[0].text)

            # List resources
            resources = await session.list_resources()

            # Read a resource
            content = await session.read_resource("data://schemas/all")
            print("Schema:", content.contents[0].text)

import asyncio
asyncio.run(use_mcp_server())

MCP in 2026: Ecosystem

Adoption

MCP has been adopted by:

MCP Registry

The MCP project runs an official registry (registry.modelcontextprotocol.io) where server publishers list their servers. Community directories such as mcp.so also index thousands of servers, but they are unvetted, so review a server before installing it.

Key Patterns in 2026

Pattern Description
MCP + RAG MCP server exposes search tools backed by vector DB
MCP + Data Catalog LLMs discover and query datasets via MCP
MCP for Agents Multi-agent systems coordinate via shared MCP servers
Secure MCP OAuth 2.0 authentication for remote MCP servers
MCP Proxies Aggregate multiple MCP servers behind a single endpoint

Remote MCP (HTTP Transport)

# Remote MCP server with authentication
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Remote Data Server", port=8080)  # host/port are server settings

@mcp.tool()
async def query_warehouse(sql: str) -> str:
    """Execute SQL against the data warehouse"""
    results = await run_query(sql)
    return json.dumps(results)

# Run as HTTP server
if __name__ == "__main__":
    mcp.run(transport="streamable-http")  # serves at http://localhost:8080/mcp
# Client connects to remote server
claude mcp add --transport http remote-warehouse https://data-server.mycompany.com/mcp

Cheat Sheet

# Claude Code MCP commands
claude mcp add <name> -- <command> [args]
claude mcp add --transport http <name> <url>
claude mcp list
claude mcp remove <name>
claude mcp get <name>

# Common servers
claude mcp add filesystem -- npx -y @modelcontextprotocol/server-filesystem /path
claude mcp add memory -- npx -y @modelcontextprotocol/server-memory
claude mcp add seq-thinking -- npx -y @modelcontextprotocol/server-sequential-thinking

# Python MCP server template
pip install mcp
# → implement list_tools() + call_tool() + main()

# TypeScript MCP server template
npm install @modelcontextprotocol/sdk
# → implement ListToolsRequestSchema + CallToolRequestSchema handlers

View or edit this guide on GitHub