sjoin_gpd
Perform spatial joins between two geospatial datasets using specified join type and spatial predicate. Save the output to a defined path for further geospatial analysis.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| how | No | inner | |
| left_path | Yes | ||
| output_path | No | ||
| predicate | No | intersects | |
| right_path | Yes |
Implementation Reference
- The primary handler function for the 'sjoin_gpd' tool. It reads two geospatial files, performs a spatial join using geopandas.sjoin with specified how and predicate, optionally saves the result, and returns metadata including preview.@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) preview = result.head(5).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 listing that includes 'sjoin_gpd' as one of the available GeoPandas join operations, serving as a discovery mechanism for the 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" ] }