opncrafter

Model Context Protocol (MCP): The USB-C for AI

Jan 1, 2026 • 20 min read

Before USB-C, every device used different connectors — Lightning for iPhone, micro-USB for Android, DisplayPort for monitors, HDMI for TVs. Connecting anything required knowing which proprietary cable to use. Before MCP, AI tool integrations had the same problem: connecting Claude to Google Drive required building a custom integration; switching from Claude to GPT-4 meant rebuilding it. MCP (Model Context Protocol) from Anthropic is the USB-C moment for AI tools — one standard that enables any MCP-compliant LLM client to talk to any MCP-compliant server.

1. The N×M Integration Problem MCP Solves

❌ Before MCP

N AI apps × M data sources = N×M custom integrations

Claude + Drive → custom code
GPT-4 + Drive → rebuild from scratch
Claude + Slack → custom code
GPT-4 + Slack → rebuild again
4 integrations for 2 apps × 2 tools

✅ With MCP

N AI clients + M MCP servers = N+M implementations

Drive MCP Server → built once
Slack MCP Server → built once
Claude (MCP client) → connects to both
GPT-4 (MCP client) → also connects to both
2+2=4 implementations, not 4 separate integrations

2. Technical Architecture: JSON-RPC over stdio

# MCP Protocol Basics:
# - JSON-RPC 2.0 message format
# - Typically runs over stdio (local) or HTTP+SSE (remote)
# - Server exposes capabilities: Resources, Tools, Prompts, Sampling

# STDIO transport (local, most common for Claude Desktop):
# AI Client (Claude) → spawns subprocess → MCP Server Python script
# Communication: JSON-RPC messages over stdin/stdout

# HTTP+SSE transport (remote, for cloud deployments):
# AI Client → HTTP POST /mcp → MCP Server (web service)
# Server responses via Server-Sent Events for streaming

# MCP Initialization handshake:
# 1. Client sends: {"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{}}}
# 2. Server responds: {"jsonrpc":"2.0","result":{"protocolVersion":"2024-11-05","capabilities":{"tools":{},"resources":{}},"serverInfo":{"name":"My Server","version":"1.0.0"}}}
# 3. Client sends: {"jsonrpc":"2.0","method":"initialized"}
# 4. Ready for tool calls!

# MCP Primitives:
# RESOURCES: Read-only data exposed as URIs (like files, database records)
# TOOLS: Executable functions that the LLM can call (like API actions)
# PROMPTS: Reusable prompt templates the user can invoke
# SAMPLING: Server asks client to run LLM completions (advanced — agent loop)

3. Building a Production MCP Server with FastMCP

pip install fastmcp

# mcp_server.py — A complete MCP server exposing database + web search
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
import sqlite3
import httpx
from typing import Optional

# Initialize server with a descriptive name (shown in Claude UI)
mcp = FastMCP(
    name="Company Knowledge Base",
    instructions="This server provides access to the company's internal knowledge base and web search. Use search_knowledge_base for internal company data before falling back to web_search.",
)

# ─── TOOLS ─────────────────────────────────────────────────────────────────

@mcp.tool()
async def search_knowledge_base(
    query: str,
    category: Optional[str] = None,
    limit: int = 5
) -> str:
    """
    Search the internal company knowledge base for relevant documents.
    Returns matching articles with their content and metadata.
    
    Args:
        query: Natural language search query
        category: Optional filter (e.g., "engineering", "hr", "finance")
        limit: Maximum number of results to return (1-20)
    """
    if limit > 20:
        raise ToolError("limit cannot exceed 20")
    
    conn = sqlite3.connect('knowledge_base.db')
    try:
        sql = "SELECT title, content, category, updated_at FROM articles WHERE content LIKE ? "
        params = [f"%{query}%"]
        if category:
            sql += "AND category = ? "
            params.append(category)
        sql += f"LIMIT {limit}"
        
        results = conn.execute(sql, params).fetchall()
        
        if not results:
            return f"No articles found matching '{query}'" + (f" in category '{category}'" if category else "")
        
        formatted = f"Found {len(results)} articles:

"
        for title, content, cat, updated in results:
            formatted += f"**{title}** [{cat}] (Updated: {updated})
"
            formatted += f"{content[:300]}...

"
        
        return formatted
    finally:
        conn.close()

@mcp.tool()
async def create_ticket(
    title: str,
    description: str,
    priority: str = "medium",
    assignee_email: Optional[str] = None
) -> dict:
    """
    Create a new support/engineering ticket.
    Returns the created ticket ID and URL.
    
    Args:
        title: Short, descriptive ticket title (max 100 chars)
        description: Detailed description of the issue or task
        priority: "low", "medium", "high", or "critical"
        assignee_email: Optional email of person to assign ticket to
    """
    valid_priorities = {"low", "medium", "high", "critical"}
    if priority not in valid_priorities:
        raise ToolError(f"priority must be one of: {', '.join(valid_priorities)}")
    
    # Call your ticketing system API (Linear, Jira, GitHub, etc.)
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.linear.app/graphql",
            json={"query": "mutation CreateIssue($input: IssueCreateInput!) { issueCreate(input: $input) { issue { id url } } }",
                  "variables": {"input": {"title": title, "description": description, "priority": {"name": priority}}}},
            headers={"Authorization": f"Bearer {LINEAR_API_KEY}"},
        )
        data = response.json()
        issue = data["data"]["issueCreate"]["issue"]
    
    return {"ticket_id": issue["id"], "url": issue["url"], "title": title}

