Model Context Protocol (MCP) – Complete Tutorial

Introduction

The Model Context Protocol (MCP) is an open standard created by Anthropic that defines how AI applications (like Claude, Cursor, or custom agents) connect to external data sources and tools. Think of MCP as a universal USB port for AI — it provides a standardized way for LLMs to interact with the outside world without building custom integrations for every tool.

Before MCP: Each AI app needed custom code for every integration (N × M problem)

Claude ──custom code──> Slack
Claude ──custom code──> GitHub
Claude ──custom code──> Database
Cursor ──custom code──> Slack     (duplicate work!)
Cursor ──custom code──> GitHub    (duplicate work!)

After MCP: Build once, connect anywhere (N + M problem)

Claude ──MCP──┐
Cursor ──MCP──┼──> MCP Server (Slack)
Agent  ──MCP──┘    MCP Server (GitHub)
                   MCP Server (Database)

Architecture

MCP follows a client-server architecture:

┌─────────────────────────────────────────────────┐
│  HOST (AI Application)                          │
│                                                 │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐     │
│  │MCP Client│  │MCP Client│  │MCP Client│     │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘     │
└───────┼──────────────┼──────────────┼───────────┘
        │              │              │
        ▼              ▼              ▼
  ┌──────────┐  ┌──────────┐  ┌──────────┐
  │MCP Server│  │MCP Server│  │MCP Server│
  │ (GitHub) │  │(Database)│  │  (Slack) │
  └──────────┘  └──────────┘  └──────────┘

Components: - Host — The AI application (Claude Desktop, Cursor, your custom app) - Client — Maintains a 1:1 connection with a server (created by the host) - Server — Exposes tools, resources, and prompts to the client

MCP Primitives

MCP servers expose three types of capabilities:

Primitive	Purpose	Example
Tools	Actions the LLM can execute	create_issue, send_email, query_db
Resources	Data the LLM can read	Files, database records, API responses
Prompts	Reusable prompt templates	"Summarize this PR", "Review this code"

Building Your First MCP Server (Python)

Setup

pip install mcp

Basic MCP Server with Tools

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json

server = Server("my-first-mcp-server")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="Get current weather for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"}
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="calculate",
            description="Evaluate a math expression",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Math expression"}
                },
                "required": ["expression"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments["city"]
        # In production, call a real weather API
        return [TextContent(type="text", text=f"Weather in {city}: 24°C, Partly cloudy")]

    elif name == "calculate":
        result = eval(arguments["expression"])
        return [TextContent(type="text", text=f"Result: {result}")]

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

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

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

Running with Claude Desktop

Add to ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

Restart Claude Desktop — your tools are now available!

MCP Server with Resources

Resources expose data that the AI can read:

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource
import os

server = Server("file-server")

@server.list_resources()
async def list_resources():
    """List available files as resources."""
    docs_dir = "/path/to/documents"
    resources = []
    for filename in os.listdir(docs_dir):
        resources.append(Resource(
            uri=f"file:///{filename}",
            name=filename,
            description=f"Document: {filename}",
            mimeType="text/plain"
        ))
    return resources

@server.read_resource()
async def read_resource(uri: str):
    """Read the content of a file resource."""
    filename = uri.replace("file:///", "")
    filepath = f"/path/to/documents/{filename}"
    with open(filepath, "r") as f:
        content = f.read()
    return content

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

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

MCP Server with Prompts

Prompts are reusable templates the AI can use:

from mcp.types import Prompt, PromptArgument, PromptMessage, TextContent

@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="code_review",
            description="Review code for bugs, style, and best practices",
            arguments=[
                PromptArgument(
                    name="code",
                    description="The code to review",
                    required=True
                ),
                PromptArgument(
                    name="language",
                    description="Programming language",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
    if name == "code_review":
        code = arguments["code"]
        language = arguments.get("language", "unknown")
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"Review this {language} code for bugs, security issues, "
                         f"and best practices. Suggest improvements:\n\n```{language}\n{code}\n```"
                )
            )
        ]

Real-World Use Cases

1. Database Query Server

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import sqlite3

server = Server("database-server")

