QGISMCP

Instructions

Implementation Reference

• src/qgis_mcp/qgis_mcp_server.py:160-165 (handler)

  Primary MCP tool handler for 'load_project'. Proxies the request to the QGIS plugin server via socket connection and returns JSON response.

  @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)

• qgis_mcp_plugin/qgis_mcp_plugin.py:133-149 (registration)

  Backend socket server registration of 'load_project' handler in QGIS plugin.

  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,
  }

• qgis_mcp_plugin/qgis_mcp_plugin.py:403-414 (handler)

  Backend QGIS plugin handler that executes the actual project loading using QgsProject.instance().read(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}")

• src/qgis_mcp/qgis_mcp_server.py:39-76 (helper)

  Helper function used by MCP tools to send commands to the QGIS plugin socket server.

  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