README.md

Build an MCP Server

Overview

MCP is just JSON-RPC 2.0 over stdio (or HTTP).

Getting Started

# Run directly (stdio mode)
python server.py

# Or run as HTTP server (for remote access)
# Change the last line to:
# mcp.run(transport="streamable_http", port=8000)

npx @modelcontextprotocol/inspector python server.py

lean_server.py

#!/usr/bin/env python3
"""
Lean MCP server implementation using only standard library.
"""

import sys
import json

from urllib.request import urlopen
from urllib.error import URLError, HTTPError

# Server metadata
SERVER_NAME = "lean_web_mcp"
SERVER_VERSION = "0.1.0"

# Define your tools
TOOLS = [
    {
        "name": "fetch_page",
        "description": "Fetch content from a URL",
        "inputSchema": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "The URL to fetch"
                }
            },
            "required": ["url"]
        }
    },
    {
        "name": "fetch_headers",
        "description": "Get HTTP headers from a URL",
        "inputSchema": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string", 
                    "description": "The URL to check"
                }
            },
            "required": ["url"]
        }
    }
]


def handle_fetch_page(args):
    """
    fetch_page tool.
    """
    url = args.get("url")
    try:
        with urlopen(url, timeout=30) as response:
            return response.read().decode('utf-8', errors='replace')
    except HTTPError as e:
        return f"HTTP Error: {e.code} {e.reason}"
    except URLError as e:
        return f"URL Error: {e.reason}"
    except Exception as e:
        return f"Error: {str(e)}"


def handle_fetch_headers(args):
    """
    fetch_headers tool.
    """
    url = args.get("url")
    try:
        with urlopen(url, timeout=10) as response:
            headers = dict(response.headers)
            return json.dumps(headers, indent=2)
    except Exception as e:
        return f"Error: {str(e)}"


# Map tool names to handlers
TOOL_HANDLERS = {
    "fetch_page": handle_fetch_page,
    "fetch_headers": handle_fetch_headers,
}


def make_response(id, result):
    """
    Create a JSON-RPC success response.
    """
    return {"jsonrpc": "2.0", "id": id, "result": result}


def make_error(id, code, message):
    """
    Create a JSON-RPC error response.
    """
    return {"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}


def handle_request(request):
    """
    Route JSON-RPC request to appropriate handler.
    """
    method = request.get("method")
    id = request.get("id")
    params = request.get("params", {})

    # Init handshake
    if method == "initialize":
        return make_response(id, {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}
        })

    # List available tools
    elif method == "tools/list":
        return make_response(id, {"tools": TOOLS})

    # Execute a tool
    elif method == "tools/call":
        tool_name = params.get("name")
        tool_args = params.get("arguments", {})
        
        handler = TOOL_HANDLERS.get(tool_name)
        if not handler:
            return make_error(id, -32602, f"Unknown tool: {tool_name}")
        
        try:
            result = handler(tool_args)
            return make_response(id, {
                "content": [{"type": "text", "text": result}]
            })
        except Exception as e:
            return make_response(id, {
                "content": [{"type": "text", "text": f"Error: {str(e)}"}],
                "isError": True
            })

    # Notifications, no response needed
    elif method == "notifications/initialized":
        return None

    # Unknown method
    else:
        return make_error(id, -32601, f"Method not found: {method}")


def main():
    """
    Main loop, which reads JSON-RPC from stdin and write responses to stdout.
    """
    # Unbuffered output
    sys.stdout.reconfigure(line_buffering=True)
    
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        
        try:
            request = json.loads(line)
            response = handle_request(request)
            
            # Notifications do not get responses
            if response is not None:
                print(json.dumps(response), flush=True)
                
        except json.JSONDecodeError as e:
            error = make_error(None, -32700, f"Parse error: {str(e)}")
            print(json.dumps(error), flush=True)


if __name__ == "__main__":
    main()

mcp.json

{
  "mcpServers": {
    "web_fetch": {
      "command": "python",
      "args": ["/path/to/your/server.py"]
    }
  }
}

server.py

#!/usr/bin/env python3

"""
Simple MCP server for fetching website data.
"""

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import Optional
import httpx


mcp = FastMCP("web_fetch_mcp")

# Input validation model
class FetchInput(BaseModel):
    url: str = Field(..., description="URL to fetch")
    selector: Optional[str] = Field(
        default=None, 
        description="Optional CSS selector to extract specific content"
    )

@mcp.tool(
    name="web_fetch_page",
    annotations={
        "readOnlyHint": True,
        "destructiveHint": False,
    }
)
async def fetch_page(params: FetchInput) -> str:
    """
    Fetch a webpage and return its content.
    
    Args:
        params: URL and optional CSS selector
        
    Returns:
        Page content as text (or extracted section if selector provided)
    """
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.get(params.url, follow_redirects=True)
            response.raise_for_status()
            
            content = response.text
            
            # If you want CSS selector support, add beautifulsoup4
            #from bs4 import BeautifulSoup
            #if params.selector:
            #    bs = BeautifulSoup(content, 'html.parser')
            #    elements = bs.select(params.selector)
            #    content = '\n'.join(str(el) for el in elements)
            
            return content
            
        except httpx.HTTPStatusError as e:
            return f"Error: HTTP {e.response.status_code}"
        except httpx.RequestError as e:
            return f"Error: {type(e).__name__} - {str(e)}"


@mcp.tool(name="web_get_headers")
async def get_headers(url: str) -> str:
    """
    Fetch just the HTTP headers from a URL (HEAD request).
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.head(url, follow_redirects=True)
            headers = dict(response.headers)
            import json
            return json.dumps(headers, indent=2)
        except httpx.RequestError as e:
            return f"Error: {str(e)}"


if __name__ == "__main__":
    mcp.run()

test.sh

#!/bin/bash

# Start the server
python server.py

# Then paste these lines
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}

{"jsonrpc":"2.0","method":"notifications/initialized"}

{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}

{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"fetch_page","arguments":{"url":"https://example.com"}}}