フリート向けローカルMCPサーバー

Universidad del Valle de Guatemala のCC3067 Redesコース、セクション10のプロジェクト1。 Fernando Hernández.

MCP（Model Context Protocol）のサーバーです。がオペレーターのマシン上で動作し、 車両フリートのデレメトリーのクエリを、言語モデルが呼び出せるツールとして公開します。 これにより、ユーザーが追跡プラットフォームを開かなくても、チャットボットが 「P-123BCD はどこにいますか？」や「今週どのユニットが最も多く走行しましたか？」 といった質問に回答できます。

プロトコルは、Python 標準ライブラリのみを使い、stdio 上にゼロから実装されています。MCP SDK も JSON-RPC を取り扱うライブラリも使っていません。これがこのプロジェクトの中心的な要件であり、 プロトコルの実装のセクションで説明します。

MCPとは

MCP は、言語モデルがどのように外部ツールを発見・呼び出すかを標準化する、アプリケーション層の プロトコルです。MCP サーバーはツールの一覧を公開します。各ツールには、名前、説明、パラメータの JSON スキーマが含まれています。クライアント（たとえば Claude Desktop）はその一覧を取得してモデルに 提示します。モデルがツールの利用を判断すると、クライアントはモデルが選んだ引数をツールで指定し、 結果をモデルに返し、モデルがそれを自然言語で説明します。

構造的に、MCP はトランスポート上の JSON-RPC 2.0 です。このプロジェクトではトランスポートは stdio です。クライアントはサーバーを子プロセスとして起動し、両者は改行区切りの JSON オブジェ クトを stdin・stdout 経由でやり取りします。セッションはハンドシェイク（initialize → 応答 → notifications/initialized）で始まり、その後クライアントは tools/list、tools/call、ping を 呼び出せるようになります。

Related MCP server: NL-to-SQL MCP

なぜフリートなのか、そしてなぜローカルで動かすのか

自前のフリートを持つ企業では、各車両に GPS と追跡プラットフォームを導入済みで、データは 存在し、しっかり揃っています。問題はートへのアクセスです。マネージャーはダッシュボードと対話し、 フィルターをかけ、レポートを作成しますが、最も現場を知る人が一番プラットフォームを使いこなせない、 というのがよくある話です。

サーバーをローカルで動かすのは、コースの要件であるためだけではありません。フリートの位置は、 取引先へのルートや顧客、営業時間を明らかにします。サーバーをオペレーターのマシンに置くことで、 モデルへは各クエリの要約結果だけが送られ、位置履歴が送られることはありません。

前提条件

• Python 3.11 以降

• git

• オプション：Google マップのキー（GOOGLE_MAPS_API_KEY）。Google でのジオコーディングとルート の生成のために使用します。これがなくても、すべて同じように動きます。

インストール

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

データベースの作成

企業の実データは使いません。生成シからない 15 台の車両を、グアテマラ市の都市部の実路でオンに 7 日間分、各ユニットの就業時間内では 15 秒ごとにレポートを出し、GPS ノイズ（σ ≈ 5 m）と、あらか じめ注入しておいたイベント（長時間の停車、速度超過、信号の消失、ジオフェンスの出入り）を加えます。

python datos/generador.py

datos/flota.db（SQLite、約 160,000 件の位置情報と約 1,300 件のイベント）を 2 秒ほどで作成します。 デフォルトでは、データセットは実行した瞬間に終了します。したがって「今日」「昨日」のような 質問は実日付を指します。再現可能なデータセットを使たい時刻を固定する方法はこちらです。

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

勤務時間帯（または --hora-tiempo）に生成するとよいでしょう。そうすれば、 unidades_detenidas の中に「停止中」の車だけでなく「走行中」の車も現れます。

ルート：オフライン版と API 版

