bonus_ebird_tools.py

"""
eBird MCP Server
Wraps eBird API v2 as reusable MCP tools
Runs locally with FastMCP and supports both stdio and streamable-http transport

Features:
- 7 core tools for bird data discovery
- Configurable tool enabling/disabling
- Support for both user input AND classifier output
- Rate limiting and error handling
- JSON responses for easy integration
- Dual transport: stdio for CLI, streamable-http for web clients (via FastAPI)
"""
import os
import sys
import requests
import json
import time
from typing import Optional, Dict, List, Any
from fastmcp import FastMCP
from dotenv import load_dotenv


# ============================================================================
# CONFIGURATION & SETUP
# ============================================================================

load_dotenv()

EBIRD_API_KEY = os.getenv("EBIRD_API_KEY")
BASE_URL = os.getenv("EBIRD_BASE_URL", "https://api.ebird.org/v2")
DEFAULT_TIMEOUT = 15
RATE_LIMIT_DELAY = 0.1  # 100ms between requests

if not EBIRD_API_KEY:
    # Print to stderr to avoid corrupting STDIO MCP protocol (stdout must be JSON-RPC only)
    print("⚠️  [WARNING]: EBIRD_API_KEY not found in .env", file=sys.stderr)
    print("   Get one from: https://ebird.org/api/keygen", file=sys.stderr)

# Authentication configuration (production only)
IS_PRODUCTION = os.getenv("ENVIRONMENT") == "production"
MCP_API_KEY = os.getenv("MCP_API_KEY")

# Tool configuration - enable/disable as needed
ENABLED_TOOLS = {
    "search_species": True,
    "get_recent_sightings_nearby": True,
    "find_hotspots_nearby": True,
    "get_location_birds": True,
    "get_species_info": True,
    "get_notable_sightings": True,
    "analyze_location": True,
}

# Initialize FastMCP server with optional auth
if IS_PRODUCTION and MCP_API_KEY:
    # Production: Enable API key authentication
    from fastmcp.server.auth.providers.debug import DebugTokenVerifier

    auth = DebugTokenVerifier(
        validate=lambda token: token == MCP_API_KEY,
        client_id="ebird-mcp-client"
    )
    mcp = FastMCP("eBird Data Explorer", auth=auth)
else:
    # Development: No authentication
    mcp = FastMCP("eBird Data Explorer")

# Rate limiting tracker
_last_request_time = 0


# ============================================================================
# HELPER FUNCTIONS
# ============================================================================

def _rate_limit():
    """Enforce rate limiting to avoid exceeding eBird's API limits"""
    global _last_request_time
    elapsed = time.time() - _last_request_time
    if elapsed < RATE_LIMIT_DELAY:
        time.sleep(RATE_LIMIT_DELAY - elapsed)
    _last_request_time = time.time()


def _make_request(endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]:
    """
    Centralized request handler with error handling and rate limiting.

    Args:
        endpoint: API endpoint path (e.g., "/data/obs/geo/recent")
        params: Query parameters dictionary

    Returns:
        JSON response data or None on error
    """
    _rate_limit()
    try:
        headers = {"X-eBirdApiToken": EBIRD_API_KEY}
        url = f"{BASE_URL}{endpoint}"
        response = requests.get(
            url,
            headers=headers,
            params=params or {},
            timeout=DEFAULT_TIMEOUT
        )

        if response.status_code == 200:
            return response.json()
        elif response.status_code == 400:
            print(f"❌ Bad Request ({url}): {response.text[:400]}", flush=True)
            return None
        elif response.status_code == 401:
            print(f"❌ Unauthorized ({url}): Check your EBIRD_API_KEY - body={response.text[:400]}", flush=True)
            return None
        elif response.status_code == 404:
            print(f"❌ Not found ({url}): Invalid endpoint or resource - body={response.text[:400]}", flush=True)
            return None
        else:
            print(
                f"❌ HTTP {response.status_code} for {url} "
                f"params={params or {}} body={response.text[:400]}",
                flush=True,
            )
            return None

    except requests.Timeout:
        print(f"❌ Request timeout after {DEFAULT_TIMEOUT}s for {endpoint}", flush=True)
        return None
    except requests.ConnectionError:
        print(f"❌ Connection error calling {endpoint} - check network", flush=True)
        return None
    except Exception as e:
        print(f"❌ Unexpected error calling {endpoint}: {str(e)}", flush=True)
        return None


