# Directory Structure ``` ├── .gitignore ├── .python-version ├── assets │ └── imgs │ ├── claude-available-tools.png │ ├── qgis-mcp-start-server.png │ └── qgis-plugins-menu.png ├── CONTRIBUTING.md ├── main.py ├── pyproject.toml ├── qgis_mcp_plugin │ ├── __init__.py │ ├── metadata.txt │ └── qgis_mcp_plugin.py ├── README.md ├── src │ └── qgis_mcp │ ├── qgis_mcp_server.py │ └── qgis_socket_client.py └── uv.lock ``` # Files -------------------------------------------------------------------------------- /.python-version: -------------------------------------------------------------------------------- ``` 3.12 ``` -------------------------------------------------------------------------------- /.gitignore: -------------------------------------------------------------------------------- ``` # Python-generated files __pycache__/ *.py[oc] build/ dist/ wheels/ *.egg-info .DS_Store # Virtual environments .venv data/ ``` -------------------------------------------------------------------------------- /README.md: -------------------------------------------------------------------------------- ```markdown # QGISMCP - QGIS Model Context Protocol Integration QGISMCP connects [QGIS](https://qgis.org/) to [Claude AI](https://claude.ai/chat) through the [Model Context Protocol (MCP)](https://modelcontextprotocol.io/docs/getting-started/intro), allowing Claude to directly interact with and control QGIS. This integration enables prompt assisted project creation, layer loading, code execution and more. This project is strongly based on the [BlenderMCP](https://github.com/ahujasid/blender-mcp/tree/main) project by [Siddharth Ahuja](https://x.com/sidahuj) ## Features - **Two-way communication**: Connect Claude AI to QGIS through a socket-based server. - **Project manipulation**: Create, load and save projects in QGIS. - **Layer manipulation**: Add and remove vector or raster layers to a project. - **Execute processing**: Execute processing algorithms ([Processing Toolbox](https://docs.qgis.org/3.40/en/docs/user_manual/processing/toolbox.html)). - **Code execution**: Run arbitrary Python code in QGIS from Claude. Very powerful, but also be very cautious using this tool. ## Components The system consists of two main components: 1. **[QGIS plugin](/qgis_mcp_plugin/)**: A QGIS plugin that creates a socket server within QGIS to receive and execute commands. 2. **[MCP Server](/src/qgis_mcp/qgis_mcp_server.py)**: A Python server that implements the Model Context Protocol and connects to the QGIS plugin. ## Installation ### Prerequisites - QGIS 3.X (only tested on 3.22) - Claude desktop - Python 3.10 or newer - uv package manager: If you're on Mac, please install uv as ```bash brew install uv ``` On Windows Powershell ```bash powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" ``` Otherwise installation instructions are on their website: [Install uv](https://docs.astral.sh/uv/getting-started/installation/) **⚠️ Do not proceed before installing UV** ### Download code Download this repo to your computer. You can clone it with: ```bash git clone [email protected]:jjsantos01/qgis_mcp.git ``` ### QGIS plugin You need to copy the folder [qgis_mcp_plugin](/qgis_mcp_plugin/) and its content on your QGIS profile plugins folder. You can get your profile folder in QGIS going to menu `Settings` -> `User profiles` -> `Open active profile folder` Then, go to `Python/plugins` and paste the folder `qgis_mcp_plugin`. > On a Windows machine the plugins folder is usually located at: `C:\Users\USER\AppData\Roaming\QGIS\QGIS3\profiles\default\python\plugins` and on MacOS: `~/Library/Application\ Support/QGIS/QGIS3/profiles/default/python/plugins` Then close QGIS and open it again. Go to the menu option `Plugins` > `Installing and Managing Plugins`, select the `All` tab and search for "QGIS MCP", then mark the QGIS MCP checkbox. ### Claude for Desktop Integration Go to `Claude` > `Settings` > `Developer` > `Edit Config` > `claude_desktop_config.json` to include the following: > If you can't find the "Developers tab" or the `claude_desktop_config.json` look at this [documentation](https://modelcontextprotocol.io/quickstart/user#2-add-the-filesystem-mcp-server). ```json { "mcpServers": { "qgis": { "command": "uv", "args": [ "--directory", "/ABSOLUTE/PATH/TO/PARENT/REPO/FOLDER/qgis_mcp/src/qgis_mcp", "run", "qgis_mcp_server.py" ] } } } ``` ## Usage ### Starting the Connection 1. In QGIS, go to `plugins` > `QGIS MCP` > `QGIS MCP`  2. Click "Start Server"  ### Using with Claude Once the config file has been set on Claude, and the server is running on QGIS, you will see a hammer icon with tools for the QGIS MCP.  #### Tools - `ping` - Simple ping command to check server connectivity - `get_qgis_info` - Get QGIS information about the current installation - `load_project` - Load a QGIS project from the specified path - `create_new_project` - Create a new project and save it - `get_project_info` - Get current project information - `add_vector_layer` - Add a vector layer to the project - `add_raster_layer` - Add a raster layer to the project - `get_layers` - Retrieve all layers in the current project - `remove_layer` - Remove a layer from the project by its ID - `zoom_to_layer` - Zoom to the extent of a specified layer - `get_layer_features` - Retrieve features from a vector layer with an optional limit - `execute_processing` - Execute a processing algorithm with the given parameters - `save_project` - Save the current project to the given path - `render_map` - Render the current map view to an image file - `execute_code` - Execute arbitrary PyQGIS code provided as a string ### Example Commands This is the example I used for the [demo](https://x.com/jjsantoso/status/1900293848271667395): ```plain You have access to the tools to work with QGIS. You will do the following: 1. Ping to check the connection. If it works, continue with the following steps. 2. Create a new project and save it at: "C:/Users/USER/GitHub/qgis_mcp/data/cdmx.qgz" 3. Load the vector layer: ""C:/Users/USER/GitHub/qgis_mcp/data/cdmx/mgpc_2019.shp" and name it "Colonias". 4. Load the raster layer: "C:/Users/USER/GitHub/qgis_mcp/data/09014.tif" and name it "BJ" 5. Zoom to the "BJ" layer. 6. Execute the centroid algorithm on the "Colonias" layer. Skip the geometry check. Save the output to "colonias_centroids.geojson". 7. Execute code to create a choropleth map using the "POB2010" field in the "Colonias" layer. Use the quantile classification method with 5 classes and the Spectral color ramp. 8. Render the map to "C:/Users/USER/GitHub/qgis_mcp/data/cdmx.png" 9. Save the project. ``` ``` -------------------------------------------------------------------------------- /CONTRIBUTING.md: -------------------------------------------------------------------------------- ```markdown # Contributing to QGIS MCP Thank you for your interest in contributing! 🎉 This project connects [QGIS](https://qgis.org/) to [Claude AI](https://claude.ai/chat) through the [Model Context Protocol (MCP)](https://modelcontextprotocol.io/docs/getting-started/intro)). Your help in improving this integration is very welcome. ## Getting Started 1. **Fork the Repository** Clone your fork locally: ```bash git clone [email protected]:YOUR-USERNAME/qgis_mcp.git cd qgis_mcp ``` 2. **Install Prerequisites** - QGIS 3.X (tested on 3.22) - Python 3.10 or newer - [uv](https://docs.astral.sh/uv/getting-started/installation/) package manager - Claude desktop On Mac: ```bash brew install uv ``` On Windows Powershell: ```bash powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" ``` 3. **Set Up the QGIS Plugin** Create a symlink from this repo’s `qgis_mcp_plugin` folder to your QGIS profile plugin directory. On Mac: ```bash ln -s $(pwd)/qgis_mcp_plugin ~/Library/Application\ Support/QGIS/QGIS3/profiles/default/python/plugins/qgis_mcp ``` On Windows Powershell: ```powershell $src = "$(pwd)\qgis_mcp_plugin" $dst = "$env:APPDATA\QGIS\QGIS3\profiles\default\python\plugins\qgis_mcp" New-Item -ItemType SymbolicLink -Path $dst -Target $src ``` Restart QGIS, go to `Plugins` > `Manage and Install Plugins`, search for **QGIS MCP**, and enable it. 4. **Configure Claude Desktop** Add the server configuration to `claude_desktop_config.json`: ```json { "mcpServers": { "qgis": { "command": "uv", "args": [ "--directory", "/ABSOLUTE/PATH/TO/qgis_mcp/src/qgis_mcp", "run", "qgis_mcp_server.py" ] } } } ``` ## Development Workflow - Start the QGIS plugin (`Plugins` > `QGIS MCP` > `Start Server`). - Run the MCP server via Claude Desktop integration. - Make your changes and test locally. ## Contributing Guidelines - Keep PRs focused on a single change. - Write clear commit messages. - Update docs if behavior changes. - Be cautious when using `execute_code` (it runs arbitrary PyQGIS). ## Reporting Issues - Use [GitHub Issues](https://github.com/jjsantos01/qgis_mcp/issues). - Include OS, QGIS version, and error logs where relevant. ``` -------------------------------------------------------------------------------- /qgis_mcp_plugin/__init__.py: -------------------------------------------------------------------------------- ```python from .qgis_mcp_plugin import classFactory ``` -------------------------------------------------------------------------------- /main.py: -------------------------------------------------------------------------------- ```python def main(): print("Hello from qgis-mcp!") if __name__ == "__main__": main() ``` -------------------------------------------------------------------------------- /pyproject.toml: -------------------------------------------------------------------------------- ```toml [project] name = "qgis-mcp" version = "0.1.0" description = "Add your description here" readme = "README.md" requires-python = ">=3.12" dependencies = [ "mcp[cli]>=1.3.0", ] ``` -------------------------------------------------------------------------------- /qgis_mcp_plugin/metadata.txt: -------------------------------------------------------------------------------- ``` [general] name=QGIS MCP description=Servidor socket para ejecutar comandos MCP en QGIS version=1.0 qgisMinimumVersion=3.0 author=TuNombre [email protected] category=Plugins [about] about=Este plugin inicia un servidor MCP en QGIS para permitir la ejecución de comandos desde un LLM. ``` -------------------------------------------------------------------------------- /src/qgis_mcp/qgis_socket_client.py: -------------------------------------------------------------------------------- ```python #!/usr/bin/env python3 """ QGIS MCP Client - Simple client to connect to the QGIS MCP server """ import socket import json import argparse import sys class QgisMCPClient: def __init__(self, host='localhost', port=9876): self.host = host self.port = port self.socket = None def connect(self): """Connect to the QGIS MCP server""" try: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.connect((self.host, self.port)) return True except Exception as e: print(f"Error connecting to server: {str(e)}") return False def disconnect(self): """Disconnect from the server""" if self.socket: self.socket.close() self.socket = None def send_command(self, command_type, params=None): """Send a command to the server and get the response""" if not self.socket: print("Not connected to server") return None # Create command command = { "type": command_type, "params": params or {} } try: # Send the command self.socket.sendall(json.dumps(command).encode('utf-8')) # Receive the response response_data = b'' while True: chunk = self.socket.recv(4096) if not chunk: break response_data += chunk # Try to decode as JSON to see if it's complete try: json.loads(response_data.decode('utf-8')) break # Valid JSON, we have the full message except json.JSONDecodeError: continue # Keep receiving # Parse and return the response return json.loads(response_data.decode('utf-8')) except Exception as e: print(f"Error sending command: {str(e)}") return None def ping(self): """Simple ping command to check server connectivity""" return self.send_command("ping") def get_qgis_info(self): """Get QGIS information""" return self.send_command("get_qgis_info") def get_project_info(self): """Get current project information""" return self.send_command("get_project_info") def execute_code(self, code): """Execute arbitrary PyQGIS code""" return self.send_command("execute_code", {"code": code}) def add_vector_layer(self, path, name=None, provider="ogr"): """Add a vector layer to the project""" params = { "path": path, "provider": provider } if name: params["name"] = name return self.send_command("add_vector_layer", params) def add_raster_layer(self, path, name=None, provider="gdal"): """Add a raster layer to the project""" params = { "path": path, "provider": provider } if name: params["name"] = name return self.send_command("add_raster_layer", params) def get_layers(self): """Get all layers in the project""" return self.send_command("get_layers") def remove_layer(self, layer_id): """Remove a layer from the project""" return self.send_command("remove_layer", {"layer_id": layer_id}) def zoom_to_layer(self, layer_id): """Zoom to a layer's extent""" return self.send_command("zoom_to_layer", {"layer_id": layer_id}) def get_layer_features(self, layer_id, limit=10): """Get features from a vector layer""" return self.send_command("get_layer_features", {"layer_id": layer_id, "limit": limit}) def execute_processing(self, algorithm, parameters): """Execute a processing algorithm""" return self.send_command("execute_processing", { "algorithm": algorithm, "parameters": parameters }) def save_project(self, path=None): """Save the current project""" params = {} if path: params["path"] = path return self.send_command("save_project", params) def load_project(self, path): """Load a project""" return self.send_command("load_project", {"path": path}) def render_map(self, path, width=800, height=600): """Render the current map view to an image""" return self.send_command("render_map", { "path": path, "width": width, "height": height }) def print_json(data): """Imprime datos JSON formateados""" print(json.dumps(data, indent=2)) def main(): # Conectar al servidor QGIS MCP client = QgisMCPClient(host='localhost', port=9876) if not client.connect(): print("No se pudo conectar al servidor QGIS MCP") return try: # Verificar conexión con ping print("Verificando conexión...") response = client.ping() if response and response.get("status") == "success": print("Conexión exitosa") else: print("Error de conexión") return # Obtener información de QGIS print("\nInformación de QGIS:") qgis_info = client.get_qgis_info() print_json(qgis_info) # Load project print("\nLoad project") load_project = client.load_project("C:/Users/jjsan/OneDrive/Consultoria/Finalizados/electoral_maps/thailand_2007/thailand_2007.qgz") print_json(load_project) # Obtener información del proyecto actual print("\nInformación del proyecto:") project_info = client.get_project_info() print_json(project_info) # Zoom to layer print("\nZoom to first layer") first_layer = project_info["result"]["layers"][0]["id"] zoom_layer = client.zoom_to_layer(first_layer) print_json(zoom_layer) # Render Map to file print("\nRendering image") render_map = client.render_map("C:/Users/jjsan/OneDrive/Consultoria/Finalizados/electoral_maps/thailand_2007/map.png") print_json(render_map) except Exception: print("Error ejecutando comandos") if __name__ == "__main__": main() ``` -------------------------------------------------------------------------------- /src/qgis_mcp/qgis_mcp_server.py: -------------------------------------------------------------------------------- ```python #!/usr/bin/env python3 """ QGIS MCP Client - Simple client to connect to the QGIS MCP server """ import logging from contextlib import asynccontextmanager import socket import json from typing import AsyncIterator, Dict, Any from mcp.server.fastmcp import FastMCP, Context logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("QgisMCPServer") class QgisMCPServer: def __init__(self, host='localhost', port=9876): self.host = host self.port = port self.socket = None def connect(self): """Connect to the QGIS MCP server""" try: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.connect((self.host, self.port)) return True except Exception as e: print(f"Error connecting to server: {str(e)}") return False def disconnect(self): """Disconnect from the server""" if self.socket: self.socket.close() self.socket = None def send_command(self, command_type, params=None): """Send a command to the server and get the response""" if not self.socket: print("Not connected to server") return None # Create command command = { "type": command_type, "params": params or {} } try: # Send the command self.socket.sendall(json.dumps(command).encode('utf-8')) # Receive the response response_data = b'' while True: chunk = self.socket.recv(4096) if not chunk: break response_data += chunk # Try to decode as JSON to see if it's complete try: json.loads(response_data.decode('utf-8')) break # Valid JSON, we have the full message except json.JSONDecodeError: continue # Keep receiving # Parse and return the response return json.loads(response_data.decode('utf-8')) except Exception as e: print(f"Error sending command: {str(e)}") return None _qgis_connection = None def get_qgis_connection(): """Get or create a persistent Qgis connection""" global _qgis_connection # If we have an existing connection, check if it's still valid if _qgis_connection is not None: # Test if the connection is still alive with a simple ping try: # Just try to send a small message to check if the socket is still connected _qgis_connection.sock.sendall(b'') return _qgis_connection except Exception as e: # Connection is dead, close it and create a new one logger.warning(f"Existing connection is no longer valid: {str(e)}") try: _qgis_connection.disconnect() except Exception: pass _qgis_connection = None # Create a new connection if needed if _qgis_connection is None: _qgis_connection = QgisMCPServer(host="localhost", port=9876) if not _qgis_connection.connect(): logger.error("Failed to connect to Qgis") _qgis_connection = None raise Exception("Could not connect to Qgis. Make sure the Qgis plugin is running.") logger.info("Created new persistent connection to Qgis") return _qgis_connection @asynccontextmanager async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]: """Manage server startup and shutdown lifecycle""" # We don't need to create a connection here since we're using the global connection # for resources and tools try: # Just log that we're starting up logger.info("QgisMCPServer server starting up") # Try to connect to Qgis on startup to verify it's available try: # This will initialize the global connection if needed qgis = get_qgis_connection() logger.info("Successfully connected to Qgis on startup") except Exception as e: logger.warning(f"Could not connect to Qgis on startup: {str(e)}") logger.warning("Make sure the Qgis addon is running before using Qgis resources or tools") # Return an empty context - we're using the global connection yield {} finally: # Clean up the global connection on shutdown global _qgis_connection if _qgis_connection: logger.info("Disconnecting from Qgis on shutdown") _qgis_connection.disconnect() _qgis_connection = None logger.info("QgisMCPServer server shut down") mcp = FastMCP( "Qgis_mcp", description="Qgis integration through the Model Context Protocol", lifespan=server_lifespan ) @mcp.tool() def ping(ctx: Context) -> str: """Simple ping command to check server connectivity""" qgis = get_qgis_connection() result = qgis.send_command("ping") return json.dumps(result, indent=2) @mcp.tool() def get_qgis_info(ctx: Context) -> str: """Get QGIS information""" qgis = get_qgis_connection() result = qgis.send_command("get_qgis_info") return json.dumps(result, indent=2) @mcp.tool() def load_project(ctx: Context, path: str) -> str: """Load a QGIS project from the specified path.""" qgis = get_qgis_connection() result = qgis.send_command("load_project", {"path": path}) return json.dumps(result, indent=2) @mcp.tool() def create_new_project(ctx: Context, path: str) -> str: """Create a new project a save it""" qgis = get_qgis_connection() result = qgis.send_command("create_new_project", {"path": path}) return json.dumps(result, indent=2) @mcp.tool() def get_project_info(ctx: Context) -> str: """Get current project information""" qgis = get_qgis_connection() result = qgis.send_command("get_project_info") return json.dumps(result, indent=2) @mcp.tool() def add_vector_layer(ctx: Context, path: str, provider: str = "ogr", name: str = None) -> str: """Add a vector layer to the project.""" qgis = get_qgis_connection() params = {"path": path, "provider": provider} if name: params["name"] = name result = qgis.send_command("add_vector_layer", params) return json.dumps(result, indent=2) @mcp.tool() def add_raster_layer(ctx: Context, path: str, provider: str = "gdal", name: str = None) -> str: """Add a raster layer to the project.""" qgis = get_qgis_connection() params = {"path": path, "provider": provider} if name: params["name"] = name result = qgis.send_command("add_raster_layer", params) return json.dumps(result, indent=2) @mcp.tool() def get_layers(ctx: Context) -> str: """Retrieve all layers in the current project.""" qgis = get_qgis_connection() result = qgis.send_command("get_layers") return json.dumps(result, indent=2) @mcp.tool() def remove_layer(ctx: Context, layer_id: str) -> str: """Remove a layer from the project by its ID.""" qgis = get_qgis_connection() result = qgis.send_command("remove_layer", {"layer_id": layer_id}) return json.dumps(result, indent=2) @mcp.tool() def zoom_to_layer(ctx: Context, layer_id: str) -> str: """Zoom to the extent of a specified layer.""" qgis = get_qgis_connection() result = qgis.send_command("zoom_to_layer", {"layer_id": layer_id}) return json.dumps(result, indent=2) @mcp.tool() def get_layer_features(ctx: Context, layer_id: str, limit: int = 10) -> str: """Retrieve features from a vector layer with an optional limit.""" qgis = get_qgis_connection() result = qgis.send_command("get_layer_features", {"layer_id": layer_id, "limit": limit}) return json.dumps(result, indent=2) @mcp.tool() def execute_processing(ctx: Context, algorithm: str, parameters: dict) -> str: """Execute a processing algorithm with the given parameters.""" qgis = get_qgis_connection() result = qgis.send_command("execute_processing", {"algorithm": algorithm, "parameters": parameters}) return json.dumps(result, indent=2) @mcp.tool() def save_project(ctx: Context, path: str = None) -> str: """Save the current project to the given path, or to the current project path if not specified.""" qgis = get_qgis_connection() params = {} if path: params["path"] = path result = qgis.send_command("save_project", params) return json.dumps(result, indent=2) @mcp.tool() def render_map(ctx: Context, path: str, width: int = 800, height: int = 600) -> str: """Render the current map view to an image file with the specified dimensions.""" qgis = get_qgis_connection() result = qgis.send_command("render_map", {"path": path, "width": width, "height": height}) return json.dumps(result, indent=2) @mcp.tool() def execute_code(ctx: Context, code: str) -> str: """Execute arbitrary PyQGIS code provided as a string.""" qgis = get_qgis_connection() result = qgis.send_command("execute_code", {"code": code}) return json.dumps(result, indent=2) def main(): """Run the MCP server""" mcp.run() if __name__ == "__main__": main() ``` -------------------------------------------------------------------------------- /qgis_mcp_plugin/qgis_mcp_plugin.py: -------------------------------------------------------------------------------- ```python import os import io import sys import json import socket import traceback from qgis.core import * from qgis.gui import * from qgis.PyQt.QtCore import QObject, pyqtSignal, QTimer, Qt, QSize from qgis.PyQt.QtWidgets import QAction, QDockWidget, QVBoxLayout, QLabel, QPushButton, QSpinBox, QWidget from qgis.PyQt.QtGui import QIcon, QColor from qgis.utils import active_plugins class QgisMCPServer(QObject): """Server class to handle socket connections and execute QGIS commands""" def __init__(self, host='localhost', port=9876, iface=None): super().__init__() self.host = host self.port = port self.iface = iface self.running = False self.socket = None self.client = None self.buffer = b'' self.timer = None def start(self): """Start the server""" self.running = True self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: self.socket.bind((self.host, self.port)) self.socket.listen(1) self.socket.setblocking(False) # Create a timer to process server operations self.timer = QTimer() self.timer.timeout.connect(self.process_server) self.timer.start(100) # 100ms interval QgsMessageLog.logMessage(f"QGIS MCP server started on {self.host}:{self.port}", "QGIS MCP") return True except Exception as e: QgsMessageLog.logMessage(f"Failed to start server: {str(e)}", "QGIS MCP", Qgis.Critical) self.stop() return False def stop(self): """Stop the server""" self.running = False if self.timer: self.timer.stop() self.timer = None if self.socket: self.socket.close() if self.client: self.client.close() self.socket = None self.client = None QgsMessageLog.logMessage("QGIS MCP server stopped", "QGIS MCP") def process_server(self): """Process server operations (called by timer)""" if not self.running: return try: # Accept new connections if not self.client and self.socket: try: self.client, address = self.socket.accept() self.client.setblocking(False) QgsMessageLog.logMessage(f"Connected to client: {address}", "QGIS MCP") except BlockingIOError: pass # No connection waiting except Exception as e: QgsMessageLog.logMessage(f"Error accepting connection: {str(e)}", "QGIS MCP", Qgis.Warning) # Process existing connection if self.client: try: # Try to receive data try: data = self.client.recv(8192) if data: self.buffer += data # Try to process complete messages try: # Attempt to parse the buffer as JSON command = json.loads(self.buffer.decode('utf-8')) # If successful, clear the buffer and process command self.buffer = b'' response = self.execute_command(command) response_json = json.dumps(response) self.client.sendall(response_json.encode('utf-8')) except json.JSONDecodeError: # Incomplete data, keep in buffer pass else: # Connection closed by client QgsMessageLog.logMessage("Client disconnected", "QGIS MCP") self.client.close() self.client = None self.buffer = b'' except BlockingIOError: pass # No data available except Exception as e: QgsMessageLog.logMessage(f"Error receiving data: {str(e)}", "QGIS MCP", Qgis.Warning) self.client.close() self.client = None self.buffer = b'' except Exception as e: QgsMessageLog.logMessage(f"Error with client: {str(e)}", "QGIS MCP", Qgis.Warning) if self.client: self.client.close() self.client = None self.buffer = b'' except Exception as e: QgsMessageLog.logMessage(f"Server error: {str(e)}", "QGIS MCP", Qgis.Critical) def execute_command(self, command): """Execute a command""" try: cmd_type = command.get("type") params = command.get("params", {}) handlers = { "ping": self.ping, "get_qgis_info": self.get_qgis_info, "load_project": self.load_project, "get_project_info": self.get_project_info, "execute_code": self.execute_code, "add_vector_layer": self.add_vector_layer, "add_raster_layer": self.add_raster_layer, "get_layers": self.get_layers, "remove_layer": self.remove_layer, "zoom_to_layer": self.zoom_to_layer, "get_layer_features": self.get_layer_features, "execute_processing": self.execute_processing, "save_project": self.save_project, "render_map": self.render_map, "create_new_project": self.create_new_project, } handler = handlers.get(cmd_type) if handler: try: QgsMessageLog.logMessage(f"Executing handler for {cmd_type}", "QGIS MCP") result = handler(**params) QgsMessageLog.logMessage(f"Handler execution complete", "QGIS MCP") return {"status": "success", "result": result} except Exception as e: QgsMessageLog.logMessage(f"Error in handler: {str(e)}", "QGIS MCP", Qgis.Critical) traceback.print_exc() return {"status": "error", "message": str(e)} else: return {"status": "error", "message": f"Unknown command type: {cmd_type}"} except Exception as e: QgsMessageLog.logMessage(f"Error executing command: {str(e)}", "QGIS MCP", Qgis.Critical) traceback.print_exc() return {"status": "error", "message": str(e)} # Command handlers def ping(self, **kwargs): """Simple ping command""" return {"pong": True} def get_qgis_info(self, **kwargs): """Get basic QGIS information""" return { "qgis_version": Qgis.version(), "profile_folder": QgsApplication.qgisSettingsDirPath(), "plugins_count": len(active_plugins) } def get_project_info(self, **kwargs): """Get information about the current QGIS project""" project = QgsProject.instance() # Get basic project information info = { "filename": project.fileName(), "title": project.title(), "layer_count": len(project.mapLayers()), "crs": project.crs().authid(), "layers": [] } # Add basic layer information (limit to 10 layers for performance) layers = list(project.mapLayers().values()) for i, layer in enumerate(layers): if i >= 10: # Limit to 10 layers break layer_info = { "id": layer.id(), "name": layer.name(), "type": self._get_layer_type(layer), "visible": layer.isValid() and project.layerTreeRoot().findLayer(layer.id()).isVisible() } info["layers"].append(layer_info) return info def _get_layer_type(self, layer): """Helper to get layer type as string""" if layer.type() == QgsMapLayer.VectorLayer: return f"vector_{layer.geometryType()}" elif layer.type() == QgsMapLayer.RasterLayer: return "raster" else: return str(layer.type()) def execute_code(self, code, **kwargs): """Execute arbitrary PyQGIS code""" # Capture stdout and stderr stdout_capture = io.StringIO() stderr_capture = io.StringIO() # Store original stdout and stderr original_stdout = sys.stdout original_stderr = sys.stderr try: # Redirect stdout and stderr sys.stdout = stdout_capture sys.stderr = stderr_capture # Create a local namespace for execution namespace = { "qgis": Qgis, "QgsProject": QgsProject, "iface": self.iface, "QgsApplication": QgsApplication, "QgsVectorLayer": QgsVectorLayer, "QgsRasterLayer": QgsRasterLayer, "QgsCoordinateReferenceSystem": QgsCoordinateReferenceSystem } # Execute the code exec(code, namespace) # Restore stdout and stderr sys.stdout = original_stdout sys.stderr = original_stderr return { "executed": True, "stdout": stdout_capture.getvalue(), "stderr": stderr_capture.getvalue() } except Exception as e: # Generate full traceback error_traceback = traceback.format_exc() # Restore stdout and stderr in case of exception sys.stdout = original_stdout sys.stderr = original_stderr return { "executed": False, "error": str(e), "traceback": error_traceback, "stdout": stdout_capture.getvalue(), "stderr": stderr_capture.getvalue() } def add_vector_layer(self, path, name=None, provider="ogr", **kwargs): """Add a vector layer to the project""" if not name: name = os.path.basename(path) # Create the layer layer = QgsVectorLayer(path, name, provider) if not layer.isValid(): raise Exception(f"Layer is not valid: {path}") # Add to project QgsProject.instance().addMapLayer(layer) return { "id": layer.id(), "name": layer.name(), "type": self._get_layer_type(layer), "feature_count": layer.featureCount() } def add_raster_layer(self, path, name=None, provider="gdal", **kwargs): """Add a raster layer to the project""" if not name: name = os.path.basename(path) # Create the layer layer = QgsRasterLayer(path, name, provider) if not layer.isValid(): raise Exception(f"Layer is not valid: {path}") # Add to project QgsProject.instance().addMapLayer(layer) return { "id": layer.id(), "name": layer.name(), "type": "raster", "width": layer.width(), "height": layer.height() } def get_layers(self, **kwargs): """Get all layers in the project""" project = QgsProject.instance() layers = [] for layer_id, layer in project.mapLayers().items(): layer_info = { "id": layer_id, "name": layer.name(), "type": self._get_layer_type(layer), "visible": project.layerTreeRoot().findLayer(layer_id).isVisible() } # Add type-specific information if layer.type() == QgsMapLayer.VectorLayer: layer_info.update({ "feature_count": layer.featureCount(), "geometry_type": layer.geometryType() }) elif layer.type() == QgsMapLayer.RasterLayer: layer_info.update({ "width": layer.width(), "height": layer.height() }) layers.append(layer_info) return layers def remove_layer(self, layer_id, **kwargs): """Remove a layer from the project""" project = QgsProject.instance() if layer_id in project.mapLayers(): project.removeMapLayer(layer_id) return {"removed": layer_id} else: raise Exception(f"Layer not found: {layer_id}") def zoom_to_layer(self, layer_id, **kwargs): """Zoom to a layer's extent""" project = QgsProject.instance() if layer_id in project.mapLayers(): layer = project.mapLayer(layer_id) self.iface.setActiveLayer(layer) self.iface.zoomToActiveLayer() return {"zoomed_to": layer_id} else: raise Exception(f"Layer not found: {layer_id}") def get_layer_features(self, layer_id, limit=10, **kwargs): """Get features from a vector layer""" project = QgsProject.instance() if layer_id in project.mapLayers(): layer = project.mapLayer(layer_id) if layer.type() != QgsMapLayer.VectorLayer: raise Exception(f"Layer is not a vector layer: {layer_id}") features = [] for i, feature in enumerate(layer.getFeatures()): if i >= limit: break # Extract attributes attrs = {} for field in layer.fields(): attrs[field.name()] = feature.attribute(field.name()) # Extract geometry if available geom = None if feature.hasGeometry(): geom = { "type": feature.geometry().type(), "wkt": feature.geometry().asWkt(precision=4) } features.append({ "id": feature.id(), "attributes": attrs, "geometry": geom }) return { "layer_id": layer_id, "feature_count": layer.featureCount(), "features": features, "fields": [field.name() for field in layer.fields()] } else: raise Exception(f"Layer not found: {layer_id}") def execute_processing(self, algorithm, parameters, **kwargs): """Execute a processing algorithm""" try: import processing result = processing.run(algorithm, parameters) return { "algorithm": algorithm, "result": {k: str(v) for k, v in result.items()} # Convert values to strings for JSON } except Exception as e: raise Exception(f"Processing error: {str(e)}") def save_project(self, path=None, **kwargs): """Save the current project""" project = QgsProject.instance() if not path and not project.fileName(): raise Exception("No project path specified and no current project path") save_path = path if path else project.fileName() if project.write(save_path): return {"saved": save_path} else: raise Exception(f"Failed to save project to {save_path}") def load_project(self, path, **kwargs): """Load a project""" project = QgsProject.instance() if project.read(path): self.iface.mapCanvas().refresh() return { "loaded": path, "layer_count": len(project.mapLayers()) } else: raise Exception(f"Failed to load project from {path}") def create_new_project(self, path, **kwargs): """ Creates a new QGIS project and saves it at the specified path. If a project is already loaded, it clears it before creating the new one. :param project_path: Full path where the project will be saved (e.g., 'C:/path/to/project.qgz') """ project = QgsProject.instance() if project.fileName(): project.clear() project.setFileName(path) self.iface.mapCanvas().refresh() # Save the project if project.write(): return { "created": f"Project created and saved successfully at: {path}", "layer_count": len(project.mapLayers()) } else: raise Exception(f"Failed to save project to {path}") def render_map(self, path, width=800, height=600, **kwargs): """Render the current map view to an image""" try: # Create map settings ms = QgsMapSettings() # Set layers to render layers = list(QgsProject.instance().mapLayers().values()) ms.setLayers(layers) # Set map canvas properties rect = self.iface.mapCanvas().extent() ms.setExtent(rect) ms.setOutputSize(QSize(width, height)) ms.setBackgroundColor(QColor(255, 255, 255)) ms.setOutputDpi(96) # Create the render render = QgsMapRendererParallelJob(ms) # Start rendering render.start() render.waitForFinished() # Get the image and save img = render.renderedImage() if img.save(path): return { "rendered": True, "path": path, "width": width, "height": height } else: raise Exception(f"Failed to save rendered image to {path}") except Exception as e: raise Exception(f"Render error: {str(e)}") class QgisMCPDockWidget(QDockWidget): """Dock widget for the QGIS MCP plugin""" closed = pyqtSignal() def __init__(self, iface): super().__init__("QGIS MCP") self.iface = iface self.server = None self.setup_ui() def setup_ui(self): """Set up the dock widget UI""" # Create widget and layout widget = QWidget() layout = QVBoxLayout() widget.setLayout(layout) # Add port selection layout.addWidget(QLabel("Server Port:")) self.port_spin = QSpinBox() self.port_spin.setMinimum(1024) self.port_spin.setMaximum(65535) self.port_spin.setValue(9876) layout.addWidget(self.port_spin) # Add server control buttons self.start_button = QPushButton("Start Server") self.start_button.clicked.connect(self.start_server) layout.addWidget(self.start_button) self.stop_button = QPushButton("Stop Server") self.stop_button.clicked.connect(self.stop_server) self.stop_button.setEnabled(False) layout.addWidget(self.stop_button) # Add status label self.status_label = QLabel("Server: Stopped") layout.addWidget(self.status_label) # Add to dock widget self.setWidget(widget) def start_server(self): """Start the server""" if not self.server: port = self.port_spin.value() self.server = QgisMCPServer(port=port, iface=self.iface) if self.server.start(): self.status_label.setText(f"Server: Running on port {self.server.port}") self.start_button.setEnabled(False) self.stop_button.setEnabled(True) self.port_spin.setEnabled(False) def stop_server(self): """Stop the server""" if self.server: self.server.stop() self.server = None self.status_label.setText("Server: Stopped") self.start_button.setEnabled(True) self.stop_button.setEnabled(False) self.port_spin.setEnabled(True) def closeEvent(self, event): """Stop server on dock close""" self.stop_server() self.closed.emit() super().closeEvent(event) class QgisMCPPlugin: """Main plugin class for QGIS MCP""" def __init__(self, iface): self.iface = iface self.dock_widget = None self.action = None def initGui(self): """Initialize GUI""" # Create action self.action = QAction( "QGIS MCP", self.iface.mainWindow() ) self.action.setCheckable(True) self.action.triggered.connect(self.toggle_dock) # Add to plugins menu and toolbar self.iface.addPluginToMenu("QGIS MCP", self.action) self.iface.addToolBarIcon(self.action) def toggle_dock(self, checked): """Toggle the dock widget""" if checked: # Create dock widget if it doesn't exist if not self.dock_widget: self.dock_widget = QgisMCPDockWidget(self.iface) self.iface.addDockWidget(Qt.RightDockWidgetArea, self.dock_widget) # Connect close event self.dock_widget.closed.connect(self.dock_closed) else: # Show existing dock widget self.dock_widget.show() else: # Hide dock widget if self.dock_widget: self.dock_widget.hide() def dock_closed(self): """Handle dock widget closed""" self.action.setChecked(False) def unload(self): """Unload plugin""" # Stop server if running if self.dock_widget: self.dock_widget.stop_server() self.iface.removeDockWidget(self.dock_widget) self.dock_widget = None # Remove plugin menu item and toolbar icon self.iface.removePluginMenu("QGIS MCP", self.action) self.iface.removeToolBarIcon(self.action) # Plugin entry point def classFactory(iface): return QgisMCPPlugin(iface) ```