# FastAPI MCP Server Tutorial voor Beginners

# Wat is Model Context Protocol (MCP)?

Model Context Protocol is een standaard die het mogelijk maakt om Large Language Models (LLMs) te verbinden met externe tools en data bronnen. Met MCP kan je LLM in LM Studio gebruik maken van:

• Externe API's
• Databases
• Bestanden op je systeem
• Real-time data
• Custom functies

# Wat gaan we bouwen?

We bouwen stap voor stap een MCP server die:

1. Start als eenvoudige FastAPI app
2. Basis MCP protocol implementeert
3. Een simpele calculator functie toevoegt
4. Bestandsoperaties toevoegt
5. Een database connectie maakt
6. Integreert met LM Studio

# Vereisten

• Python 3.8 of hoger
• Basis kennis van Python
• LM Studio geïnstalleerd
• Een teksteditor (VSCode aanbevolen)

# Stap 1: Project Setup

# 1.1 Maak een nieuwe folder aan

mkdir mijn-mcp-server
cd mijn-mcp-server

# 1.2 Maak een virtual environment

python -m venv venv

# Windows:
venv\Scripts\activate

# Mac/Linux:
source venv/bin/activate

# 1.3 Installeer benodigde packages

pip install fastapi uvicorn python-multipart pydantic

# 1.4 Maak de basis bestanden

Maak deze bestanden in je project folder:

• main.py (hoofdapplicatie)
• requirements.txt (dependencies)
• README.md (documentatie)

requirements.txt:

fastapi==0.104.1
uvicorn==0.24.0
python-multipart==0.0.6
pydantic==2.5.0

# Stap 2: Eerste FastAPI App

# 2.1 Basis FastAPI server

main.py:

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

# Maak FastAPI app instance
app = FastAPI(
    title="Mijn MCP Server",
    description="Een MCP server voor LM Studio integratie",
    version="1.0.0"
)

# Basis health check endpoint
@app.get("/")
async def root():
    return {
        "message": "MCP Server is actief!",
        "status": "running",
        "protocol": "MCP v1.0"
    }

# Health check voor monitoring
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# 2.2 Test je server

python main.py

Open je browser en ga naar: http://localhost:8000

Je zou moeten zien:

{
    "message": "MCP Server is actief!",
    "status": "running", 
    "protocol": "MCP v1.0"
}

🎉 Gefeliciteerd! Je eerste FastAPI server draait!

# Stap 3: MCP Protocol Basis

# 3.1 MCP Message Models

Voeg deze code toe aan main.py (boven de app definitie):

from typing import Dict, List, Any, Optional
from enum import Enum

# MCP Message types
class MCPMessageType(str, Enum):
    INITIALIZE = "initialize"
    TOOLS_LIST = "tools/list"
    TOOLS_CALL = "tools/call"

# Base MCP Message
class MCPMessage(BaseModel):
    jsonrpc: str = "2.0"
    id: Optional[str] = None
    method: str
    params: Optional[Dict[str, Any]] = None

# MCP Response
class MCPResponse(BaseModel):
    jsonrpc: str = "2.0"
    id: Optional[str] = None
    result: Optional[Dict[str, Any]] = None
    error: Optional[Dict[str, Any]] = None

# Tool Definition
class MCPTool(BaseModel):
    name: str
    description: str
    inputSchema: Dict[str, Any]

# 3.2 MCP Initialize Endpoint

Voeg deze endpoint toe:

@app.post("/mcp/initialize")
async def mcp_initialize(message: MCPMessage):
    """Initialiseer MCP verbinding"""
    return MCPResponse(
        id=message.id,
        result={
            "protocolVersion": "1.0.0",
            "serverInfo": {
                "name": "Mijn MCP Server",
                "version": "1.0.0"
            },
            "capabilities": {
                "tools": {}
            }
        }
    )

# 3.3 Test MCP Initialize

Test met curl of een HTTP client:

curl -X POST http://localhost:8000/mcp/initialize \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "1",
    "method": "initialize",
    "params": {}
  }'

# Stap 4: Je Eerste Tool - Calculator

# 4.1 Calculator Tool toevoegen

Voeg deze code toe aan main.py:

import math

