QGISMCP

Instructions

Implementation Reference

• src/qgis_mcp/qgis_mcp_server.py:256-261 (handler)

  MCP tool handler for 'execute_code' that proxies the request to the QGIS plugin server socket connection.
  @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)
• qgis_mcp_plugin/qgis_mcp_plugin.py:221-239 (handler)

  Core handler in QGIS plugin that executes the PyQGIS code using exec() with a predefined safe namespace containing QGIS essentials.
  def execute_code(self, code, **kwargs): """Execute arbitrary PyQGIS code""" try: # 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) return {"executed": True} except Exception as e: raise Exception(f"Code execution error: {str(e)}")
• qgis_mcp_plugin/qgis_mcp_plugin.py:133-149 (registration)

  Registration of all command handlers in the QGIS plugin server, including 'execute_code'.
  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, }
• src/qgis_mcp/qgis_mcp_server.py:79-108 (helper)

  Helper function to get or create a persistent socket connection to the QGIS plugin server, used by all MCP tools including execute_code.
  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