Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/index.py:25-95 (handler)

  The handler function implementing the tool logic. It constructs and executes a SQL++ query using the Couchbase ADVISOR function to analyze the provided query and return index recommendations, including current used indexes, recommended indexes, and covering indexes.
  def get_index_advisor_recommendations( ctx: Context, bucket_name: str, scope_name: str, query: str ) -> dict[str, Any]: """Get index recommendations from Couchbase Index Advisor for a given SQL++ query. The Index Advisor analyzes the query and provides recommendations for optimal indexes. This tool works with SELECT, UPDATE, DELETE, or MERGE queries. The queries will be run on the specified scope in the specified bucket. Returns a dictionary with: - current_used_indexes: Array of currently used indexes (if any) - recommended_indexes: Array of recommended secondary indexes (if any) - recommended_covering_indexes: Array of recommended covering indexes (if any) Each index object contains: - index: The CREATE INDEX SQL++ command - statements: Array of statement objects with the query and run count """ try: # Build the ADVISOR query advisor_query = f"SELECT ADVISOR('{query}') AS advisor_result" logger.info("Running Index Advisor for the provided query") # Execute the ADVISOR function at cluster level using run_sql_plus_plus_query advisor_results = run_sql_plus_plus_query( ctx, bucket_name, scope_name, advisor_query ) if not advisor_results: return { "message": "No recommendations available", "current_used_indexes": [], "recommended_indexes": [], "recommended_covering_indexes": [], } # The result is wrapped in advisor_result key advisor_data = advisor_results[0].get("advisor_result", {}) # Extract the relevant fields with defaults response = { "current_used_indexes": advisor_data.get("current_used_indexes", []), "recommended_indexes": advisor_data.get("recommended_indexes", []), "recommended_covering_indexes": advisor_data.get( "recommended_covering_indexes", [] ), } # Add summary information for better user experience response["summary"] = { "current_indexes_count": len(response["current_used_indexes"]), "recommended_indexes_count": len(response["recommended_indexes"]), "recommended_covering_indexes_count": len( response["recommended_covering_indexes"] ), "has_recommendations": bool( response["recommended_indexes"] or response["recommended_covering_indexes"] ), } logger.info( f"Index Advisor completed. Found {response['summary']['recommended_indexes_count']} recommended indexes" ) return response except Exception as e: logger.error(f"Error running Index Advisor: {e!s}", exc_info=True) raise
• src/tools/__init__.py:42-64 (registration)

  The tool function is imported from index.py (line 8) and included in the ALL_TOOLS list, which is used for bulk registration of all tools in the MCP server.
  ALL_TOOLS = [ get_buckets_in_cluster, get_server_configuration_status, test_cluster_connection, get_scopes_and_collections_in_bucket, get_collections_in_scope, get_scopes_in_bucket, get_document_by_id, upsert_document_by_id, delete_document_by_id, get_schema_for_collection, run_sql_plus_plus_query, get_index_advisor_recommendations, list_indexes, get_cluster_health_and_services, get_queries_not_selective, get_queries_not_using_covering_index, get_queries_using_primary_index, get_queries_with_large_result_count, get_queries_with_largest_response_sizes, get_longest_running_queries, get_most_frequent_queries, ]
• src/mcp_server.py:175-178 (registration)

  The MCP server registers all tools from ALL_TOOLS, including get_index_advisor_recommendations, by calling mcp.add_tool(tool) in a loop.
  # Register all tools for tool in ALL_TOOLS: mcp.add_tool(tool)