MCP/mcblmplss
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Surface Management Agency and Mining Claims tools

Browse Source 
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.
This commit is contained in:
Ryan Malloy 2025-12-03 15:31:17 -07:00
parent b1617104ee
commit 0ef2c70af4
8 changed files with 781 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
README.md
View File

@ -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
```

8
src/mcblmplss/__init__.py
View File

@ -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

113
src/mcblmplss/client.py Normal file
View File

@ -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)

11
src/mcblmplss/mixins/__init__.py Normal file
View File

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

181
src/mcblmplss/mixins/mining_claims.py Normal file
View File

@ -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),
)

174
src/mcblmplss/mixins/plss.py Normal file
View File

@ -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,
)

183
src/mcblmplss/mixins/surface_management.py Normal file
View File

@ -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.",
)

247
src/mcblmplss/server.py
View File

@ -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()