What Is the Model Context Protocol (MCP) and How to Build MCP Servers

The Model Context Protocol (MCP) is an open standard that enables LLM assistants to discover, invoke, and safely interact with external tools through a standardized handshake mechanism, OAuth 2.1 scope-based authorization, and Streamable HTTP transport.

The Model Context Protocol (MCP) has emerged as the de-facto standard for connecting large language models to external tools and resources. According to the rohitg00/ai-engineering-from-scratch repository, MCP replaces ad-hoc tool schemas with a unified discovery and security framework that supports enterprise-scale deployments. This guide explains the core architecture and provides production-ready code for building your own MCP servers.

What Is the Model Context Protocol?

MCP is a specification that standardizes how AI assistants discover and use external capabilities. Instead of hard-coding tool definitions for each provider, an MCP server publishes a capability manifest that describes every tool, its required OAuth scopes, destructiveness flags, and JSON Schema inputs. This allows any compatible client—whether Claude, GPT-4, or Gemini—to dynamically understand and invoke tools without prior configuration.

The protocol specifies Streamable HTTP as the default transport, enabling stateless, low-latency JSON-RPC communication that scales horizontally. By 2026, MCP is shipped by Anthropic, OpenAI, Google, and most IDE extensions, making it the lingua franca for tool-use in generative AI applications.

Core Architecture of MCP Servers

Capability Manifest and Handshake

Every MCP server exposes a .well-known/mcp-capabilities endpoint that returns a manifest describing available tools. In phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py (lines 48–61), the server constructs this document dynamically:

• Tool name and description
• Required OAuth 2.1 scope (e.g., postgres:query:readonly)
• Destructiveness flag (boolean indicating if the tool modifies state)
• Input JSON Schema for validation

This manifest enables zero-configuration discovery by clients and registries.

Streamable HTTP Transport

MCP uses Streamable HTTP as its wire format. The server exposes a single POST endpoint that receives JSON-RPC requests, streams JSON responses, and maintains open connections for server-sent notifications. This design is stateless, making horizontal scaling trivial. The transport layer is implemented in phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py (lines 3–5).

Scope-Based Authorization

Security is enforced through OAuth 2.1 scopes. Each tool declares a required scope in its schema, and the client must supply an access token containing that scope. Destructive tools require an additional time-bounded approved:by:human scope. As shown in lines 68–78 of the main server file, the server validates these claims before execution.

OPA-Style Policy Decisions

Before invoking any tool, the server runs a lightweight policy function inspired by Open Policy Agent (OPA). Located at phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py (lines 85–97), this function checks:

• Token scope membership
• Fresh human approval for destructive operations
• Payload size limits (default 8 KB maximum)

The policy returns a boolean decision and a human-readable reason string.

Audit Logging and Data Redaction

Every request—allowed or denied—is logged as structured JSON Lines (JSONL) entries. Sensitive fields such as email addresses, SSNs, and phone numbers are redacted using regex patterns similar to Microsoft Presidio. This audit pipeline appears in phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py (lines 104–119).

Registry and Discovery

A separate registry service polls the .well-known/mcp-capabilities endpoints of registered servers, validates manifests against the MCP specification (2025-11-25), and indexes tools for global discovery. The registry implementation in lines 150–164 supports search(query) operations, forming the backbone of enterprise deployments.

Building an MCP Server: Step-by-Step Implementation

Step 1: Define Tool Schemas

Create a dataclass to represent tool metadata and a type alias for handlers. This example from phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py (lines 71–76) defines a read-only Postgres query tool:

from dataclasses import dataclass
from typing import Callable

@dataclass
class ToolSchema:
    name: str
    required_scope: str
    destructive: bool
    description: str
    input_schema: dict

Handler = Callable[[dict], dict]

# Define a read-only tool

postgres_tool = ToolSchema(
    name="postgres.readonly",
    required_scope="postgres:query:readonly",
    destructive=False,
    description="Read-only Postgres query",
    input_schema={"type": "object", "properties": {"sql": {"type": "string"}}},
)

def postgres_handler(args: dict) -> dict:
    return {"rows": [[1]], "sql_echo": args.get("sql", "")}

Step 2: Implement the MCPServer Class

The server class manages tool registration and generates the capability manifest. From lines 37–62:

from dataclasses import dataclass, field

@dataclass
class MCPServer:
    name: str
    url: str
    tools: dict[str, ToolSchema] = field(default_factory=dict)
    handlers: dict[str, Handler] = field(default_factory=dict)

    def register(self, schema: ToolSchema, handler: Handler) -> None:
        self.tools[schema.name] = schema
        self.handlers[schema.name] = handler

    def capabilities(self) -> dict:
        return {
            "server": self.name,
            "transport": "streamable_http",
            "url": self.url,
            "tools": [
                {
                    "name": t.name,
                    "scope": t.required_scope,
                    "destructive": t.destructive,
                    "description": t.description,
                    "input_schema": t.input_schema,
                }
                for t in self.tools.values()
            ],
        }