def _format_success_response(data: Any, **kwargs) -> str:
    """Format a successful response as JSON"""
    response = {"status": "success", "data": data}
    response.update(kwargs)
    return json.dumps(response)


def _format_error_response(error: str) -> str:
    """Format an error response as JSON"""
    return json.dumps({"status": "error", "error": error})


# ============================================================================
# TOOL 1: search_species
# ============================================================================
# Use case: User types "cardinal" or classifier returns "Northern Cardinal"
# This tool finds the species code needed for other tools

def search_species(search_term: str, max_results: int = 10) -> str:
    """
    Search for bird species by common or scientific name.

    This tool finds species codes needed for other lookups. Accepts:
    - Common names: "cardinal", "blue jay", "bald eagle"
    - Partial matches: "car" -> "Northern Cardinal", "Carolina Parakeet", etc.
    - Scientific names: "Cardinalis cardinalis"

    Can accept:
    - User input: Direct species search
    - Classifier output: e.g., "Northern Cardinal" from image classification

    Args:
        search_term: Bird name (common or scientific)
        max_results: Maximum matches to return (default: 10)

    Returns:
        JSON with matched species and their codes for other tools

    Example:
        search_species("cardinal")
        -> Returns all cardinals with species codes (norcar, carcar, etc.)
    """
    if not search_term or len(search_term.strip()) < 2:
        return _format_error_response("Search term must be at least 2 characters")

    try:
        endpoint = "/ref/taxonomy/ebird"
        params = {"fmt": "json"}

        data = _make_request(endpoint, params)

        if not data:
            return _format_error_response("Failed to fetch species database")

        search_lower = search_term.lower()

        # Filter: match in common name OR scientific name, main species only
        matches = [
            {
                "common_name": s['comName'],
                "scientific_name": s['sciName'],
                "species_code": s['speciesCode'],
                "family": s.get('familyComName', 'Unknown'),
                "order": s.get('order', 'Unknown'),
                "category": s.get('category', 'Unknown')
            }
            for s in data
            if (search_lower in s['comName'].lower() or search_lower in s['sciName'].lower()) and s.get('category') == 'species'
        ]

        if not matches:
            return _format_error_response(f"No species found matching '{search_term}'")

        return _format_success_response(
            matches[:max_results],
            count=len(matches[:max_results]),
            search_term=search_term
        )

    except Exception as e:
        return _format_error_response(f"Search failed: {str(e)}")

# Register as MCP tool
mcp.tool()(search_species)


# ============================================================================
# TOOL 2: get_recent_sightings_nearby
# ============================================================================
# Use case: After identifying a bird, find recent sightings near user

