Lokaler MCP-Server für Flottentelemetrie

Projekt 1 des Kurses CC3067 Redes, Sektion 10, Universidad del Valle de Guatemala. Fernando Hernández.

Ein MCP-Server (Model Context Protocol), der auf dem Rechner des Bedieners läuft und die Telemetrieabfragen einer Fahrzeugflotte als Werkzeuge bereitstellt, die ein Sprachmodell aufrufen kann. Mit ihm beantwortet ein Chatbot Fragen wie „Wo ist das P-123BCD?“ oder „Welches Fahrzeug hat diese Woche die meisten Kilometer zurückgelegt?“, ohne dass der Benutzer die Tracking-Plattform öffnen muss.

Das Protokoll ist von Grund auf über stdio mit der Standardbibliothek von Python implementiert. Ich verwende weder das MCP-SDK noch eine Bibliothek, die JSON-RPC verarbeitet; das ist die zentrale Anforderung des Projekts, und ich erkläre es im Abschnitt Implementierung des Protokolls.

Was ist MCP

MCP ist ein Anwendungsschichtprotokoll, das standardisiert, wie ein Sprachmodell externe Werkzeuge entdeckt und aufruft. Ein MCP-Server veröffentlicht eine Liste von Werkzeugen, jedes mit einem Namen, einer Beschreibung und einem JSON-Schema seiner Parameter. Der Client (z. B. Claude Desktop) ruft diese Liste ab, zeigt sie dem Modell, und wenn das Modell beschließt, ein Werkzeug zu verwenden, ruft der Client es mit den vom Modell gewählten Argumenten auf und gibt das Ergebnis zurück, damit es in natürlicher Sprache erklärt wird.

Mechanisch ist MCP JSON-RPC 2.0 über einen Transport. In diesem Projekt ist der Transport stdio: Der Client startet den Server als Kindprozess, und die beiden tauschen durch Zeilenumbrüche getrennte JSON-Objekte über stdin und stdout aus. Die Sitzung beginnt mit einem Handshake (initialize → Antwort → notifications/initialized), danach kann der Client tools/list, tools/call und ping aufrufen.

Related MCP server: NL-to-SQL MCP

Warum eine Flotte, und warum lokal

Unternehmen mit eigener Flotte haben bereits GPS in ihren Fahrzeugen und eine Tracking-Plattform; die Daten existieren und sind vollständig. Das Problem ist der Zugriff: Heute muss man durch Dashboards navigieren, Filter anwenden und Berichte erstellen, und wer den Betrieb am besten kennt, beherrscht oft die Plattform am wenigsten.

Der Server läuft bewusst lokal und nicht nur wegen der Kursanforderung: Die Positionen einer Flotte offenbaren Handelsrouten, Kunden und Zeitpläne. Mit dem Server auf dem Rechner des Bedieners reist zum Modell nur das aggregierte Ergebnis jeder Abfrage, niemals der Positionsverlauf.

Voraussetzungen

• Python 3.11 oder höher

• git

• Optional: ein Google-Maps-Schlüssel (GOOGLE_MAPS_API_KEY) für die Geokodierung mit Google und zum Neugenerieren von Routen. Ohne ihn funktioniert alles genauso.

Installation

git clone https://github.com/FerAHMz/mcp-local-redes.git
cd mcp-local-redes
python3.11 -m venv .venv
source .venv/bin/activate        # en Windows: .venv\Scripts\activate
pip install -r requirements.txt

Die Datenbank erzeugen

Ich verwende keine echten Daten von irgendeinem Unternehmen. Der Generator simuliert 15 Fahrzeuge über 7 Tage auf echten Routen des Ballungsraums von Guatemala, einen Bericht alle 15 Sekunden innerhalb der Arbeitszeit jedes Fahrzeugs, mit GPS-Rauschen von σ ≈ 5 m und injizierten Ereignissen, die ich im Voraus kenne (längere Stopps, Geschwindigkeitsüberschreitungen, Signalverluste, Geo-Fence-Ein- und -Austritte).

python datos/generador.py

Erzeugt datos/flota.db (SQLite, ~160 000 Positionen und ~1 300 Ereignisse) in ein paar Sekunden. Standardmäßig endet der Datensatz in dem Moment, in dem er erzeugt wird, sodass sich „heute“ und „gestern“ in den Fragen auf echte Daten beziehen. Für einen reproduzierbaren Datensatz wird der Endzeitpunkt festgelegt:

python datos/generador.py --ahora 2026-08-19T15:30

Es ist ratsam, es während der Arbeitszeit zu erzeugen (oder ein --ahora mit Arbeitszeit zu übergeben), damit es bei unidades_detenidas Fahrzeuge auf der Strecke gibt und nicht nur ausgeschaltete.

Routen: Offline-Modus und Modus mit API

