An MCP server to provide stop and search data from the UK Police APIs

uk_police_mcp.py

"""
UK Police Stop and Search Data MCP Server

This server provides tools to fetch stop and search data from the UK Police Database API.
"""

from typing import Optional
from datetime import datetime
import httpx
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("UK Police Stop and Search")

# Available police forces (subset - can be extended)
POLICE_FORCES = [
    "metropolitan",
    "greater-manchester",
    "west-midlands",
    "west-yorkshire",
    "thames-valley",
    "merseyside",
    "essex",
    "avon-and-somerset",
    "kent",
    "hampshire",
]


@mcp.tool()
async def get_stop_and_search_data(
    year: int,
    month: int,
    force: str = "metropolitan"
) -> dict:
    """
    Fetch stop and search data from the UK Police Database for a specified month and year.
    
    Args:
        year: Year (e.g., 2024)
        month: Month (1-12)
        force: Police force name (default: metropolitan). Common forces include:
               metropolitan, greater-manchester, west-midlands, west-yorkshire,
               thames-valley, merseyside, essex, avon-and-somerset, kent, hampshire
    
    Returns:
        Dictionary containing stop and search records and summary statistics
    """
    # Validate month
    if not 1 <= month <= 12:
        return {
            "error": "Invalid month. Must be between 1 and 12.",
            "data": []
        }
    
    # Validate year
    current_year = datetime.now().year
    if year < 2010 or year > current_year:
        return {
            "error": f"Invalid year. Must be between 2010 and {current_year}.",
            "data": []
        }
    
    # Format date as YYYY-MM
    date_str = f"{year}-{month:02d}"
    
    # Build API URL
    url = f"https://data.police.uk/api/stops-force"
    params = {
        "date": date_str,
        "force": force
    }
    
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(url, params=params)
            response.raise_for_status()
            data = response.json()
            
            # Calculate summary statistics
            total_searches = len(data)
            
            # Age range breakdown
            age_ranges = {}
            for record in data:
                age = record.get("age_range", "Unknown")
                age_ranges[age] = age_ranges.get(age, 0) + 1
            
            # Gender breakdown
            genders = {}
            for record in data:
                gender = record.get("gender", "Unknown")
                genders[gender] = genders.get(gender, 0) + 1
            
            # Outcome breakdown
            outcomes = {}
            for record in data:
                outcome = record.get("outcome", "Unknown")
                outcomes[outcome] = outcomes.get(outcome, 0) + 1
            
            # Object of search breakdown
            search_objects = {}
            for record in data:
                obj = record.get("object_of_search", "Unknown")
                search_objects[obj] = search_objects.get(obj, 0) + 1
            
            return {
                "date": date_str,
                "force": force,
                "total_records": total_searches,
                "summary": {
                    "age_ranges": age_ranges,
                    "genders": genders,
                    "outcomes": outcomes,
                    "search_objects": search_objects
                },
                "records": data
            }
            
    except httpx.HTTPStatusError as e:
        return {
            "error": f"HTTP error occurred: {e.response.status_code}",
            "message": str(e),
            "data": []
        }
    except httpx.RequestError as e:
        return {
            "error": "Request failed",
            "message": str(e),
            "data": []
        }
    except Exception as e:
        return {
            "error": "Unexpected error occurred",
            "message": str(e),
            "data": []
        }


@mcp.tool()
async def list_available_forces() -> dict:
    """
    List commonly available police forces in the UK.
    
    Returns:
        Dictionary containing a list of police force identifiers
    """
    return {
        "forces": POLICE_FORCES,
        "note": "This is a subset of available forces. Check the UK Police API documentation for a complete list."
    }


@mcp.tool()
async def get_search_statistics(
    year: int,
    month: int,
    force: str = "metropolitan"
) -> dict:
    """
    Get summary statistics for stop and search data without full records.
    
    Args:
        year: Year (e.g., 2024)
        month: Month (1-12)
        force: Police force name (default: metropolitan)
    
    Returns:
        Dictionary containing only summary statistics
    """
    result = await get_stop_and_search_data(year, month, force)
    
    if "error" in result:
        return result
    
    # Return only summary information, not full records
    return {
        "date": result["date"],
        "force": result["force"],
        "total_records": result["total_records"],
        "summary": result["summary"]
    }


if __name__ == "__main__":
    # Run the MCP server
    mcp.run()