Index Ai Academy
Project tutorials · 6 steps

Build an MCP server in Python from scratch

A working Model Context Protocol server with tools, resources and prompts, written without an SDK so you can see exactly what goes over the wire.

What you'll build

The Model Context Protocol is how AI apps connect to tools and data. An SDK can hide the details, but it helps to see them at least once. This tutorial builds a read-only support server for a shop, with tools, resources and prompts, as one function that takes a JSON-RPC message and returns one. It follows the 2026-07-28 version of the protocol.

What you'll learn

  • Plan an MCP server: which tools, resources and prompts, and how each is described
  • Wire the pieces of this phase into one dispatcher that follows the 2026-07-28 rules
  • Serve it over stdio framing and test it end to end through a client
  • Pass a conformance suite and review the server against a security checklist

What you need

Python functions, classes and decorators, and what JSON is. It uses only the standard library, so it runs anywhere with Python 3. In production you'd normally use an official SDK, and knowing what's underneath makes that much easier to debug.

Step 1: Set up the protocol basics

MCP messages are JSON-RPC 2.0: a request has an id and a method, and a response has the same id with a result or an error. In this version each request says which protocol version it's using in _meta, so the server keeps no session. ok and error build the two response shapes, so nothing else has to think about them.

import inspect, json, re
from string import Template

PV = "io.modelcontextprotocol/protocolVersion"
CAPS = "io.modelcontextprotocol/clientCapabilities"
SUPPORTED = ["2026-07-28"]
SERVER_INFO = {"name": "shop-support", "version": "1.0.0"}
CAPABILITIES = {"tools": {}, "resources": {}, "prompts": {}}
INSTRUCTIONS = "Read-only support tools for the shop. Look up the order before drafting replies."
CACHE = {"ttlMs": 300000, "cacheScope": "public"}


class RpcError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code, self.message = code, message


def ok(msg, result):
    return {"jsonrpc": "2.0", "id": msg["id"],
            "result": {"resultType": "complete", **result, "_meta": {"io.modelcontextprotocol/serverInfo": SERVER_INFO}}}


def error(msg, code, message, data=None):
    err = {"code": code, "message": message}
    if data is not None:
        err["data"] = data
    return {"jsonrpc": "2.0", "id": msg.get("id"), "error": err}

Step 2: Build a tool registry

A decorator turns a plain Python function into a tool. It reads the function's signature for the input schema and its docstring for the description, so the two can't drift apart. Notice how call_tool separates two kinds of failure: a call to an unknown tool is a protocol error, while a tool that ran and has bad news returns a normal result with isError, which the model can read and act on.

# ── Data ──
ORDERS = {"ord_51": {"status": "shipped", "items": ["TENT-2"], "total": 249.0},
          "ord_52": {"status": "processing", "items": ["MUG-1", "MUG-1"], "total": 25.0}}
STOCK = {"TENT-2": 12, "MUG-1": 0, "BOOT-9": 4}
ARTICLES = {"doc_shipping": ("Shipping", "We ship in 2 to 4 business days."),
            "doc_returns": ("Returns", "Returns are free within 30 days.")}


# ── Tools (from "Build a server: tools") ──
JSON_TYPES = {str: "string", int: "integer", float: "number", bool: "boolean"}


class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def tool(self, annotations=None):
        def register(fn):
            sig = inspect.signature(fn)
            props = {n: {"type": JSON_TYPES[p.annotation]} for n, p in sig.parameters.items()}
            required = [n for n, p in sig.parameters.items() if p.default is inspect.Parameter.empty]
            d = {"name": fn.__name__, "description": inspect.getdoc(fn) or "",
                 "inputSchema": {"type": "object", "properties": props, "required": required, "additionalProperties": False}}
            if annotations:
                d["annotations"] = annotations
            self.tools[fn.__name__] = (d, fn)
            return fn
        return register

    def list_tools(self, cursor=None):
        if cursor is not None:
            raise RpcError(-32602, "Invalid cursor")
        return {"tools": [self.tools[n][0] for n in sorted(self.tools)], **CACHE}

    def call_tool(self, name, arguments):
        if name not in self.tools:
            raise RpcError(-32602, f"Unknown tool: {name}")
        d, fn = self.tools[name]
        missing = [k for k in d["inputSchema"]["required"] if k not in arguments]
        unknown = [k for k in arguments if k not in d["inputSchema"]["properties"]]
        if missing or unknown:
            return {"content": [{"type": "text", "text": f"Missing {missing} / unknown {unknown} arguments."}], "isError": True}
        value = fn(**arguments)
        if isinstance(value, str):
            return {"content": [{"type": "text", "text": value}], "isError": value.startswith("No ")}
        return {"content": [{"type": "text", "text": json.dumps(value)}], "structuredContent": value, "isError": False}