# Calculator functie
def calculate(operation: str, a: float, b: float = None) -> float:
    """Voer basis wiskundige operaties uit"""
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else None,
        "power": lambda x, y: x ** y,
        "sqrt": lambda x, y: math.sqrt(x),
        "abs": lambda x, y: abs(x)
    }

    if operation not in operations:
        raise ValueError(f"Onbekende operatie: {operation}")

    if operation in ["sqrt", "abs"]:
        return operations[operation](a, None)
    else:
        if b is None:
            raise ValueError(f"Operatie {operation} heeft twee getallen nodig")
        return operations[operation](a, b)

# Tool Call Model
class ToolCall(BaseModel):
    name: str
    arguments: Dict[str, Any]

# 4.2 Tools List Endpoint

@app.post("/mcp/tools/list")
async def mcp_tools_list(message: MCPMessage):
    """Geef lijst van beschikbare tools terug"""
    tools = [
        {
            "name": "calculator",
            "description": "Voer basis wiskundige operaties uit",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["add", "subtract", "multiply", "divide", "power", "sqrt", "abs"],
                        "description": "De wiskundige operatie om uit te voeren"
                    },
                    "a": {
                        "type": "number",
                        "description": "Het eerste getal"
                    },
                    "b": {
                        "type": "number",
                        "description": "Het tweede getal (optioneel voor sqrt en abs)"
                    }
                },
                "required": ["operation", "a"]
            }
        }
    ]

    return MCPResponse(
        id=message.id,
        result={"tools": tools}
    )

# 4.3 Tool Call Endpoint

@app.post("/mcp/tools/call")
async def mcp_tools_call(message: MCPMessage):
    """Voer een tool uit"""
    try:
        tool_name = message.params.get("name")
        arguments = message.params.get("arguments", {})

        if tool_name == "calculator":
            operation = arguments.get("operation")
            a = arguments.get("a")
            b = arguments.get("b")

            result = calculate(operation, a, b)

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": f"Resultaat: {operation}({a}, {b if b is not None else ''}) = {result}"
                        }
                    ]
                }
            )
        else:
            return MCPResponse(
                id=message.id,
                error={
                    "code": -1,
                    "message": f"Onbekende tool: {tool_name}"
                }
            )

    except Exception as e:
        return MCPResponse(
            id=message.id,
            error={
                "code": -1,
                "message": str(e)
            }
        )

# 4.4 Test de Calculator

# Test tools list
curl -X POST http://localhost:8000/mcp/tools/list \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "2",
    "method": "tools/list"
  }'

# Test calculator
curl -X POST http://localhost:8000/mcp/tools/call \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "3",
    "method": "tools/call",
    "params": {
      "name": "calculator",
      "arguments": {
        "operation": "add",
        "a": 5,
        "b": 3
      }
    }
  }'

🎉 Je hebt nu een werkende calculator tool!

# Stap 5: Bestandsoperaties Tool

# 5.1 File Operations toevoegen

import os
import json
from pathlib import Path

def read_file(filepath: str) -> str:
    """Lees een tekstbestand"""
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        raise Exception(f"Kon bestand niet lezen: {e}")

def write_file(filepath: str, content: str) -> str:
    """Schrijf naar een tekstbestand"""
    try:
        # Maak directory aan als deze niet bestaat
        os.makedirs(os.path.dirname(filepath), exist_ok=True)

        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"Bestand succesvol geschreven naar {filepath}"
    except Exception as e:
        raise Exception(f"Kon bestand niet schrijven: {e}")

def list_files(directory: str) -> List[str]:
    """Lijst bestanden in een directory"""
    try:
        path = Path(directory)
        if not path.exists():
            raise Exception("Directory bestaat niet")

        files = []
        for item in path.iterdir():
            files.append({
                "name": item.name,
                "type": "directory" if item.is_dir() else "file",
                "size": item.stat().st_size if item.is_file() else None
            })
        return files
    except Exception as e:
        raise Exception(f"Kon directory niet lezen: {e}")

# 5.2 Update Tools List

Update de mcp_tools_list functie:

