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
- What is MCP?
- Why MCP Matters
- Architecture
- Core Primitives
- MCP in Claude Code
- Popular MCP Servers
- Building an MCP Server (Python)
- Building an MCP Server (TypeScript)
- MCP Client (Using a Server)
- MCP in 2026: Ecosystem
What is MCP?
Before MCP, connecting an LLM to external tools was bespoke:
- Every LLM had its own tool-use format
- Every integration had to be written from scratch per-LLM
- Context/memory was per-conversation
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:
- Standardized: build once, use everywhere
- Secure: servers run locally or in controlled environments
- Composable: mix and match servers
- Discoverable: clients can query server capabilities at runtime
- Stateful: maintain context across a session
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"]
}
}
}
Popular MCP Servers
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:
- Anthropic: Claude Desktop, Claude Code, Claude API
- OpenAI: ChatGPT supports MCP servers
- Google: Gemini MCP support
- Microsoft: GitHub Copilot, VS Code
- Cursor, Windsurf, Zed: AI code editors
- LangChain, LlamaIndex: frameworks integrate MCP
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