get_layer_features
Retrieve features from vector layers in QGIS to analyze spatial data attributes and geometry with configurable result limits.
Instructions
Retrieve features from a vector layer with an optional limit.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| layer_id | Yes | ||
| limit | No |
Implementation Reference
- src/qgis_mcp/qgis_mcp_server.py:222-227 (handler)MCP tool handler for 'get_layer_features' that proxies the command to the QGIS socket server via get_qgis_connection().@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)
- Core implementation of get_layer_features that retrieves and serializes features from a QGIS vector layer, including attributes and geometry.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}")
- qgis_mcp_plugin/qgis_mcp_plugin.py:133-149 (registration)Registers 'get_layer_features' as a command handler in the QGIS MCP socket server's execute_command method.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, }
- Helper method used by the MCP handler to send the get_layer_features command 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
- Client-side helper method mirroring the get_layer_features command for the socket client.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})