College Football MCP

""" Research MCP Client This module provides a client interface to call the research-mcp server for AI-powered research capabilities. """ import httpx import logging import os from typing import Optional, Dict, Any logger = logging.getLogger(__name__) # Research MCP server URL - defaults to external URL, can be overridden via env RESEARCH_MCP_URL = os.environ.get("RESEARCH_MCP_URL", "https://prompt80.com/research-mcp") RESEARCH_MCP_API_KEY = os.environ.get("RESEARCH_MCP_API_KEY", "") class ResearchClient: """Client for interacting with the research-mcp server""" def __init__(self, base_url: str = None, api_key: str = None): """ Initialize Research client. Args: base_url: Base URL of the research-mcp server (defaults to RESEARCH_MCP_URL env var) api_key: API key for authentication (optional, defaults to RESEARCH_MCP_API_KEY env var) """ self.base_url = (base_url or RESEARCH_MCP_URL).rstrip('/') self.api_key = api_key or RESEARCH_MCP_API_KEY self.client = httpx.AsyncClient(timeout=120.0) # Research can take longer async def close(self): """Close the HTTP client""" await self.client.aclose() async def run_research( self, topic: str, iterations: int = 1, query_count: int = 5, page: int = 1, page_size: int = 10 ) -> Optional[Dict[str, Any]]: """ Run a research query on a topic. Args: topic: Research topic/question iterations: Number of research iterations (default: 1) query_count: Number of search queries to generate (default: 5) page: Page number for pagination (default: 1) page_size: Results per page (default: 10) Returns: Research results with synthesis or None if error """ try: url = f"{self.base_url}/tools/run_research" headers = { "Content-Type": "application/json" } # Add auth header if API key is configured if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" payload = { "topic": topic, "iterations": iterations, "query_count": query_count, "page": page, "page_size": page_size } logger.info(f"Calling research-mcp with topic: {topic[:50]}...") response = await self.client.post(url, json=payload, headers=headers) response.raise_for_status() result = response.json() return result.get("result") if isinstance(result, dict) and "result" in result else result except httpx.HTTPStatusError as e: logger.error(f"HTTP error calling research-mcp: {e.response.status_code} - {e.response.text}") return None except Exception as e: logger.error(f"Error calling research-mcp: {e}") return None