finn-mcp/validate_mcp_tools.py

#!/usr/bin/env python3
"""
Validate that all MCP tool definitions correctly match their service layer functions.
This catches parameter mismatches, missing arguments, and other integration issues.
"""

import inspect
from typing import get_type_hints
from finn_eiendom import mcp_server, service, eiendom_no

# Define the mapping of MCP tools to their service/module functions
TOOL_MAPPINGS = {
    # Tool name: (service function, expected params to check)
    "finn_analyze_search": (
        service.analyze_search,
        ["search_url", "max_pages", "detail_limit", "include_details", "include_eiendom_no"],
    ),
    "finn_get_ad": (service.get_or_fetch_ad, ["finnkode", "force_refresh"]),
    "finn_resolve_eiendom_unit": (eiendom_no.search_unit_from_finn_url, ["finn_url"]),
    "finn_get_eiendom_unit": (service.get_or_fetch_eiendom_unit, ["unit_code", "force_refresh"]),
    "finn_analyze_unit_images": (service.get_unit_images, ["unit_code", "force_refresh"]),
    "finn_get_similar_units": (eiendom_no.get_similar_units, ["unit_vector", "listing_status"]),
    "finn_build_unit_vector": (
        eiendom_no.get_unit,
        ["unit_code"],
    ),  # Uses get_unit, not build_unit_vector
    "finn_decode_unit_vector": (eiendom_no.decode_unit_vector, ["unit_vector"]),
    "finn_analyze_ad": (
        service.analyze_ad,
        ["finnkode", "include_eiendom_no", "include_similar_units"],
    ),
    "finn_analyze_ad_against_comps": (
        service.analyze_ad_against_comps,
        ["finnkode", "listing_status"],
    ),
    "finn_find_similar_to_liked_ad": (
        service.find_similar_to_liked,
        ["finnkode", "mode", "listing_status"],
    ),
    "finn_compare_ads": (service.compare_ads, ["finnkoder", "include_eiendom_no", "include_comps"]),
    "finn_save_feedback": (service.save_feedback, ["finnkode", "verdict", "notes"]),
    "finn_get_shortlist": (service.get_shortlist, ["run_id", "limit"]),
    "finn_get_new_ads_since_last_run": (service.get_new_ads_since_last_run, ["search_url"]),
}


def get_function_params(func) -> dict:
    """Extract parameter names and defaults from a function."""
    sig = inspect.signature(func)
    params = {}
    for name, param in sig.parameters.items():
        if name in ("self", "cls"):
            continue
        params[name] = {
            "default": param.default,
            "annotation": param.annotation,
            "kind": param.kind.name,
        }
    return params


def validate_tool_mapping(
    tool_name: str, service_func, expected_params: list[str]
) -> tuple[bool, list[str]]:
    """Validate that an MCP tool correctly maps to its service function."""
    errors = []

    # Get the MCP tool function
    mcp_tool = getattr(mcp_server, tool_name, None)
    if not mcp_tool:
        errors.append(f"MCP tool '{tool_name}' not found in mcp_server module")
        return False, errors

    # Get function signatures
    mcp_params = get_function_params(mcp_tool)
    service_params = get_function_params(service_func)

    # Check that expected parameters exist in both
    for param in expected_params:
        if param not in mcp_params:
            errors.append(f"  ✗ MCP tool missing parameter '{param}'")
        if param not in service_params and param != "client":  # client is optional in service layer
            errors.append(f"  ✗ Service function missing parameter '{param}'")

    # Check that MCP tool doesn't pass unknown parameters
    # (skip return annotation)
    for param_name, param_info in mcp_params.items():
        if param_name not in service_params and param_name not in ["return"]:
            # This might be OK if it's a tool-specific parameter, but warn
            pass

    if errors:
        return False, errors
    return True, []


async def validate_service_imports():
    """Validate that all imported service functions exist and are callable."""
    imported_funcs = [
        ("analyze_ad", service.analyze_ad),
        ("analyze_ad_against_comps", service.analyze_ad_against_comps),
        ("analyze_search", service.analyze_search),
        ("compare_ads", service.compare_ads),
        ("find_similar_to_liked", service.find_similar_to_liked),
        ("get_new_ads_since_last_run", service.get_new_ads_since_last_run),
        ("get_or_fetch_ad", service.get_or_fetch_ad),
        ("get_or_fetch_eiendom_unit", service.get_or_fetch_eiendom_unit),
        ("get_shortlist", service.get_shortlist),
        ("get_unit_images", service.get_unit_images),
        ("save_feedback", service.save_feedback),
    ]

    errors = []
    for name, func in imported_funcs:
        if not callable(func):
            errors.append(f"Service function '{name}' is not callable")

    return errors


def main():
    """Run validation checks."""
    print("=" * 80)
    print("MCP Tool Parameter Validation")
    print("=" * 80)

    all_passed = True
    total_checks = 0
    passed_checks = 0

    for tool_name, (service_func, expected_params) in TOOL_MAPPINGS.items():
        total_checks += 1
        passed, errors = validate_tool_mapping(tool_name, service_func, expected_params)

        if passed:
            print(f"✓ {tool_name}")
            passed_checks += 1
        else:
            print(f"✗ {tool_name}")
            for error in errors:
                print(f"  {error}")
            all_passed = False

    print("\n" + "=" * 80)
    print(f"Results: {passed_checks}/{total_checks} tools validated")
    print("=" * 80)

    return 0 if all_passed else 1


if __name__ == "__main__":
    exit(main())