Baidu Vector Database MCP Server

Instructions

Perform vector similarity search combining vector similarity and scalar attribute filtering in the Mochow instance.

Args:
    table_name (str): Name of the table to search.
    vector (list[float]): Search vector.
    vector_field (str): Target field containing vectors to search. Defaults to "vector".
    limit (int): Maximum number of results. Defaults to 10.
    output_fields (Optional[list[str]]): Fields to return in the results. Defaults to None.
    filter_expr (Optional[str]): Filter expression for scalar attributes. Defaults to None.
    params: Additional vector search parameters

Returns:
    str: A string containing the vector search results.

Implementation Reference

• src/mochow_mcp_server/server.py:682-714 (handler)

  The main handler function for the 'vector_search' MCP tool. It uses the MochowConnector to perform the vector search and formats the results as a string.

  @mcp.tool()
  async def vector_search(
      table_name: str, 
      vector: list[float],
      vector_field: str = "vector",
      filter_expr: Optional[str] = None,
      limit: int = 10,
      output_fields: Optional[list[str]] = None,
      ctx: Context = None,
  ) -> str:
      """
      Perform vector similarity search combining vector similarity and scalar attribute filtering in the Mochow instance.
  
      Args:
          table_name (str): Name of the table to search.
          vector (list[float]): Search vector.
          vector_field (str): Target field containing vectors to search. Defaults to "vector".
          limit (int): Maximum number of results. Defaults to 10.
          output_fields (Optional[list[str]]): Fields to return in the results. Defaults to None.
          filter_expr (Optional[str]): Filter expression for scalar attributes. Defaults to None.
          params: Additional vector search parameters
  
      Returns:
          str: A string containing the vector search results.
      """
      connector = ctx.request_context.lifespan_context.connector
      search_results = await connector.vector_search(table_name, vector, vector_field, limit, output_fields, filter_expr)
  
      output = f"Vector search results for '{table_name}':\n"
      for row in search_results.rows:
          output += f"{str(row["row"])}\n"
  
      return output

• src/mochow_mcp_server/server.py:682-682 (registration)

  The @mcp.tool() decorator registers the vector_search function as an MCP tool.

  @mcp.tool()

• src/mochow_mcp_server/server.py:366-400 (helper)

  Helper method in MochowConnector class that constructs the VectorTopkSearchRequest and calls the underlying database.vector_search method to execute the vector search.

  async def vector_search(
      self,
      table_name: str, 
      vector: list[float],
      vector_field: str = "vector",
      limit: int = 10,
      output_fields: Optional[list[str]] = None,
      filter_expr: Optional[str] = None,
      params: Optional[dict[str, Any]] = None,
  ) -> HttpResponse:
      """
      Perform vector similarity search combining vector similarity and scalar attribute filtering.
  
      Args:
          table_name (str): Name of the table to search.
          vector (list[float]): Search vector.
          vector_field (str): Target field containing vectors to search. Defaults to "vector".
          limit (int): Maximum number of results. Defaults to 10.
          output_fields (Optional[list[str]]): Fields to return in the results. Defaults to None.
          filter_expr (Optional[str]): Filter expression for scalar attributes. Defaults to None.
          params (Optional[dict[str, Any]]): Additional vector search parameters. Defaults to None.
  
      Returns:
          HttpResponse: The HTTP response containing the search results.
      """
      if self.database is None:
          raise ValueError("Switch to the database before perform vector search.")
  
      request = VectorTopkSearchRequest(vector_field=vector_field, vector=FloatVector(vector),
                                        limit=limit, filter=filter_expr,
                                        config=VectorSearchConfig(ef=params.get("ef", 200)))
      try:
          return self.database.table(table_name).vector_search(request=request, projections=output_fields)
      except Exception as e:
          raise ValueError(f"Failed to perform vector search: {str(e)}")

• src/mochow_mcp_server/server.py:31-31 (schema)

  Imports schema classes used in the vector_search implementation, such as VectorTopkSearchRequest for input validation and VectorSearchConfig for parameters.

  from pymochow.model.table import VectorTopkSearchRequest, VectorSearchConfig, FloatVector, BM25SearchRequest