Die Basisrouten sind in datos/rutas/*.json als codierte Polylinien gespeichert (das gleiche Format, das die Google Directions API zurückgibt), zusammen mit den Stopps jeder Route. Der Generator liest sie von dort und benötigt weder Netzwerk noch Schlüssel.

Um sie erneut von der Directions API anzufordern (z. B. um die Stopps durch Bearbeiten der JSON-Dateien zu ändern):

export GOOGLE_MAPS_API_KEY=...
python datos/generador.py --regenerar-rutas

Den Server mit dem Testclient ausführen

Der Server ist für sich genommen nicht interaktiv: Er liest JSON von stdin und schreibt JSON nach stdout. Um ihn in Aktion zu sehen, habe ich cliente_prueba.py geschrieben, das ihn als Unterprozess startet, den Handshake durchführt, die Werkzeuge auflistet und es ermöglicht, sie aufzurufen, wobei jede Nachricht so gedruckt wird, wie sie in jede Richtung reist.

python cliente_prueba.py          # interactivo
python cliente_prueba.py --demo   # las herramientas de texto y tres casos de error, de corrido

Im interaktiven Modus gibt man die Nummer des Werkzeugs ein, beantwortet seine Parameter und sieht die Anfrage, die Antwort und das Ergebnis. Es akzeptiert auch ping und lista.

Der Server kann auch von Hand getestet werden:

printf '{"jsonrpc":"2.0","id":1,"method":"ping"}\n' | python -m servidor.main

Die Logs des Servers gehen nach stderr; mit --verbose druckt er zusätzlich jede Nachricht, die ein- und ausgeht.

Mit Claude Desktop verbinden

Die Konfigurationsdatei von Claude Desktop bearbeiten:

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

• Windows: %APPDATA%\Claude\claude_desktop_config.json

und den Server mit den absoluten Pfaden des Repositorys hinzufügen:

{
  "mcpServers": {
    "flota": {
      "command": "/ruta/absoluta/mcp-local-redes/.venv/bin/python",
      "args": ["/ruta/absoluta/mcp-local-redes/servidor/main.py"]
    }
  }
}

Unter Windows ist command C:\\ruta\\mcp-local-redes\\.venv\\Scripts\\python.exe. Wenn man Geokodierung mit Google möchte, fügt man "env": {"GOOGLE_MAPS_API_KEY": "..."} innerhalb von "flota" hinzu.

Nach einem Neustart von Claude Desktop erscheinen die sieben Werkzeuge, und man kann in natürlicher Sprache fragen. Die Datenbank wird in datos/flota.db relativ zum Repository gesucht; sie kann mit der Variable MCP_FLOTA_DB geändert werden.

Werkzeuge

Daten im Format AAAA-MM-DD. Alarmtypen: exceso_velocidad, parada_prolongada, perdida_senal, geocerca_entrada, geocerca_salida.

Im synthetischen Datensatz definierte Geo-Fences: CEDIS Zona 12, Bodega Villa Nueva, Bodega Mixco, CD Zona 18, Bodega Carretera a El Salvador und Centro Histórico. verificar_geocerca akzeptiert den vollständigen Namen oder einen Teil („cedis“, „mixco“).

Kein Werkzeug liefert Rohdaten. Sieben Tage lang berichten fünfzehn Fahrzeuge alle fünfzehn Sekunden – das sind Hunderttausende von Zeilen; sie an das Modell zu senden ist unpraktikabel und unnötig. Jedes Werkzeug aggregiert in SQL oder in pandas und gibt das berechnete Ergebnis zurück. Das Zeilenlimit pro Antwort ist die Konstante MAX_FILAS = 200 in servidor/registro.py, und es gibt einen Test, der das für jedes Werkzeug überprüft.

Die Routenkarte

mapa_recorrido ist das einzige Werkzeug, das mehr als Text zurückgibt: Sein Ergebnis enthält zwei Inhaltsblöcke, einen text mit der Zusammenfassung und ein image mit dem PNG in Base64, das Claude Desktop direkt im Chat anzeigt. Die Karte wird mit matplotlib gezeichnet; die Hintergrundkacheln werden mit urllib von OpenStreetMap heruntergeladen, und wenn keine Verbindung besteht, wird die Route auf einem flachen Hintergrund gezeichnet.

Beispielfragen

• Wo ist das P-123BCD gerade?

• Gibt es Fahrzeuge, die seit mehr als einer Stunde stehen?

• Gib mir die Zusammenfassung der Route der P-456DEF von gestern.

• Zeig mir auf einer Karte, wo die P-789GHJ gestern unterwegs war.

• Wie viele Stopps machte die P-234KLM am Montag und wo war der längste?

• Gab es diese Woche Geschwindigkeitsüberschreitungen? Welches Fahrzeug hatte die meisten?

• Welches Fahrzeug hat in den letzten sieben Tagen das Signal verloren?

• Ist die P-456DEF gestern ins CEDIS gefahren? Um wie viel Uhr und wie lange war sie dort?

• Welches Fahrzeug ist diese Woche die meisten Kilometer gefahren?

• Wie viele Kilometer ist die gesamte Flotte von Montag bis Freitag gefahren?

Tests

python -m pytest tests -v

Zwei Gruppen:

• tests/test_protocolo.py: korrekter Handshake, Ablehnung von Methoden vor initialize, fehlerhaftes JSON → -32700, ungültige Anfrage → -32600, nicht vorhandene Methode → -32601, ungültige Argumente → -32602, eine Benachrichtigung erzeugt keine Antwort, die id der Antwort stimmt mit der der Anfrage überein, eine Antwort enthält nie gleichzeitig result und error, und ein echter Start des Prozesses über stdio mit sauberem Ende bei EOF.

• tests/test_herramientas.py: jedes Werkzeug gegen einen in einem temporären Verzeichnis mit fester Seed erzeugten Datensatz, wobei gegen die Ereignisse geprüft wird, die der Generator absichtlich injiziert hat (die Fahrzeuge, die ich stehen ließ, die längeren Stopps, die Signallücken, die Geschwindigkeitsüberschreitungen), die Geschäftsfehler und dass keine Antwort MAX_FILAS überschreitet.

Implementierung des Protokolls

Alles, was das Protokoll betrifft, ist von Hand mit sys, json und logging geschrieben. pandas, shapely, geopy und matplotlib sind Geschäftslogik; requests verwendet nur der Datengenerator.

• servidor/main.py, Transport. Liest stdin Zeile für Zeile, schreibt jede Antwort nach stdout gefolgt von \n und flush(). Jedes Log geht nach stderr, weil stdout der Kanal des Protokolls ist und ein einziges zusätzliches Byte es zerstört. Bei EOF schließt er die Datenbank und beendet mit Code 0.

• servidor/jsonrpc.py, JSON-RPC 2.0. Parst und validiert jede Nachricht, unterscheidet Request von Benachrichtigung anhand des Vorhandenseins des Schlüssels id (nicht anhand seines Werts, denn null ist eine gültige id), und erstellt Antworten und Fehler mit den Standardcodes -32700, -32600, -32601, -32602 und -32603.

• servidor/protocolo.py, MCP. Initialisierungs-Handshake mit Zustandsmaschine (NUEVA → INICIALIZANDO → LISTA): Jede Methode, die nicht initialize oder ping ist, wird abgelehnt, bis notifications/initialized eintrifft. Versionsverhandlung: Wenn der Client eine Version anfordert, die ich unterstütze, gebe ich sie zurück; wenn nicht, gebe ich die neueste zurück, die ich unterstütze. tools/list, tools/call und ping. Benachrichtigungen, die ich nicht behandle, werden still ignoriert, denn auf eine Benachrichtigung zu antworten zerstört den Client.

• servidor/registro.py. Liste der Werkzeuge mit ihrem inputSchema und Validierung von Argumenten dagegen (Typen, required, enum). Hier lebt MAX_FILAS.

Ich habe beschlossen, JSON-RPC von MCP zu trennen, weil es zwei verschiedene Ebenen des Protokolls sind: JSON-RPC definiert die Form der Nachrichten und MCP definiert, welche Methoden es gibt und in welcher Reihenfolge. Durch die Trennung konnte ich die Nachrichtenvalidierung ohne eine Sitzung und die Zustandsmaschine ohne stdin testen.

Die Unterscheidung, auf die ich am meisten geachtet habe, liegt in tools/call: Wenn das Tool nicht existiert oder die Argumente nicht dem Schema entsprechen, ist es ein Protokollfehler und wird als JSON-RPC-error mit -32602 zurückgegeben; wenn das Tool existiert und läuft, aber das Ergebnis ein Geschäftsfehler ist (nicht vorhandenes Kennzeichen, Tag ohne Daten), wird es als result mit isError: true und einer lesbaren Nachricht zurückgegeben, damit das Modell es dem Benutzer erklären kann.

Die vollständige Ablaufverfolgung einer echten Sitzung mit dem exakten JSON jeder Nachricht befindet sich in docs/protocolo.md.

Struktur des Repositorys

mcp-local-redes/
├── servidor/
│   ├── main.py              # punto de entrada, bucle de stdio
│   ├── jsonrpc.py           # construcción y validación de mensajes JSON-RPC 2.0
│   ├── protocolo.py         # handshake, máquina de estados, despacho de métodos
│   ├── registro.py          # registro de herramientas, validación de argumentos, MAX_FILAS
│   └── herramientas/
│       ├── comun.py         # consultas compartidas
│       ├── geocodificacion.py
│       ├── posicion.py
│       ├── detenidas.py
│       ├── recorrido.py
│       ├── mapa.py
│       ├── alertas.py
│       ├── geocercas.py
│       └── kilometraje.py
├── datos/
│   ├── generador.py         # set sintético
│   ├── esquema.sql
│   └── rutas/               # polilíneas guardadas para modo offline
├── cliente_prueba.py
├── tests/
│   ├── test_protocolo.py
│   └── test_herramientas.py
├── docs/
│   └── protocolo.md
└── requirements.txt