Rather write it yourself?

In the lesson you write this from a starter, in your browser with nothing to install, and checks tell you what's wrong as you go. A free account opens it and all 244 lessons.

Create a free account

Step 3: Write the tools

The docstring is the prompt: it's all the model knows about when to use the tool. A good error message says what would work, as "Order ids look like ord_51" does, so the model can correct itself. readOnlyHint tells a client that the tool changes nothing.

TOOLS = ToolRegistry()

@TOOLS.tool(annotations={"readOnlyHint": True})
def get_order(order_id: str):
    """Get one order's status, items and total by id, such as ord_51. Use before answering any order question."""
    if order_id not in ORDERS:
        return f"No order {order_id!r}. Order ids look like ord_51."
    return {"order_id": order_id, **ORDERS[order_id]}

@TOOLS.tool(annotations={"readOnlyHint": True})
def check_stock(sku: str):
    """Check how many units of a product are in stock, by sku such as TENT-2."""
    if sku not in STOCK:
        return f"No sku {sku!r}. Known skus: {', '.join(sorted(STOCK))}."
    return {"sku": sku, "in_stock": STOCK[sku]}

Step 4: Add resources and prompts

Tools are for the model to call. Resources are data that an application can read, such as help articles at an address like docs://help/doc_returns. Prompts are templates that a user picks, such as drafting a reply. Having all three in the protocol is what lets a client treat them differently.

# ── Resources and prompts (from "Resources and prompts") ──
class Resources:
    def list(self):
        return {"resources": [{"uri": f"docs://help/{k}", "name": k, "title": t, "mimeType": "text/plain"}
                              for k, (t, _) in sorted(ARTICLES.items())], **CACHE}

    def templates(self):
        return {"resourceTemplates": [{"uriTemplate": "docs://help/{id}", "name": "help_article", "mimeType": "text/plain"}], **CACHE}

    def read(self, uri):
        m = re.fullmatch(r"docs://help/(?P<id>[^/]+)", uri)
        if not m or m["id"] not in ARTICLES:
            raise RpcError(-32602, f"Resource not found: {uri}")
        title, text = ARTICLES[m["id"]]
        return {"contents": [{"uri": uri, "mimeType": "text/plain", "text": f"{title}\n\n{text}"}], **CACHE}


class Prompts:
    PROMPTS = {"draft_reply": {"description": "Draft a reply to a customer about their order.",
                               "arguments": [{"name": "order_id", "required": True}, {"name": "tone", "required": False}],
                               "template": Template("Draft a $tone reply to a customer asking about order $order_id. "
                                                    "Use get_order first, and keep it under 80 words.")}}

    def list(self):
        return {"prompts": [{"name": n, "description": p["description"], "arguments": p["arguments"]}
                            for n, p in self.PROMPTS.items()], **CACHE}

    def get(self, name, arguments):
        p = self.PROMPTS.get(name)
        if p is None:
            raise RpcError(-32602, f"Unknown prompt: {name}")
        if "order_id" not in arguments:
            raise RpcError(-32602, "Missing required argument: order_id")
        text = p["template"].substitute({"tone": "friendly", **arguments})
        return {"description": p["description"], "messages": [{"role": "user", "content": {"type": "text", "text": text}}]}


RESOURCES, PROMPTS = Resources(), Prompts()

Step 5: Route the messages

This is the server. A message with no id is a notification and gets no reply. Missing or unsupported version information gets a precise error code, and an unknown method gets -32601. Anything raised by a handler becomes a clean error, so a bug can never crash the connection.

ROUTES = {
    "server/discover": lambda p: {"supportedVersions": SUPPORTED, "capabilities": CAPABILITIES, "instructions": INSTRUCTIONS},
    "tools/list": lambda p: TOOLS.list_tools(p.get("cursor")),
    "tools/call": lambda p: TOOLS.call_tool(p["name"], p.get("arguments", {})),
    "resources/list": lambda p: RESOURCES.list(),
    "resources/templates/list": lambda p: RESOURCES.templates(),
    "resources/read": lambda p: RESOURCES.read(p["uri"]),
    "prompts/list": lambda p: PROMPTS.list(),
    "prompts/get": lambda p: PROMPTS.get(p["name"], p.get("arguments", {})),
}


def dispatch(msg):
    if "id" not in msg:
        return None
    params = msg.get("params") or {}
    meta = params.get("_meta") or {}
    missing = [k for k in (PV, CAPS) if k not in meta]
    if missing:
        return error(msg, -32602, "Missing required _meta fields: " + ", ".join(missing))
    if meta[PV] not in SUPPORTED:
        return error(msg, -32022, "Unsupported protocol version", {"supported": SUPPORTED, "requested": meta[PV]})
    route = ROUTES.get(msg.get("method"))
    if route is None:
        return error(msg, -32601, f"Method not found: {msg.get('method')}")
    try:
        return ok(msg, route(params))
    except RpcError as e:
        return error(msg, e.code, e.message)
    except Exception:
        return error(msg, -32603, "Internal error")

