Baidu Vector Database MCP Server

Instructions

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