How MCP servers work: a beginner's guide

MCP servers expose tools, resources, and prompts to LLM clients through a standardized JSON-RPC 2.0 interface. If you’re building an integration that lets an agent query your database, trigger a deployment, or search internal docs, this is the protocol you implement. The spec is small but the operational details — transport, capability negotiation, error handling — are where things break.

What an MCP server actually does

An MCP server is a JSON-RPC 2.0 service that speaks three primitive types: tools (functions the model can call), resources (read-only data the model can fetch), and prompts (templated interactions). The client — typically an agent framework or IDE extension — discovers capabilities at startup via initialize and tools/list, resources/list, prompts/list. After that, the model decides what to call and the server executes.

The transport layer is pluggable. The spec defines stdio (subprocess pipes) and HTTP+SSE. Stdio is simpler for local tools; HTTP is required for remote servers and multi-tenant deployments. Most production servers end up supporting both.

# Minimal stdio server skeleton (Python, using mcp library)
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("my-internal-tools")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query_postgres",
            description="Run a read-only SQL query against the analytics warehouse",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SELECT statement only"}
                },
                "required": ["sql"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "query_postgres":
        sql = arguments["sql"]
        # validate, execute, format results
        rows = run_readonly_query(sql)
        return [TextContent(type="text", text=format_rows(rows))]
    raise ValueError(f"Unknown tool: {name}")

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

Capability negotiation and versioning

The initialize handshake is where clients and servers agree on protocol version and capabilities. A client sends:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {},
      "resources": {},
      "prompts": {}
    },
    "clientInfo": {"name": "my-agent", "version": "1.0.0"}
  }
}

The server responds with its supported version and capabilities. If versions mismatch, the client should degrade gracefully — but many don’t. Pin your server to a specific protocol version and test against the client versions you support. The spec evolves; 2024-11-05 is current as of this writing.

Pitfall: Clients often omit clientInfo or send malformed capability objects. Validate incoming initialize params and return a proper JSON-RPC error (-32602 invalid params) rather than crashing.

Tools: the execution surface

Tools are the primary way models affect state. Each tool has a name, description, and JSON Schema for inputs. The description is prompt engineering — it’s what the model reads to decide whether to call your tool. Be specific.

{
  "name": "create_jira_ticket",
  "description": "Create a Jira ticket in the ENG project. Requires summary and description. Optionally accepts assignee (email) and labels (array of strings). Returns the ticket key (e.g., ENG-4231).",
  "inputSchema": {
    "type": "object",
    "properties": {
      "summary": {"type": "string", "maxLength": 200},
      "description": {"type": "string"},
      "assignee": {"type": "string", "format": "email"},
      "labels": {"type": "array", "items": {"type": "string"}}
    },
    "required": ["summary", "description"]
  }
}

Tradeoff: Rich schemas help models call tools correctly but increase token overhead. For high-frequency tools, consider a simpler schema and validate in the handler. For destructive operations, always require explicit confirmation fields — models hallucinate required params.

Tool execution lifecycle

1. Client sends tools/call with name and arguments
2. Server validates arguments against the schema (do this yourself; the protocol doesn’t enforce it)
3. Server executes the operation — this is your code
4. Server returns CallToolResult with content array (text, image, or embedded resource)

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    try:
        validated = ToolInputSchema(**arguments)  # pydantic validation
    except ValidationError as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Invalid arguments: {e}")],
            isError=True
        )
    
    result = await execute_tool(name, validated)
    return CallToolResult(content=[TextContent(type="text", text=result)])

Return isError: true for expected failures (validation, not found, permission denied). Reserve JSON-RPC errors for protocol violations. This distinction lets clients surface user-facing messages vs. retry logic.

Resources: read-only context

Resources are URI-addressable data the model can read without side effects. Think: file contents, API responses, database snapshots. The client calls resources/read with a URI; the server returns content and MIME type.

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="postgres://analytics/customers/recent",
            name="Recent customers",
            description="Last 100 customers from the analytics warehouse",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> ReadResourceResult:
    if uri == "postgres://analytics/customers/recent":
        data = fetch_recent_customers(100)
        return ReadResourceResult(contents=[
            ResourceContents(uri=uri, mimeType="application/json", text=json.dumps(data))
        ])
    raise ValueError(f"Unknown resource: {uri}")

Pitfall: Resources can be large. The protocol supports pagination via resources/templates but most clients don’t implement it yet. If your resource exceeds ~50KB, consider a tool that accepts query parameters instead.

Resource templates for parameterized access

Resource templates let clients discover parameterized URIs:

@app.list_resource_templates()
async def list_templates() -> list[ResourceTemplate]:
    return [
        ResourceTemplate(
            uriTemplate="postgres://analytics/customers/{customer_id}",
            name="Customer by ID",
            description="Fetch a single customer record",
            mimeType="application/json"
        )
    ]

The client fills in {customer_id} and calls resources/read. This is cleaner than tools for pure reads, but tool support is more mature in current clients.

Prompts: reusable interaction patterns

