MCP: How the 'USB-C for AI' Became the Standard Nobody Expected

┌─────────────┐     MCP Protocol     ┌─────────────┐
│   AI Host   │◄────────────────────►│  MCP Server │
│  (Claude,   │                      │  (Tool/DB/  │
│   ChatGPT)  │                      │   Service)  │
└─────────────┘                      └─────────────┘
       │                                    │
       │       Standardized JSON-RPC        │
       │      over stdio/SSE/WebSocket      │
       ▼                                    ▼
   Any AI App                          Any Data Source

{
  "uri": "file:///path/to/document.md",
  "name": "Project README",
  "mimeType": "text/markdown"
}

{
  "name": "run_query",
  "description": "Execute a SQL query",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {"type": "string"}
    }
  }
}

{
  "name": "explain_code",
  "description": "Explain a code snippet",
  "arguments": [
    {"name": "code", "required": true}
  ]
}

from mcp import Server, Resource, Tool
server = Server("my-server")
@server.resource("config://app") async def get_config():     """Expose application configuration."""     return Resource(         uri="config://app",         name="App Configuration",         text=open("config.json").read()     )
@server.tool("search_logs") async def search_logs(query: str, limit: int = 100):     """Search application logs."""     results = log_search(query, limit)     return {"matches": results}
if __name__ == "__main__":     server.run()

import { Server } from "@modelcontextprotocol/sdk";
const server = new Server({   name: "my-server",   version: "1.0.0" });
server.addResource({   uri: "docs://readme",   name: "Documentation",   async read() {     return { text: await fs.readFile("README.md", "utf-8") };   } });
server.addTool({   name: "deploy",   description: "Deploy to production",   inputSchema: {     type: "object",     properties: {       environment: { type: "string", enum: ["staging", "production"] }     }   },   async execute({ environment }) {     return await deployTo(environment);   } });
server.start();

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/files"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "mcp-server-github"],
      "env": {
        "GITHUB_TOKEN": "your-token-here"
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://..."
      }
    }
  }
}

{
  "mcp.servers": {
    "project": {
      "command": "node",
      "args": ["./mcp-server/index.js"]
    }
  }
}

{
  "mcpServers": {
    "safe-fs": {
      "command": "npx",
      "args": [
        "@modelcontextprotocol/server-filesystem",
        "/safe/directory/only"
      ],
      "sandbox": true
    }
  }
}

import logging
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
@server.tool("query_database") async def query_database(sql: str):     with tracer.start_as_current_span("mcp_query_database") as span:         span.set_attribute("sql.query", sql[:100])
start = time.time()         result = await db.execute(sql)
span.set_attribute("result.count", len(result))         logging.info(f"Query completed in {time.time()-start:.2f}s")
return result

from asyncio import Semaphore
rate_limiter = Semaphore(10)  # Max 10 concurrent
@server.tool("expensive_operation") async def expensive_operation(params):     async with rate_limiter:         return await do_expensive_thing(params)

from functools import lru_cache
@lru_cache(maxsize=1000) def cached_lookup(key: str):     return database.get(key)
@server.resource("data://{key}") async def get_data(key: str):     return Resource(         uri=f"data://{key}",         text=json.dumps(cached_lookup(key))     )