MCP server to download road network data from OpenStreetMap

mcp_server_road_network.py

# /// script
# dependencies = [
#     "mcp",
#     "osmnx",
#     "eclipse-sumo",
# ]
# ///
import http.client as httplib
import os
import subprocess

import osmnx as ox
import sumo  # to set os.environ["SUMO_HOME"]

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "RoadNetwork",
    instructions="Download road network data of specified area (bounding box) from OpenStreetMap(OSM)."
)


@mcp.tool()
async def bounding_box_from_point(
    center_latitude: float,
    center_longitude: float,
    distance: float,
) -> (float, float, float, float):
    """
    Create a bounding box around a (latitude, longitude) point.

    Create a bounding box some distance (in meters) in each direction (top, bottom, right, and left) from the center point.

    Args:
        center_latitude: latitude of center point to create the bounding box around. (decimal degrees)
        center_longitude: longitude of center point to create the bounding box around. (decimal degrees)
        distance: Bounding box distance in meters from the center point.

    Returns:
        (float, float, float, float): created bounding box. tuple of (left, bottom, right, top). tuple of (west_longitude, south_latitude, east_longitude, north_latitude)
    """
    return ox.utils_geo.bbox_from_point((center_latitude, center_longitude), distance)


@mcp.tool()
async def download_road_network_data(
    min_longitude: float,
    min_latitude: float,
    max_longitude: float,
    max_latitude: float
) -> str:
    """
    Download road network data of specified area (bounding box) from OpenStreetMap(OSM).

    Args:
        min_longitude: West boundary of area (decimal degrees).
        min_latitude: South boundary of area (decimal degrees).
        max_longitude: East boundary of area (decimal degrees).
        max_latitude: North boundary of area (decimal degrees).

    Returns:
        str: file path of downloaded OSM road network data (osm.xml.gz).
    """
    # cf. SUMO tools/osmGet.py
    q = f'''<osm-script timeout="240" element-limit="1073741824">
        <union>
         <bbox-query n="{max_latitude}" s="{min_latitude}" w="{min_longitude}" e="{max_longitude}"/>
         <recurse type="node-relation" into="rels"/>
         <recurse type="node-way"/>
         <recurse type="way-relation"/>
        </union>
        <union>
          <item/>
          <recurse type="way-node"/>
        </union>
        <print mode="body"/>
    </osm-script>'''
    try:
        conn = httplib.HTTPSConnection("www.overpass-api.de")
        conn.request("POST", "/api/interpreter", q, headers={'Accept-Encoding': 'gzip'})
        filename = None
        response = conn.getresponse()
        if response.status == 200:
            filename = 'osm.xml'
            if response.getheader('Content-Encoding') == 'gzip':
                filename += '.gz'
            lines = response.read()
            with open(filename, "wb") as out:
                out.write(lines)
        else:
            raise Exception(f"Failed to download road network data: {response.status}")
    finally:
        conn.close()
    return os.path.abspath(filename)


@mcp.tool()
async def convert_osm_to_sumo_network(
    osm_filepath: str,
    lefthand: bool = False
) -> str:
    """
    Import OpenStreetMap (OSM) road network data and convert/generate road network data for SUMO (Simulation of Urban MObility).

    Args:
        osm_filepath: file path of OpenStreetMap (OSM) road network data (osm.xml.gz).
        lefthand: Assumes left-hand traffic on the road network. default: False (assumes networks to follow right-hand traffic rules.).

    Returns:
        str: file path of generated road network data for SUMO (map.net.xml.gz).
    """
    if not os.path.isfile(osm_filepath):
        raise FileNotFoundError(f"specified osm_filepath ({osm_filepath}) does not exist.")
    opts = [
        "--speed-in-kmh",
        "--junctions.join",
        "--osm.sidewalks",
        # "--osm.crossings",  # XXX: not success
        "--no-internal-links",
        "--remove-edges.isolated",
    ]
    if lefthand:
        opts.append("--lefthand")
    if "SUMO_HOME" in os.environ:
        execpath = os.path.join(os.environ["SUMO_HOME"], "bin", "netconvert")
    else:
        execpath = "netconvert"
    output_filename = "map.net.xml.gz"
    cmd = [execpath, "--osm-files", osm_filepath, "-o", output_filename] + opts
    result = subprocess.run(cmd, check=True, capture_output=True, text=True)
    return output_filename


if __name__ == "__main__":
    mcp.run()