From 0ef2c70af49719476b3edaf38c22039f656a1814 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 3 Dec 2025 15:31:17 -0700 Subject: [PATCH] Add Surface Management Agency and Mining Claims tools Refactor to use FastMCP mixin pattern for composable tools: - PLSSMixin: Section/Township/Range queries - SurfaceManagementMixin: Land manager (BLM, USFS, NPS, Private, etc.) - MiningClaimsMixin: Active/closed mining claims from MLRS Shared BLM API client handles identify/query operations against multiple ArcGIS REST endpoints. --- README.md | 88 +++++++- src/mcblmplss/__init__.py | 8 +- src/mcblmplss/client.py | 113 ++++++++++ src/mcblmplss/mixins/__init__.py | 11 + src/mcblmplss/mixins/mining_claims.py | 181 +++++++++++++++ src/mcblmplss/mixins/plss.py | 174 +++++++++++++++ src/mcblmplss/mixins/surface_management.py | 183 +++++++++++++++ src/mcblmplss/server.py | 247 +++------------------ 8 files changed, 781 insertions(+), 224 deletions(-) create mode 100644 src/mcblmplss/client.py create mode 100644 src/mcblmplss/mixins/__init__.py create mode 100644 src/mcblmplss/mixins/mining_claims.py create mode 100644 src/mcblmplss/mixins/plss.py create mode 100644 src/mcblmplss/mixins/surface_management.py diff --git a/README.md b/README.md index d0ce9e3..9e8fc1e 100644 --- a/README.md +++ b/README.md @@ -1,32 +1,98 @@ # mcblmplss -FastMCP server for querying BLM Public Land Survey System (PLSS) data by coordinates. +FastMCP server for querying BLM (Bureau of Land Management) land data by coordinates. -## Usage +## Features + +- **PLSS** - Public Land Survey System (Section, Township, Range) +- **Surface Management** - Who manages the land (BLM, Forest Service, NPS, Private, etc.) +- **Mining Claims** - Active and closed mining claims from MLRS database + +## Installation ```bash -# Install and run +# Run directly with uvx uvx mcblmplss # Add to Claude Code -claude mcp add plss "uvx mcblmplss" +claude mcp add blm-land "uvx mcblmplss" ``` ## Tools -- **get_plss_location** - Get Section/Township/Range for lat/long coordinates -- **get_plss_details** - Get full structured PLSS data +### PLSS (Public Land Survey System) -## Example +| Tool | Description | +|------|-------------| +| `get_plss_location` | Get Section/Township/Range as human-readable text | +| `get_plss_details` | Get full PLSS data as structured object | ``` -> get_plss_location(latitude=40.0, longitude=-105.0) +> get_plss_location(40.0, -105.0) -Section 10, Township 1N, Range 68W, 6th Meridian +Section 9, Township 1N, Range 68W, 6th Meridian State: CO -PLSS ID: CO060010N0680W0SN100 +PLSS ID: CO060010S0680W0SN090 +``` + +### Surface Management Agency + +| Tool | Description | +|------|-------------| +| `get_land_manager` | Determine who manages the land | +| `get_land_manager_details` | Get full SMA data as structured object | + +``` +> get_land_manager(38.5, -110.5) + +Bureau of Land Management (BLM) - Department of the Interior +Unit: Bureau of Land Management +State: UT +Status: Federal, Public land +``` + +### Mining Claims + +| Tool | Description | +|------|-------------| +| `get_mining_claims` | Find mining claims at/near location | +| `get_mining_claims_details` | Get full claim data as structured objects | + +``` +> get_mining_claims(39.5, -117.0) + +Found 42 mining claim(s): + +MAGA #6 + Serial: NV105221817 + Type: Lode Claim + Status: Active + Acres: 20.66 +... ``` ## Coverage -PLSS data is available for 30 states where federal land surveys were conducted (western/midwestern US). Not available for eastern seaboard states or Texas. +Data available for **30 western/midwestern states** where federal land surveys were conducted. Not available for eastern seaboard states or Texas (different survey systems). + +## Data Sources + +All data queried from official BLM ArcGIS REST services: + +- [BLM National PLSS](https://gis.blm.gov/arcgis/rest/services/Cadastral/BLM_Natl_PLSS_CadNSDI/MapServer) +- [Surface Management Agency](https://gis.blm.gov/arcgis/rest/services/lands/BLM_Natl_SMA_LimitedScale/MapServer) +- [Mining Claims (MLRS)](https://gis.blm.gov/nlsdb/rest/services/Mining_Claims/MiningClaims/MapServer) + +## Architecture + +Uses FastMCP's mixin pattern for composable tool modules: + +``` +src/mcblmplss/ +├── server.py # Main FastMCP server +├── client.py # Shared BLM API client +└── mixins/ + ├── plss.py # PLSS tools + ├── surface_management.py # SMA tools + └── mining_claims.py # Mining claims tools +``` diff --git a/src/mcblmplss/__init__.py b/src/mcblmplss/__init__.py index a567393..4f825a7 100644 --- a/src/mcblmplss/__init__.py +++ b/src/mcblmplss/__init__.py @@ -1,8 +1,10 @@ """ -mcblmplss - FastMCP server for BLM Public Land Survey System queries. +mcblmplss - FastMCP server for BLM land data queries. -Query PLSS data (Township, Range, Section) from coordinates using the -BLM National Cadastral API. +Query BLM data by coordinates: +- PLSS (Public Land Survey System) - Section, Township, Range +- Surface Management Agency - Who manages the land +- Mining Claims - Active and closed mining claims """ from mcblmplss.server import main, mcp diff --git a/src/mcblmplss/client.py b/src/mcblmplss/client.py new file mode 100644 index 0000000..299071a --- /dev/null +++ b/src/mcblmplss/client.py @@ -0,0 +1,113 @@ +""" +Shared HTTP client for BLM ArcGIS REST API queries. + +Provides common identify/query operations against BLM MapServer endpoints. +""" + +import httpx +from typing import Any + + +class BLMClient: + """Async HTTP client for BLM ArcGIS REST services.""" + + def __init__(self, timeout: float = 30.0): + self.timeout = timeout + + async def identify( + self, + base_url: str, + latitude: float, + longitude: float, + layers: str, + tolerance: int = 1, + return_geometry: bool = False, + ) -> list[dict[str, Any]]: + """ + Perform an identify operation on a MapServer. + + Args: + base_url: MapServer URL (e.g., https://gis.blm.gov/.../MapServer) + latitude: WGS84 latitude + longitude: WGS84 longitude + layers: Layer specification (e.g., "all:1,2" or "visible:0") + tolerance: Pixel tolerance for hit detection + return_geometry: Whether to include geometry in response + + Returns: + List of result dictionaries with layerId and attributes + """ + params = { + "f": "json", + "geometry": f"{longitude},{latitude}", + "geometryType": "esriGeometryPoint", + "sr": "4326", + "layers": layers, + "tolerance": str(tolerance), + "mapExtent": f"{longitude-1},{latitude-1},{longitude+1},{latitude+1}", + "imageDisplay": "100,100,96", + "returnGeometry": "true" if return_geometry else "false", + } + + async with httpx.AsyncClient(timeout=self.timeout) as client: + response = await client.get(f"{base_url}/identify", params=params) + response.raise_for_status() + data = response.json() + + return data.get("results", []) + + async def query( + self, + base_url: str, + layer_id: int, + where: str = "1=1", + geometry: dict | None = None, + out_fields: str = "*", + return_geometry: bool = False, + result_offset: int = 0, + result_record_count: int = 100, + ) -> list[dict[str, Any]]: + """ + Perform a query operation on a specific layer. + + Args: + base_url: MapServer URL + layer_id: Layer ID to query + where: SQL WHERE clause + geometry: Optional geometry filter (point, envelope, etc.) + out_fields: Fields to return (* for all) + return_geometry: Whether to include geometry + result_offset: Pagination offset + result_record_count: Max records to return + + Returns: + List of feature attribute dictionaries + """ + params = { + "f": "json", + "where": where, + "outFields": out_fields, + "returnGeometry": "true" if return_geometry else "false", + "resultOffset": str(result_offset), + "resultRecordCount": str(result_record_count), + } + + if geometry: + params["geometry"] = str(geometry) + params["geometryType"] = geometry.get("type", "esriGeometryPoint") + params["spatialRel"] = "esriSpatialRelIntersects" + params["inSR"] = "4326" + + async with httpx.AsyncClient(timeout=self.timeout) as client: + response = await client.get( + f"{base_url}/{layer_id}/query", params=params + ) + response.raise_for_status() + data = response.json() + + features = data.get("features", []) + return [f.get("attributes", {}) for f in features] + + +# Shared client instance - longer timeout for slower services like mining claims +blm_client = BLMClient(timeout=60.0) diff --git a/src/mcblmplss/mixins/__init__.py b/src/mcblmplss/mixins/__init__.py new file mode 100644 index 0000000..c502b07 --- /dev/null +++ b/src/mcblmplss/mixins/__init__.py @@ -0,0 +1,11 @@ +""" +MCP Mixins for BLM data services. + +Each mixin provides tools for a specific BLM data domain. +""" + +from mcblmplss.mixins.plss import PLSSMixin +from mcblmplss.mixins.surface_management import SurfaceManagementMixin +from mcblmplss.mixins.mining_claims import MiningClaimsMixin + +__all__ = ["PLSSMixin", "SurfaceManagementMixin", "MiningClaimsMixin"] diff --git a/src/mcblmplss/mixins/mining_claims.py b/src/mcblmplss/mixins/mining_claims.py new file mode 100644 index 0000000..84e80a7 --- /dev/null +++ b/src/mcblmplss/mixins/mining_claims.py @@ -0,0 +1,181 @@ +""" +Mining Claims mixin for BLM MCP server. + +Provides tools for querying active and closed mining claims from BLM's MLRS database. +""" + +from pydantic import BaseModel, Field +from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool + +from mcblmplss.client import blm_client + +# Mining Claims MapServer (separate server from main BLM arcgis) +MINING_URL = "https://gis.blm.gov/nlsdb/rest/services/Mining_Claims/MiningClaims/MapServer" +LAYER_ACTIVE = 1 +LAYER_CLOSED = 2 + +# Claim type codes +CLAIM_TYPES = { + "384101": "Lode Claim", + "384102": "Placer Claim", + "384103": "Mill Site", + "384104": "Tunnel Site", +} + + +class MiningClaim(BaseModel): + """Individual mining claim record.""" + + claim_name: str = Field(..., description="Geographic/claim name") + serial_number: str = Field(..., description="BLM case serial number") + claim_type: str = Field(..., description="Type of claim (Lode, Placer, Mill Site, etc.)") + status: str = Field(..., description="Case disposition (Active, Closed)") + acres: float | None = Field(None, description="Claim acreage") + legal_description: str | None = Field(None, description="PLSS legal description") + + +class MiningClaimsResult(BaseModel): + """Result from mining claims query.""" + + latitude: float + longitude: float + claims: list[MiningClaim] = Field(default_factory=list) + total_found: int = 0 + error: str | None = None + + +def _parse_claim(attrs: dict, status: str) -> MiningClaim: + """Parse mining claim attributes from API response.""" + # API returns human-readable field names + claim_type_code = attrs.get("BLM Product Code", "") + claim_type = attrs.get("BLM Product", "") or CLAIM_TYPES.get(claim_type_code, "Unknown") + + acres_str = attrs.get("Case Acres", "") + try: + acres = float(acres_str) if acres_str else None + except (ValueError, TypeError): + acres = None + + return MiningClaim( + claim_name=attrs.get("Geographic Name", "Unnamed"), + serial_number=attrs.get("Case Serial Number", ""), + claim_type=claim_type, + status=status, + acres=acres, + legal_description=attrs.get("Case Metadata"), + ) + + +class MiningClaimsMixin(MCPMixin): + """MCP tools for mining claims queries.""" + + @mcp_tool( + name="get_mining_claims", + description="Find active mining claims at or near coordinates. Returns claim names, serial numbers, and types.", + ) + async def get_mining_claims( + self, + latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), + longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), + include_closed: bool = Field( + default=False, description="Include closed/void claims in results" + ), + tolerance: int = Field( + default=10, description="Search radius in pixels (larger = wider search)" + ), + ) -> str: + """ + Find mining claims at a location. + + Mining claims include: + - Lode Claims: Hard rock mineral deposits (veins, ledges) + - Placer Claims: Loose mineral deposits (gold in streams) + - Mill Sites: Processing facilities + - Tunnel Sites: Underground access + + Returns claim names, BLM serial numbers, and acreage. + """ + result = await self._query_claims(latitude, longitude, include_closed, tolerance) + + if result.error: + return f"Error: {result.error}" + + if not result.claims: + return "No mining claims found at this location." + + lines = [f"Found {result.total_found} mining claim(s):"] + for claim in result.claims[:20]: # Limit display to 20 + lines.append(f"\n{claim.claim_name}") + lines.append(f" Serial: {claim.serial_number}") + lines.append(f" Type: {claim.claim_type}") + lines.append(f" Status: {claim.status}") + if claim.acres: + lines.append(f" Acres: {claim.acres:.2f}") + + if result.total_found > 20: + lines.append(f"\n... and {result.total_found - 20} more claims") + + return "\n".join(lines) + + @mcp_tool( + name="get_mining_claims_details", + description="Get detailed mining claims data as structured objects.", + ) + async def get_mining_claims_details( + self, + latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), + longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), + include_closed: bool = Field( + default=False, description="Include closed/void claims" + ), + tolerance: int = Field(default=10, description="Search radius in pixels"), + ) -> MiningClaimsResult: + """Get full mining claims data including legal descriptions.""" + return await self._query_claims(latitude, longitude, include_closed, tolerance) + + async def _query_claims( + self, + latitude: float, + longitude: float, + include_closed: bool, + tolerance: int, + ) -> MiningClaimsResult: + """Query BLM Mining Claims API.""" + layers = f"all:{LAYER_ACTIVE}" + if include_closed: + layers = f"all:{LAYER_ACTIVE},{LAYER_CLOSED}" + + results = await blm_client.identify( + MINING_URL, latitude, longitude, layers, tolerance=tolerance + ) + + if not results: + return MiningClaimsResult( + latitude=latitude, + longitude=longitude, + claims=[], + total_found=0, + ) + + claims = [] + seen_serials = set() + + for result in results: + layer_id = result.get("layerId") + attrs = result.get("attributes", {}) + + # Determine status from layer + status = "Active" if layer_id == LAYER_ACTIVE else "Closed" + + # Dedupe by serial number + serial = attrs.get("Case Serial Number", "") + if serial and serial not in seen_serials: + seen_serials.add(serial) + claims.append(_parse_claim(attrs, status)) + + return MiningClaimsResult( + latitude=latitude, + longitude=longitude, + claims=claims, + total_found=len(claims), + ) diff --git a/src/mcblmplss/mixins/plss.py b/src/mcblmplss/mixins/plss.py new file mode 100644 index 0000000..dbebd47 --- /dev/null +++ b/src/mcblmplss/mixins/plss.py @@ -0,0 +1,174 @@ +""" +PLSS (Public Land Survey System) mixin for BLM MCP server. + +Provides tools for querying Section, Township, and Range from coordinates. +""" + +from pydantic import BaseModel, Field +from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool + +from mcblmplss.client import blm_client + +# BLM Cadastral MapServer +PLSS_URL = "https://gis.blm.gov/arcgis/rest/services/Cadastral/BLM_Natl_PLSS_CadNSDI/MapServer" +LAYER_TOWNSHIP = 1 +LAYER_SECTION = 2 + + +class PLSSLocation(BaseModel): + """Public Land Survey System location.""" + + section: str | None = Field(None, description="Section number (1-36)") + township: str = Field(..., description="Township with direction (e.g., '4N')") + range: str = Field(..., description="Range with direction (e.g., '6E')") + principal_meridian: str = Field(..., description="Principal meridian name") + state: str = Field(..., description="State abbreviation") + label: str = Field(..., description="Human-readable label") + plss_id: str = Field(..., description="Unique PLSS identifier") + + +class PLSSResult(BaseModel): + """Result from PLSS coordinate query.""" + + latitude: float + longitude: float + township: PLSSLocation | None = None + section: PLSSLocation | None = None + error: str | None = None + + +def _parse_township(attrs: dict) -> PLSSLocation: + """Parse township attributes from API response.""" + twp_num = attrs.get("TWNSHPNO", "").lstrip("0") or "0" + twp_dir = attrs.get("TWNSHPDIR", "") + rng_num = attrs.get("RANGENO", "").lstrip("0") or "0" + rng_dir = attrs.get("RANGEDIR", "") + + return PLSSLocation( + section=None, + township=f"{twp_num}{twp_dir}", + range=f"{rng_num}{rng_dir}", + principal_meridian=attrs.get("PRINMER", "Unknown"), + state=attrs.get("STATEABBR", ""), + label=attrs.get("TWNSHPLAB", ""), + plss_id=attrs.get("PLSSID", ""), + ) + + +def _parse_section(attrs: dict, township: PLSSLocation | None) -> PLSSLocation: + """Parse section attributes from API response.""" + sec_num = ( + attrs.get("First Division Number", "") or attrs.get("FRSTDIVNO", "") + ).lstrip("0") or "0" + div_id = attrs.get("First Division Identifier", "") or attrs.get("FRSTDIVID", "") + + if township: + return PLSSLocation( + section=sec_num, + township=township.township, + range=township.range, + principal_meridian=township.principal_meridian, + state=township.state, + label=f"Section {sec_num}, {township.label}", + plss_id=div_id, + ) + + twp_id = attrs.get("Township Identifier", "") or attrs.get("PLSSID", "") + return PLSSLocation( + section=sec_num, + township="Unknown", + range="Unknown", + principal_meridian="Unknown", + state=twp_id[:2] if len(twp_id) >= 2 else "", + label=f"Section {sec_num}", + plss_id=div_id, + ) + + +class PLSSMixin(MCPMixin): + """MCP tools for Public Land Survey System queries.""" + + @mcp_tool( + name="get_plss_location", + description="Get Section/Township/Range for coordinates. Returns the PLSS legal land description.", + ) + async def get_plss_location( + self, + latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), + longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), + ) -> str: + """ + Get PLSS location for coordinates. + + Example: "Section 12, Township 4N, Range 6E, 6th Meridian" + + Coverage: 30 western/midwestern states. Not available for eastern + seaboard states or Texas (different survey systems). + """ + result = await self._query_plss(latitude, longitude) + + if result.error: + return f"Error: {result.error}" + + if result.section: + return ( + f"Section {result.section.section}, " + f"Township {result.section.township}, " + f"Range {result.section.range}, " + f"{result.section.principal_meridian}\n" + f"State: {result.section.state}\n" + f"PLSS ID: {result.section.plss_id}" + ) + elif result.township: + return ( + f"Township {result.township.township}, " + f"Range {result.township.range}, " + f"{result.township.principal_meridian}\n" + f"State: {result.township.state}\n" + f"(Section data not available)" + ) + return "Unable to determine PLSS location." + + @mcp_tool( + name="get_plss_details", + description="Get detailed PLSS data as structured object with all metadata.", + ) + async def get_plss_details( + self, + latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), + longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), + ) -> PLSSResult: + """Get full PLSS details including township and section objects.""" + return await self._query_plss(latitude, longitude) + + async def _query_plss(self, latitude: float, longitude: float) -> PLSSResult: + """Query BLM PLSS API for location.""" + results = await blm_client.identify( + PLSS_URL, latitude, longitude, f"all:{LAYER_TOWNSHIP},{LAYER_SECTION}" + ) + + if not results: + return PLSSResult( + latitude=latitude, + longitude=longitude, + error="No PLSS data found. Location may be outside surveyed areas.", + ) + + township_data: PLSSLocation | None = None + section_data: PLSSLocation | None = None + + for result in results: + layer_id = result.get("layerId") + attrs = result.get("attributes", {}) + + if layer_id == LAYER_TOWNSHIP and township_data is None: + township_data = _parse_township(attrs) + elif layer_id == LAYER_SECTION and section_data is None: + section_data = _parse_section(attrs, township_data) + + return PLSSResult( + latitude=latitude, + longitude=longitude, + township=township_data, + section=section_data, + ) diff --git a/src/mcblmplss/mixins/surface_management.py b/src/mcblmplss/mixins/surface_management.py new file mode 100644 index 0000000..69660a3 --- /dev/null +++ b/src/mcblmplss/mixins/surface_management.py @@ -0,0 +1,183 @@ +""" +Surface Management Agency mixin for BLM MCP server. + +Provides tools for determining who manages federal lands (BLM, USFS, NPS, etc.). +""" + +from pydantic import BaseModel, Field +from fastmcp.contrib.mcp_mixin import MCPMixin, mcp_tool + +from mcblmplss.client import blm_client + +# Surface Management Agency MapServer +SMA_URL = "https://gis.blm.gov/arcgis/rest/services/lands/BLM_Natl_SMA_LimitedScale/MapServer" +LAYER_SMA = 1 # Main SMA layer with all agencies + +# Agency code mappings +AGENCY_NAMES = { + "BLM": "Bureau of Land Management", + "NPS": "National Park Service", + "USFS": "U.S. Forest Service", + "FWS": "U.S. Fish and Wildlife Service", + "USBR": "Bureau of Reclamation", + "BIA": "Bureau of Indian Affairs", + "DOD": "Department of Defense", + "DOE": "Department of Energy", + "TVA": "Tennessee Valley Authority", + "PVT": "Private", + "ST": "State", + "LG": "Local Government", + "UND": "Undetermined", +} + +DEPT_NAMES = { + "DOI": "Department of the Interior", + "USDA": "Department of Agriculture", + "DOD": "Department of Defense", + "DOE": "Department of Energy", + "PVT": "Private", + "ST": "State", + "LG": "Local Government", +} + + +class LandManager(BaseModel): + """Surface management agency information.""" + + agency_code: str = Field(..., description="Agency code (BLM, USFS, NPS, etc.)") + agency_name: str = Field(..., description="Full agency name") + department_code: str | None = Field(None, description="Department code (DOI, USDA, etc.)") + department_name: str | None = Field(None, description="Full department name") + admin_unit_name: str | None = Field(None, description="Administrative unit name") + admin_unit_type: str | None = Field(None, description="Administrative unit type") + state: str = Field(..., description="State abbreviation") + is_federal: bool = Field(..., description="Whether this is federal land") + is_public: bool = Field(..., description="Whether this is publicly accessible") + + +class SurfaceManagementResult(BaseModel): + """Result from surface management query.""" + + latitude: float + longitude: float + manager: LandManager | None = None + error: str | None = None + + +def _parse_sma(attrs: dict) -> LandManager: + """Parse surface management attributes.""" + agency_code = attrs.get("ADMIN_AGENCY_CODE", "UND") + dept_code = attrs.get("ADMIN_DEPT_CODE", "") + + # Determine if federal and public + federal_depts = {"DOI", "USDA", "DOD", "DOE"} + public_agencies = {"BLM", "USFS", "NPS", "FWS", "USBR"} + + is_federal = dept_code in federal_depts + is_public = agency_code in public_agencies + + return LandManager( + agency_code=agency_code, + agency_name=AGENCY_NAMES.get(agency_code, agency_code), + department_code=dept_code if dept_code and dept_code != "Null" else None, + department_name=DEPT_NAMES.get(dept_code) if dept_code else None, + admin_unit_name=attrs.get("ADMIN_UNIT_NAME") if attrs.get("ADMIN_UNIT_NAME") != "Null" else None, + admin_unit_type=attrs.get("ADMIN_UNIT_TYPE") if attrs.get("ADMIN_UNIT_TYPE") != "Null" else None, + state=attrs.get("ADMIN_ST", ""), + is_federal=is_federal, + is_public=is_public, + ) + + +class SurfaceManagementMixin(MCPMixin): + """MCP tools for Surface Management Agency queries.""" + + @mcp_tool( + name="get_land_manager", + description="Determine who manages the land at given coordinates (BLM, Forest Service, NPS, Private, etc.)", + ) + async def get_land_manager( + self, + latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), + longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), + ) -> str: + """ + Get the surface management agency for a location. + + Returns the federal agency, state agency, or private designation + for who manages the surface rights at the given coordinates. + + Examples: + - "Bureau of Land Management (BLM) - Department of the Interior" + - "U.S. Forest Service (USFS) - Department of Agriculture" + - "Private land" + """ + result = await self._query_sma(latitude, longitude) + + if result.error: + return f"Error: {result.error}" + + if result.manager: + mgr = result.manager + lines = [f"{mgr.agency_name} ({mgr.agency_code})"] + + if mgr.department_name: + lines[0] += f" - {mgr.department_name}" + + if mgr.admin_unit_name: + lines.append(f"Unit: {mgr.admin_unit_name}") + + lines.append(f"State: {mgr.state}") + + status = [] + if mgr.is_federal: + status.append("Federal") + if mgr.is_public: + status.append("Public") + if status: + lines.append(f"Status: {', '.join(status)} land") + + return "\n".join(lines) + + return "Unable to determine land manager." + + @mcp_tool( + name="get_land_manager_details", + description="Get detailed surface management data as structured object.", + ) + async def get_land_manager_details( + self, + latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), + longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), + ) -> SurfaceManagementResult: + """Get full surface management details including agency codes and flags.""" + return await self._query_sma(latitude, longitude) + + async def _query_sma(self, latitude: float, longitude: float) -> SurfaceManagementResult: + """Query BLM SMA API for location.""" + results = await blm_client.identify( + SMA_URL, latitude, longitude, f"all:{LAYER_SMA}" + ) + + if not results: + return SurfaceManagementResult( + latitude=latitude, + longitude=longitude, + error="No surface management data found for this location.", + ) + + # Use first result from main SMA layer + for result in results: + if result.get("layerId") == LAYER_SMA: + attrs = result.get("attributes", {}) + return SurfaceManagementResult( + latitude=latitude, + longitude=longitude, + manager=_parse_sma(attrs), + ) + + return SurfaceManagementResult( + latitude=latitude, + longitude=longitude, + error="No surface management data in response.", + ) diff --git a/src/mcblmplss/server.py b/src/mcblmplss/server.py index cef8d15..2d375b9 100644 --- a/src/mcblmplss/server.py +++ b/src/mcblmplss/server.py @@ -1,240 +1,67 @@ """ -FastMCP server for querying BLM Public Land Survey System (PLSS) data. +FastMCP server for querying BLM land data. -Converts lat/long coordinates to Section, Township, Range using the -BLM National Cadastral ArcGIS REST service. +Provides tools for: +- PLSS (Public Land Survey System) - Section, Township, Range +- Surface Management Agency - Who manages the land +- Mining Claims - Active and closed mining claims + +All data is queried from official BLM ArcGIS REST services. """ -import httpx from fastmcp import FastMCP -from pydantic import BaseModel, Field - -# BLM Cadastral MapServer endpoint -BLM_PLSS_URL = "https://gis.blm.gov/arcgis/rest/services/Cadastral/BLM_Natl_PLSS_CadNSDI/MapServer" - -# Layer IDs in the MapServer -LAYER_TOWNSHIP = 1 -LAYER_SECTION = 2 - - -class PLSSLocation(BaseModel): - """Public Land Survey System location description.""" - - section: str | None = Field(None, description="Section number (1-36)") - township: str = Field(..., description="Township number with direction (e.g., '4N')") - range: str = Field(..., description="Range number with direction (e.g., '6E')") - principal_meridian: str = Field(..., description="Principal meridian name") - state: str = Field(..., description="State abbreviation") - label: str = Field(..., description="Human-readable PLSS label") - plss_id: str = Field(..., description="Unique PLSS identifier") - - -class PLSSQueryResult(BaseModel): - """Result from a PLSS coordinate query.""" - - latitude: float - longitude: float - township: PLSSLocation | None = None - section: PLSSLocation | None = None - error: str | None = None - - -def parse_township_attributes(attrs: dict) -> PLSSLocation: - """Parse township attributes from BLM API response.""" - township_num = attrs.get("TWNSHPNO", "").lstrip("0") or "0" - township_dir = attrs.get("TWNSHPDIR", "") - range_num = attrs.get("RANGENO", "").lstrip("0") or "0" - range_dir = attrs.get("RANGEDIR", "") - - return PLSSLocation( - section=None, - township=f"{township_num}{township_dir}", - range=f"{range_num}{range_dir}", - principal_meridian=attrs.get("PRINMER", "Unknown"), - state=attrs.get("STATEABBR", ""), - label=attrs.get("TWNSHPLAB", ""), - plss_id=attrs.get("PLSSID", ""), - ) - - -def parse_section_attributes(attrs: dict, township: PLSSLocation | None) -> PLSSLocation: - """Parse section attributes from BLM API response. - - Note: Section layer uses human-readable field names like "First Division Number" - while Township layer uses abbreviated names like "TWNSHPNO". - """ - # Try both field name formats (API returns friendly names for section layer) - section_num = ( - attrs.get("First Division Number", "") or attrs.get("FRSTDIVNO", "") - ).lstrip("0") or "0" - - # Get the division ID (also uses friendly name) - division_id = attrs.get("First Division Identifier", "") or attrs.get("FRSTDIVID", "") - - if township: - return PLSSLocation( - section=section_num, - township=township.township, - range=township.range, - principal_meridian=township.principal_meridian, - state=township.state, - label=f"Section {section_num}, {township.label}", - plss_id=division_id, - ) - - # Fallback: extract state from Township Identifier if no township data - township_id = attrs.get("Township Identifier", "") or attrs.get("PLSSID", "") - return PLSSLocation( - section=section_num, - township="Unknown", - range="Unknown", - principal_meridian="Unknown", - state=township_id[:2] if len(township_id) >= 2 else "", - label=f"Section {section_num}", - plss_id=division_id, - ) - - -async def query_plss(latitude: float, longitude: float) -> PLSSQueryResult: - """ - Query BLM PLSS API for location at given coordinates. - - Uses the ArcGIS REST identify operation to find township and section - features that contain the specified point. - """ - # Build identify request parameters - # Note: BLM service accepts WGS84 (EPSG:4326) coordinates directly - params = { - "f": "json", - "geometry": f"{longitude},{latitude}", - "geometryType": "esriGeometryPoint", - "sr": "4326", # WGS84 - "layers": f"all:{LAYER_TOWNSHIP},{LAYER_SECTION}", - "tolerance": "1", - "mapExtent": f"{longitude-1},{latitude-1},{longitude+1},{latitude+1}", - "imageDisplay": "100,100,96", - "returnGeometry": "false", - } - - async with httpx.AsyncClient(timeout=30.0) as client: - response = await client.get(f"{BLM_PLSS_URL}/identify", params=params) - response.raise_for_status() - data = response.json() - - results = data.get("results", []) - - if not results: - return PLSSQueryResult( - latitude=latitude, - longitude=longitude, - error="No PLSS data found for this location. The point may be outside surveyed areas.", - ) - - # Parse results by layer - township_data: PLSSLocation | None = None - section_data: PLSSLocation | None = None - - for result in results: - layer_id = result.get("layerId") - attrs = result.get("attributes", {}) - - if layer_id == LAYER_TOWNSHIP and township_data is None: - township_data = parse_township_attributes(attrs) - elif layer_id == LAYER_SECTION and section_data is None: - section_data = parse_section_attributes(attrs, township_data) - - return PLSSQueryResult( - latitude=latitude, - longitude=longitude, - township=township_data, - section=section_data, - ) +from mcblmplss.mixins import PLSSMixin, SurfaceManagementMixin, MiningClaimsMixin # Initialize FastMCP server mcp = FastMCP( name="mcblmplss", instructions=""" - Query the U.S. Public Land Survey System (PLSS) using coordinates. + Query U.S. Bureau of Land Management (BLM) land data by coordinates. - The PLSS divides land into: - - Townships: 6x6 mile squares identified by Township (N/S) and Range (E/W) - - Sections: 1 square mile (640 acres), numbered 1-36 within townships - - Principal Meridians: Reference lines for the survey system + **PLSS (Public Land Survey System)** + The survey grid dividing western/midwestern US into: + - Townships: 6x6 mile squares (N/S from baseline, E/W from meridian) + - Sections: 1 square mile (640 acres), numbered 1-36 + - Principal Meridians: Reference lines for survey system - This server queries the BLM National Cadastral database for authoritative PLSS data. - Coverage: 30 western and midwestern states (not eastern seaboard or Texas). + **Surface Management Agency** + Who manages the land surface: + - Federal: BLM, Forest Service, NPS, Fish & Wildlife, etc. + - State, Local, Tribal, or Private + + **Mining Claims** + Mineral rights claims from BLM's MLRS database: + - Lode Claims: Hard rock minerals (veins, ledges) + - Placer Claims: Loose deposits (gold in streams) + - Mill Sites, Tunnel Sites + + Coverage: 30 western/midwestern states. Eastern seaboard and Texas + use different survey systems. """, ) +# Create mixin instances and register tools +plss_mixin = PLSSMixin() +sma_mixin = SurfaceManagementMixin() +mining_mixin = MiningClaimsMixin() -@mcp.tool() -async def get_plss_location( - latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), - longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), -) -> str: - """ - Get the Public Land Survey System (PLSS) location for coordinates. - - Returns Section, Township, Range, and Principal Meridian for the given - latitude/longitude. Example: "Section 12, Township 4N, Range 6E, 6th PM" - - Note: PLSS coverage is limited to 30 states where federal land surveys - were conducted. Eastern states and Texas use different systems. - """ - result = await query_plss(latitude, longitude) - - if result.error: - return f"Error: {result.error}" - - if result.section: - return ( - f"Section {result.section.section}, " - f"Township {result.section.township}, " - f"Range {result.section.range}, " - f"{result.section.principal_meridian}\n" - f"State: {result.section.state}\n" - f"PLSS ID: {result.section.plss_id}" - ) - elif result.township: - return ( - f"Township {result.township.township}, " - f"Range {result.township.range}, " - f"{result.township.principal_meridian}\n" - f"State: {result.township.state}\n" - f"(Section data not available for this location)" - ) - else: - return "Unable to determine PLSS location for these coordinates." - - -@mcp.tool() -async def get_plss_details( - latitude: float = Field(description="Latitude in decimal degrees (WGS84)"), - longitude: float = Field(description="Longitude in decimal degrees (WGS84)"), -) -> PLSSQueryResult: - """ - Get detailed PLSS information as structured data. - - Returns a PLSSQueryResult object with full township and section details - including PLSS IDs, principal meridian, and state information. - - Useful when you need programmatic access to all PLSS fields. - """ - return await query_plss(latitude, longitude) +# Register all mixin tools with the server +plss_mixin.register_all(mcp) +sma_mixin.register_all(mcp) +mining_mixin.register_all(mcp) def main(): """Entry point for the mcblmplss MCP server.""" try: from importlib.metadata import version - package_version = version("mcblmplss") except Exception: package_version = "dev" - print(f"🗺️ mcblmplss v{package_version} - BLM PLSS Query Server") - print("📍 Query Section/Township/Range from coordinates") + print(f"🗺️ mcblmplss v{package_version} - BLM Land Data Server") + print("📍 PLSS | 🏛️ Surface Management | ⛏️ Mining Claims") mcp.run()