@app.post("/mcp/tools/list")
async def mcp_tools_list(message: MCPMessage):
    """Geef lijst van beschikbare tools terug"""
    tools = [
        {
            "name": "calculator",
            "description": "Voer basis wiskundige operaties uit",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["add", "subtract", "multiply", "divide", "power", "sqrt", "abs"],
                        "description": "De wiskundige operatie om uit te voeren"
                    },
                    "a": {
                        "type": "number",
                        "description": "Het eerste getal"
                    },
                    "b": {
                        "type": "number",
                        "description": "Het tweede getal (optioneel voor sqrt en abs)"
                    }
                },
                "required": ["operation", "a"]
            }
        },
        {
            "name": "read_file",
            "description": "Lees de inhoud van een tekstbestand",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "Pad naar het bestand om te lezen"
                    }
                },
                "required": ["filepath"]
            }
        },
        {
            "name": "write_file", 
            "description": "Schrijf tekst naar een bestand",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "Pad waar het bestand moet worden opgeslagen"
                    },
                    "content": {
                        "type": "string",
                        "description": "De inhoud om naar het bestand te schrijven"
                    }
                },
                "required": ["filepath", "content"]
            }
        },
        {
            "name": "list_files",
            "description": "Toon alle bestanden in een directory",
            "inputSchema": {
                "type": "object", 
                "properties": {
                    "directory": {
                        "type": "string",
                        "description": "Pad naar de directory om te bekijken"
                    }
                },
                "required": ["directory"]
            }
        }
    ]

    return MCPResponse(
        id=message.id,
        result={"tools": tools}
    )

# 5.3 Update Tool Call Handler

Update de mcp_tools_call functie om bestandsoperaties te ondersteunen:

@app.post("/mcp/tools/call")
async def mcp_tools_call(message: MCPMessage):
    """Voer een tool uit"""
    try:
        tool_name = message.params.get("name")
        arguments = message.params.get("arguments", {})

        if tool_name == "calculator":
            operation = arguments.get("operation")
            a = arguments.get("a")
            b = arguments.get("b")

            result = calculate(operation, a, b)

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": f"Resultaat: {operation}({a}, {b if b is not None else ''}) = {result}"
                        }
                    ]
                }
            )

        elif tool_name == "read_file":
            filepath = arguments.get("filepath")
            content = read_file(filepath)

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": f"Inhoud van {filepath}:\n\n{content}"
                        }
                    ]
                }
            )

        elif tool_name == "write_file":
            filepath = arguments.get("filepath")
            content = arguments.get("content")
            result = write_file(filepath, content)

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": result
                        }
                    ]
                }
            )

        elif tool_name == "list_files":
            directory = arguments.get("directory")
            files = list_files(directory)

            files_text = "\n".join([
                f"{'📁' if f['type'] == 'directory' else '📄'} {f['name']}" + 
                (f" ({f['size']} bytes)" if f['size'] is not None else "")
                for f in files
            ])

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": f"Bestanden in {directory}:\n\n{files_text}"
                        }
                    ]
                }
            )
        else:
            return MCPResponse(
                id=message.id,
                error={
                    "code": -1,
                    "message": f"Onbekende tool: {tool_name}"
                }
            )

    except Exception as e:
        return MCPResponse(
            id=message.id,
            error={
                "code": -1,
                "message": str(e)
            }
        )

🎉 Nu kun je bestanden lezen, schrijven en mappen bekijken!

# Stap 6: Database Connectie (SQLite)

# 6.1 Database Dependencies

pip install sqlite3 # (meestal al geïnstalleerd)

Update requirements.txt:

fastapi==0.104.1
uvicorn==0.24.0
python-multipart==0.0.6
pydantic==2.5.0

# 6.2 Database Setup

Voeg toe aan main.py:

import sqlite3
from datetime import datetime

# Database initialisatie
def init_database():
    """Initialiseer de database met een voorbeeldtabel"""
    conn = sqlite3.connect('mcp_data.db')
    cursor = conn.cursor()

    # Maak een notities tabel
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            content TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')

    conn.commit()
    conn.close()

def save_note(title: str, content: str) -> str:
    """Sla een notitie op in de database"""
    try:
        conn = sqlite3.connect('mcp_data.db')
        cursor = conn.cursor()

        cursor.execute(
            'INSERT INTO notes (title, content) VALUES (?, ?)',
            (title, content)
        )

        note_id = cursor.lastrowid
        conn.commit()
        conn.close()

        return f"Notitie opgeslagen met ID: {note_id}"
    except Exception as e:
        raise Exception(f"Kon notitie niet opslaan: {e}")

def get_notes() -> List[Dict]:
    """Haal alle notities op uit de database"""
    try:
        conn = sqlite3.connect('mcp_data.db')
        cursor = conn.cursor()

        cursor.execute('SELECT id, title, content, created_at, updated_at FROM notes ORDER BY created_at DESC')
        rows = cursor.fetchall()

        notes = []
        for row in rows:
            notes.append({
                'id': row[0],
                'title': row[1],
                'content': row[2],
                'created_at': row[3],
                'updated_at': row[4]
            })

        conn.close()
        return notes
    except Exception as e:
        raise Exception(f"Kon notities niet ophalen: {e}")