def get_recent_sightings_nearby(
    species_code: str,
    latitude: float,
    longitude: float,
    radius_km: int = 50,
    max_results: int = 10
) -> str:
    """
    Get recent sightings of a specific bird near a location.

    Returns observations from other birdwatchers in the eBird network.

    Can accept:
        - User input: Coordinates from address lookup, species code from search
        - Classifier output: Species code (after search_species lookup)

    Args:
        species_code: eBird species code (e.g., "norcar" for Northern Cardinal)
        latitude: Location latitude
        longitude: Location longitude
        radius_km: Search radius in kilometers (max 50)
        max_results: Maximum observations to return

    Returns:
        JSON with recent observations near location

    Example:
        get_recent_sightings_nearby("norcar", 40.7829, -73.9654, 25, 10)
        -> Recent cardinal sightings in Central Park area
    """
    if not species_code:
        return _format_error_response("Species code required")

    if not -90 <= latitude <= 90:
        return _format_error_response("Latitude must be between -90 and 90")

    if not -180 <= longitude <= 180:
        return _format_error_response("Longitude must be between -180 and 180")

    try:
        endpoint = f"/data/obs/geo/recent/{species_code}"
        params = {
            "lat": latitude,
            "lng": longitude,
            "dist": min(radius_km, 50),
            "maxResults": max_results
        }

        data = _make_request(endpoint, params)

        if data is None:
            return _format_error_response("Failed to fetch sightings")

        if not data:
            return _format_success_response(
                [],
                count=0,
                location={"lat": latitude, "lng": longitude},
                radius_km=radius_km,
                species_code=species_code
            )

        sightings = [
            {
                "common_name": obs['comName'],
                "scientific_name": obs['sciName'],
                "location": obs['locName'],
                "location_id": obs['locId'],
                "date": obs['obsDt'],
                "count": obs.get('howMany'),
                "latitude": obs.get('lat'),
                "longitude": obs.get('lng')
            }
            for obs in data
        ]

        return _format_success_response(
            sightings,
            count=len(sightings),
            location={"lat": latitude, "lng": longitude},
            radius_km=radius_km
        )

    except Exception as e:
        return _format_error_response(f"Lookup failed: {str(e)}")

# Register as MCP tool
mcp.tool()(get_recent_sightings_nearby)


# ============================================================================
# TOOL 3: find_hotspots_nearby
# ============================================================================
# Use case: Find popular birding locations near user

def find_hotspots_nearby(
    latitude: float,
    longitude: float,
    radius_km: int = 50,
    max_results: int = 15
) -> str:
    """
    Find popular birding hotspots (known locations) near a location.

    Hotspots are locations frequented by birders where many species recorded.
    Great for planning birding trips.

    Can accept:
        - User input: Coordinates from address lookup
        - Classifier output: Not directly, but used after location analysis

    Args:
        latitude: Location latitude
        longitude: Location longitude
        radius_km: Search radius in kilometers
        max_results: Maximum hotspots to return

    Returns:
        JSON with nearby hotspots and their details

    Example:
        find_hotspots_nearby(40.7829, -73.9654, 25, 10)
        -> Popular birding locations near Central Park
    """
    if not -90 <= latitude <= 90:
        return _format_error_response("Latitude must be between -90 and 90")

    if not -180 <= longitude <= 180:
        return _format_error_response("Longitude must be between -180 and 180")

    try:
        endpoint = "/ref/hotspot/geo"
        params = {
            "lat": latitude,
            "lng": longitude,
            "dist": radius_km,
            "fmt": "json"
        }

        data = _make_request(endpoint, params)

        if data is None:
            return _format_error_response("Failed to fetch hotspots")

        if not data:
            return _format_success_response(
                [],
                count=0,
                location={"lat": latitude, "lng": longitude},
                radius_km=radius_km,
                message="No hotspots found nearby"
            )

        hotspots = [
            {
                "name": hotspot['locName'],
                "location_id": hotspot['locId'],
                "latitude": hotspot['lat'],
                "longitude": hotspot['lng'],
                "species_recorded": hotspot.get('numSpeciesAllTime', 0),
                "latest_obs_date": hotspot.get('latestObsDt', 'Unknown')
            }
            for hotspot in data[:max_results]
        ]

        return _format_success_response(
            hotspots,
            count=len(hotspots),
            location={"lat": latitude, "lng": longitude},
            radius_km=radius_km
        )

    except Exception as e:
        return _format_error_response(f"Lookup failed: {str(e)}")

# Register as MCP tool
mcp.tool()(find_hotspots_nearby)

# ============================================================================
# TOOL 4: get_location_birds
# ============================================================================
# Use case: See ALL birds being seen at a location right now

