append_gpd

Instructions

Implementation Reference

• src/gis_mcp/geopandas_functions.py:74-119 (handler)

  The append_gpd function is the core handler that reads two shapefiles, ensures matching CRS, concatenates them vertically using pd.concat, and saves the result. It is registered as an MCP tool via @gis_mcp.tool() decorator.

  @gis_mcp.tool()
  def append_gpd(shapefile1_path: str, shapefile2_path: str, output_path: str) -> Dict[str, Any]:
      """ Reads two shapefiles directly, concatenates them vertically."""
      try:
          # Configure a basic logger for demonstration
          logging.basicConfig(level=logging.INFO)
          logger = logging.getLogger(__name__)
  
          # Step 1: Read the two shapefiles into GeoDataFrames.
          logger.info(f"Reading {shapefile1_path}...")
          gdf1 = gpd.read_file(shapefile1_path)
          
          logger.info(f"Reading {shapefile2_path}...")
          gdf2 = gpd.read_file(shapefile2_path)
  
          # Step 2: Ensure the Coordinate Reference Systems (CRS) match.
          if gdf1.crs != gdf2.crs:
              logger.warning(
                  f"CRS mismatch: GDF1 has '{gdf1.crs}' and GDF2 has '{gdf2.crs}'. "
                  "Reprojecting GDF2."
              )
              gdf2 = gdf2.to_crs(gdf1.crs)
  
          # Step 3: Concatenate the two GeoDataFrames.
          combined_gdf = pd.concat([gdf1, gdf2], ignore_index=True)
  
          # Step 4: Save the combined GeoDataFrame to a new shapefile.
          output_path_resolved = resolve_path(output_path, relative_to_storage=True)
          output_path_resolved.parent.mkdir(parents=True, exist_ok=True)
          logger.info(f"Saving combined shapefile to {output_path_resolved}...")
          combined_gdf.to_file(str(output_path_resolved), driver='ESRI Shapefile')
  
          return {
              "status": "success",
              "message": f"Shapefiles concatenated successfully into '{output_path_resolved}'.",
              "info": {
                  "output_path": str(output_path_resolved),
                  "num_features": len(combined_gdf),
                  "crs": str(combined_gdf.crs),
                  "columns": list(combined_gdf.columns)
              }
          }
      
      except Exception as e:
          logger.error(f"Error processing shapefiles: {str(e)}")
          raise ValueError(f"Failed to process shapefiles: {str(e)}")

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

  The string "append_gpd" is listed as an available operation under the gis://geopandas/joins resource, serving as a registration/reference entry point for this tool.

  @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"
          ]
      }

• tests/test_geopandas_functions.py:52-73 (helper)

  Test for append_gpd: creates two GeoDataFrames, calls the tool via client.call_tool('append_gpd', ...), and asserts success.

  async def test_append_gpd(self, sample_geodataframe, temp_dir):
      """Test appending GeoDataFrames."""
      _, file1 = sample_geodataframe
      # Create second file
      gdf2 = gpd.GeoDataFrame(
          {'id': [4, 5], 'name': ['D', 'E'], 'value': [40, 50]},
          geometry=[Point(3, 3), Point(4, 4)],
          crs='EPSG:4326'
      )
      file2 = os.path.join(temp_dir, "test2.shp")
      gdf2.to_file(file2)
      
      output_path = os.path.join(temp_dir, "appended.shp")
      async with Client(gis_mcp) as client:
          result = await client.call_tool("append_gpd", {
              "shapefile1_path": file1,
              "shapefile2_path": file2,
              "output_path": output_path
          })
          result_data = get_result_data(result)
          assert result_data["status"] == "success"
          assert os.path.exists(output_path)

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

  The FastMCP instance 'gis_mcp' is created here. The @gis_mcp.tool() decorator on append_gpd uses this instance to register the tool.

  # MCP imports using the new SDK patterns
  from fastmcp import FastMCP
  
  
  gis_mcp = FastMCP("GIS MCP")