Prompts are parameterized templates the client can invoke to structure a conversation. They’re less common in practice but useful for standardized workflows — “analyze this error log,” “write a PR description from this diff.”

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="analyze_error",
            description="Analyze an error log and suggest fixes",
            arguments=[
                PromptArgument(name="log", description="Error log text", required=True),
                PromptArgument(name="context", description="Additional context", required=False)
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    if name == "analyze_error":
        log = arguments["log"]
        context = arguments.get("context", "")
        return GetPromptResult(
            description="Error analysis",
            messages=[
                PromptMessage(role="user", content=TextContent(
                    type="text",
                    text=f"Analyze this error and suggest fixes:\n\n{log}\n\nContext: {context}"
                ))
            ]
        )
    raise ValueError(f"Unknown prompt: {name}")

Transport: stdio vs HTTP+SSE

Stdio (local subprocess)

The client spawns your server as a child process and communicates over stdin/stdout. Each line is a complete JSON-RPC message. This works well for:

• Developer tools (IDE extensions, CLI agents)
• Single-user, ephemeral servers
• Environments where network ports are restricted

# Client launches server like this
npx -y @modelcontextprotocol/server-postgres \
  --connection-string "$DATABASE_URL"

Operational note: The client owns the process lifecycle. If your server crashes, the client restarts it. Design for fast startup (<500ms) and statelessness.

HTTP+SSE (remote servers)

For multi-tenant, persistent, or network-accessible servers, use HTTP. The client POSTs to /mcp for requests; the server streams responses via SSE on the same connection.

# FastAPI + MCP HTTP transport
from mcp.server.streamable_http import StreamableHTTPSessionManager
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()
session_manager = StreamableHTTPSessionManager(app, "/mcp")

@app.post("/mcp")
async def mcp_endpoint(request: Request):
    return await session_manager.handle_request(request)

Pitfall: SSE connections are long-lived. Configure your load balancer/proxy timeouts accordingly (at least 5 minutes idle). Handle client disconnects gracefully — cancel in-flight tool calls when the SSE stream closes.

Authentication and authorization

The MCP spec doesn’t define auth. For HTTP transport, you add it at the transport layer:

# Bearer token validation middleware
async def verify_token(request: Request):
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(401, "Missing bearer token")
    token = auth[7:]
    if not validate_token(token):
        raise HTTPException(403, "Invalid token")
    request.state.user = get_user_from_token(token)

For stdio, auth happens out-of-band (SSH, local user context, mTLS sidecar). Don’t invent custom auth in the JSON-RPC layer — it breaks client compatibility.

Error handling that doesn’t suck

JSON-RPC 2.0 defines standard error codes. Use them:

For tool-level errors, return success (JSON-RPC 2.0 result) with isError: true in the payload:

{
  "jsonrpc": "2.0",
  "id": 42,
  "result": {
    "content": [{"type": "text", "text": "Customer not found: CUST-999"}],
    "isError": true
  }
}

This lets the model see the error and decide whether to retry, ask the user, or try a different approach.

Common mistake: Throwing exceptions that become -32603 internal errors. The model sees “internal error” and has no actionable information. Catch expected failures and return structured tool errors.

Testing your server

Test the protocol directly, not just your business logic. The mcp CLI can act as a client:

# List tools
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' | python -m my_server

# Call a tool
echo '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"query_postgres","arguments":{"sql":"SELECT 1"}}}' | python -m my_server

For HTTP servers, use curl with SSE:

curl -N -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{}}}'

Automate this in CI. Protocol conformance tests catch breaking changes before your clients do.

Deployment patterns

Sidecar per user (stdio)

Each developer runs their own server instance. Zero shared state, simple scaling. Used by Cursor, Windsurf, Claude Desktop.

Shared HTTP pool

A fleet of stateless servers behind a load balancer. Requires:

• Session affinity (sticky sessions) for SSE, or
• Stateless design with external session store (Redis)

# Stateless tool execution with Redis session store
import redis
r = redis.Redis()

@app.call_tool()
async def call_tool(name: str, arguments: dict, session_id: str) -> CallToolResult:
    # Session state in Redis, not memory
    ctx = json.loads(r.get(f"mcp:{session_id}") or "{}")
    result = await execute_with_context(name, arguments, ctx)
    r.setex(f"mcp:{session_id}", 3600, json.dumps(result.context))
    return result

Gateway mode

An MCP server that proxies to other MCP servers, aggregating tools. This is where n4n.ai fits — it can act as an MCP gateway that routes tool calls to downstream providers while handling auth, rate limiting, and observability centrally.

Observability

Log every tool call with: session ID, tool name, arguments (redacted), latency, success/error. Emit structured JSON logs:

{
  "timestamp": "2024-12-15T10:23:45.123Z",
  "level": "INFO",
  "session_id": "sess_abc123",
  "tool": "query_postgres",
  "duration_ms": 142,
  "status": "success",
  "rows_returned": 47
}

Trace IDs should propagate from the client through your server to downstream systems. The MCP spec doesn’t mandate this — add a trace_id field to your tool arguments convention and thread it through.

Common pitfalls checklist

• No input validation — models send malformed args. Validate with Pydantic/Zod/equivalent.
• Blocking the event loop — synchronous DB calls in async handlers. Use thread pools or async drivers.
• Unbounded resource reads — returning 10MB JSON crashes clients. Implement pagination or size limits.
• Leaking secrets in errors — stack traces with API keys. Sanitize error messages.
• Ignoring client disconnects — long-running tools keep executing after SSE closes. Check request.is_disconnected().
• Version drift — server updates break older clients. Version your tool schemas and support multiple versions simultaneously.
• No timeouts — a hung downstream API blocks the server forever. Set aggressive timeouts (30s default) on all external calls.

What’s next

The protocol is stabilizing but clients vary in capability support. Test against the specific clients your users run. If you’re building a platform that serves multiple MCP servers, consider a gateway layer that normalizes capabilities, enforces policies, and provides a single endpoint for clients — this avoids the N×M integration problem as you add servers and clients.

Start with stdio for local development. Move to HTTP when you need multi-user access, centralized auth, or observability. Keep your tool handlers pure and testable; the transport layer is interchangeable.