def get_location_birds(
    latitude: float,
    longitude: float,
    radius_km: int = 50,
    max_results: int = 50
) -> str:
    """
    Get ALL recent bird sightings at a location (no species filter).

    Returns comprehensive view of bird activity - what's being seen right now

    Can accept:
        - User input: Coordinates from address lookup
        - Classifier output: Not directly, but provides context for found species

    Args:
        latitude: Location latitude
        longitude: Location longitude
        radius_km: Search radius in kilometers
        max_results: Maximum sightings to return

    Returns:
        JSON with all recent sightings and summary statistics

    Example:
        get_location_birds(40.7829, -73.9654, 25, 60)
        -> All birds being seen in Central Park area right now
    """
    if not -90 <= latitude <= 90:
        return _format_error_response("Latitude must be between -90 and 90")

    if not -180 <= longitude <= 180:
        return _format_error_response("Longitude must be between -180 and 180")

    try:
        endpoint = "/data/obs/geo/recent"
        params = {
            "lat": latitude,
            "lng": longitude,
            "dist": radius_km,
            "maxResults": max_results
        }

        data = _make_request(endpoint, params)

        if data is None:
            return _format_error_response("Failed to fetch sightings")

        if not data:
            return _format_success_response(
                [],
                count=0,
                unique_species=0,
                location={"lat": latitude, "lng": longitude},
                radius_km=radius_km,
                message="No sightings found at this location"
            )

        sightings = [
            {
                "common_name": obs['comName'],
                "scientific_name": obs['sciName'],
                "species_code": obs['speciesCode'],
                "location": obs['locName'],
                "date": obs['obsDt'],
                "count": obs.get('howMany'),
                "latitude": obs.get('lat'),
                "longitude": obs.get('lng')
            }
            for obs in data
        ]

        # Calculate unique species count
        unique_species = len(set(obs['common_name'] for obs in sightings))

        # Find most common birds
        bird_counts = {}
        for obs in sightings:
            bird_counts[obs['common_name']] = bird_counts.get(obs['common_name'], 0) + 1
        top_birds = sorted(bird_counts.items(), key=lambda x: x[1], reverse=True)[:5]

        return _format_success_response(
            sightings,
            count=len(sightings),
            unique_species=unique_species,
            location={"lat": latitude, "lng": longitude},
            radius_km=radius_km,
            top_birds=[{"species": name, "observations": count} for name, count in top_birds]
        )

    except Exception as e:
        return _format_error_response(f"Lookup failed: {str(e)}")

# Register as MCP tool
mcp.tool()(get_location_birds)


# ============================================================================
# TOOL 5: get_species_info
# ============================================================================
# Use case: Get taxonomy and detailed info about a species

def get_species_info(species_code: str) -> str:
    """
    Get detailed taxonomy and metadata for a bird species.

    Returns scientific classification, family, order, and other details.

    Can accept:
        - User input: Species code from search_species tool
        - Classifier output: Species code (after search_species lookup)

    Args:
        species_code: eBird species code (e.g., "norcar")

    Returns:
        JSON with complete species information

    Example:
        get_species_info("norcar")
        -> Northern Cardinal taxonomy, family, order, etc.
    """
    if not species_code or len(species_code.strip()) < 2:
        return _format_error_response("Species code required")

    try:
        endpoint = "/ref/taxonomy/ebird"
        params = {
            "fmt": "json",
            "species": species_code
        }

        data = _make_request(endpoint, params)

        if data is None:
            return _format_error_response("Failed to fetch taxonomy")

        # Find main species (not subspecies)
        species = None
        for s in data:
            if s.get('speciesCode') == species_code and s.get('category') == 'species':
                species = s
                break

        if not species:
            return _format_error_response(f"Species code '{species_code}' not found")

        info = {
            "common_name": species['comName'],
            "scientific_name": species['sciName'],
            "species_code": species['speciesCode'],
            "family": species.get('familyComName', 'Unknown'),
            "family_sci_name": species.get('familySciName', 'Unknown'),
            "order": species.get('order', 'Unknown'),
            "category": species.get('category', 'Unknown')
        }

        return _format_success_response(info, species_code=species_code)

    except Exception as e:
        return _format_error_response(f"Lookup failed: {str(e)}")

# Register as MCP tool
mcp.tool()(get_species_info)


