sjoin_nearest_gpd
Perform spatial joins between geospatial datasets by matching features based on nearest proximity, with configurable distance thresholds and join types.
Instructions
Nearest neighbor spatial join using geopandas.sjoin_nearest. Args: left_path: Path to the left geospatial file. right_path: Path to the right geospatial file. how: Type of join ('left', 'right'). max_distance: Optional maximum search distance. output_path: Optional path to save the result. Returns: Dictionary with status, message, and output info.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| left_path | Yes | ||
| right_path | Yes | ||
| how | No | left | |
| max_distance | No | ||
| output_path | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- Core handler function that loads two geospatial files, performs nearest neighbor spatial join with geopandas.sjoin_nearest, handles CRS alignment, optional distance limit and output saving, returns preview and metadata.
@gis_mcp.tool() def sjoin_nearest_gpd(left_path: str, right_path: str, how: str = "left", max_distance: float = None, output_path: str = None) -> Dict[str, Any]: """ Nearest neighbor spatial join using geopandas.sjoin_nearest. Args: left_path: Path to the left geospatial file. right_path: Path to the right geospatial file. how: Type of join ('left', 'right'). max_distance: Optional maximum search distance. 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) kwargs = {"how": how} if max_distance is not None: kwargs["max_distance"] = max_distance result = gpd.sjoin_nearest(left, right, **kwargs) 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"Nearest spatial join ({how}) 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_nearest_gpd: {str(e)}") return {"status": "error", "message": str(e)} - src/gis_mcp/main.py:66-72 (registration)Imports the geopandas_functions module (among others), which executes the @gis_mcp.tool decorators to register the sjoin_nearest_gpd tool with the MCP server.
from . import ( geopandas_functions, shapely_functions, rasterio_functions, pyproj_functions, pysal_functions, ) - src/gis_mcp/geopandas_functions.py:28-39 (registration)MCP resource that lists 'sjoin_nearest_gpd' as one of the available GeoPandas join operations for 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-5 (registration)Creates the FastMCP instance 'gis_mcp' used for tool and resource registration via decorators.
gis_mcp = FastMCP("GIS MCP")