def delete_note(note_id: int) -> str:
    """Verwijder een notitie uit de database"""
    try:
        conn = sqlite3.connect('mcp_data.db')
        cursor = conn.cursor()

        cursor.execute('DELETE FROM notes WHERE id = ?', (note_id,))

        if cursor.rowcount > 0:
            result = f"Notitie {note_id} succesvol verwijderd"
        else:
            result = f"Notitie {note_id} niet gevonden"

        conn.commit()
        conn.close()

        return result
    except Exception as e:
        raise Exception(f"Kon notitie niet verwijderen: {e}")

# Initialiseer database bij start
init_database()

# 6.3 Database Tools toevoegen

Update je tools list en call handler met deze database tools:

# In mcp_tools_list, voeg toe aan tools array:
        {
            "name": "save_note",
            "description": "Sla een notitie op in de database",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "Titel van de notitie"
                    },
                    "content": {
                        "type": "string", 
                        "description": "Inhoud van de notitie"
                    }
                },
                "required": ["title", "content"]
            }
        },
        {
            "name": "get_notes",
            "description": "Haal alle opgeslagen notities op",
            "inputSchema": {
                "type": "object",
                "properties": {}
            }
        },
        {
            "name": "delete_note",
            "description": "Verwijder een notitie op basis van ID",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "note_id": {
                        "type": "integer",
                        "description": "ID van de notitie om te verwijderen"
                    }
                },
                "required": ["note_id"]
            }
        }

# In mcp_tools_call, voeg toe aan de elif chain:
        elif tool_name == "save_note":
            title = arguments.get("title")
            content = arguments.get("content")
            result = save_note(title, content)

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": result
                        }
                    ]
                }
            )

        elif tool_name == "get_notes":
            notes = get_notes()

            if not notes:
                notes_text = "Geen notities gevonden."
            else:
                notes_text = "Opgeslagen notities:\n\n"
                for note in notes:
                    notes_text += f"**ID {note['id']}: {note['title']}**\n"
                    notes_text += f"{note['content']}\n"
                    notes_text += f"Aangemaakt: {note['created_at']}\n\n"

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": notes_text
                        }
                    ]
                }
            )

        elif tool_name == "delete_note":
            note_id = arguments.get("note_id")
            result = delete_note(note_id)

            return MCPResponse(
                id=message.id,
                result={
                    "content": [
                        {
                            "type": "text",
                            "text": result
                        }
                    ]
                }
            )

🎉 Nu heb je database functionaliteit!

# Stap 7: Integratie met LM Studio

# 7.1 MCP Client voor LM Studio

Maak een nieuw bestand mcp_client.py:

import requests
import json

class MCPClient:
    def __init__(self, server_url="http://localhost:8000"):
        self.server_url = server_url
        self.request_id = 1

    def _send_request(self, method: str, params: dict = None):
        """Stuur een MCP request naar de server"""
        message = {
            "jsonrpc": "2.0",
            "id": str(self.request_id),
            "method": method,
            "params": params or {}
        }

        self.request_id += 1

        response = requests.post(
            f"{self.server_url}/mcp/{method.replace('/', '/')}",
            json=message,
            headers={"Content-Type": "application/json"}
        )

        return response.json()

    def initialize(self):
        """Initialiseer verbinding met MCP server"""
        return self._send_request("initialize")

    def get_tools(self):
        """Krijg lijst van beschikbare tools"""
        return self._send_request("tools/list")

    def call_tool(self, tool_name: str, arguments: dict):
        """Roep een tool aan"""
        return self._send_request("tools/call", {
            "name": tool_name,
            "arguments": arguments
        })

# Voorbeeld gebruik
if __name__ == "__main__":
    client = MCPClient()

    # Test verbinding
    print("=== Initialisatie ===")
    init_result = client.initialize()
    print(json.dumps(init_result, indent=2))

    # Krijg tools
    print("\n=== Beschikbare Tools ===")
    tools_result = client.get_tools()
    print(json.dumps(tools_result, indent=2))

    # Test calculator
    print("\n=== Calculator Test ===")
    calc_result = client.call_tool("calculator", {
        "operation": "multiply",
        "a": 7,
        "b": 6
    })
    print(json.dumps(calc_result, indent=2))