Step 3: Add Policy Enforcement

Implement the OPA-style policy check (lines 85–97):

import json
import time

def policy_decide(server: MCPServer, tool: str, token, args: dict, now: float) -> tuple[bool, str]:
    if tool not in server.tools:
        return False, f"no such tool: {tool}"
    schema = server.tools[tool]
    if not token.has_scope(schema.required_scope):
        return False, f"missing scope: {schema.required_scope}"
    if schema.destructive and not token.fresh_approval(now):
        return False, "destructive tool requires fresh human approval"
    if len(json.dumps(args)) > 8192:
        return False, "payload too large (> 8 KB)"
    return True, "ok"

Step 4: Handle Dispatch and Audit Logging

The dispatch function orchestrates policy checks, execution, and audit trails (lines 126–143):

def dispatch(server: MCPServer, token, tool: str, args: dict, audit: list) -> dict:
    now = time.time()
    ok, reason = policy_decide(server, tool, token, args, now)
    if not ok:
        audit.append({"ts": now, "user": token.user, "tool": tool, 
                      "status": f"denied:{reason}", "args": args})
        return {"error": {"code": 403, "message": reason}}

    handler = server.handlers[tool]
    try:
        result = handler(args)
        audit.append({"ts": now, "user": token.user, "tool": tool, 
                      "status": "ok", "args": args, "result": result})
        return {"result": result}
    except Exception as exc:
        audit.append({"ts": now, "user": token.user, "tool": tool, 
                      "status": f"error:{exc}", "args": args})
        return {"error": {"code": 500, "message": str(exc)}}

Step 5: Assemble and Register Servers

Combine read-only and destructive servers with a registry (lines 171–190):

def build_readonly_server() -> MCPServer:
    s = MCPServer(name="internal-readonly-mcp",
                  url="https://mcp.internal/readonly")
    s.register(
        ToolSchema("postgres.readonly", "postgres:query:readonly", False,
                   "Read-only Postgres query",
                   {"type": "object", "properties": {"sql": {"type": "string"}}}),
        lambda a: {"rows": [[1]], "sql_echo": a.get("sql", "")},
    )
    return s

def build_destructive_server() -> MCPServer:
    s = MCPServer(name="internal-destructive-mcp",
                  url="https://mcp.internal/destructive")
    s.register(
        ToolSchema("jira.create", "jira:write", True,
                   "Create Jira issue",
                   {"type": "object", "properties": {"title": {"type": "string"}}}),
        lambda a: {"id": "PROJ-99", "created": True},
    )
    return s

# Production note: Replace with persistent storage and JWT introspection

registry = Registry()
registry.register(build_readonly_server())
registry.register(build_destructive_server())

Summary

• MCP standardizes tool discovery through a .well-known/mcp-capabilities manifest that eliminates hard-coded integrations.
• Streamable HTTP provides stateless, scalable transport for JSON-RPC communication between clients and servers.
• OAuth 2.1 scopes enforce least-privilege access, with additional human-approval requirements for destructive operations.
• OPA-style policies enable fine-grained authorization decisions based on token claims, payload size, and tool destructiveness.
• Audit logging with redaction ensures compliance by recording all requests and removing sensitive PII before storage.
• Registry services enable enterprise-scale discovery by indexing multiple MCP servers and validating their manifests.

Frequently Asked Questions

What is the Model Context Protocol used for?

The Model Context Protocol enables LLM assistants to discover and invoke external tools—such as databases, APIs, and file systems—through a standardized interface. It allows AI systems to dynamically understand tool capabilities without hard-coded configurations, supporting use cases from read-only data queries to destructive operations like creating Jira tickets or modifying database records.

How does MCP authentication differ from traditional API keys?

MCP uses OAuth 2.1 scope-based authorization rather than simple API keys. Each tool declares required scopes (e.g., postgres:query:readonly), and the client must present an access token containing those scopes. Destructive tools additionally require a fresh approved:by:human scope that expires after a short time window, adding a human-in-the-loop safety mechanism absent from traditional API authentication.

What security features are built into MCP servers?

MCP servers implement defense in depth through multiple layers: OPA-style policy decisions that validate scopes and payload sizes before execution; audit logging that records every request as structured JSONL with sensitive data redacted; and capability manifests that explicitly flag destructive operations. The protocol also references RFC 7591 and RFC 8707 for OAuth implementation standards.

Can I build an MCP server in any programming language?

Yes. While the rohitg00/ai-engineering-from-scratch repository provides Python examples in phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py, MCP is transport-agnostic and uses standard JSON-RPC over HTTP. You can implement the protocol in any language that supports HTTP servers and JSON processing, including TypeScript, Go, Rust, or Java. For production use, validate your implementation against the official MCP conformance suite.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}