Perplexity MCP Server

Instructions

Implementation Reference

• server.py:36-76 (handler)

  Core implementation of the perplexity_small tool. This handler function is decorated with @mcp.tool() for automatic registration in the FastMCP server. It processes the input query and optional messages, appends the query to messages, retrieves the 'small' tool configuration (sonar-pro model), calls the PerplexityClient for chat completion, formats the response, and handles errors.
  @mcp.tool() def perplexity_small(query: str, messages: List[Dict[str, str]] = None) -> Dict[str, Any]: """ Quick and reliable queries using Perplexity's sonar-pro model. Best for: Fast factual questions, basic research, immediate answers. Uses default parameters for optimal speed and cost-effectiveness. 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["small"] # 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_small") return { "error": "tool_error", "message": f"Failed to process query: {str(e)}" }
• config.py:27-29 (schema)

  Schema/configuration for the perplexity_small tool specifying the model 'sonar-pro' used in the API call.
  "small": { "model": "sonar-pro" },
• server.py:28-34 (helper)

  Helper function to lazily initialize and retrieve the shared PerplexityClient instance used by perplexity_small.
  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
• client.py:99-140 (helper)

  Helper method in PerplexityClient used by perplexity_small to format the raw API response into MCP-compatible output with content and citations, cleaning reasoning tags.
  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)}" }
• client.py:29-98 (helper)

  Core API interaction helper in PerplexityClient called by perplexity_small to send the chat completion request to Perplexity API using the specified model and config.
  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)}" }