# ============================================================================
# TOOL 6: get_notable_sightings
# ============================================================================
# Use case: Find rare/unusual birds in a region

def get_notable_sightings(
    region_code: str = "US",
    max_results: int = 10
) -> str:
    """
    Get rare or notable bird sightings in a region.

    Notable sightings are birds that are unusual/rare for the region.
    Great for discovering unexpected species.

    Can accept:
        - User input: Region code (e.g., "US", "US-NY", "CA-ON")
        - Classifier output: Not directly, but region can be derived from location

    Args:
        region_code: Region code (country, state, province)
        max_results: Maximum notable sightings to return

    Returns:
        JSON with recent notable/rare sightings

    Example:
        get_notable_sightings("US-NY", 10)
        -> Rare/unusual birds spotted in New York recently
    """
    if not region_code:
        return _format_error_response("Region code required")

    try:
        endpoint = f"/data/obs/{region_code}/recent/notable"
        params = {"maxResults": max_results}

        data = _make_request(endpoint, params)

        if data is None:
            return _format_error_response("Failed to fetch notable sightings")

        if not data:
            return _format_success_response(
                [],
                count=0,
                region_code=region_code,
                message="No notable sightings found"
            )

        notable = [
            {
                "common_name": obs['comName'],
                "scientific_name": obs['sciName'],
                "species_code": obs['speciesCode'],
                "location": obs['locName'],
                "location_id": obs['locId'],
                "date": obs['obsDt'],
                "count": obs.get('howMany'),
                "latitude": obs.get('lat'),
                "longitude": obs.get('lng')
            }
            for obs in data
        ]

        return _format_success_response(
            notable,
            count=len(notable),
            region_code=region_code
        )

    except Exception as e:
        return _format_error_response(f"Lookup failed: {str(e)}")

# Register as MCP tool
mcp.tool()(get_notable_sightings)

# ============================================================================
# TOOL 7: analyze_location
# ============================================================================
# Use case: Comprehensive location analysis - all birds + hotspots + summary

def analyze_location(
    latitude: float,
    longitude: float,
    radius_km: int = 50
) -> str:
    """
    Comprehensive location analysis combining all bird data.

    This is a "power tool" that combines multiple API calls to give
    complete view of birding activity: recent sightings, hotspots, stats.

    Can accept:
        - User input: Coordinates from address lookup
        - Classifier output: Not directly, but provides full context

    Args:
        latitude: Location latitude
        longitude: Location longitude
        radius_km: Search radius in kilometers

    Returns:
        JSON with sightings, hotspots, and comprehensive statistics

    Example:
        analyze_location(40.7820, -73.9654, 25)
        -> Complete birding report for Central Park area
    """
    if not -90 <= latitude <= 90:
        return _format_error_response("Latitude must be between -90 and 90")

    if not -180 <= longitude <= 180:
        return _format_error_response("Longitude must be between -180 and 180")

    try:
        # Get all recent observations
        obs_endpoint = "/data/obs/geo/recent"
        obs_params = {
            "lat": latitude,
            "lng": longitude,
            "dist": radius_km,
            "maxResults": 100
        }

        sightings_data = _make_request(obs_endpoint, obs_params) or []

        # Get hotspots (max 50)
        hotspots_endpoint = "/ref/hotspot/geo"
        hotspots_params = {
            "lat": latitude,
            "lng": longitude,
            "dist": radius_km,
            "fmt": "json"
        }

        hotspots_data = _make_request(hotspots_endpoint, hotspots_params) or []

        # Format sightings
        sightings = [
            {
                "common_name": obs['comName'],
                "scientific_name": obs['sciName'],
                "species_code": obs['speciesCode'],
                "location": obs['locName'],
                "date": obs['obsDt'],
                "count": obs.get('howMany'),
                "latitude": obs.get('lat'),
                "longitude": obs.get('lng')
            }
            for obs in sightings_data
        ]

        # Format hotspots
        hotspots = [
            {
                "name": hotspot['locName'],
                "location_id": hotspot['locId'],
                "latitude": hotspot['lat'],
                "longitude": hotspot['lng'],
                "species_recorded": hotspot.get('numSpeciesAllTime', 0),
                "latest_obs_date": hotspot.get('latestObsDt', 'Unknown')
            }
            for hotspot in hotspots_data[:15]
        ]

        # Calculate statistics
        unique_species = len(set(obs['common_name'] for obs in sightings))

        # Find top species
        bird_counts = {}
        for obs in sightings:
            bird_counts[obs['common_name']] = bird_counts.get(obs['common_name'], 0) + 1
        top_birds = sorted(bird_counts.items(), key=lambda x: x[1], reverse=True)[:10]

        analysis = {
            "location": {
                "latitude": latitude,
                "longitude": longitude,
                "radius_km": radius_km,
            },
            "sightings": sightings,
            "hotspots": hotspots,
            "summary": {
                "total_sightings": len(sightings),
                "unique_species": unique_species,
                "total_hotspots": len(hotspots),
                "top_species": [{"name": name, "observations": count} for name, count in top_birds]
            }
        }

        return _format_success_response(analysis)

    except Exception as e:
        return _format_error_response(f"Analysis failed: {str(e)}")

