Grants Search MCP Server

"""Agency landscape analysis tool for mapping agencies and their funding focus areas.""" import logging import time from collections import defaultdict from typing import Any, Dict, List, Optional from src.mcp_server.models.grants_schemas import AgencyV1, GrantsAPIResponse, OpportunityV1 from src.mcp_server.tools.utils.api_client import APIError, SimplerGrantsAPIClient from src.mcp_server.tools.utils.cache_manager import InMemoryCache from src.mcp_server.tools.utils.cache_utils import CacheKeyGenerator logger = logging.getLogger(__name__) def analyze_agency_portfolio( agency_code: str, opportunities: List[OpportunityV1] ) -> Dict[str, Any]: """ Analyze an agency's grant portfolio. Args: agency_code: Agency code to analyze opportunities: List of opportunities from this agency Returns: Analysis of the agency's portfolio """ portfolio = { "agency_code": agency_code, "total_opportunities": len(opportunities), "status_breakdown": defaultdict(int), "category_breakdown": defaultdict(int), "funding_stats": { "total_estimated_funding": 0, "average_award_ceiling": 0, "average_award_floor": 0, "min_award": None, "max_award": None, }, "deadline_distribution": defaultdict(int), "eligibility_patterns": defaultdict(int), } award_ceilings = [] award_floors = [] total_funding = 0 for opp in opportunities: # Status breakdown portfolio["status_breakdown"][opp.opportunity_status] += 1 # Category breakdown if opp.category: portfolio["category_breakdown"][opp.category] += 1 # Funding analysis summary = opp.summary if summary.award_ceiling: award_ceilings.append(summary.award_ceiling) if portfolio["funding_stats"]["max_award"] is None: portfolio["funding_stats"]["max_award"] = summary.award_ceiling else: portfolio["funding_stats"]["max_award"] = max( portfolio["funding_stats"]["max_award"], summary.award_ceiling ) if summary.award_floor: award_floors.append(summary.award_floor) if portfolio["funding_stats"]["min_award"] is None: portfolio["funding_stats"]["min_award"] = summary.award_floor else: portfolio["funding_stats"]["min_award"] = min( portfolio["funding_stats"]["min_award"], summary.award_floor ) if summary.estimated_total_program_funding: total_funding += summary.estimated_total_program_funding # Deadline distribution if summary.close_date: try: # Extract month from close date month = summary.close_date.split("-")[1] if "-" in summary.close_date else "Unknown" portfolio["deadline_distribution"][f"Month_{month}"] += 1 except: pass # Eligibility patterns (simplified) if summary.applicant_types: for applicant_type in summary.applicant_types: portfolio["eligibility_patterns"][applicant_type] += 1 # Calculate averages if award_ceilings: portfolio["funding_stats"]["average_award_ceiling"] = sum(award_ceilings) / len(award_ceilings) if award_floors: portfolio["funding_stats"]["average_award_floor"] = sum(award_floors) / len(award_floors) portfolio["funding_stats"]["total_estimated_funding"] = total_funding # Convert defaultdicts to regular dicts for JSON serialization portfolio["status_breakdown"] = dict(portfolio["status_breakdown"]) portfolio["category_breakdown"] = dict(portfolio["category_breakdown"]) portfolio["deadline_distribution"] = dict(portfolio["deadline_distribution"]) portfolio["eligibility_patterns"] = dict(portfolio["eligibility_patterns"]) return portfolio def identify_cross_agency_patterns( agency_profiles: Dict[str, Dict[str, Any]] ) -> Dict[str, Any]: """ Identify patterns across multiple agencies. Args: agency_profiles: Dictionary of agency profiles Returns: Cross-agency analysis """ patterns = { "overlap_areas": [], "unique_specializations": {}, "collaboration_patterns": {}, "funding_comparison": {}, } # Find overlapping categories all_categories = defaultdict(list) for agency_code, profile in agency_profiles.items(): for category in profile.get("category_breakdown", {}).keys(): all_categories[category].append(agency_code) # Identify overlaps and unique specializations for category, agencies in all_categories.items(): if len(agencies) > 1: patterns["overlap_areas"].append({ "category": category, "agencies": agencies, "count": len(agencies) }) elif len(agencies) == 1: agency = agencies[0] if agency not in patterns["unique_specializations"]: patterns["unique_specializations"][agency] = [] patterns["unique_specializations"][agency].append(category) # Compare funding levels for agency_code, profile in agency_profiles.items(): funding_stats = profile.get("funding_stats", {}) patterns["funding_comparison"][agency_code] = { "average_ceiling": funding_stats.get("average_award_ceiling", 0), "average_floor": funding_stats.get("average_award_floor", 0), "total_opportunities": profile.get("total_opportunities", 0), } return patterns def format_agency_landscape_report( agencies: List[AgencyV1], agency_profiles: Dict[str, Dict[str, Any]], cross_agency_analysis: Dict[str, Any], funding_landscape: Dict[str, Any] ) -> str: """ Format the agency landscape analysis as a readable report. Args: agencies: List of agencies agency_profiles: Individual agency analyses cross_agency_analysis: Cross-agency patterns funding_landscape: Overall funding landscape Returns: Formatted report string """ report = """ AGENCY LANDSCAPE ANALYSIS ========================= OVERVIEW --------""" report += f"\nTotal Active Agencies: {funding_landscape['total_active_agencies']}" report += f"\nTotal Opportunities Analyzed: {sum(p['total_opportunities'] for p in agency_profiles.values())}" # Top agencies by opportunity count top_agencies = sorted( agency_profiles.items(), key=lambda x: x[1]['total_opportunities'], reverse=True )[:5] report += "\n\nTOP AGENCIES BY OPPORTUNITY COUNT\n" + "-" * 35 for agency_code, profile in top_agencies: agency_name = next((a.agency_name for a in agencies if a.agency_code == agency_code), agency_code) report += f"\n{agency_code}: {agency_name}" report += f"\n • Opportunities: {profile['total_opportunities']}" report += f"\n • Categories: {', '.join(profile['category_breakdown'].keys())[:100]}" if profile['funding_stats']['average_award_ceiling']: report += f"\n • Avg Award Ceiling: ${profile['funding_stats']['average_award_ceiling']:,.0f}" # Cross-agency patterns if cross_agency_analysis['overlap_areas']: report += "\n\nCROSS-AGENCY COLLABORATION AREAS\n" + "-" * 33 for overlap in cross_agency_analysis['overlap_areas'][:5]: report += f"\n• {overlap['category']}: {', '.join(overlap['agencies'])}" # Unique specializations if cross_agency_analysis['unique_specializations']: report += "\n\nUNIQUE AGENCY SPECIALIZATIONS\n" + "-" * 30 for agency, specializations in list(cross_agency_analysis['unique_specializations'].items())[:5]: report += f"\n{agency}: {', '.join(specializations[:3])}" # Funding distribution report += "\n\nFUNDING LANDSCAPE\n" + "-" * 17 total_funding = sum( p['funding_stats']['total_estimated_funding'] for p in agency_profiles.values() ) if total_funding > 0: report += f"\nTotal Estimated Funding: ${total_funding:,.0f}" # Category distribution if funding_landscape.get('category_specialization'): report += "\n\nFUNDING BY CATEGORY\n" + "-" * 19 for category, count in list(funding_landscape['category_specialization'].items())[:5]: report += f"\n• {category}: {count} opportunities" report += "\n\n" + "=" * 60 return report def register_agency_landscape_tool(mcp: Any, context: Dict[str, Any]) -> None: """ Register the agency landscape tool with the MCP server. Args: mcp: FastMCP instance context: Server context containing cache, API client, etc. """ cache: InMemoryCache = context["cache"] api_client: SimplerGrantsAPIClient = context["api_client"] @mcp.tool async def agency_landscape( include_opportunities: bool = True, focus_agencies: Optional[List[str]] = None, funding_category: Optional[str] = None, max_agencies: int = 10 ) -> str: """ Map agencies and their funding focus areas with comprehensive analysis. This tool provides insights into agency funding patterns, specializations, and cross-agency collaboration opportunities. Args: include_opportunities: Include opportunity analysis (default: True) focus_agencies: Specific agency codes to analyze (e.g., ["NSF", "NIH"]) funding_category: Filter by funding category max_agencies: Maximum number of agencies to analyze (default: 10) Returns: Comprehensive agency landscape analysis report """ try: start_time = time.time() # Generate optimized cache key cache_key = CacheKeyGenerator.generate_simple( "agency_landscape", include_opportunities=include_opportunities, focus_agencies=focus_agencies, funding_category=funding_category, max_agencies=max_agencies ) # Check cache cached_result = cache.get(cache_key) if cached_result: logger.info("Cache hit for agency landscape analysis") return cached_result["report"] logger.info(f"Analyzing agency landscape (max_agencies={max_agencies})") # Prepare filters for agency search agency_filters = {} if focus_agencies: # Note: API might not support direct agency code filtering in agency search # We'll filter the results manually pass # Search for agencies logger.debug("Fetching agencies from API") agency_response = await api_client.search_agencies( filters=agency_filters, pagination={"page_size": 100, "page_offset": 1} ) # Parse response api_response = GrantsAPIResponse(**agency_response) all_agencies = api_response.get_agencies() # Filter agencies if specific ones requested if focus_agencies: agencies = [a for a in all_agencies if a.agency_code in focus_agencies] else: agencies = all_agencies[:max_agencies] logger.info(f"Analyzing {len(agencies)} agencies") # Analyze each agency's opportunities agency_profiles = {} if include_opportunities: for agency in agencies: try: # Search for opportunities from this agency opp_filters = { "agency_code": agency.agency_code, "opportunity_status": { "one_of": ["posted", "forecasted"] } } if funding_category: opp_filters["category"] = funding_category opp_response = await api_client.search_opportunities( filters=opp_filters, pagination={"page_size": 50, "page_offset": 1} ) opp_api_response = GrantsAPIResponse(**opp_response) opportunities = opp_api_response.get_opportunities() # Analyze this agency's portfolio agency_profiles[agency.agency_code] = analyze_agency_portfolio( agency.agency_code, opportunities ) agency_profiles[agency.agency_code]["agency_name"] = agency.agency_name except Exception as e: logger.warning(f"Error analyzing agency {agency.agency_code}: {e}") # Create minimal profile agency_profiles[agency.agency_code] = { "agency_code": agency.agency_code, "agency_name": agency.agency_name, "total_opportunities": 0, "error": str(e) } # Cross-agency analysis cross_agency_analysis = identify_cross_agency_patterns(agency_profiles) # Overall funding landscape funding_landscape = { "total_active_agencies": len(agencies), "funding_distribution": {}, "category_specialization": defaultdict(int), } # Aggregate category specialization for profile in agency_profiles.values(): for category, count in profile.get("category_breakdown", {}).items(): funding_landscape["category_specialization"][category] += count funding_landscape["category_specialization"] = dict( sorted( funding_landscape["category_specialization"].items(), key=lambda x: x[1], reverse=True ) ) # Generate report report = format_agency_landscape_report( agencies, agency_profiles, cross_agency_analysis, funding_landscape ) # Prepare result for caching result = { "agencies": agencies, "agency_profiles": agency_profiles, "cross_agency_analysis": cross_agency_analysis, "funding_landscape": funding_landscape, "report": report, "metadata": { "analysis_time": time.time() - start_time, "agencies_analyzed": len(agencies), "cache_used": False } } # Cache the result cache.set(cache_key, result) return report except APIError as e: logger.error(f"API error during agency landscape analysis: {e}") return f"Error analyzing agency landscape: {e}" except Exception as e: logger.error(f"Unexpected error during agency landscape analysis: {e}", exc_info=True) return f"An unexpected error occurred: {e}" logger.info("Registered agency_landscape tool")