Step 6: Send it a request

A tools/call request goes in and a full response comes out, as text for the model and as structured data for a program that wants it.

meta = {PV: "2026-07-28", CAPS: {}}
print(dispatch({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "get_order", "arguments": {"order_id": "ord_51"}, "_meta": meta}}))
What it prints
{'jsonrpc': '2.0', 'id': 1, 'result': {'resultType': 'complete', 'content': [{'type': 'text', 'text': '{"order_id": "ord_51", "status": "shipped", "items": ["TENT-2"], "total": 249.0}'}], 'structuredContent': {'order_id': 'ord_51', 'status': 'shipped', 'items': ['TENT-2'], 'total': 249.0}, 'isError': False, '_meta': {'io.modelcontextprotocol/serverInfo': {'name': 'shop-support', 'version': '1.0.0'}}}}

To connect it to a client over stdio, read one JSON message per line from standard input, pass it to dispatch, and write any reply to standard output as one line. Anything else you print to standard output will corrupt the stream, so send logs to standard error.

Where to take it next

  • Serve it over stdio or Streamable HTTP, and connect it to a real client.
  • Add authentication before exposing a remote server. The protocol uses OAuth for this.
  • Treat anything a tool returns as untrusted when it reaches a model, because it can carry prompt injection.

Mistakes to watch for

  • Routing by method before checking the version and metadata.
  • Letting an unexpected exception escape the dispatcher.
  • Shipping write tools without user confirmation.
  • Testing only by hand in one host.
The complete program
import inspect, json, re
from string import Template

PV = "io.modelcontextprotocol/protocolVersion"
CAPS = "io.modelcontextprotocol/clientCapabilities"
SUPPORTED = ["2026-07-28"]
SERVER_INFO = {"name": "shop-support", "version": "1.0.0"}
CAPABILITIES = {"tools": {}, "resources": {}, "prompts": {}}
INSTRUCTIONS = "Read-only support tools for the shop. Look up the order before drafting replies."
CACHE = {"ttlMs": 300000, "cacheScope": "public"}


class RpcError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code, self.message = code, message


def ok(msg, result):
    return {"jsonrpc": "2.0", "id": msg["id"],
            "result": {"resultType": "complete", **result, "_meta": {"io.modelcontextprotocol/serverInfo": SERVER_INFO}}}


def error(msg, code, message, data=None):
    err = {"code": code, "message": message}
    if data is not None:
        err["data"] = data
    return {"jsonrpc": "2.0", "id": msg.get("id"), "error": err}


# ── Data ──
ORDERS = {"ord_51": {"status": "shipped", "items": ["TENT-2"], "total": 249.0},
          "ord_52": {"status": "processing", "items": ["MUG-1", "MUG-1"], "total": 25.0}}
STOCK = {"TENT-2": 12, "MUG-1": 0, "BOOT-9": 4}
ARTICLES = {"doc_shipping": ("Shipping", "We ship in 2 to 4 business days."),
            "doc_returns": ("Returns", "Returns are free within 30 days.")}


# ── Tools (from "Build a server: tools") ──
JSON_TYPES = {str: "string", int: "integer", float: "number", bool: "boolean"}


class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def tool(self, annotations=None):
        def register(fn):
            sig = inspect.signature(fn)
            props = {n: {"type": JSON_TYPES[p.annotation]} for n, p in sig.parameters.items()}
            required = [n for n, p in sig.parameters.items() if p.default is inspect.Parameter.empty]
            d = {"name": fn.__name__, "description": inspect.getdoc(fn) or "",
                 "inputSchema": {"type": "object", "properties": props, "required": required, "additionalProperties": False}}
            if annotations:
                d["annotations"] = annotations
            self.tools[fn.__name__] = (d, fn)
            return fn
        return register

    def list_tools(self, cursor=None):
        if cursor is not None:
            raise RpcError(-32602, "Invalid cursor")
        return {"tools": [self.tools[n][0] for n in sorted(self.tools)], **CACHE}

    def call_tool(self, name, arguments):
        if name not in self.tools:
            raise RpcError(-32602, f"Unknown tool: {name}")
        d, fn = self.tools[name]
        missing = [k for k in d["inputSchema"]["required"] if k not in arguments]
        unknown = [k for k in arguments if k not in d["inputSchema"]["properties"]]
        if missing or unknown:
            return {"content": [{"type": "text", "text": f"Missing {missing} / unknown {unknown} arguments."}], "isError": True}
        value = fn(**arguments)
        if isinstance(value, str):
            return {"content": [{"type": "text", "text": value}], "isError": value.startswith("No ")}
        return {"content": [{"type": "text", "text": json.dumps(value)}], "structuredContent": value, "isError": False}


