""" GIS / QGIS Processing MCP Server ================================ This is a small, practical MCP server for running QGIS Processing tools through the qgis_process command line. It is intended as a stable batch companion to QGIS MCP Plugin + qgis-mcp Server. Requirements ------------ - QGIS installed and qgis_process available in PATH - Python 3.12+ - uv or pip environment with: mcp[cli] Example ------- uv init uv add "mcp[cli]" uv run python gis_qgis_process_mcp_server.py Security note ------------- This server exposes predefined GIS operations only. Avoid exposing arbitrary shell or PyQGIS execution to an AI client in production environments. """ from __future__ import annotations import json import os import subprocess from pathlib import Path from typing import Any from mcp.server.fastmcp import FastMCP mcp = FastMCP("GIS QGIS Processing MCP Server") QGIS_PROCESS = os.environ.get("QGIS_PROCESS", "qgis_process") DEFAULT_TIMEOUT_SEC = int(os.environ.get("QGIS_MCP_TIMEOUT_SEC", "900")) def _run_command(cmd: list[str], timeout_sec: int = DEFAULT_TIMEOUT_SEC) -> str: """Run a command and return a readable result string.""" try: result = subprocess.run( cmd, text=True, capture_output=True, shell=False, timeout=timeout_sec, check=False, ) except FileNotFoundError: return ( f"qgis_process が見つかりません: {QGIS_PROCESS}\n" "QGIS の bin フォルダを PATH に追加するか、環境変数 QGIS_PROCESS にフルパスを指定してください。" ) except subprocess.TimeoutExpired: return f"処理がタイムアウトしました。timeout={timeout_sec} sec" parts = [f"COMMAND: {' '.join(cmd)}", f"RETURN_CODE: {result.returncode}"] if result.stdout: parts.append("STDOUT:\n" + result.stdout.strip()) if result.stderr: parts.append("STDERR:\n" + result.stderr.strip()) return "\n\n".join(parts) def _ensure_parent(path: str) -> None: """Create parent directory for an output path when possible.""" if path and path not in {"TEMPORARY_OUTPUT", "memory:"}: Path(path).expanduser().parent.mkdir(parents=True, exist_ok=True) def _check_input(path: str) -> str | None: """Return an error message if a file path does not exist.""" if not Path(path).expanduser().exists(): return f"入力ファイルが存在しません: {path}" return None def _json_parameters_to_qgis_args(parameters: dict[str, Any]) -> list[str]: """Convert a parameter dictionary to qgis_process KEY=VALUE arguments.""" args: list[str] = [] for key, value in parameters.items(): if isinstance(value, bool): value_text = "true" if value else "false" elif isinstance(value, (dict, list)): value_text = json.dumps(value, ensure_ascii=False) else: value_text = str(value) args.append(f"{key}={value_text}") return args @mcp.tool() def qgis_process_version() -> str: """qgis_process のバージョンを確認する。""" return _run_command([QGIS_PROCESS, "--version"], timeout_sec=60) @mcp.tool() def list_qgis_processing_algorithms(filter_text: str = "") -> str: """QGIS Processing アルゴリズム一覧を取得する。filter_text で絞り込みも可能。""" output = _run_command([QGIS_PROCESS, "list"], timeout_sec=120) if not filter_text: return output lines = [line for line in output.splitlines() if filter_text.lower() in line.lower()] return "\n".join(lines) if lines else f"一致するアルゴリズムが見つかりません: {filter_text}" @mcp.tool() def qgis_algorithm_help(algorithm_id: str) -> str: """QGIS Processing アルゴリズムのヘルプを表示する。例: native:buffer""" return _run_command([QGIS_PROCESS, "help", algorithm_id], timeout_sec=120) @mcp.tool() def run_qgis_algorithm(algorithm_id: str, parameters_json: str) -> str: """ 任意の QGIS Processing アルゴリズムを実行する。 parameters_json は JSON 文字列で指定する。 例: {"INPUT":"C:/gis/points.gpkg","DISTANCE":20,"SEGMENTS":8,"OUTPUT":"C:/gis/out/buffer.gpkg"} """ try: parameters = json.loads(parameters_json) except json.JSONDecodeError as exc: return f"parameters_json が JSON として読めません: {exc}" if not isinstance(parameters, dict): return "parameters_json は JSON object にしてください。" output_value = parameters.get("OUTPUT") or parameters.get("OUTPUT_LAYER") if isinstance(output_value, str): _ensure_parent(output_value) cmd = [QGIS_PROCESS, "run", algorithm_id, "--"] + _json_parameters_to_qgis_args(parameters) return _run_command(cmd) @mcp.tool() def create_buffer( input_path: str, output_path: str, distance: float = 20.0, segments: int = 8, dissolve: bool = False, ) -> str: """ベクタレイヤにバッファを作成する。""" error = _check_input(input_path) if error: return error _ensure_parent(output_path) params = { "INPUT": input_path, "DISTANCE": distance, "SEGMENTS": segments, "END_CAP_STYLE": 0, "JOIN_STYLE": 0, "MITER_LIMIT": 2, "DISSOLVE": dissolve, "OUTPUT": output_path, } return run_qgis_algorithm("native:buffer", json.dumps(params, ensure_ascii=False)) @mcp.tool() def clip_vector_by_polygon(input_path: str, overlay_path: str, output_path: str) -> str: """ポリゴンでベクタレイヤをクリップする。""" for path in [input_path, overlay_path]: error = _check_input(path) if error: return error _ensure_parent(output_path) params = {"INPUT": input_path, "OVERLAY": overlay_path, "OUTPUT": output_path} return run_qgis_algorithm("native:clip", json.dumps(params, ensure_ascii=False)) @mcp.tool() def intersection(input_path: str, overlay_path: str, output_path: str) -> str: """2つのベクタレイヤの交差を作成する。""" for path in [input_path, overlay_path]: error = _check_input(path) if error: return error _ensure_parent(output_path) params = {"INPUT": input_path, "OVERLAY": overlay_path, "OUTPUT": output_path} return run_qgis_algorithm("native:intersection", json.dumps(params, ensure_ascii=False)) @mcp.tool() def raster_contour(input_raster: str, output_path: str, interval: float = 10.0, field_name: str = "ELEV") -> str: """DEM等のラスタから等高線を作成する。""" error = _check_input(input_raster) if error: return error _ensure_parent(output_path) params = { "INPUT": input_raster, "BAND": 1, "INTERVAL": interval, "FIELD_NAME": field_name, "CREATE_3D": False, "OUTPUT": output_path, } return run_qgis_algorithm("gdal:contour", json.dumps(params, ensure_ascii=False)) @mcp.tool() def clip_raster_by_mask(input_raster: str, mask_layer: str, output_path: str) -> str: """マスクポリゴンでラスタを切り出す。""" for path in [input_raster, mask_layer]: error = _check_input(path) if error: return error _ensure_parent(output_path) params = { "INPUT": input_raster, "MASK": mask_layer, "SOURCE_CRS": None, "TARGET_CRS": None, "NODATA": None, "ALPHA_BAND": False, "CROP_TO_CUTLINE": True, "KEEP_RESOLUTION": True, "OUTPUT": output_path, } return run_qgis_algorithm("gdal:cliprasterbymasklayer", json.dumps(params, ensure_ascii=False)) if __name__ == "__main__": mcp.run()