# 7.2 LM Studio Configuratie

1. Open LM Studio
2. Ga naar Settings > Developer
3. Schakel "Enable Extensions" in
4. Voeg je MCP server toe:

Maak een bestand lm_studio_config.json:

{
  "mcpServers": {
    "mijn-mcp-server": {
      "command": "python",
      "args": ["main.py"],
      "cwd": "/pad/naar/jouw/project",
      "env": {}
    }
  }
}

# 7.3 Web Interface voor Testing

Maak web_interface.py:

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
import uvicorn

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/", response_class=HTMLResponse)
async def web_interface(request: Request):
    return """
    <!DOCTYPE html>
    <html>
    <head>
        <title>MCP Server Test Interface</title>
        <style>
            body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
            .tool-section { border: 1px solid #ccc; margin: 10px 0; padding: 15px; border-radius: 5px; }
            input, textarea, select { width: 100%; padding: 8px; margin: 5px 0; }
            button { background: #007cba; color: white; padding: 10px 20px; border: none; border-radius: 3px; cursor: pointer; }
            button:hover { background: #005a8b; }
            .result { background: #f0f0f0; padding: 10px; margin: 10px 0; border-radius: 3px; }
        </style>
    </head>
    <body>
        <h1>MCP Server Test Interface</h1>

        <div class="tool-section">
            <h2>Calculator</h2>
            <select id="calc-operation">
                <option value="add">Optellen</option>
                <option value="subtract">Aftrekken</option>
                <option value="multiply">Vermenigvuldigen</option>
                <option value="divide">Delen</option>
                <option value="power">Macht</option>
                <option value="sqrt">Wortel</option>
                <option value="abs">Absolute waarde</option>
            </select>
            <input type="number" id="calc-a" placeholder="Eerste getal" step="any">
            <input type="number" id="calc-b" placeholder="Tweede getal (optioneel)" step="any">
            <button onclick="callCalculator()">Bereken</button>
            <div id="calc-result" class="result"></div>
        </div>

        <div class="tool-section">
            <h2>Notities</h2>
            <input type="text" id="note-title" placeholder="Titel van notitie">
            <textarea id="note-content" placeholder="Inhoud van notitie" rows="4"></textarea>
            <button onclick="saveNote()">Notitie Opslaan</button>
            <button onclick="getNotes()">Alle Notities Tonen</button>
            <div id="notes-result" class="result"></div>
        </div>

        <div class="tool-section">
            <h2>Bestanden</h2>
            <input type="text" id="file-path" placeholder="Bestandspad">
            <button onclick="readFile()">Bestand Lezen</button>
            <button onclick="listFiles()">Bestanden Tonen</button>
            <div id="file-result" class="result"></div>
        </div>

        <script>
        async function callTool(toolName, arguments) {
            const response = await fetch('/mcp/tools/call', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({
                    jsonrpc: "2.0",
                    id: Date.now().toString(),
                    method: "tools/call",
                    params: { name: toolName, arguments: arguments }
                })
            });
            return await response.json();
        }

        async function callCalculator() {
            const operation = document.getElementById('calc-operation').value;
            const a = parseFloat(document.getElementById('calc-a').value);
            const b = parseFloat(document.getElementById('calc-b').value);

            const result = await callTool('calculator', { 
                operation: operation, 
                a: a, 
                b: isNaN(b) ? null : b 
            });

            document.getElementById('calc-result').innerHTML = 
                result.result ? result.result.content[0].text : 
                'Fout: ' + (result.error ? result.error.message : 'Onbekende fout');
        }

        async function saveNote() {
            const title = document.getElementById('note-title').value;
            const content = document.getElementById('note-content').value;

            if (!title || !content) {
                alert('Vul beide velden in!');
                return;
            }

            const result = await callTool('save_note', { title: title, content: content });

            document.getElementById('notes-result').innerHTML = 
                result.result ? result.result.content[0].text : 
                'Fout: ' + (result.error ? result.error.message : 'Onbekende fout');

            // Clear form
            document.getElementById('note-title').value = '';
            document.getElementById('note-content').value = '';
        }

        async function getNotes() {
            const result = await callTool('get_notes', {});

            document.getElementById('notes-result').innerHTML = 
                result.result ? result.result.content[0].text.replace(/\n/g, '<br>') : 
                'Fout: ' + (result.error ? result.error.message : 'Onbekende fout');
        }

        async function readFile() {
            const filepath = document.getElementById('file-path').value;

            if (!filepath) {
                alert('Voer een bestandspad in!');
                return;
            }

            const result = await callTool('read_file', { filepath: filepath });

            document.getElementById('file-result').innerHTML = 
                result.result ? '<pre>' + result.result.content[0].text + '</pre>' : 
                'Fout: ' + (result.error ? result.error.message : 'Onbekende fout');
        }

        async function listFiles() {
            const directory = document.getElementById('file-path').value || '.';

            const result = await callTool('list_files', { directory: directory });

            document.getElementById('file-result').innerHTML = 
                result.result ? result.result.content[0].text.replace(/\n/g, '<br>') : 
                'Fout: ' + (result.error ? result.error.message : 'Onbekende fout');
        }
        </script>
    </body>
    </html>
    """

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