TOOLS = ToolRegistry()

@TOOLS.tool(annotations={"readOnlyHint": True})
def get_order(order_id: str):
    """Get one order's status, items and total by id, such as ord_51. Use before answering any order question."""
    if order_id not in ORDERS:
        return f"No order {order_id!r}. Order ids look like ord_51."
    return {"order_id": order_id, **ORDERS[order_id]}

@TOOLS.tool(annotations={"readOnlyHint": True})
def check_stock(sku: str):
    """Check how many units of a product are in stock, by sku such as TENT-2."""
    if sku not in STOCK:
        return f"No sku {sku!r}. Known skus: {', '.join(sorted(STOCK))}."
    return {"sku": sku, "in_stock": STOCK[sku]}


# ── Resources and prompts (from "Resources and prompts") ──
class Resources:
    def list(self):
        return {"resources": [{"uri": f"docs://help/{k}", "name": k, "title": t, "mimeType": "text/plain"}
                              for k, (t, _) in sorted(ARTICLES.items())], **CACHE}

    def templates(self):
        return {"resourceTemplates": [{"uriTemplate": "docs://help/{id}", "name": "help_article", "mimeType": "text/plain"}], **CACHE}

    def read(self, uri):
        m = re.fullmatch(r"docs://help/(?P<id>[^/]+)", uri)
        if not m or m["id"] not in ARTICLES:
            raise RpcError(-32602, f"Resource not found: {uri}")
        title, text = ARTICLES[m["id"]]
        return {"contents": [{"uri": uri, "mimeType": "text/plain", "text": f"{title}\n\n{text}"}], **CACHE}


class Prompts:
    PROMPTS = {"draft_reply": {"description": "Draft a reply to a customer about their order.",
                               "arguments": [{"name": "order_id", "required": True}, {"name": "tone", "required": False}],
                               "template": Template("Draft a $tone reply to a customer asking about order $order_id. "
                                                    "Use get_order first, and keep it under 80 words.")}}

    def list(self):
        return {"prompts": [{"name": n, "description": p["description"], "arguments": p["arguments"]}
                            for n, p in self.PROMPTS.items()], **CACHE}

    def get(self, name, arguments):
        p = self.PROMPTS.get(name)
        if p is None:
            raise RpcError(-32602, f"Unknown prompt: {name}")
        if "order_id" not in arguments:
            raise RpcError(-32602, "Missing required argument: order_id")
        text = p["template"].substitute({"tone": "friendly", **arguments})
        return {"description": p["description"], "messages": [{"role": "user", "content": {"type": "text", "text": text}}]}


RESOURCES, PROMPTS = Resources(), Prompts()


ROUTES = {
    "server/discover": lambda p: {"supportedVersions": SUPPORTED, "capabilities": CAPABILITIES, "instructions": INSTRUCTIONS},
    "tools/list": lambda p: TOOLS.list_tools(p.get("cursor")),
    "tools/call": lambda p: TOOLS.call_tool(p["name"], p.get("arguments", {})),
    "resources/list": lambda p: RESOURCES.list(),
    "resources/templates/list": lambda p: RESOURCES.templates(),
    "resources/read": lambda p: RESOURCES.read(p["uri"]),
    "prompts/list": lambda p: PROMPTS.list(),
    "prompts/get": lambda p: PROMPTS.get(p["name"], p.get("arguments", {})),
}


def dispatch(msg):
    if "id" not in msg:
        return None
    params = msg.get("params") or {}
    meta = params.get("_meta") or {}
    missing = [k for k in (PV, CAPS) if k not in meta]
    if missing:
        return error(msg, -32602, "Missing required _meta fields: " + ", ".join(missing))
    if meta[PV] not in SUPPORTED:
        return error(msg, -32022, "Unsupported protocol version", {"supported": SUPPORTED, "requested": meta[PV]})
    route = ROUTES.get(msg.get("method"))
    if route is None:
        return error(msg, -32601, f"Method not found: {msg.get('method')}")
    try:
        return ok(msg, route(params))
    except RpcError as e:
        return error(msg, e.code, e.message)
    except Exception:
        return error(msg, -32603, "Internal error")


meta = {PV: "2026-07-28", CAPS: {}}
print(dispatch({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "get_order", "arguments": {"order_id": "ord_51"}, "_meta": meta}}))

Write it yourself

In the lesson you write this from a starter, in your browser with nothing to install, and checks tell you what's wrong as you go. A free account opens it and all 244 lessons.

Create a free account