# ─── RESOURCES ─────────────────────────────────────────────────────────────

@mcp.resource("kb://categories")
async def list_knowledge_categories() -> str:
    """List all available knowledge base categories."""
    conn = sqlite3.connect('knowledge_base.db')
    categories = conn.execute("SELECT DISTINCT category, COUNT(*) as count FROM articles GROUP BY category").fetchall()
    conn.close()
    return "
".join([f"- {cat}: {count} articles" for cat, count in categories])

@mcp.resource("kb://article/{article_id}")
async def get_article(article_id: str) -> str:
    """Read a specific article by ID."""
    conn = sqlite3.connect('knowledge_base.db')
    result = conn.execute("SELECT title, content, author, updated_at FROM articles WHERE id = ?", [article_id]).fetchone()
    conn.close()
    if not result:
        return f"Article {article_id} not found"
    title, content, author, updated = result
    return f"# {title}

Author: {author} | Last Updated: {updated}

{content}"

# ─── PROMPTS ───────────────────────────────────────────────────────────────

@mcp.prompt()
async def analyze_and_create_ticket(symptoms: str) -> str:
    """Template for diagnosing a problem and creating a ticket."""
    return f"""Search the knowledge base for solutions to this issue: "{symptoms}"

If found, summarize the solution. If not found or the solution doesn't work:
1. Create a support ticket with the title "Issue: {symptoms[:50]}"
2. Include the search results you found in the description
3. Set priority based on impact (critical if production is down, high if blocking work)"""

if __name__ == "__main__":
    mcp.run()  # Runs in stdio mode by default

4. Connecting to Claude Desktop

# MacOS/Linux config path:
# ~/Library/Application Support/Claude/claude_desktop_config.json

# Windows config path:
# %APPDATA%/Claude/claude_desktop_config.json

{
    "mcpServers": {
        "company-kb": {
            "command": "python",
            "args": ["/Users/yourname/projects/mcp_server.py"],
            "env": {
                "LINEAR_API_KEY": "lin_api_...",
                "DB_PATH": "/Users/yourname/data/knowledge_base.db"
            }
        },
        
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/yourname/Documents"]
        },
        
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {
                "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."
            }
        }
    }
}

# After saving, restart Claude Desktop
# You'll see a 🔌 icon in the UI confirming MCP servers are connected
# Ask Claude: "Search the knowledge base for our onboarding process"
# → Claude calls your search_knowledge_base tool and returns results!

Frequently Asked Questions

What's the security model for MCP? Can the LLM do anything to my system?

MCP has a layered security model. Resources (read-only data) are generally considered safe since they don't execute side effects — they're analogous to reading a file. Tools (executable actions) require explicit user approval in Claude Desktop: before calling any tool, Claude shows the tool name, parameters, and asks "Allow this tool call?" The user must click Allow. This user-in-the-loop model prevents the LLM from autonomously taking destructive actions. However, you must still implement authorization in your tool code: validate that the current user has permission to perform the action (create a ticket, delete a file), never expose root filesystem access, and rate-limit tool calls to prevent abuse.

Can I deploy an MCP server to the cloud instead of running it locally?

Yes — use the HTTP+SSE transport instead of stdio. FastMCP supports this with mcp.run(transport='http', host='0.0.0.0', port=8000). Deploy to any cloud provider (Railway, Fly.io, AWS Lambda). Update the Claude Desktop config with "command": "npx" and "args": ["@modelcontextprotocol/client-http", "https://your-server.railway.app/mcp"]. Remote MCP servers also enable multi-user scenarios where multiple clients share the same server, unlike stdio which spawns one process per client. Authentication for remote servers is typically handled via Bearer tokens in the HTTP transport headers.

Conclusion

MCP's standardization of AI tool integration is a similar inflection point to what REST APIs were for web services in the 2000s. Once an MCP server is built for a data source (GitHub, Notion, Linear, PostgreSQL, your internal systems), any MCP-compliant AI client can use it — today that includes Claude, Cursor, Zed, Continue.dev, and an expanding ecosystem. The FastMCP Python library reduces server implementation to a few decorated functions with docstrings. The three-primitive model (Resources for data, Tools for actions, Prompts for templates) covers virtually every AI integration pattern. For teams building internal AI tools, MCP is the most leverage-efficient integration investment: build once, connect everywhere.

Continue Reading

👨‍💻
Written by

Vivek

AI Engineer

Full-stack AI engineer with 4+ years building LLM-powered products, autonomous agents, and RAG pipelines. I've shipped AI features to production for startups and worked hands-on with GPT-4o, LangChain, LlamaIndex, and the Vercel AI SDK. I started OpnCrafter to share everything I wish I had when learning — no fluff, just working code and real-world context.

GPT-4oLangChainNext.jsVector DBsRAGVercel AI SDK