DB_PATH = "/path/to/company.db"

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_database",
            description="Execute a read-only SQL query against the company database. "
                        "Tables: customers(id, name, email, plan), "
                        "orders(id, customer_id, amount, date, status), "
                        "products(id, name, price, category)",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL SELECT query"}
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="list_tables",
            description="List all tables and their schemas in the database",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()

    if name == "list_tables":
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = cursor.fetchall()
        result = []
        for table in tables:
            cursor.execute(f"PRAGMA table_info({table[0]})")
            cols = cursor.fetchall()
            result.append(f"{table[0]}: {', '.join(c[1] for c in cols)}")
        conn.close()
        return [TextContent(type="text", text="\n".join(result))]

    elif name == "query_database":
        sql = arguments["sql"].strip()
        # Safety: only allow SELECT
        if not sql.upper().startswith("SELECT"):
            return [TextContent(type="text", text="Error: Only SELECT queries allowed")]
        cursor.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        conn.close()
        # Format as table
        header = " | ".join(columns)
        data = "\n".join(" | ".join(str(v) for v in row) for row in rows[:50])
        return [TextContent(type="text", text=f"{header}\n{'─'*len(header)}\n{data}")]

    conn.close()

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

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

2. GitHub Integration Server

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
import os

server = Server("github-server")
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
HEADERS = {"Authorization": f"token {GITHUB_TOKEN}", "Accept": "application/vnd.github.v3+json"}

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="list_repos",
            description="List repositories for a GitHub user or organization",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string", "description": "GitHub username or org"}
                },
                "required": ["owner"]
            }
        ),
        Tool(
            name="get_pull_requests",
            description="List open pull requests for a repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"}
                },
                "required": ["owner", "repo"]
            }
        ),
        Tool(
            name="create_issue",
            description="Create a new issue in a repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "title": {"type": "string"},
                    "body": {"type": "string"}
                },
                "required": ["owner", "repo", "title"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient() as client:
        if name == "list_repos":
            resp = await client.get(
                f"https://api.github.com/users/{arguments['owner']}/repos",
                headers=HEADERS
            )
            repos = resp.json()
            result = "\n".join(f"• {r['name']} - {r.get('description', 'No description')}"
                              for r in repos[:10])
            return [TextContent(type="text", text=result)]

        elif name == "get_pull_requests":
            resp = await client.get(
                f"https://api.github.com/repos/{arguments['owner']}/{arguments['repo']}/pulls",
                headers=HEADERS
            )
            prs = resp.json()
            result = "\n".join(f"#{pr['number']} {pr['title']} by {pr['user']['login']}"
                              for pr in prs[:10])
            return [TextContent(type="text", text=result or "No open PRs")]

        elif name == "create_issue":
            resp = await client.post(
                f"https://api.github.com/repos/{arguments['owner']}/{arguments['repo']}/issues",
                headers=HEADERS,
                json={"title": arguments["title"], "body": arguments.get("body", "")}
            )
            issue = resp.json()
            return [TextContent(type="text", text=f"Created issue #{issue['number']}: {issue['html_url']}")]

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

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

3. Kubernetes Cluster Management

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from kubernetes import client, config

server = Server("k8s-server")
config.load_kube_config()
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="list_pods",
            description="List pods in a namespace",
            inputSchema={
                "type": "object",
                "properties": {
                    "namespace": {"type": "string", "default": "default"}
                }
            }
        ),
        Tool(
            name="get_pod_logs",
            description="Get recent logs from a pod",
            inputSchema={
                "type": "object",
                "properties": {
                    "pod_name": {"type": "string"},
                    "namespace": {"type": "string", "default": "default"},
                    "lines": {"type": "integer", "default": 50}
                },
                "required": ["pod_name"]
            }
        ),
        Tool(
            name="scale_deployment",
            description="Scale a deployment to a specified number of replicas",
            inputSchema={
                "type": "object",
                "properties": {
                    "deployment": {"type": "string"},
                    "replicas": {"type": "integer"},
                    "namespace": {"type": "string", "default": "default"}
                },
                "required": ["deployment", "replicas"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    ns = arguments.get("namespace", "default")

    if name == "list_pods":
        pods = v1.list_namespaced_pod(ns)
        result = "\n".join(
            f"{p.metadata.name} | {p.status.phase} | {p.spec.containers[0].image}"
            for p in pods.items
        )
        return [TextContent(type="text", text=result or "No pods found")]

    elif name == "get_pod_logs":
        logs = v1.read_namespaced_pod_log(
            arguments["pod_name"], ns, tail_lines=arguments.get("lines", 50)
        )
        return [TextContent(type="text", text=logs)]

    elif name == "scale_deployment":
        body = {"spec": {"replicas": arguments["replicas"]}}
        apps_v1.patch_namespaced_deployment_scale(
            arguments["deployment"], ns, body
        )
        return [TextContent(type="text",
            text=f"Scaled {arguments['deployment']} to {arguments['replicas']} replicas")]

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

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

4. Slack Notification Server

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
import os

server = Server("slack-server")
SLACK_TOKEN = os.environ["SLACK_BOT_TOKEN"]

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="send_message",
            description="Send a message to a Slack channel",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {"type": "string", "description": "Channel name (e.g., #general)"},
                    "message": {"type": "string", "description": "Message text"}
                },
                "required": ["channel", "message"]
            }
        ),
        Tool(
            name="list_channels",
            description="List available Slack channels",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient() as client:
        if name == "list_channels":
            resp = await client.get(
                "https://slack.com/api/conversations.list",
                headers={"Authorization": f"Bearer {SLACK_TOKEN}"}
            )
            channels = resp.json()["channels"]
            result = "\n".join(f"#{c['name']} - {c.get('purpose', {}).get('value', '')}"
                              for c in channels[:20])
            return [TextContent(type="text", text=result)]

        elif name == "send_message":
            resp = await client.post(
                "https://slack.com/api/chat.postMessage",
                headers={"Authorization": f"Bearer {SLACK_TOKEN}"},
                json={"channel": arguments["channel"], "text": arguments["message"]}
            )
            if resp.json().get("ok"):
                return [TextContent(type="text", text=f"Message sent to {arguments['channel']}")]
            return [TextContent(type="text", text=f"Error: {resp.json().get('error')}")]

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

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

MCP over REST API (HTTP + SSE Transport)

By default, MCP uses stdio (standard input/output) for local communication. For remote or web-based integrations, MCP supports HTTP with Server-Sent Events (SSE) transport — allowing you to expose your MCP server as a REST API.

Why REST API Transport?

• Deploy MCP servers as web services accessible from anywhere
• Enable multiple clients to connect simultaneously
• Integrate with cloud-hosted AI applications
• Use behind load balancers and API gateways

MCP Server with SSE Transport

# rest_mcp_server.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.requests import Request
from starlette.responses import JSONResponse
import uvicorn

server = Server("rest-mcp-server")
sse = SseServerTransport("/messages/")

# Define tools
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_user",
            description="Get user details by ID",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "string", "description": "The user ID"}
                },
                "required": ["user_id"]
            }
        ),
        Tool(
            name="create_order",
            description="Create a new order for a user",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "string"},
                    "product": {"type": "string"},
                    "quantity": {"type": "integer"}
                },
                "required": ["user_id", "product", "quantity"]
            }
        ),
        Tool(
            name="search_products",
            description="Search products by keyword",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "max_results": {"type": "integer", "default": 10}
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_user":
        # Simulate database lookup
        user = {"id": arguments["user_id"], "name": "John Doe", "email": "john@example.com", "plan": "pro"}
        return [TextContent(type="text", text=str(user))]

    elif name == "create_order":
        order_id = "ORD-12345"
        return [TextContent(type="text",
            text=f"Order {order_id} created: {arguments['quantity']}x {arguments['product']} for user {arguments['user_id']}")]

    elif name == "search_products":
        products = [
            {"name": "Widget Pro", "price": 29.99},
            {"name": "Widget Lite", "price": 9.99}
        ]
        return [TextContent(type="text", text=str(products))]

# SSE endpoint for MCP clients
async def handle_sse(request: Request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await server.run(streams[0], streams[1])

# Health check endpoint
async def health(request: Request):
    return JSONResponse({"status": "ok", "server": "rest-mcp-server"})

# Create ASGI app
app = Starlette(
    routes=[
        Route("/health", health),
        Route("/sse", handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ]
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Running the REST MCP Server

pip install mcp starlette uvicorn
python rest_mcp_server.py
# Server running at http://localhost:8000
# SSE endpoint: http://localhost:8000/sse
# Health check: http://localhost:8000/health

Connecting a Client to the REST MCP Server

# client.py
from mcp.client import ClientSession
from mcp.client.sse import sse_client

async def main():
    async with sse_client("http://localhost:8000/sse") as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print("Available tools:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")

            # Call a tool
            result = await session.call_tool("get_user", {"user_id": "usr_001"})
            print(f"\nUser info: {result.content[0].text}")

            # Search products
            result = await session.call_tool("search_products", {"query": "widget"})
            print(f"Products: {result.content[0].text}")

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

Registering Existing REST APIs as MCP Servers

The most common real-world scenario: you have existing REST APIs and want to expose them via MCP so AI agents can use them.

Pattern: REST API Wrapper

# Wrap any REST API as an MCP server
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.requests import Request
from starlette.responses import JSONResponse
import httpx
import uvicorn

server = Server("api-gateway-mcp")
sse = SseServerTransport("/messages/")

# Configuration: your existing REST APIs
API_BASE = "https://api.yourcompany.com/v1"
API_KEY = "your-api-key"

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_customers",
            description="Fetch customers. Optional filters: status, plan, limit",
            inputSchema={
                "type": "object",
                "properties": {
                    "status": {"type": "string", "enum": ["active", "inactive", "all"]},
                    "plan": {"type": "string", "enum": ["free", "pro", "enterprise"]},
                    "limit": {"type": "integer", "default": 10}
                }
            }
        ),
        Tool(
            name="get_customer_by_id",
            description="Fetch a single customer by their ID",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"}
                },
                "required": ["customer_id"]
            }
        ),
        Tool(
            name="create_customer",
            description="Create a new customer record",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "email": {"type": "string", "format": "email"},
                    "plan": {"type": "string", "enum": ["free", "pro", "enterprise"]}
                },
                "required": ["name", "email", "plan"]
            }
        ),
        Tool(
            name="update_customer",
            description="Update an existing customer's information",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "name": {"type": "string"},
                    "email": {"type": "string"},
                    "plan": {"type": "string"}
                },
                "required": ["customer_id"]
            }
        ),
        Tool(
            name="get_invoices",
            description="Get invoices for a customer",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "status": {"type": "string", "enum": ["paid", "pending", "overdue"]}
                },
                "required": ["customer_id"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

    async with httpx.AsyncClient() as client:
        if name == "get_customers":
            params = {k: v for k, v in arguments.items() if v is not None}
            resp = await client.get(f"{API_BASE}/customers", headers=headers, params=params)

        elif name == "get_customer_by_id":
            resp = await client.get(
                f"{API_BASE}/customers/{arguments['customer_id']}", headers=headers
            )

        elif name == "create_customer":
            resp = await client.post(f"{API_BASE}/customers", headers=headers, json=arguments)

        elif name == "update_customer":
            cid = arguments.pop("customer_id")
            resp = await client.patch(
                f"{API_BASE}/customers/{cid}", headers=headers, json=arguments
            )

        elif name == "get_invoices":
            cid = arguments["customer_id"]
            params = {"status": arguments.get("status")}
            resp = await client.get(
                f"{API_BASE}/customers/{cid}/invoices", headers=headers, params=params
            )

        else:
            return [TextContent(type="text", text=f"Unknown tool: {name}")]

        if resp.status_code >= 400:
            return [TextContent(type="text", text=f"API Error {resp.status_code}: {resp.text}")]

        return [TextContent(type="text", text=resp.text)]

async def handle_sse(request: Request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await server.run(streams[0], streams[1])

async def health(request: Request):
    return JSONResponse({"status": "ok"})

app = Starlette(routes=[
    Route("/health", health),
    Route("/sse", handle_sse),
    Mount("/messages/", app=sse.handle_post_message),
])

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Auto-Generating MCP Server from OpenAPI Spec

If your REST API has an OpenAPI/Swagger spec, you can auto-generate the MCP server:

# openapi_to_mcp.py
import yaml
import httpx
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.requests import Request
from starlette.responses import JSONResponse
import uvicorn

# Load OpenAPI spec
with open("openapi.yaml") as f:
    spec = yaml.safe_load(f)

server = Server("openapi-mcp-server")
sse = SseServerTransport("/messages/")
BASE_URL = spec["servers"][0]["url"]

def build_tools_from_spec(spec):
    """Convert OpenAPI paths to MCP tools."""
    tools = []
    tool_map = {}  # name -> (method, path, params)

    for path, methods in spec.get("paths", {}).items():
        for method, details in methods.items():
            if method not in ["get", "post", "put", "patch", "delete"]:
                continue
            operation_id = details.get("operationId", f"{method}_{path.replace('/', '_')}")
            description = details.get("summary", details.get("description", ""))

            # Build input schema from parameters and request body
            properties = {}
            required = []
            for param in details.get("parameters", []):
                properties[param["name"]] = {
                    "type": param["schema"].get("type", "string"),
                    "description": param.get("description", "")
                }
                if param.get("required"):
                    required.append(param["name"])

            if "requestBody" in details:
                body_schema = details["requestBody"]["content"]["application/json"]["schema"]
                if "properties" in body_schema:
                    properties.update(body_schema["properties"])
                    required.extend(body_schema.get("required", []))

            tools.append(Tool(
                name=operation_id,
                description=description,
                inputSchema={"type": "object", "properties": properties, "required": required}
            ))
            tool_map[operation_id] = (method, path, details.get("parameters", []))

    return tools, tool_map

TOOLS, TOOL_MAP = build_tools_from_spec(spec)

@server.list_tools()
async def list_tools():
    return TOOLS

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name not in TOOL_MAP:
        return [TextContent(type="text", text=f"Unknown tool: {name}")]

    method, path, params = TOOL_MAP[name]

    # Replace path parameters
    url = BASE_URL + path
    query_params = {}
    body = {}
    for param in params:
        value = arguments.get(param["name"])
        if value is None:
            continue
        if param["in"] == "path":
            url = url.replace(f"{{{param['name']}}}", str(value))
        elif param["in"] == "query":
            query_params[param["name"]] = value

    # Remaining arguments go to request body
    for key in arguments:
        if key not in [p["name"] for p in params]:
            body[key] = arguments[key]

    async with httpx.AsyncClient() as client:
        resp = await getattr(client, method)(url, params=query_params, json=body or None)
        return [TextContent(type="text", text=resp.text)]

async def handle_sse(request: Request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await server.run(streams[0], streams[1])

app = Starlette(routes=[
    Route("/health", lambda r: JSONResponse({"status": "ok"})),
    Route("/sse", handle_sse),
    Mount("/messages/", app=sse.handle_post_message),
])

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Client Configuration for Remote MCP Servers

Claude Desktop (remote SSE server):

{
  "mcpServers": {
    "my-api": {
      "url": "http://your-server.com:8000/sse"
    }
  }
}

Cursor IDE:

{
  "mcpServers": {
    "my-api": {
      "url": "http://your-server.com:8000/sse"
    }
  }
}

Deploying MCP REST Servers

Docker Deployment

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "rest_mcp_server:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - API_KEY=${API_KEY}
    restart: unless-stopped

Deploy on AWS (ECS/Fargate)

# Build and push to ECR
aws ecr get-login-password | docker login --username AWS --password-stdin $ECR_URI
docker build -t mcp-server .
docker tag mcp-server:latest $ECR_URI/mcp-server:latest
docker push $ECR_URI/mcp-server:latest

# Deploy with Fargate (use your existing ECS cluster)
aws ecs update-service --cluster prod --service mcp-server --force-new-deployment

Security Best Practices

1. Authentication — Always require API keys or JWT tokens for remote MCP servers
2. Rate Limiting — Prevent abuse with request limits per client
3. Input Validation — Validate all tool arguments before executing
4. Read-Only by Default — Start with read-only tools; add write access carefully
5. Audit Logging — Log every tool call with client identity and arguments
6. Network Security — Use HTTPS in production, restrict access via security groups

# Example: Adding authentication middleware
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path == "/health":
            return await call_next(request)
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if token != os.environ["MCP_AUTH_TOKEN"]:
            return Response("Unauthorized", status_code=401)
        return await call_next(request)

app = Starlette(
    routes=[...],
    middleware=[Middleware(AuthMiddleware)]
)

Summary

Concept	Description
MCP	Open protocol for AI-to-tool communication
Server	Exposes tools, resources, prompts
Client	Connects to servers (managed by host app)
stdio transport	Local communication (default)
SSE transport	Remote/REST-based communication
Tools	Actions the AI can execute
Resources	Data the AI can read
Prompts	Reusable templates

MCP is rapidly becoming the standard for connecting AI to the real world. By building MCP servers around your existing REST APIs, you make them instantly usable by Claude, Cursor, and any MCP-compatible AI agent.