Model Context Protocol (MCP)

"""MCP concepts: a client and server exchanging JSON-RPC messages

The Model Context Protocol (MCP) uses JSON-RPC 2.0 messages between a client
(inside the AI application) and a server (which exposes capabilities).
A server can offer three kinds of capabilities:
  * Tools: functions the model can call (tools/list, tools/call)
  * Resources: data the application can read (resources/list, resources/read)
  * Prompts: reusable prompt templates (prompts/list, prompts/get)

Illustrative simulation: this runs both sides in one process and skips the
transport (stdio or HTTP) and parts of the initialization handshake. Message
shapes are simplified. See the next example for the official Python SDK.
"""
import json


class TinyMCPServer:
    """A toy server exposing one tool, one resource, and one prompt."""

    def __init__(self):
        self.tools = {
            "get_course_schedule": {
                "description": "Return the schedule for a course module",
                "inputSchema": {"type": "object", "properties": {"module": {"type": "integer"}}, "required": ["module"]},
            }
        }
        self.resources = {"course://syllabus": "Module 1: Evolution of AI ... Module 10: Governance"}
        self.prompts = {"summarize_module": "Summarize module {module} for a beginner in 3 bullet points."}

    def handle(self, request: dict) -> dict:
        method, params = request["method"], request.get("params", {})
        if method == "initialize":
            result = {"serverInfo": {"name": "course-server"}, "capabilities": {"tools": {}, "resources": {}, "prompts": {}}}
        elif method == "tools/list":
            result = {"tools": [{"name": n, **spec} for n, spec in self.tools.items()]}
        elif method == "tools/call":
            module = params["arguments"]["module"]
            result = {"content": [{"type": "text", "text": f"Module {module} runs on Sunday at 4 pm"}]}
        elif method == "resources/list":
            result = {"resources": [{"uri": u} for u in self.resources]}
        elif method == "resources/read":
            uri = params["uri"]
            result = {"contents": [{"uri": uri, "text": self.resources[uri]}]}
        elif method == "prompts/get":
            text = self.prompts[params["name"]].format(**params["arguments"])
            result = {"messages": [{"role": "user", "content": {"type": "text", "text": text}}]}
        else:
            return {"jsonrpc": "2.0", "id": request["id"], "error": {"code": -32601, "message": "Method not found"}}
        return {"jsonrpc": "2.0", "id": request["id"], "result": result}


class TinyMCPClient:
    def __init__(self, server):
        self.server, self.next_id = server, 1

    def request(self, method: str, params: dict | None = None) -> dict:
        msg = {"jsonrpc": "2.0", "id": self.next_id, "method": method}
        if params:
            msg["params"] = params
        self.next_id += 1
        print("client ->", json.dumps(msg))
        reply = self.server.handle(msg)
        print("server <-", json.dumps(reply), "\n")
        return reply


client = TinyMCPClient(TinyMCPServer())
client.request("initialize")
client.request("tools/list")
client.request("tools/call", {"name": "get_course_schedule", "arguments": {"module": 5}})
client.request("resources/read", {"uri": "course://syllabus"})
client.request("prompts/get", {"name": "summarize_module", "arguments": {"module": 5}})

"""A real MCP server and client with the official Python SDK

This uses the official `mcp` Python SDK (https://github.com/modelcontextprotocol/python-sdk),
version 2 or later. In version 2 the high-level server class is MCPServer
(it was called FastMCP in version 1).
Part 1 writes a small server with one tool, one resource, and one prompt.
Part 2 starts the server as a subprocess over stdio and calls it from a client.

Run it in Google Colab or locally (Python 3.10+). In a plain script, run the
server and client as two separate files; the notebook does both for you.
"""
# %%
# !pip -q install "mcp[cli]>=2"

# %%
# %%writefile course_server.py
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Course Helper")


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b


@mcp.resource("course://modules/{number}")
def module_title(number: int) -> str:
    """Title of a course module."""
    titles = {1: "Evolution of AI", 2: "Components of AI Agents", 5: "Model Context Protocol"}
    return titles.get(int(number), "Unknown module")


@mcp.prompt()
def explain(topic: str) -> str:
    """A reusable prompt template."""
    return f"Explain {topic} to a beginner using one everyday analogy."


if __name__ == "__main__":
    mcp.run()          # stdio transport by default

# %%
# Part 2: a client that launches the server and talks to it.
import asyncio
import sys

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    params = StdioServerParameters(command=sys.executable, args=["course_server.py"])
    # Server log messages go to a file (notebooks do not always expose a real stderr).
    with open("server.log", "w") as log:
        async with stdio_client(params, errlog=log) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()

                tools = await session.list_tools()
                print("Tools:", [t.name for t in tools.tools])

                result = await session.call_tool("add", arguments={"a": 2, "b": 3})
                print("add(2, 3) ->", result.content[0].text)

                resource = await session.read_resource("course://modules/5")
                print("Resource:", resource.contents[0].text)

                prompt = await session.get_prompt("explain", arguments={"topic": "MCP"})
                print("Prompt:", prompt.messages[0].content.text)


asyncio.run(main())