GIS MCP Server

Instructions

Implementation Reference

• src/gis_mcp/geopandas_functions.py:337-377 (handler)

  Main handler function for the 'sjoin_gpd' tool, performing spatial join on two GeoDataFrames using geopandas.sjoin.

  @gis_mcp.tool()
  def sjoin_gpd(left_path: str, right_path: str, how: str = "inner", predicate: str = "intersects", output_path: str = None) -> Dict[str, Any]:
      """
      Spatial join between two GeoDataFrames using geopandas.sjoin.
      Args:
          left_path: Path to the left geospatial file.
          right_path: Path to the right geospatial file.
          how: Type of join ('left', 'right', 'inner').
          predicate: Spatial predicate ('intersects', 'within', 'contains', etc.).
          output_path: Optional path to save the result.
      Returns:
          Dictionary with status, message, and output info.
      """
      try:
          left = gpd.read_file(left_path)
          right = gpd.read_file(right_path)
          if left.crs != right.crs:
              right = right.to_crs(left.crs)
          result = gpd.sjoin(left, right, how=how, predicate=predicate)
          if output_path:
              output_path_resolved = resolve_path(output_path, relative_to_storage=True)
              output_path_resolved.parent.mkdir(parents=True, exist_ok=True)
              result.to_file(str(output_path_resolved))
              output_path = str(output_path_resolved)
          # Convert geometry to WKT for serialization
          preview_df = result.head(5).copy()
          if 'geometry' in preview_df.columns:
              preview_df['geometry'] = preview_df['geometry'].apply(lambda g: g.wkt if g is not None else None)
          preview = preview_df.to_dict(orient="records")
          return {
              "status": "success",
              "message": f"Spatial join ({how}, {predicate}) completed successfully.",
              "num_features": len(result),
              "crs": str(result.crs),
              "columns": list(result.columns),
              "preview": preview,
              "output_path": output_path,
          }
      except Exception as e:
          logger.error(f"Error in sjoin_gpd: {str(e)}")
          return {"status": "error", "message": str(e)}

• src/gis_mcp/geopandas_functions.py:28-39 (registration)

  Resource function listing 'sjoin_gpd' among available GeoPandas join operations, serving as tool discovery.

  @gis_mcp.resource("gis://geopandas/joins")
  def get_geopandas_joins() -> Dict[str, List[str]]:
      """List available GeoPandas join operations."""
      return {
          "operations": [
              "append_gpd",
              "merge_gpd",
              "sjoin_gpd",
              "sjoin_nearest_gpd",
              "point_in_polygon"
          ]
      }

• src/gis_mcp/mcp.py:5-6 (registration)

  FastMCP server instance 'gis_mcp' that handles tool registration via decorators like @gis_mcp.tool().

  gis_mcp = FastMCP("GIS MCP")