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