Model Context Protocol (MCP): Complete Technical Guide

from mcp.server.stdio import stdio_server

async def main():
    # stdio_server() returns (read_stream, write_stream) from stdin/stdout
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

@mcp.resource("file://{path}")
async def read_file(path: str) -> str:
    """
    Read a file from the allowed directories.
    The 'path' variable is extracted from the resource URI.
    """
    # ALWAYS resolve and validate before reading
    resolved = resolve_safe_path(path, allowed_roots=["/home/user/project"])
    if resolved is None:
        raise ValueError(f"Path {path!r} is outside allowed directories")
    return resolved.read_text(encoding="utf-8")


@mcp.resource("postgres://mydb/{table}")
async def read_table_schema(table: str) -> str:
    """Return the schema for a given table."""
    schema = await db.get_table_schema(table)
    return schema.to_json()

{
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "Classify this customer message as: billing, technical, general\n\nMessage: My invoice shows double charges for March."
        }
      }
    ],
    "modelPreferences": {
      "hints": [{"name": "claude-haiku-4-5"}],
      "costPriority": 0.8,
      "speedPriority": 0.9
    },
    "systemPrompt": "You are a customer support classifier. Respond with only: billing, technical, or general.",
    "maxTokens": 10
  }
}

from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

# Create the server instance with a descriptive name
app = Server("my-company-server")


@app.list_tools()
async def list_tools() -> list[types.Tool]:
    """Return all tools this server exposes."""
    return [
        types.Tool(
            name="get_customer",
            description=(
                "Retrieve a customer record by ID or email address. "
                "Use when asked about a specific customer's account details, "
                "order history, subscription status, or contact information."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "identifier": {
                        "type": "string",
                        "description": (
                            "Customer ID (format: CUST-xxxxx) "
                            "or email address (e.g. jane@example.com)"
                        )
                    }
                },
                "required": ["identifier"]
            }
        ),
        types.Tool(
            name="list_recent_orders",
            description=(
                "List the most recent orders for a customer. "
                "Use when asked about purchase history or recent activity."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "limit": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 50,
                        "default": 10
                    }
                },
                "required": ["customer_id"]
            }
        )
    ]


@app.call_tool()
async def call_tool(
    name: str,
    arguments: dict
) -> list[types.TextContent]:
    """Dispatch tool calls to their handlers."""
    if name == "get_customer":
        customer = await db.get_customer(arguments["identifier"])
        if not customer:
            return [types.TextContent(
                type="text",
                text=f"No customer found for identifier: {arguments['identifier']!r}"
            )]
        return [types.TextContent(
            type="text",
            text=(
                f"Customer: {customer.name}\n"
                f"Email: {customer.email}\n"
                f"Status: {customer.status}\n"
                f"Since: {customer.created_at.strftime('%Y-%m-%d')}"
            )
        )]

    elif name == "list_recent_orders":
        orders = await db.get_orders(
            arguments["customer_id"],
            limit=arguments.get("limit", 10)
        )
        if not orders:
            return [types.TextContent(type="text", text="No orders found.")]
        lines = [f"- {o.date} | {o.id} | {o.total} | {o.status}" for o in orders]
        return [types.TextContent(type="text", text="\n".join(lines))]

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


async def main():
    async with stdio_server() as streams:
        await app.run(*streams, app.create_initialization_options())


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

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  ListToolsRequestSchema,
  CallToolRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "my-company-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "get_customer",
      description: "Retrieve customer record by ID or email.",
      inputSchema: {
        type: "object",
        properties: {
          identifier: { type: "string", description: "Customer ID or email" }
        },
        required: ["identifier"]
      }
    }
  ]
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  if (name === "get_customer") {
    const customer = await db.getCustomer(args?.identifier as string);
    return {
      content: [{ type: "text", text: customer ? JSON.stringify(customer) : "Not found" }]
    };
  }
  throw new Error(`Unknown tool: ${name}`);
});

const transport = new StdioServerTransport();
await server.connect(transport);

{
  "mcpServers": {
    "my-company": {
      "command": "python",
      "args": ["-m", "my_company_mcp"],
      "env": {
        "DB_URL": "postgresql://user:password@localhost/mydb",
        "API_KEY": "sk-..."
      }
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/home/user/Documents"
      ]
    }
  }
}