query_features
Extract vector layer data from GeoServer using CQL filters to specify query conditions, return properties, and limit results.
Instructions
Query features from a vector layer using CQL filter.
Args:
workspace: The workspace containing the layer
layer: The layer to query
filter: Optional CQL filter expression
properties: Optional list of properties to return
max_features: Maximum number of features to return
Returns:
GeoJSON FeatureCollection with query resultsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| filter | No | ||
| layer | Yes | ||
| max_features | No | ||
| properties | No | ||
| workspace | Yes |
Implementation Reference
- src/geoserver_mcp/main.py:309-370 (handler)Implementation of the 'query_features' tool handler. Queries features from a GeoServer vector layer via WFS GetFeature request with optional CQL filter, property selection, and max features limit. Returns a GeoJSON FeatureCollection. Registered via @mcp.tool() decorator.
def query_features( workspace: str, layer: str, filter: Optional[str] = None, properties: Optional[List[str]] = None, max_features: Optional[int] = 10 ) -> Dict[str, Any]: """Query features from a vector layer using CQL filter. Args: workspace: The workspace containing the layer layer: The layer to query filter: Optional CQL filter expression properties: Optional list of properties to return max_features: Maximum number of features to return Returns: GeoJSON FeatureCollection with query results """ geo = get_geoserver() if geo is None: raise ValueError("Not connected to GeoServer") if not workspace or not layer: raise ValueError("Workspace and layer name are required") try: # Construct WFS GetFeature request URL url = f"{geo.service_url}/wfs" params = { "service": "WFS", "version": "1.0.0", "request": "GetFeature", "typeName": f"{workspace}:{layer}", "outputFormat": "application/json", "maxFeatures": max_features or 10 } # Add CQL filter if provided if filter: params["CQL_FILTER"] = filter # Add property names if provided if properties: params["propertyName"] = ",".join(properties) # Make the request import requests response = requests.get(url, params=params, auth=(geo.username, geo.password)) response.raise_for_status() # Parse the GeoJSON response features = response.json() return { "type": "FeatureCollection", "features": features.get("features", []) } except Exception as e: logger.error(f"Error querying features: {str(e)}") raise ValueError(f"Failed to query features: {str(e)}")