mcp-implementation-patterns

MCP Implementation Patterns for OSM Server

FastMCP Server Structure

The server.py uses FastMCP framework with this pattern:

from mcp.server.fastmcp import FastMCP

# Initialize server
mcp = FastMCP("osm-edit-mcp")

# Define tools with decorator
@mcp.tool()
async def tool_name(param: type) -> Dict[str, Any]:
    """Tool description for AI assistant."""
    # Implementation

MCP Tool Categories

1. Read Operations (Safe)

Tools that only retrieve data from OSM:

• get_osm_node, get_osm_way, get_osm_relation
• get_osm_elements_in_area, get_changeset_history
• find_nearby_amenities, search_osm_elements

2. Validation and Utility Tools

Non-destructive operations for data validation:

• validate_coordinates, validate_osm_data
• get_place_info, smart_geocode
• get_server_info, export_osm_data

3. Write Operations (OAuth Required)

Tools that modify OSM data:

• create_osm_node, update_osm_node, delete_osm_node
• create_osm_way, update_osm_way, delete_osm_way
• create_osm_relation, update_osm_relation, delete_osm_relation
• create_changeset, close_changeset

4. Natural Language Processing

Advanced tools for natural language interaction:

• parse_natural_language_osm_request
• create_place_from_description
• find_and_update_place, delete_place_from_description
• bulk_create_places

Tool Implementation Pattern

Standard Tool Structure

@mcp.tool()
async def tool_name(param1: type, param2: Optional[type] = None) -> Dict[str, Any]:
    """Clear description of what the tool does.

    Args:
        param1: Description of required parameter
        param2: Description of optional parameter

    Returns:
        Dictionary containing result data and status
    """
    try:
        # Log the operation
        logger.debug(f"Operation description with params")

        # Validate inputs
        if not validate_input(param1):
            return {
                "success": False,
                "error": "Validation error message",
                "message": "Human-readable error"
            }

        # Build API URL using configuration
        url = f"{config.current_api_base_url}/endpoint"

        # Make API call
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            response.raise_for_status()

        # Process response
        result_data = process_response(response)

        return {
            "success": True,
            "data": result_data,
            "message": "Success message"
        }

    except Exception as e:
        logger.error(f"Tool failed: {str(e)}")
        return {
            "success": False,
            "error": str(e),
            "message": f"Failed to perform operation"
        }