Perplexity MCP Server

Instructions

Implementation Reference

• server.py:78-118 (handler)

  The main handler function for the 'perplexity_medium' tool. It prepares messages, fetches the medium configuration from TOOL_CONFIGS, calls the PerplexityClient's chat_completion, formats the response, and handles errors.
  @mcp.tool() def perplexity_medium(query: str, messages: List[Dict[str, str]] = None) -> Dict[str, Any]: """ Enhanced reasoning with moderate search depth using sonar-reasoning-pro. Best for: Complex questions requiring analysis, moderate research depth, technical explanations with citations. Uses medium reasoning effort and search context size. 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["medium"] # 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_medium") return { "error": "tool_error", "message": f"Failed to process query: {str(e)}" }
• config.py:30-36 (schema)

  Schema/configuration defining parameters for the perplexity_medium tool, including model, reasoning effort, and web search options.
  "medium": { "model": "sonar-reasoning-pro", "reasoning_effort": "medium", "web_search_options": { "search_context_size": "medium" } },
• client.py:16-140 (helper)

  The PerplexityClient class providing chat_completion and format_response methods, which perform the actual API interaction and response processing used by perplexity_medium.
  class PerplexityClient: """Client for interacting with Perplexity API.""" def __init__(self): """Initialize the Perplexity client.""" self.api_key = get_api_key() self.base_url = PERPLEXITY_BASE_URL self.timeout = PERPLEXITY_TIMEOUT self.headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } 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)

  Helper function to lazily initialize and retrieve the shared PerplexityClient instance used by all perplexity tools including medium.
  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
• server.py:22-23 (registration)

  Initialization of the FastMCP server instance where tools like perplexity_medium are registered via decorators.
  mcp = FastMCP("Perplexity MCP")