Agentic AI with MCP

Instructions

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