# Stap 8: Production Ready maken

# 8.1 Environment Configuration

Maak .env bestand:

# Server configuratie
HOST=0.0.0.0
PORT=8000
DEBUG=False

# Database
DATABASE_URL=sqlite:///mcp_data.db

# Security
SECRET_KEY=jouw-geheime-sleutel-hier

# LM Studio
LM_STUDIO_URL=http://localhost:1234

# 8.2 Configuration Management

Maak config.py:

import os
from pydantic import BaseSettings

class Settings(BaseSettings):
    host: str = "0.0.0.0"
    port: int = 8000
    debug: bool = False
    database_url: str = "sqlite:///mcp_data.db"
    secret_key: str = "development-key"
    lm_studio_url: str = "http://localhost:1234"

    class Config:
        env_file = ".env"

settings = Settings()

# 8.3 Logging toevoegen

Update main.py met logging:

import logging
from config import settings

# Logging setup
logging.basicConfig(
    level=logging.INFO if not settings.debug else logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Update je FastAPI app
app = FastAPI(
    title="MCP Server",
    description="Model Context Protocol server voor LM Studio",
    version="1.0.0",
    debug=settings.debug
)

# Add logging to endpoints
@app.middleware("http")
async def log_requests(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    logger.info(
        f"{request.method} {request.url.path} - "
        f"Status: {response.status_code} - "
        f"Time: {process_time:.4f}s"
    )
    return response

# 8.4 Error Handling verbeteren

from fastapi import HTTPException
from fastapi.exception_handlers import http_exception_handler

@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
    logger.error(f"Onverwachte fout: {exc}")
    return MCPResponse(
        error={
            "code": -1,
            "message": "Interne server fout" if not settings.debug else str(exc)
        }
    )

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request, exc):
    logger.warning(f"HTTP Exception: {exc.detail}")
    return await http_exception_handler(request, exc)

# Stap 9: LM Studio Connectie

# 9.1 LM Studio MCP Integration

Maak lm_studio_connector.py:

import asyncio
import json
import websockets
from typing import Dict, Any
import logging

logger = logging.getLogger(__name__)

class LMStudioConnector:
    def __init__(self, mcp_server_url: str = "http://localhost:8000"):
        self.mcp_server_url = mcp_server_url
        self.tools = {}

    async def register_with_lm_studio(self):
        """Registreer MCP server bij LM Studio"""
        try:
            # Haal tools op van onze MCP server
            import requests
            response = requests.post(
                f"{self.mcp_server_url}/mcp/tools/list",
                json={
                    "jsonrpc": "2.0",
                    "id": "1",
                    "method": "tools/list"
                }
            )

            if response.status_code == 200:
                result = response.json()
                self.tools = result.get("result", {}).get("tools", [])
                logger.info(f"Geregistreerd {len(self.tools)} tools bij LM Studio")
                return True
            else:
                logger.error(f"Kon tools niet ophalen: {response.status_code}")
                return False

        except Exception as e:
            logger.error(f"Fout bij registratie met LM Studio: {e}")
            return False

    def get_tool_schema(self) -> Dict[str, Any]:
        """Geef tool schema terug voor LM Studio"""
        return {
            "type": "mcp_server",
            "url": self.mcp_server_url,
            "tools": self.tools
        }

# Gebruik in main.py
connector = LMStudioConnector()

@app.on_event("startup")
async def startup_event():
    """Initialisatie bij opstarten"""
    logger.info("MCP Server wordt opgestart...")

    # Registreer bij LM Studio
    success = await connector.register_with_lm_studio()
    if success:
        logger.info("✅ Succesvol geregistreerd bij LM Studio")
    else:
        logger.warning("⚠️  Kon niet registreren bij LM Studio")

    logger.info("🚀 MCP Server is klaar!")

@app.get("/lm-studio/schema")
async def get_lm_studio_schema():
    """Geef schema terug voor LM Studio integratie"""
    return connector.get_tool_schema()

# 9.2 LM Studio Configuration File

Maak mcp_server.json voor LM Studio:

{
  "name": "Mijn MCP Server",
  "description": "Custom MCP server met calculator, bestanden en database functies",
  "version": "1.0.0",
  "mcp": {
    "transport": {
      "type": "http",
      "url": "http://localhost:8000"
    },
    "capabilities": {
      "tools": true
    }
  },
  "tools": [
    {
      "name": "calculator",
      "description": "Wiskundige berekeningen uitvoeren"
    },
    {
      "name": "read_file", 
      "description": "Bestanden lezen van het systeem"
    },
    {
      "name": "write_file",
      "description": "Bestanden schrijven naar het systeem"  
    },
    {
      "name": "save_note",
      "description": "Notities opslaan in database"
    },
    {
      "name": "get_notes", 
      "description": "Opgeslagen notities ophalen"
    }
  ]
}

# Stap 10: Testing & Deployment

# 10.1 Unit Tests

Maak test_mcp_server.py:

import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_health_check():
    """Test basis health check"""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json()["status"] == "running"

def test_mcp_initialize():
    """Test MCP initialize endpoint"""
    response = client.post("/mcp/initialize", json={
        "jsonrpc": "2.0",
        "id": "1", 
        "method": "initialize"
    })
    assert response.status_code == 200
    result = response.json()
    assert result["result"]["protocolVersion"] == "1.0.0"

def test_tools_list():
    """Test tools list endpoint"""
    response = client.post("/mcp/tools/list", json={
        "jsonrpc": "2.0",
        "id": "2",
        "method": "tools/list"
    })
    assert response.status_code == 200
    result = response.json()
    assert "tools" in result["result"]
    assert len(result["result"]["tools"]) > 0

def test_calculator_tool():
    """Test calculator tool"""
    response = client.post("/mcp/tools/call", json={
        "jsonrpc": "2.0",
        "id": "3",
        "method": "tools/call",
        "params": {
            "name": "calculator",
            "arguments": {
                "operation": "add",
                "a": 5,
                "b": 3
            }
        }
    })
    assert response.status_code == 200
    result = response.json()
    assert "8" in result["result"]["content"][0]["text"]

# Voer tests uit met: pytest test_mcp_server.py

# 10.2 Docker Deployment

Maak Dockerfile:

FROM python:3.11-slim

WORKDIR /app

# Kopieer requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Kopieer applicatie
COPY . .

# Maak database directory
RUN mkdir -p /app/data

# Expose port
EXPOSE 8000

# Start command
CMD ["python", "main.py"]

Maak docker-compose.yml:

version: '3.8'

services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - HOST=0.0.0.0
      - PORT=8000
      - DEBUG=false
    volumes:
      - ./data:/app/data
    restart: unless-stopped

  web-interface:
    build: .
    command: python web_interface.py
    ports:
      - "8001:8001"
    depends_on:
      - mcp-server
    restart: unless-stopped

# 10.3 Production Startup Script

Maak start.sh:

#!/bin/bash
echo "🚀 Starting MCP Server..."

# Activeer virtual environment
source venv/bin/activate

# Installeer dependencies
pip install -r requirements.txt

# Start server
echo "✅ MCP Server gestart op http://localhost:8000"
echo "🌐 Web interface beschikbaar op http://localhost:8001"

# Start beide servers
python main.py &
python web_interface.py &

wait

# Stap 11: Gebruik in LM Studio

# 11.1 LM Studio Setup

1. Start je MCP Server:

   python main.py

2. Open LM Studio

3. Ga naar Chat tab

4. Klik op het Tools icoon (🔧)

5. Add Custom Tool:

   • Name: "Mijn MCP Server"
   • Type: "HTTP API"
   • Base URL: http://localhost:8000
   • Schema: Upload mcp_server.json

# 11.2 Voorbeeld Chat Sessies

Chat 1 - Calculator:

Gebruiker: Kun je 125 * 47 voor me uitrekenen?

LM Studio: Ik ga de calculator tool gebruiken om dat voor je uit te rekenen.

[Tool Call: calculator with operation="multiply", a=125, b=47]

Het resultaat van 125 × 47 = 5875

Chat 2 - Notities:

Gebruiker: Sla een notitie op met de titel "Boodschappen" en de inhoud "Melk, brood, kaas, appels"

LM Studio: Ik ga je notitie opslaan in de database.

[Tool Call: save_note with title="Boodschappen", content="Melk, brood, kaas, appels"]

Je notitie "Boodschappen" is succesvol opgeslagen!

Chat 3 - Bestanden:

Gebruiker: Lees het bestand config.py voor me

LM Studio: Ik ga het bestand config.py voor je lezen.

[Tool Call: read_file with filepath="config.py"]

Hier is de inhoud van config.py:
[bestandsinhoud wordt getoond]

# Stap 12: Uitbreidingen en Tips

# 12.1 Meer Tools Toevoegen

Web Scraping Tool:

import requests
from bs4 import BeautifulSoup

def scrape_website(url: str) -> str:
    """Scrape tekst van een website"""
    try:
        response = requests.get(url, timeout=10)
        soup = BeautifulSoup(response.content, 'html.parser')
        return soup.get_text()[:1000]  # Eerste 1000 karakters
    except Exception as e:
        raise Exception(f"Kon website niet scrapen: {e}")

Weather Tool (met API):

async def get_weather(city: str) -> str:
    """Haal weer informatie op"""
    # Gebruik een weather API zoals OpenWeatherMap
    api_key = "jouw-api-key"
    url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"

    response = requests.get(url)
    data = response.json()

    return f"Het weer in {city}: {data['weather'][0]['description']}, {data['main']['temp']}°C"

# 12.2 Security Tips

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer
import jwt

security = HTTPBearer()

def verify_token(token: str = Depends(security)):
    """Verificeer authentication token"""
    try:
        payload = jwt.decode(token.credentials, settings.secret_key, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token verlopen")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Ongeldige token")

# Gebruik in endpoints:
@app.post("/mcp/tools/call")
async def mcp_tools_call(message: MCPMessage, user=Depends(verify_token)):
    # Tool call logic here
    pass

# 12.3 Performance Tips

from functools import lru_cache
import asyncio

# Cache voor dure operaties
@lru_cache(maxsize=128)
def cached_calculation(operation: str, a: float, b: float):
    return calculate(operation, a, b)

# Async database operaties
import aiosqlite

async def async_save_note(title: str, content: str):
    async with aiosqlite.connect('mcp_data.db') as db:
        await db.execute(
            'INSERT INTO notes (title, content) VALUES (?, ?)',
            (title, content)
        )
        await db.commit()

# Troubleshooting

# Veel Voorkomende Problemen

1. "ModuleNotFoundError"

# Zorg dat je in de juiste virtual environment zit
source venv/bin/activate  # Mac/Linux
venv\Scripts\activate     # Windows

# Installeer dependencies opnieuw
pip install -r requirements.txt

2. "Port already in use"

# Wijzig de port in main.py
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)  # Andere port

