GeoServer MCP Server

Instructions

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

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)}")