元となるルートは、datos/rutas/*.json に、エンコードされたポリライン（Google Directions API が 返すのと同じ形式）として、各ルートのバス停と一緒に保存されています。ジェネレーターはそこから 読み込むだけです。ネットワークもAPIも不要です。

API から出し直す場合（例：JSON を編集してバス停を変更するため）：

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

テストクライアントでサーバーを動かす

サーバーは単体では対話型ではありません。stdin から JSON を読み、stdout に JSON を出力します。 実際の動作を確認できるように、cliente_prueba.py を書きました。これはサーバーをサブプロセスと して起動し、ハンドシェイクを行い、ツールを一覧し、呼び出すことができます。送受信される各メッセ ージを表示します。

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

対話型では、ツール番号を入力し、パラメータを答えると、リクエスト、レスポンス、実行結果が表示されます。 ping と list も受け付けます。

サーバーを手動で試すこともできます。

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

サーバーのログは stderr に出力されます。--verbose を付けると、入ってくるすべてのメッセージと 出力されるメッセージも表示されます。

Claude Desktop への接続

Claude Desktop の設定ファイルを編集します。

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

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

そして、リポジトリの絶対パスでサーバーを追加します。

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

Windows では command は C:\\ruta\\mcp-local-redes\\.venv\\Scripts\\python.exe です。 Google のジオコーディングを使うには、"flota" の中に "env": {"GOOGLE_MAPS_API_KEY": "..."} を追加します。

Claude Desktop を再起動すると、7 個のツールが表示され、自然言語で質問できます。 データベースはリポジトリ相対の datos/flota.db を探します。 これは環境変数 MCP_FLOTA_DB で変更できます。

ツール

日付は AAAA-MM-DD 形式です。アラートの種類：exceso_velocidad、parada_prolongada、 perdida_senal、geocerca_entrada、geocerca_salida。

合成データセットに定義されているバッフェンスは、CEDIS Zona 12、Bodega Villa Nueva、Bodega Mixco、 CD Zona 18、Bodega Carretera a El Salvador、Centro Histórico です。verificar_geocerca は完全名 または部分名（“cedis”、“mixco” など）を受け付けます。

どのツールも裸のデータを返しません。 45秒ごとに報告 15 台 × 7 日間は数十万行になります。 とてもモデルに送ることはできませんし、不要でもあります。各ツールは SQL または pandas で集計し、 計算済みの結果を返します。一つの回答の最大行数は servidor/registro.py の中の定数 MAX_FILAS = 200 によって制限されており、どのツールにもこれを確認するテストがあります。

走行マップ

mapa_recorrido は、テキスト以外のものを返す唯一のツールです。その結果は2つのコンテンツブロック を含んでおり、text の要約と、base64 でエンコードした PNG を持つ image です。 Mobile Client で直接表示されます。地図は matplotlib で描かれ、背景のタイルは urllib で OpenStreetMap からダウンロードします。ネットワークが届かないときは、平に背景に道筋を描きます。

質問の例

• 今、P-123BCD はどこにいますか。

• 1時間以上も動いていないユニットはありますか？

• 昨日の P-456DEF の走行の要約を得たい。

• 昨日 P-789GHJ の通ったルートを地図で見せてください。

• 月曜日に P-234KLM は何回停車しましたか？一番長い停車はどこでしたか？

• 今週のスピードの落後はありましたか？どのユニットが一番多かったですか？

• 過去7日間で信号の消失があったユニットはどれですか？

• 昨日 P-456F は CEDIS に入りましたか？何時に入って、何時間いましたか？

• 今週最も走行っていたユニットはどこですか。

• 月曜から金曜までのフリート全体の走行合計キロはどれだけですか。

テスト

python -m pytest tests -v

次の2つのグループです。

• tests/test_protocolo.py：正しいハンドシェイク、initialize 前のメソッドの拒否、 JSON 壊れ →-32700、無効リクエスト → -32600、存在しないメソッド → -32601、 不正な引数 →-32602、通知にレスポンスを返さないこと、応答の id はリクエストの id と一致すること、 応答は result と error を同時に含まないこと、そして stdio を介して実際に起動し、EOF で 正常終了すること。

• tests/test_herramientas.py：固定シードで一時ディレクトリに作成したデータセットを使い、各ツールが、 意図的に注入したイベント（停止させたユニット、長時間の停車、信号消失、スピード違反）、 ビジネスロジックのエラー、そしてどの応答も MAX_FILAS を超えないことを検証します。

プロトコルの実装

プロトコルに直接関係するすべての部分は、sys, json, logging だけを使って手書きしています。 pandas, shapely, geopy, matplotlib はビジネスロ

私はJSON-RPCとMCPを分離することにしました。なぜなら、それらはプロトコルの2つの異なるレベルだからです： JSON-RPCはメッセージの形式を定義し、MCPはどのメソッドが存在し、どの順序で呼び出されるかを定義します。それらを分離することで、セッションなしでメッセージ検証をテストし、stdinなしでステートマシンをテストすることができました。

私が最も注意を払った区別はtools/callにあります：ツールが存在しないか、引数がスキーマに適合しない場合は、プロトコルエラーであり、-32602のJSON-RPC errorとして返されます。一方、ツールが存在して実行されるが、結果がビジネス上の失敗（存在しないプレート、データのない日付など）である場合は、isError: trueと人間が読めるメッセージを含むresultとして返され、モデルがそれをユーザーに説明できるようにします。

実際のセッションの完全なトレース（各メッセージの正確なJSONを含む）は、 docs/protocolo.mdにあります。

リポジトリ構造

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