# Register as MCP tool
mcp.tool()(analyze_location)

# ============================================================================
# SERVER STARTUP WITH DUAL TRANSPORT SUPPORT
# ============================================================================

def main():
    """Start the MCP server with dual transport support."""
    # Determine transport mode first
    is_http_mode = "--http" in sys.argv or "--streamable-http" in sys.argv

    # For STDIO mode, all informational output must go to stderr (stdout is for JSON-RPC only)
    output = sys.stdout if is_http_mode else sys.stderr

    print("\n" + "=" * 70, file=output)
    print("🦅 [eBird MCP SERVER] - Starting...", file=output)
    print("=" * 70, file=output)
    print(f"[API KEY]:    {'✅ Configured' if EBIRD_API_KEY else '❌ Missing'}", file=output)
    print("\n[AVAILABLE TOOLS]:", file=output)

    tools_list = [
        "1. search_species - Find species by name",
        "2. get_recent_sightings_nearby - Recent sightings near location",
        "3. find_hotspots_nearby - Find popular birding locations",
        "4. get_location_birds - All birds at a location",
        "5. get_species_info - Taxonomy and species details",
        "6. get_notable_sightings - Rare/unusual birds in region",
        "7. analyze_location - Comprehensive location analysis"
    ]

    for tool in tools_list:
        print(f"  ✓ {tool}", file=output)

    print("\n" + "=" * 70, file=output)

    if is_http_mode:
        # Extract port from command line args
        port = 8000
        host = "127.0.0.1"

        for i, arg in enumerate(sys.argv):
            if arg == "--port" and i + 1 < len(sys.argv):
                port = int(sys.argv[i + 1])
            elif arg == "--host" and i + 1 < len(sys.argv):
                host = sys.argv[i + 1]

        print("[TRANSPORT]: Starting streamable-http MCP server", file=output)
        print(f"[HOST]:      {host}", file=output)
        print(f"[PORT]:      {port}", file=output)
        print(f"[URL]:       http://{host}:{port}", file=output)
        print(f"[AUTH]:      {'🔒 Enabled (production)' if IS_PRODUCTION and MCP_API_KEY else '🔓 Disabled (development)'}", file=output)
        print("[NOTE]:      This is proper MCP over HTTP", file=output)
        print("=" * 70 + "\n", file=output)

        # Run with streamable-http transport (built-in MCP support)
        mcp.run(transport="streamable-http", host=host, port=port)
    else:
        print("[TRANSPORT]: Running as stdio MCP server", file=output)
        print("[NOTE]:      For HTTP transport, use: python ebird_tools.py --http", file=output)
        print("=" * 70 + "\n", file=output)

        # Run as stdio MCP server (default)
        mcp.run(transport="stdio")


if __name__ == "__main__":
    main()