internet_search
Search the internet to find current information and web content using the Tavily API, enabling AI systems to access real-time data for answering queries.
Instructions
Search the internet using Tavily API.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| limit | No | ||
| include_raw_content | No |
Implementation Reference
- server.py:89-127 (handler)The core handler implementation for the 'internet_search' MCP tool. It sends a search query to the Tavily API, processes the response to extract answers, results with titles, URLs, and truncated content, and optional follow-up questions. The @mcp.tool() decorator registers it with the MCP server. Input schema inferred from type hints: query (str), limit (int=3), include_raw_content (bool=False); output: str.@mcp.tool() async def internet_search(query: str, limit: int = 3, include_raw_content: bool = False) -> str: """Search the internet using Tavily API.""" try: request_data = { "api_key": TAVILY_API_KEY, "query": query, "search_depth": "basic", "include_answer": True, "include_raw_content": include_raw_content, "include_images": False, "max_results": limit } data = await make_api_request(f"{TAVILY_API_BASE}/search", json=request_data) if not data: return "Failed to perform internet search. Please try again later." results = [] if data.get("answer"): results.append(f"Quick Answer: {data['answer']}") if data.get("results"): for idx, result in enumerate(data["results"][:limit], 1): result_str = f"{idx}. {result.get('title', 'No title')}\n URL: {result.get('url', 'No URL')}" if result.get("content"): result_str += f"\n Content: {result['content'][:500]}..." # Truncate content results.append(result_str) if data.get("follow_up_questions"): results.append("\nSuggested follow-up questions:") results.extend(f"- {q}" for q in data["follow_up_questions"]) return "\n\n".join(results) if results else "No results found." except Exception as e: logger.error(f"Error in internet_search: {e}") return "Failed to perform internet search due to an internal error."
- server.py:34-58 (helper)Supporting helper utility function used by internet_search (and other tools) to make HTTP requests to APIs with timeout, error handling, and JSON parsing.async def make_api_request(url: str, params: dict = None, headers: dict = None, json: dict = None) -> dict[str, Any] | None: """Make a generic API request with proper error handling.""" default_headers = { "User-Agent": USER_AGENT, "Accept": "application/json" } if headers: default_headers.update(headers) async with httpx.AsyncClient(timeout=30.0) as client: try: if json: response = await client.post(url, json=json, headers=default_headers) else: response = await client.get(url, params=params, headers=default_headers) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: logger.error(f"HTTP error for {url}: {e}") except httpx.RequestError as e: logger.error(f"Request failed for {url}: {e}") except Exception as e: logger.error(f"Unexpected error for {url}: {e}") return None