Perplexity MCP Server

Instructions

Comprehensive research with maximum depth using sonar-deep-research.

Best for: Deep research tasks, comprehensive analysis, complex multi-step reasoning,
academic research, detailed technical investigations.
Uses high reasoning effort and search context size.

WARNING: This tool may take significantly longer (potentially 10-30 minutes) 
and may timeout on very complex queries.

Args:
    query: The question or prompt to send to Perplexity
    messages: Optional conversation context (list of {"role": "user/assistant", "content": "..."})

Returns:
    Dictionary with content and citations

Implementation Reference

• server.py:121-168 (handler)

  The primary handler function for the 'perplexity_large' tool. Decorated with @mcp.tool() for automatic registration in FastMCP. Prepares messages, retrieves large config, calls PerplexityClient.chat_completion, formats response, and handles errors. Includes comprehensive docstring serving as schema description.

  @mcp.tool()
  def perplexity_large(query: str, messages: List[Dict[str, str]] = None) -> Dict[str, Any]:
      """
      Comprehensive research with maximum depth using sonar-deep-research.
      
      Best for: Deep research tasks, comprehensive analysis, complex multi-step reasoning,
      academic research, detailed technical investigations.
      Uses high reasoning effort and search context size.
      
      WARNING: This tool may take significantly longer (potentially 10-30 minutes) 
      and may timeout on very complex queries.
      
      Args:
          query: The question or prompt to send to Perplexity
          messages: Optional conversation context (list of {"role": "user/assistant", "content": "..."})
      
      Returns:
          Dictionary with content and citations
      """
      try:
          client = get_perplexity_client()
          
          # Prepare messages
          if messages is None:
              messages = []
          
          # Add the current query
          messages.append({"role": "user", "content": query})
          
          # Get tool configuration
          config = TOOL_CONFIGS["large"]
          
          # Log warning about potential timeout
          logger.warning(f"Starting deep research query - this may take 10-30 minutes")
          
          # Make API request
          response = client.chat_completion(messages=messages, **config)
          
          # Format and return response
          return client.format_response(response)
          
      except Exception as e:
          logger.exception("Error in perplexity_large")
          return {
              "error": "tool_error", 
              "message": f"Failed to process query: {str(e)}"
          }

• config.py:26-43 (helper)

  Global TOOL_CONFIGS dictionary defining parameters for all tools, specifically TOOL_CONFIGS['large'] configures the sonar-deep-research model with high reasoning effort and search context size for the perplexity_large tool.

  TOOL_CONFIGS = {
      "small": {
          "model": "sonar-pro"
      },
      "medium": {
          "model": "sonar-reasoning-pro",
          "reasoning_effort": "medium",
          "web_search_options": {
              "search_context_size": "medium"
          }
      },
      "large": {
          "model": "sonar-deep-research",
          "reasoning_effort": "high", 
          "web_search_options": {
              "search_context_size": "high"
          }
      }

• client.py:29-140 (helper)

  PerplexityClient methods chat_completion (invoked with large config) and format_response (used to process the API response for the tool). Handles HTTP requests to Perplexity API, error handling, logging, token usage, and response formatting including removal of <think> blocks specific to reasoning models like large.

  def chat_completion(self,messages: List[Dict[str, str]],model: str,**kwargs) -> Dict[str, Any]:
      """
      Send a chat completion request to Perplexity API.
      
      Args:
          messages: List of message objects with role and content
          model: Model name (e.g., "sonar-pro", "sonar-reasoning-pro", "sonar-deep-research")
          **kwargs: Additional parameters for the API request
          
      Returns:
          Dict containing the API response
      """
      try:
          # Prepare request payload
          payload = {
              "model": model,
              "messages": messages,
              **kwargs
          }
          
          # Log request details to stderr
          logger.info(f"Making request to Perplexity API with model: {model}")
          logger.debug(f"Request payload: {payload}")
          
          # Make the API request
          with httpx.Client(timeout=self.timeout) as client:
              response = client.post(
                  f"{self.base_url}/chat/completions",
                  headers=self.headers,
                  json=payload
              )
              
              # Check for HTTP errors
              response.raise_for_status()
              
              # Parse response
              result = response.json()
              
              # Log response details to stderr
              logger.info(f"Received response from Perplexity API")
              if "usage" in result:
                  usage = result["usage"]
                  logger.info(f"Token usage - Prompt: {usage.get('prompt_tokens', 0)}, "
                            f"Completion: {usage.get('completion_tokens', 0)}, "
                            f"Total: {usage.get('total_tokens', 0)}")
              
              return result
              
      except httpx.HTTPStatusError as e:
          logger.error(f"HTTP error from Perplexity API: {e.response.status_code}")
          logger.error(f"Response content: {e.response.text}")
          return {
              "error": f"HTTP {e.response.status_code}",
              "message": f"API request failed: {e.response.text}"
          }
          
      except httpx.TimeoutException:
          logger.error(f"Request timeout after {self.timeout} seconds")
          return {
              "error": "timeout",
              "message": f"Request timed out after {self.timeout} seconds"
          }
          
      except Exception as e:
          logger.exception("Unexpected error in chat_completion")
          return {
              "error": "unexpected_error",
              "message": f"Unexpected error: {str(e)}"
          }
  
  def format_response(self, api_response: Dict[str, Any]) -> Dict[str, Any]:
      """
      Format API response for MCP tool return.
      
      Args:
          api_response: Raw API response from Perplexity
          
      Returns:
          Formatted response for MCP tool with only content and citations
      """
      # Handle error responses
      if "error" in api_response:
          return api_response
      
      try:
          # Extract main content
          content = ""
          if "choices" in api_response and api_response["choices"]:
              content = api_response["choices"][0]["message"]["content"]
              
              # Remove <think>...</think> sections for reasoning models
              # This removes the thinking tokens that appear in medium/large responses
              content = re.sub(r'<think>.*?</think>', '', content, flags=re.DOTALL)
              
              # Clean up any extra whitespace left after removing think tags
              content = re.sub(r'\n\s*\n\s*\n', '\n\n', content)
              content = content.strip()
          
          # Format response - only include content and citations
          formatted = {
              "content": content,
              "citations": api_response.get("citations", [])
          }
          
          return formatted
          
      except Exception as e:
          logger.exception("Error formatting API response")
          return {
              "error": "format_error",
              "message": f"Failed to format response: {str(e)}"
          }

• server.py:28-34 (helper)

  Utility function to lazily initialize and retrieve the shared PerplexityClient instance used by all perplexity_* tools including perplexity_large.

  def get_perplexity_client() -> PerplexityClient:
      """Get or create the Perplexity client instance."""
      global perplexity_client
      if perplexity_client is None:
          perplexity_client = PerplexityClient()
      return perplexity_client