3. "Database locked"

# Gebruik connection pooling
import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    conn = sqlite3.connect('mcp_data.db', timeout=20.0)
    try:
        yield conn
    finally:
        conn.close()

4. LM Studio kan server niet vinden

• Controleer of server draait: curl http://localhost:8000
• Check firewall instellingen
• Zorg dat beide applicaties op hetzelfde netwerk draaien

# Volgende Stappen

Nu je een werkende MCP server hebt, kun je:

1. Meer tools toevoegen - Email, SMS, API integraties
2. Database uitbreiden - PostgreSQL, MongoDB
3. Authentication toevoegen - JWT tokens, OAuth
4. Monitoring - Logging, metrics, health checks
5. Deployment - Docker, cloud providers
6. Performance - Caching, async operations

# Conclusie

🎉 Gefeliciteerd! Je hebt succesvol een MCP server gebouwd met FastAPI die:

• ✅ Integreert met LM Studio
• ✅ Calculator functionaliteit biedt
• ✅ Bestandsoperaties uitvoert
• ✅ Database operaties ondersteunt
• ✅ Een web interface heeft
• ✅ Production-ready is

Je LLM in LM Studio kan nu gebruik maken van externe tools en data, wat de mogelijkheden enorm uitbreidt!

# Resources

• FastAPI Documentatie
• Model Context Protocol Specificatie
• LM Studio Documentatie
• Python Virtual Environments

Happy coding! 🚀