Vectara MCP server

Instructions

Run a RAG query using Vectara, returning search results with a generated response.

Args:
    query: str, The user query to run - required.
    corpus_keys: list[str], List of Vectara corpus keys to use for the search - required. Please ask the user to provide one or more corpus keys. 
    api_key: str, The Vectara API key - required.
    n_sentences_before: int, Number of sentences before the answer to include in the context - optional, default is 2.
    n_sentences_after: int, Number of sentences after the answer to include in the context - optional, default is 2.
    lexical_interpolation: float, The amount of lexical interpolation to use - optional, default is 0.005.
    max_used_search_results: int, The maximum number of search results to use - optional, default is 10.
    generation_preset_name: str, The name of the generation preset to use - optional, default is "vectara-summary-table-md-query-ext-jan-2025-gpt-4o".
    response_language: str, The language of the response - optional, default is "eng".

Returns:
    The response from Vectara, including the generated answer and the search results.

Implementation Reference

• vectara_mcp/server.py:483-575 (handler)

  The primary handler function for the 'ask_vectara' tool. Registered via @mcp.tool() decorator. Includes type-hinted parameters serving as input schema, comprehensive docstring describing args/returns, full execution logic including validation, API payload construction, Vectara query execution, response processing with citations extraction, and error handling.

  @mcp.tool()
  async def ask_vectara(
      query: str,
      ctx: Context,
      corpus_keys: list[str],
      n_sentences_before: int = 2,
      n_sentences_after: int = 2,
      lexical_interpolation: float = 0.005,
      max_used_search_results: int = 10,
      generation_preset_name: str = "vectara-summary-table-md-query-ext-jan-2025-gpt-4o",
      response_language: str = "eng",
  ) -> dict:
      """
      Run a RAG query using Vectara, returning search results with a generated response.
  
      Args:
          query: str, The user query to run - required.
          corpus_keys: list[str], List of Vectara corpus keys to use for the search - required. Please ask the user to provide one or more corpus keys.
          n_sentences_before: int, Number of sentences before the answer to include in the context - optional, default is 2.
          n_sentences_after: int, Number of sentences after the answer to include in the context - optional, default is 2.
          lexical_interpolation: float, The amount of lexical interpolation to use - optional, default is 0.005.
          max_used_search_results: int, The maximum number of search results to use - optional, default is 10.
          generation_preset_name: str, The name of the generation preset to use - optional, default is "vectara-summary-table-md-query-ext-jan-2025-gpt-4o".
          response_language: str, The language of the response - optional, default is "eng".
  
      Note: API key must be configured first using 'setup_vectara_api_key' tool
  
      Returns:
          dict: Structured response containing:
              - "summary": Generated AI summary with markdown citations
              - "citations": List of citation objects with score, text, and metadata
              - "factual_consistency_score": Score indicating factual consistency (if available)
          On error, returns dict with "error" key containing error message.
      """
      # Validate parameters
      validation_error = _validate_common_parameters(query, corpus_keys)
      if validation_error:
          return {"error": validation_error}
  
      if ctx:
          ctx.info(f"Running Vectara RAG query: {query}")
  
      try:
          payload = _build_query_payload(
              query=query,
              corpus_keys=corpus_keys,
              n_sentences_before=n_sentences_before,
              n_sentences_after=n_sentences_after,
              lexical_interpolation=lexical_interpolation,
              max_used_search_results=max_used_search_results,
              generation_preset_name=generation_preset_name,
              response_language=response_language,
              enable_generation=True
          )
  
          result = await _call_vectara_query(payload, ctx)
  
          # Extract the generated summary from the response
          summary_text = ""
          if "summary" in result:
              summary_text = result["summary"]
          elif "answer" in result:
              summary_text = result["answer"]
          else:
              return {"error": f"Unexpected response format: {json.dumps(result, indent=2)}"}
  
          # Build citations list
          citations = []
          if "search_results" in result and result["search_results"]:
              for i, search_result in enumerate(result["search_results"], 1):
                  citation = {
                      "id": i,
                      "score": search_result.get("score", 0.0),
                      "text": search_result.get("text", ""),
                      "document_metadata": search_result.get("document_metadata", {})
                  }
                  citations.append(citation)
  
          # Build response dict
          response = {
              "summary": summary_text,
              "citations": citations
          }
  
          # Add factual consistency score if available
          if "factual_consistency_score" in result:
              response["factual_consistency_score"] = result["factual_consistency_score"]
  
          return response
  
      except Exception as e:
          return {"error": _format_error("Vectara RAG query", e)}

• vectara_mcp/server.py:285-297 (helper)

  Helper function that makes the actual API request to Vectara's /query endpoint, used by ask_vectara.

  async def _call_vectara_query(
      payload: dict,
      ctx: Context = None,
      api_key_override: str = None
  ) -> dict:
      """Make API call to Vectara query endpoint"""
      return await _make_api_request(
          f"{VECTARA_BASE_URL}/query",
          payload,
          ctx,
          api_key_override,
          "query"
      )

• vectara_mcp/server.py:234-283 (helper)

  Helper function that constructs the detailed JSON payload for the Vectara query API, incorporating all parameters from ask_vectara.

  def _build_query_payload(
      query: str,
      corpus_keys: list[str],
      n_sentences_before: int = 2,
      n_sentences_after: int = 2,
      lexical_interpolation: float = 0.005,
      max_used_search_results: int = 10,
      generation_preset_name: str = "vectara-summary-table-md-query-ext-jan-2025-gpt-4o",
      response_language: str = "eng",
      enable_generation: bool = True
  ) -> dict:
      """Build the query payload for Vectara API"""
      payload = {
          "query": query,
          "search": {
              "limit": 100,
              "corpora": [
                  {
                      "corpus_key": corpus_key,
                      "lexical_interpolation": lexical_interpolation
                  } for corpus_key in corpus_keys
              ],
              "context_configuration": {
                  "sentences_before": n_sentences_before,
                  "sentences_after": n_sentences_after
              },
              "reranker": {
                  "type": "customer_reranker",
                  "reranker_name": "Rerank_Multilingual_v1",
                  "limit": 100,
                  "cutoff": 0.2
              }
          },
          "save_history": True,
      }
  
      if enable_generation:
          payload["generation"] = {
              "generation_preset_name": generation_preset_name,
              "max_used_search_results": max_used_search_results,
              "response_language": response_language,
              "citations": {
                  "style": "markdown",
                  "url_pattern": "{doc.url}",
                  "text_pattern": "{doc.title}"
              },
              "enable_factual_consistency_score": True
          }
  
      return payload

• vectara_mcp/server.py:98-114 (helper)

  Helper function for parameter validation (query, corpus_keys, API key), called at the start of ask_vectara.

  def _validate_common_parameters(query: str = "", corpus_keys: list[str] = None) -> str | None:
      """Validate common parameters used across Vectara tools.
  
      Returns:
          str: Error message if validation fails, None if valid
      """
      if not query:
          return "Query is required."
      if not corpus_keys:
          return "Corpus keys are required. Please ask the user to provide one or more corpus keys."
  
      # Check API key availability
      api_key = _get_api_key()
      if not api_key:
          return API_KEY_ERROR_MESSAGE
  
      return None