OpenStreetMap MCP Server

Instructions

Implementation Reference

• src/osm_mcp_server/server.py:1226-1457 (handler)

  The primary handler function decorated with @mcp.tool() that implements the analyze_neighborhood tool. It performs detailed neighborhood analysis by querying OSM Overpass API for various amenity categories (groceries, restaurants, healthcare, etc.), calculates distances, scores livability factors, and computes overall and walkability scores.
  async def analyze_neighborhood( latitude: float, longitude: float, ctx: Context, radius: float = 1000 ) -> Dict[str, Any]: """ Generate a comprehensive neighborhood analysis focused on livability factors. This advanced analysis tool evaluates a neighborhood based on multiple livability factors, including amenities, transportation options, green spaces, and services. Results include counts and proximity scores for various categories, helping to assess the overall quality and convenience of a residential area. Invaluable for real estate decisions, relocation planning, and neighborhood comparisons. Args: latitude: Center point latitude (decimal degrees) longitude: Center point longitude (decimal degrees) radius: Analysis radius in meters (defaults to 1000m/1km) Returns: Comprehensive neighborhood profile including: - Overall neighborhood score - Walkability assessment - Public transportation access - Nearby amenities (shops, restaurants, services) - Green spaces and recreation - Education and healthcare facilities - Detailed counts and distance metrics for each category """ osm_client = ctx.request_context.lifespan_context.osm_client # Get address information for the center point address_info = await osm_client.reverse_geocode(latitude, longitude) # Categories to analyze for neighborhood quality categories = [ # Essential services {"name": "groceries", "tags": ["shop=supermarket", "shop=convenience", "shop=grocery"]}, {"name": "restaurants", "tags": ["amenity=restaurant", "amenity=cafe", "amenity=fast_food"]}, {"name": "healthcare", "tags": ["amenity=hospital", "amenity=doctors", "amenity=pharmacy"]}, {"name": "education", "tags": ["amenity=school", "amenity=kindergarten", "amenity=university"]}, # Transportation {"name": "public_transport", "tags": ["public_transport=stop_position", "railway=station", "amenity=bus_station"]}, # Recreation {"name": "parks", "tags": ["leisure=park", "leisure=garden", "leisure=playground"]}, {"name": "sports", "tags": ["leisure=sports_centre", "leisure=fitness_centre", "leisure=swimming_pool"]}, # Culture and entertainment {"name": "entertainment", "tags": ["amenity=theatre", "amenity=cinema", "amenity=arts_centre"]}, # Other amenities {"name": "shopping", "tags": ["shop=mall", "shop=department_store", "shop=clothes"]}, {"name": "services", "tags": ["amenity=bank", "amenity=post_office", "amenity=atm"]} ] # Build overpass queries and collect results results = {} scores = {} for i, category in enumerate(categories): await ctx.report_progress(i, len(categories)) ctx.info(f"Analyzing {category['name']} in neighborhood...") # Convert radius to bounding box lat_delta = radius / 111000 lon_delta = radius / (111000 * math.cos(math.radians(latitude))) bbox = ( longitude - lon_delta, latitude - lat_delta, longitude + lon_delta, latitude + lat_delta ) # Build Overpass query overpass_url = "https://overpass-api.de/api/interpreter" # Create query for category tags tag_filters = [] for tag in category["tags"]: key, value = tag.split("=") tag_filters.append(f'node["{key}"="{value}"]({{bbox}});') tag_filters.append(f'way["{key}"="{value}"]({{bbox}});') query = f""" [out:json]; ( {" ".join(tag_filters)} ); out body; """ query = query.replace("{bbox}", f"{bbox[1]},{bbox[0]},{bbox[3]},{bbox[2]}") try: async with aiohttp.ClientSession() as session: async with session.post(overpass_url, data={"data": query}) as response: if response.status == 200: data = await response.json() features = data.get("elements", []) else: ctx.warning(f"Failed to analyze {category['name']}: {response.status}") features = [] # Process and calculate metrics feature_list = [] distances = [] for feature in features: tags = feature.get("tags", {}) # Get coordinates based on feature type coords = {} if feature.get("type") == "node": coords = { "latitude": feature.get("lat"), "longitude": feature.get("lon") } elif "center" in feature: coords = { "latitude": feature.get("center", {}).get("lat"), "longitude": feature.get("center", {}).get("lon") } # Skip if no valid coordinates if not coords: continue # Calculate distance from center point from math import radians, sin, cos, sqrt, asin def haversine(lat1, lon1, lat2, lon2): R = 6371000 # Earth radius in meters dLat = radians(lat2 - lat1) dLon = radians(lon2 - lon1) a = sin(dLat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dLon/2)**2 c = 2 * asin(sqrt(a)) return R * c distance = haversine(latitude, longitude, coords["latitude"], coords["longitude"]) distances.append(distance) feature_list.append({ "id": feature.get("id"), "name": tags.get("name", "Unnamed"), "type": feature.get("type"), "coordinates": coords, "distance": round(distance, 1), "tags": tags }) # Sort by distance feature_list.sort(key=lambda x: x["distance"]) # Calculate metrics count = len(feature_list) avg_distance = sum(distances) / count if count > 0 else None min_distance = min(distances) if count > 0 else None # Score this category (0-10) # Higher score for more amenities and closer proximity if count == 0: category_score = 0 else: # Base score on count and proximity count_score = min(count / 5, 1) * 5 # Up to 5 points for count proximity_score = 5 - min(min_distance / radius, 1) * 5 # Up to 5 points for proximity category_score = count_score + proximity_score # Store results results[category["name"]] = { "count": count, "features": feature_list[:10], # Limit to top 10 "metrics": { "total_count": count, "avg_distance": round(avg_distance, 1) if avg_distance else None, "min_distance": round(min_distance, 1) if min_distance else None } } scores[category["name"]] = category_score except Exception as e: ctx.warning(f"Error analyzing {category['name']}: {str(e)}") results[category["name"]] = {"error": str(e)} scores[category["name"]] = 0 # Calculate overall neighborhood score if scores: overall_score = sum(scores.values()) / len(scores) else: overall_score = 0 # Calculate walkability score based on amenities within walking distance (500m) walkable_amenities = 0 walkable_categories = 0 for category_name, category_data in results.items(): if "metrics" in category_data: # Count amenities within walking distance walking_count = sum(1 for feature in category_data.get("features", []) if feature.get("distance", float("inf")) <= 500) if walking_count > 0: walkable_amenities += walking_count walkable_categories += 1 walkability_score = min(walkable_amenities + walkable_categories, 10) # Report completion await ctx.report_progress(len(categories), len(categories)) return { "location": { "coordinates": { "latitude": latitude, "longitude": longitude }, "address": address_info.get("display_name", "Unknown location") }, "scores": { "overall": round(overall_score, 1), "walkability": walkability_score, "categories": {k: round(v, 1) for k, v in scores.items()} }, "categories": results, "analysis_radius": radius, "timestamp": datetime.now().isoformat() }
• src/osm_mcp_server/server.py:1226-1226 (registration)

  The @mcp.tool() decorator registers the analyze_neighborhood function as an MCP tool.
  async def analyze_neighborhood(